وكلاء قادرون على الدفع: ACP وAP2 وx402 وMPP في الإنتاج

دورة مكثفة عن البروتوكولات الأربعة التي تجعل أنظمة OpenAI Agents SDK قادرة على إنفاق المال: عند التجار، وأمام واجهات API، ومع وكلاء آخرين، وعبر الاقتصاد المفتوح. هذه الدورة للمهندسين الذين أطلقوا وكلاء بالفعل ويحتاجون الآن إلى جعلهم يدفعون.

19 مفهوما. 5 قرارات. 3 مخططات. أربعة مسارات تعلم. يستغرق مسار القارئ 2-3 ساعات من القراءة الخالصة: مكدس الطبقات الأربع، وكل بروتوكول بعمق، وقواعد التركيب، بلا إعداد. تضيف مسارات المبتدئ والمتوسط والمتقدم عمقا عمليا متزايدا في ربط الوكلاء بالبروتوكولات، وتشغيل التركيب بمتانة، وإدارة الهوية والإنفاق والنزاعات. تستغرق هذه المسارات نحو يوم واحد، ثم 2-3 أيام، ثم 4-5 أيام. التقدير الصريح: 2-3 ساعات للقراءة، و4-5 أيام حتى يجعل فريق هذا المكدس عادة عملية. اختر مسارك قبل مختبر القرار في الجزء 5.

تقوم هذه الدورة على فكرة واحدة: البروتوكولات الأربعة ليست متنافسة. إنها طبقات. تسأل معظم المقالات: "ACP أم x402؟" وهذا سؤال يخطئ في فهم طبيعة هذه البروتوكولات. النظام الحقيقي الذي يصل إلى الإنتاج في 2026 يستخدم عدة بروتوكولات معا، لأن كل واحد منها يحل طبقة مختلفة من مشكلة تجارة الوكلاء. يستخدم وكيل التسوق للمستهلك ACP في طبقة التجارة وسكك البطاقات في التسوية. ويستخدم الوكيل الذي يدفع مقابل واجهات API بروتوكول x402 في الطبقتين، لأن مدفوعات الآلة الصغيرة تجعل هاتين الطبقتين تنطبقان في طبقة واحدة. ويستخدم وكيل مشتريات المؤسسات تفويضات AP2 ليثبت أن الإنسان فوّض الإنفاق، ثم يستخدم Stripe MPP في التسوية. ستتعلم كيف تقرأ حالة الاستخدام وتختار التركيب الصحيح.

إذا لم تدرس دورات Agent Factory الأخرى

تذكر هذه الدورة بضعة أشقاء: الدورة المكثفة في بناء وكلاء الذكاء الاصطناعي (أساسيات SDK)، والدورة المكثفة في عامل الإنتاج (تشغيل الوكلاء بمتانة مع Inngest)، والدورة المكثفة في التطوير المدفوع بالتقييم، والدورة المكثفة في اختيار المعماريات الوكيلة. يمكنك قراءة هذه الدورة من دونها. فالبروتوكولات الأربعة، والتقسيم الطبقي، وكود SDK، وانضباط القرار كلها تقف وحدها. الشيء الوحيد الذي يساعدك: كود الجزء 3 يفترض أنك تستطيع قراءة Agent وRunner.run() و@function_tool. إذا كانت هذه جديدة عليك، فاقرأ قراءة سريعة الدورة المكثفة في بناء وكلاء الذكاء الاصطناعي أو وثائق OpenAI Agents SDK أولا، ثم عد.

ما يقابل كل بروتوكول إذا كنت على مكدس مختلف

إذا لم تكن تستخدم OpenAI Agents SDK مع Stripe وCoinbase وCloudflare، فهذا الجدول يربط كل تنفيذ مرجعي ببدائل شائعة. مواصفات البروتوكولات محايدة تجاه المكدس؛ الذي يتغير هو أسماء البدائيات فقط.

البروتوكول	SDK المرجعي الأساسي (2026)	بدائل شائعة	الترخيص / الحوكمة
ACP (Agentic Commerce Protocol)	Stripe SDK (stripe) مع OpenAI Agents SDK؛ أدوات خادم PayPal ACP؛ بيانات اعتماد Worldpay	Adyen ACP، وواجهة checkout API أصلية من Shopify	Apache 2.0، OpenAI مع Stripe في github.com/agentic-commerce-protocol/agentic-commerce-protocol
AP2 (Agent Payments Protocol)	Google ADK مع تنفيذات مرجعية في Python وTypeScript وKotlin وGo	LangGraph مع توقيع تفويضات مخصص، أو AutoGen مع a2a-x402	Apache 2.0، Google مع أكثر من 60 شريكا في github.com/google-agentic-commerce/AP2
x402	x402-client (Python)، و@x402/client (JS/TS)، وCoinbase Developer Platform، وCloudflare withX402Client، وAgentPay MCP	تنفيذات مباشرة ل EIP-3009؛ Lobster.cash؛ محافظ وكلاء من Crossmint	Apache 2.0، أنشأته Coinbase، وهو الآن تحت x402 Foundation التابعة ل Linux Foundation
MPP (Machine Payments Protocol)	Stripe PaymentIntents مع امتدادات MPP؛ وTempo blockchain SDK	Lightning Network مباشرة؛ Tempo native SDKs	Apache 2.0، Stripe مع Tempo؛ المواصفات في mpp.dev
A2A (Agent2Agent، الطبقة التي يوسعها AP2)	Google ADK	تنفيذات A2A مخصصة	Apache 2.0، Google مع Linux Foundation
MCP (Model Context Protocol، طبقة الاكتشاف)	خوادم وعملاء Anthropic MCP؛ ودعم MCP في openai-agents	LangChain MCP	MIT، Anthropic

طريقة استخدام الجدول: عندما تقول الدورة "اربط @function_tool الخاصة بالوكيل بنقطة نهاية Stripe ACP" وأنت تستخدم Adyen مع LangGraph، فاقرأها هكذا: "اربط أداة LangGraph المكافئة بنقطة نهاية Adyen ACP." الحجة نفسها؛ الأسماء هي التي تتغير. لا تحتاج إلى تعلم مكدس Stripe لمجرد قراءة هذه الدورة. طابق البدائيات، واتبع الإطار، وطبقه على مكدسك.

المسرد

📖 المصطلحات التي تستخدمها هذه الدورة، افتحها في القراءة الأولى وارجع إليها لاحقا

البروتوكولات الأربعة الرئيسية

• بروتوكول ACP (Agentic Commerce Protocol). بروتوكول تسوق المستهلك، بنته OpenAI وStripe. يحكم كيف يكمل الوكيل عملية checkout لدى تاجر حقيقي نيابة عن شخص. يشغل ChatGPT Instant Checkout. يعيش غالبا في طبقة التجارة. ترخيصه Apache 2.0.
• بروتوكول AP2 (Agent Payments Protocol). بروتوكول التفويض، بنته Google مع أكثر من 60 شريكا. ينتج "تفويضات" موقعة تثبت أن إنسانا سمح للوكيل بالإنفاق. لا ينقل المال بنفسه؛ بل يثبت أن الإنفاق كان مفوضا. ترخيصه Apache 2.0.
• بروتوكول x402. بروتوكول تسوية أصيل في HTTP، بنته Coinbase وتحكمه الآن Linux Foundation. يحيي رمز الحالة HTTP 402 "Payment Required" غير المستخدم حتى يستطيع الوكيل دفع تكلفة استدعاء API خلال ثانية إلى ثانيتين بعملة مستقرة. ترخيصه Apache 2.0.
• بروتوكول MPP (Machine Payments Protocol). بروتوكول التسوية من Stripe وTempo. حيلته هي "الجلسة": يوافق الوكيل مسبقا على سقف إنفاق، ثم يبث مدفوعات صغيرة كثيرة مقابله. متعدد السكك (عملة مستقرة، وLightning، وبطاقات). ترخيصه Apache 2.0.

الطبقات، وهي عمود الدورة كلها

• طبقة الاكتشاف. الطبقة التي يعرف فيها الوكيل ما يستطيع شراءه. تجيب عنها خوادم MCP، وA2A، وأدلة الوكلاء، أو واجهات التسوق داخل منتجات الذكاء الاصطناعي. السؤال: "ما المتاح؟"
• طبقة التفويض، وتسمى أيضا الهوية والتفويض. الطبقة التي تثبت شيئين قبل انتقال المال: أن الإنسان سمح بهذا الإنفاق، وأن الوكيل هو من يدعي أنه هو. السؤال: "هل يسمح لي بالإنفاق؟"
• طبقة التجارة. الطبقة التي تشغل الشراء الكامل: السلة، وcheckout، والتنفيذ، والنزاع، والاسترداد. السؤال: "ما دورة حياة الشراء كلها؟" وتتجاوز كليا في استدعاءات API البسيطة.
• طبقة التسوية. الطبقة التي ينتقل فيها المال فعلا. السؤال: "أين يتحرك المال حقا؟"
• سكة التسوية، أو السكة. الأنبوب الفعلي الذي ينتقل المال عبره: شبكات البطاقات (Visa/Mastercard عبر Stripe)، أو العملات المستقرة على blockchain، أو التحويل البنكي (ACH/SEPA)، أو Lightning. ومعنى "اختر سكة" هو "اختر كيف يتحرك المال."

بدائيات التفويض

• تفويض (AP2). إثبات رقمي موقّع بأن إنسانا فوّض نوعا محددا من الإنفاق. لدى AP2 ثلاثة: Intent وCart وPayment (أدناه). تشكل معا سلسلة يمكنك تدقيقها لاحقا.
• تفويض Intent Mandate. أول تفويض، يوقّعه المستخدم قبل أن يبدأ الوكيل: قواعد المهمة. مثال: "اشتر حذاء بسعر أقل من 120 دولارا." يضع الحدود التي يجب أن يبقى الوكيل داخلها.
• تفويض Cart Mandate. التفويض الأوسط، يوقّعه المستخدم بعد أن يبني الوكيل سلة محددة: "نعم، هذه العناصر بالضبط وبهذا السعر بالضبط." يستخدم في التدفقات التي يكون فيها إنسان حاضر للموافقة.
• تفويض Payment Mandate. التفويض الأخير، يوقّع، أو يولد آليا مقابل Intent Mandate، عند لحظة الدفع: "فوّض هذه الدفعة بالضبط على هذه السكة بالضبط."
• رمز SPT (Shared Payment Token). بدائية ACP. رمز لمرة واحدة من معالج الدفع (Stripe)، مقفل على تاجر واحد، وسقف مبلغ واحد، ونافذة زمنية قصيرة واحدة. إذا حاول وكيل مسموح له بإنفاق 50 دولارا أن ينفق 1000 دولار، يفشل SPT ببساطة. إنه ابن عم Payment Mandate على سكك البطاقات.
• غير قابل للإنكار. سجل موقّع لا يستطيع الموقّع أن ينكر لاحقا أنه أنشأه. سلسلة تفويضات AP2 غير قابلة للإنكار: عبارة "لم أفوض هذا" لا تصمد أمام توقيع المستخدم التشفيري نفسه.

بدائيات التسوية والعملات المشفرة

• عملة مستقرة. عملة مشفرة مربوطة بقيمة مستقرة، غالبا دولار أمريكي واحد. يستخدمها الوكلاء حتى تبقى "دفعة 0.05 دولار" مساوية لخمسة سنتات بين الإرسال والتسوية.
• عملة USDC. العملة المستقرة المرتبطة بالدولار التي تستخدمها معظم مدفوعات x402. يفترض أن يساوي USDC واحد دولارا أمريكيا واحدا. تصدرها Circle.
• رمز HTTP 402 Payment Required. رمز حالة HTTP محجوز منذ 1997 لكنه بقي غير مستخدم إلى أن أحياه x402. يرد الخادم ب "402" مع متطلبات الدفع؛ ثم يعيد العميل المحاولة ومعه إثبات دفع موقّع.
• معيار EIP-3009 (transferWithAuthorization). معيار Ethereum الذي يبنى عليه x402. يسمح للمشتري بتوقيع دفعة خارج السلسلة يرسلها طرف آخر إلى السلسلة، فلا يدفع المشتري رسوم gas ولا يلمس blockchain مباشرة.
• دور Facilitator (x402). طرف ثالث اختياري يتحقق من التوقيع ويرسل الدفع على السلسلة لصالح التاجر، حتى لا يضطر التاجر إلى تشغيل سباكة blockchain الخاصة به. تشغل Coinbase وCloudflare facilitators.
• معيار CAIP-2. طريقة قياسية لتسمية blockchain حتى يبقى البروتوكول محايدا تجاه السلسلة. يكتب x402 السلاسل بصيغة CAIP-2.
• مخطط EIP-155 / chain id. نظام الترقيم داخل CAIP-2 للسلاسل المشابهة ل Ethereum. يعني eip155:8453 سلسلة Base؛ والرقم هو chain id. (سترى eip155:8453 في كود x402؛ ومعناه ببساطة "Base.")
• محفظة عقد ذكي. محفظة crypto تفرض قواعد إنفاقها (سقف كل معاملة، سقف يومي، سقف كل مستلم) بكود على blockchain نفسها. وبما أن السلسلة تفرض هذه الحدود، فإنها تصمد حتى إذا اختل كود الوكيل نفسه.
• جلسة MPP. الحركة المميزة في MPP. بدلا من توقيع كل دفعة على حدة، يفتح الوكيل "جلسة" بسقف إنفاق وحد زمني، ثم يبث مدفوعات صغيرة مقاسة مقابلها حتى تغلق. تخيلها "حسابا مسبق الدفع للوكيل."
• نموذج الجلسات (MPP). الاسم العام للنمط السابق: فوّض سقفا ومدة مسبقا، ثم قس عدة رسوم صغيرة مقابله. أرخص من توقيع كل دفعة صغيرة عندما تكون الاستدعاءات متكررة.

مفاهيم التجارة والمال

• التسوية. لحظة انتقال المال فعلا من المشتري إلى البائع. كل ما قبل التسوية هو تنسيق؛ ولا تتم الصفقة إلا عندما تكتمل التسوية.
• مفهوم Merchant of record (MoR). النشاط التجاري المسؤول قانونيا عن المعاملة: يتعامل مع الضرائب والنزاعات ودعم العملاء. في ACP يبقى التاجر هو MoR. وفي استدعاءات x402 البسيطة بين الآلات لا يوجد MoR غالبا.
• آلية Chargeback. عندما يعكس بنك المشتري دفعة بطاقة، غالبا بعد نزاع. تدعم سكك البطاقات chargebacks؛ ولا يدعم x402 الخالص ذلك. والحاجة إلى chargebacks تحدد كثيرا أي بروتوكول تستخدمه.
• النزاع. اعتراض رسمي من المشتري على الرسوم ("لم أفوض هذا" / "لم يصل المنتج"). تحل البروتوكولات المختلفة النزاعات بطرق مختلفة؛ وغالبا يفرض هذا الاختلاف اختيار البروتوكول.
• خاصية Idempotency. خاصية تجعل تنفيذ العملية نفسها مرتين يعطي أثر تنفيذها مرة واحدة. تهم لأن Stripe يعيد webhooks بالمعرف نفسه event id؛ ومن دون idempotency key يمكن لكودك أن يسترد أو يخصم مرتين.

مكدس الوكيل والبروتوكولات المجاورة

• تجارة الوكلاء. أي معاملة يكون فيها وكيل ذكاء اصطناعي مستقل هو المشتري أو البائع أو الاثنين معا، من دون إنسان يضغط زر الشراء في تلك اللحظة. تختلف عن التسوق "المساعد بالذكاء الاصطناعي" حيث يبقى الشخص هو من يضغط الزر.
• عدة OpenAI Agents SDK. عدة Python/JavaScript لبناء حلقات وكلاء باستخدام Agent وRunner.run() و@function_tool وحواجز الأمان. في هذه الدورة هو "العميل الموحد": يصبح كل بروتوكول دفع أداة واحدة أو أكثر يستطيع الوكيل استدعاءها.
• معيار MCP (Model Context Protocol). معيار Anthropic المفتوح لتعريض الأدوات والسياق للوكلاء. في تجارة الوكلاء يكون غالبا طبقة الاكتشاف: تعثر الوكلاء على الخدمات القابلة للشراء عبر خوادم MCP. مرخص MIT.
• بروتوكول A2A (Agent2Agent). بروتوكول Google لتخاطب الوكلاء واكتشاف بعضهم. يبنى AP2 فوق A2A: ينتقل التفويض كرسالة A2A. ترخيصه Apache 2.0.
• بروتوكول UCP (Universal Commerce Protocol). بروتوكول طبقة التجارة من Google، النظير ل ACP، ومبني حول واجهات التسوق لدى Google (Gemini وGoogle AI Mode). ينافس ACP في طبقة التجارة.
• بروتوكول TAP (Trusted Agent Protocol). بروتوكول Visa وCloudflare لإثبات هوية الوكيل داخل ترويسات طلب HTTP. يثبت من هو الوكيل لا ما المسموح له بإنفاقه، لذلك يضاف عادة إلى بروتوكول مصادقة آخر بدلا من أن يحل محله.
• معيار ERC-8004. معيار على السلسلة لهوية الوكلاء وسمعتهم: سجل عام للوكلاء مع تاريخ معاملاتهم، حتى يستطيع وكيل لا تربطه علاقة سابقة أن يفحص سجل وكيل آخر قبل الوثوق به.
• tool_input_guardrail. حاجز أمان في OpenAI Agents SDK يعمل قبل تنفيذ الأداة ويمكنه رفض الاستدعاء. إنه طريقة SDK الأصلية لإيقاف الدفع قبل حدوثه. وهو عمود هذه الدورة: يظهر في سبع أدوات دفع. (قارن ذلك ب output_guardrail، الذي يعمل على الرد النهائي للوكيل، أي بعد فوات الأوان لإيقاف الدفع.)

المتطلبات السابقة

ستستفيد من هذه الدورة أكثر إذا كان لديك:

1. الدورة المكثفة في بناء وكلاء الذكاء الاصطناعي، أو خبرة مكافئة في SDK. تعرض تكاملات البروتوكولات على شكل كود SDK، لذلك تحتاج إلى قراءة Agent وRunner.run() و@function_tool براحة. راجع الدورة المكثفة في بناء وكلاء الذكاء الاصطناعي.
2. الدورة المكثفة في اختيار المعماريات الوكيلة، أو حس تصميم مكافئ. يبني إطار "أي تركيب لأي حالة استخدام" في الجزء 5 على انضباط اختيار الأنماط. راجع الدورة المكثفة في اختيار المعماريات الوكيلة.
3. أساسيات HTTP. رموز الحالة، ودورة الطلب/الاستجابة، والترويسات. يعمل x402 خصوصا على مستوى HTTP.
4. مفردات دفع أساسية. التاجر، والتسوية، والنزاع، وchargeback. تشرح الدورة الأجزاء الخاصة بالوكلاء، لكنها تفترض أنك تعرف معنى "merchant of record".

لا تحتاج إلى خبرة blockchain أو smart-contract (تعلّمك الدورة ما يكفي من x402 وEIP-3009 للمتابعة؛ لا Solidity)، ولا تحتاج إلى خبرة سابقة بأي من البروتوكولات الأربعة. ستتعلمها كلها من المصادر الأولية.

مسارات التعلم الأربعة

تعمل هذه الدورة على أربعة أعماق. اختر مسارك قبل الجزء 5.

المسار	الوقت	ما ستفعله	الأنسب له
القارئ	2-3 ساعات	اقرأ كل المفاهيم والقرارات، وتجاوز تشغيل الكود.	مهندسون يقررون هل يستثمرون بعمق أكبر. ومديرو منتجات ومعماريون يحتاجون الإطار لتقييم عروض الموردين.
المبتدئ	نحو يوم	مسار القارئ مع تشغيل أمثلة عميل x402 ومعاملة ACP اختبارية واحدة في Stripe test mode.	مهندسون جدد في تجارة الوكلاء يريدون وقتا عمليا مع أبسط بروتوكول (x402) وأكثرها جاهزية للإنتاج (ACP).
المتوسط	2-3 أيام	مسار المبتدئ مع بناء وكيل يستخدم ACP لنوع من المعاملات وx402 لنوع آخر، ومع ربط فحوص Intent Mandate من AP2.	مهندسون يطلقون نظاما حقيقيا عليه تركيب عدة بروتوكولات.
المتقدم	4-5 أيام	مسار المتوسط مع تشغيل النظام المركب بمتانة (غلاف Inngest)، وربط حدود الإنفاق وبوابات موافقة الإنسان، وقياس مقاييس trace والكلفة والنزاعات، والتعامل مع دورة استرداد كاملة.	مهندسون مسؤولون عن أنظمة إنتاج تنقل مالا حقيقيا لمستخدمين حقيقيين.

اختبار ذاتي مفيد: "في حالة الاستخدام التي أفكر فيها، ما أصغر تركيب بروتوكولات يطلق قيمة؟" إذا لم تستطع الإجابة بعد الجزء 4، فأعد قراءة الجزء 4. وإذا استطعت، فمسارك يصبح مجرد سؤال: إلى أي مدى تريد الانتقال من "أصغر تركيب" إلى "تركيب صالح للإنتاج"؟

مكدس الطبقات الأربع

هذا هو المخطط الذي يتعلق به كل شيء بعده.

تلمس كل حالة استخدام في تجارة الوكلاء الطبقات الأربع كلها، لكن الحالات المختلفة تركب بروتوكولات مختلفة في كل طبقة. وكيل تسوق للمستهلك: MCP للاكتشاف، ورمز دفع مشترك من ACP للتفويض، وACP للتجارة، وسكك البطاقات للتسوية. وكيل يدفع مقابل API: دليل وكلاء للاكتشاف، وتوقيع EIP-3009 للتفويض، ولا طبقة تجارة إطلاقا، وx402 للتسوية. وكيل مشتريات مؤسسة: دليل A2A، وتفويض AP2، وACP أو UCP للتجارة، وStripe MPP للتسوية. القاعدة بسيطة: اختر بروتوكولا واحدا لكل طبقة، واجعل حالة الاستخدام تبرر كل اختيار.

الجزء 1: لماذا تحتاج تجارة الوكلاء إلى بروتوكولات جديدة

هل تقرأ الجزء 1 قراءة خفيفة؟

يقدم الجزء 1 ستة أسماء (ACP وAP2 وx402 وMPP وMCP وA2A) وثلاث طبقات بسرعة. هذا مقصود: هذا الجزء هو موجز تمهيدي لا غوص عميق. الحد الأدنى الذي تحتاج إلى حمله: ACP هو تسوق المستهلك (Stripe مع OpenAI)، وAP2 هو تفويضات authorization (Google مع أكثر من 60 شريكا)، وx402 هو دفع عملة مستقرة لكل طلب عبر HTTP (Coinbase)، وMPP هو دفع متعدد السكك قائم على الجلسات (Stripe مع Tempo). أما MCP وA2A فهما كيف تعثر الوكلاء على بعضها وتتخاطب، وليسا بروتوكولي دفع. احمل الأسماء بخفة. يعود الجزء 2 إلى كل واحد منها بعمق، وستثبت في القراءة الثانية.

المفهوم 1: الافتراض الذي انكسر

في سطر واحد: افترضت أنظمة الدفع أن إنسانا يضغط زر الشراء، ويكسر الوكلاء هذا الافتراض بثلاث طرق معا.

بنيت أنظمة الدفع على افتراض هادئ واحد: إنسان يجلس أمام لوحة المفاتيح ويضغط الشراء. كل شاشة، وكل فحص احتيال، وكل عملية نزاع، وكل نموذج تسجيل صمم للبشر. تكسر وكلاء الذكاء الاصطناعي هذا الافتراض بثلاث طرق في الوقت نفسه.

الكسر 1: لا تملك الوكلاء عناوين بريد إلكتروني. تريد تدفقات الدفع الاستهلاكية حسابا. والحساب يريد بريدا إلكترونيا ورقم هاتف، وغالبا اسما. لا يملك الوكيل المستقل أيا من ذلك. إذا زورتها، فقد أنشأت كيانا يفشل في KYC لحظة نظر اكتشاف الاحتيال إليه. يفترض تدفق التسجيل إنسانا يتقدم لعلاقة؛ أما الوكيل فيحتاج إلى شيء آخر.

الكسر 2: تعمل الوكلاء آلاف المرات في الثانية. ترصد أنظمة الاحتيال السلوك الغريب بالمعدل والموقع والنمط. يبدو وكيل يجري 1,000 استدعاء API في دقيقة واحدة كأنه هجوم credential-stuffing تماما. ما هو طبيعي للوكيل إنذار للإنسان، والسكك مضبوطة للبشر.

الكسر 3: لا تستطيع الوكلاء رفع الهاتف. تفترض تسوية النزاعات إمكانية الوصول إلى المشتري: "هل فوّضت هذا؟" الوكيل الذي فوّض رسما لا يستطيع الرد، وقد لا يعرف الإنسان خلفه أن الرسوم حدثت أصلا. تحتاج النزاعات إلى نموذج مختلف عندما يكون المشتري برمجيات.

يحتاج كل كسر إلى إصلاح على مستوى البروتوكول، لا إلى طبقة طلاء على واجهة المستخدم:

الكسر	ما لا يستطيع إصلاح السكة البشرية فعله	ما يعطيك إياه إصلاح سكة الوكلاء
لا بريد ولا حساب	إجبار الوكلاء على إنشاء حسابات مزيفة	هوية تشفيرية (TAP وERC-8004) أو رموز محددة النطاق (ACP SPT وAP2 Mandate)
سلوك عالي التكرار	حجب حركة مرور تشبه الهجوم	دفع أصيل في HTTP لكل طلب (x402) أو جلسات مفوضة مسبقا (MPP)
لا هاتف للنزاعات	إرسال نزاعات بالبريد لا يستطيع الوكيل قراءتها	تفويض قائم على mandates (AP2) مع مسار تدقيق غير قابل للإنكار

فشل مسار "غلف المدفوعات القديمة بتجربة وكيل أجمل" بالفعل. جرّبت عدة شركات ناشئة في 2024 و2025 أن تعطي الوكلاء حسابات تشبه حسابات البشر بهويات مخترعة. قبضت عليها أنظمة كشف الاحتيال، وتراكمت chargebacks، وانهارت علاقات التجار. اتضح أن إصلاحات مستوى البروتوكول مطلوبة لا اختيارية. لذلك ظهرت ACP وAP2 وx402 وMPP كلها خلال الاثني عشر شهرا نفسها.

المفهوم 2: لماذا لا يستطيع بروتوكول واحد الفوز

في سطر واحد: تحدث الكسور في أربع طبقات مختلفة مع أربعة أصحاب نفوذ مختلفين، لذلك قسمت البروتوكولات العمل حسب الطبقة بدلا من أن يبتلع واحد منها كل شيء.

سؤال عادل: إذا ظهرت البروتوكولات الأربعة كلها لإصلاح الكسور الثلاثة نفسها، فلماذا لم يفز واحد منها ببساطة؟ الإجابة بنيوية. تقع الكسور في طبقات مختلفة، والبروتوكول الواحد الذي يحاول إصلاحها كلها سيكون أكبر من أن يتبناه أحد.

فكر في ما كان يجب أن يحدده بروتوكول موحد واحد:

1. كيف تعثر الوكلاء على التجار والخدمات المتاحة. هذه هي طبقة الاكتشاف.
2. كيف تثبت الوكلاء من هي وأن إنسانا فوّض الإنفاق. هذه هي طبقة التفويض.
3. كيف تدير الوكلاء شراء كاملا، بما في ذلك النزاعات والاستردادات. هذه هي طبقة التجارة.
4. كيف يتحرك المال فعلا بين الأطراف. هذه هي طبقة التسوية.

لكل طبقة أصحاب نفوذ أقوياء بالفعل. الاكتشاف لمحركات البحث وواجهات API. والهوية ل OAuth وسلطات الشهادات. والتجارة ل Stripe وAdyen وShopify. والتسوية ل Visa وMastercard وACH، إضافة إلى سكك crypto الجديدة. كان بروتوكول موحد سيحتاج إلى اتفاق كل صاحب نفوذ في كل طبقة. لم يكن ذلك سيحدث.

ما حدث بدلا من ذلك هو أن كل بروتوكول اختار الطبقة التي يملك راعيه فيها أكبر نفوذ:

البروتوكول	أين يملك راعيه النفوذ	الطبقة التي أخذها
ACP	تملك OpenAI قناة التسوق (ChatGPT)؛ وتملك Stripe تكامل التجار	التجارة، لتدفقات مشتري بشري عبر ذكاء اصطناعي
AP2	لدى Google محافظ Android وتحالف من 60 شريكا	التفويض، mandates كبيانات اعتماد موقعة
x402	لدى Coinbase بنية عملات مستقرة؛ ولدى Cloudflare حافة HTTP	التسوية، لمدفوعات صغيرة بين الآلات
MPP	لدى Stripe علاقات التجار؛ ولدى Tempo blockchain	التسوية، لتدفقات المؤسسات ومتعددة السكك

لذلك لا تتقاتل البروتوكولات داخل طبقة؛ بل تتنافس على تعريف طبقة. في التسوية، يتنافس x402 وMPP فعلا، وتفوت معظم مقالات "x402 ضد MPP" أن هذا هو المكان الوحيد الذي يتداخلان فيه حقا. وفي التجارة، يتنافس ACP وUCP. وفي التفويض، تتنافس AP2 وTAP وERC-8004.

وهنا تصل الفكرة التي تقوم عليها الدورة كلها. داخل الطبقة، تختار بروتوكولا واحدا. وعبر الطبقات، تركب عدة بروتوكولات. البروتوكولات الأربعة ليست بدائل تختار بينها؛ إنها طبقات ترصها. وكيل تسوق للمستهلك يشغّل ACP في التجارة وسكك البطاقات في التسوية. وكيل يدفع ل API يشغّل x402 في التسوية ويتجاوز التجارة كليا. وكيل مؤسسة يشغّل تفويضات AP2 في التفويض وMPP في التسوية. تمشي شجرة القرار في الجزء 5 طبقة بطبقة. تمسك بهذه الفكرة: كل قسم بعد هذا يفترضها.

المفهوم 3: OpenAI Agents SDK بوصفه العميل الموحد

في سطر واحد: لا تتعامل مع أربعة بروتوكولات بأربع طرق مختلفة؛ يصبح كل بروتوكول أداة يستدعيها الوكيل، ويكون SDK هو العميل الواحد الذي يربطها كلها.

السؤال العملي: مع أربعة بروتوكولات في أربع طبقات، كيف يستخدمها الوكيل فعلا؟ الإجابة في 2026 هي أن إطار عمل الوكيل يصبح العميل الموحد. يعرّض كل بروتوكول SDK أو نقطة نهاية HTTP، ويربطه إطار العمل كأداة. هذا صحيح في OpenAI Agents SDK وLangGraph وAutoGen وCrewAI على السواء. سنستخدم OpenAI Agents SDK طوال الدورة.

بالنسبة إلى SDK، يتبع كل تكامل بروتوكول الشكل نفسه: غلف البروتوكول في دالة واحدة أو أكثر من دوال @function_tool، ومررها إلى Agent، ودع Runner.run يقود الحلقة.

note

استدعاءات stripe.PaymentTokens.create وX402Client(wallet=...) وfrom ap2 import ... أدناه توضيحية: تبين شكل تكامل البروتوكول. أما طبقة Agent وRunner و@function_tool حولها فهي حقيقية وتعمل. راجع الملاحظة بعد الكتلة لمعرفة ما هو بديل تعليمي.

from agents import Agent, Runner, function_tool
import stripe  # for ACP/MPP
from x402_client import X402Client  # for x402
from ap2 import IntentMandate, CartMandate  # for AP2
from decimal import Decimal
from .models import PaymentToolResult, X402PaymentResult  # the shared result models

# Each protocol becomes one or more @function_tool decorated functions
@function_tool
async def acp_checkout(merchant_id: str, items: list, max_amount: Decimal) -> PaymentToolResult:
    """Complete an ACP checkout at a merchant with the given items."""
    # Mint a one-time payment token scoped to this merchant, then POST the order
    spt = stripe.PaymentTokens.create(
        amount=int(max_amount * 100),   # cents as int
        currency="usd",
        merchant_id=merchant_id,
        max_uses=1,
    )
    response = await acp_post(merchant_id, items, spt.token)
    return PaymentToolResult(
        status="success" if response.status == "confirmed" else "failed",
        details={"order_id": response.order_id, "merchant_status": response.status},
    )

@function_tool
async def x402_fetch(url: str, max_payment_usdc: Decimal) -> X402PaymentResult:
    """Fetch a URL that may require x402 payment up to max_payment_usdc."""
    client = X402Client(wallet=agent_wallet, max_per_request=max_payment_usdc)
    response = await client.get(url)
    return X402PaymentResult(
        content=response.content,
        amount_paid_usdc=response.amount_paid_usdc,
        tx_hash=response.payment_proof,
    )

# Compose the tools into an agent
shopping_agent = Agent(
    name="ShoppingAgent",
    instructions="Help the user find and purchase items. Use acp_checkout for retail goods, x402_fetch for paid APIs.",
    tools=[acp_checkout, x402_fetch],
    model="gpt-5.5",
)

# Run the agent. The SDK handles tool selection, the loop, and retries.
result = await Runner.run(shopping_agent, "Buy me a red t-shirt under $30")

ما يعمل وما هو بديل هنا: ربط Agent وRunner.run و@function_tool هو SDK الحقيقي ويعمل كما هو مكتوب. عملاء الدفع بدائل تعليمية. stripe.PaymentTokens.create ليس استدعاء Stripe حقيقيا (يدمج ACP في الإنتاج عبر نقاط نهاية Stripe ACP الحية)، والباني الغني X402Client(wallet=..., max_per_request=...) توضيحي أيضا. الحزمة الحقيقية من جهة المشتري هي x402-client، والاستدعاء الحي يحتاج إلى حساب ممول ونقطة نهاية 402 حقيقية، وهو ما لا تفعله هذه الدورة. يعطي الجزء 3 لكل بروتوكول خلفية mock قابلة للتشغيل حتى ترى الغلاف يعمل من البداية إلى النهاية من دون نقل مال حقيقي.

الشكل متطابق عبر البروتوكولات الأربعة كلها. SDK هو العميل الموحد؛ كل بروتوكول ليس إلا أداة يفكر فيها الوكيل ويستدعيها عندما يحتاج إليها. تهم ثلاثة أجزاء من بنية SDK للمدفوعات:

1. قيمة إرجاع ذات نوع محدد لنتائج الدفع. عندما ترجع أداة الدفع نموذج Pydantic، يحصل مستدل الوكيل على معلومات نوعية نظيفة عن ما نجح، وما فشل، وما الخطوة التالية. يعرّف الجزء 3 نماذج النتائج المشتركة هذه مرة واحدة.
2. Runner.run(..., context=...) لسياق الدفع. يحتاج الوكيل غالبا إلى هوية المستخدم، وحدود الإنفاق، ومقبض المحفظة. مرر هذه عبر معامل context في SDK بدلا من حشرها في التعليمات. context خاص بكل تشغيل وبكل مستخدم.
3. tool_input_guardrail لحدود الإنفاق. يعمل حاجز إدخال الأداة قبل تنفيذ كل أداة ويمكنه رفض الاستدعاء. إنها طريقة SDK الأصلية لحجب الدفع قبل حدوثه، لا بعده. يشرح المفهوم 15 فرض الحدود على ثلاثة مستويات. لا يحل output_guardrail على مستوى الوكيل هذه المشكلة، لأنه يعمل على الرد النهائي للوكيل بعد أن تكون أي أداة دفع قد عملت بالفعل.

الجزء 2: الطبقات الأربع بعمق

رأيت الطبقات الأربع في الجزء 1؛ وهنا نقترب من كل واحدة. تجيب كل طبقة عن سؤال مختلف، ولها بروتوكولات متنافسة خاصة بها، وتفرض قرارا واحدا: في حالة الاستخدام هذه، أي بروتوكول يناسب الطبقة هذه أفضل؟ اقرأ الجزء 2 قبل الجزء 3. framing الطبقات هنا هو ما يجعل تفاصيل البروتوكولات في الجزء 3 متماسكة بدلا من أن تبدو قائمة منافسين.

المفهوم 4: الطبقة 1، الاكتشاف (كيف تعثر الوكلاء على ما يمكنها شراءه)

في سطر واحد: الاكتشاف هو المكان الذي يعرف فيه الوكيل ما هو متاح أصلا للشراء، والآلية الصحيحة تعتمد على مكان وجود تلك الخدمات فعلا.

قبل أن يستطيع الوكيل إجراء معاملة، عليه أن يعرف ما الموجود. يحتاج وكيل التسوق إلى تجار يحملون المنتج. ويحتاج وكيل يدفع ل API إلى معرفة أي نقاط نهاية لديها البيانات وما تكلفتها. ويحتاج وكيل المشتريات إلى موردين يجتازون قواعد الامتثال. يجيب الاكتشاف عن سؤال "ما المتاح؟"

تتنافس أربعة خيارات جدية هنا في 2026:

البروتوكول	كيف يعمل	الأنسب له
MCP (Anthropic)	تعرّض خوادم الأدوات دوالا قابلة للاستدعاء؛ يتصل الوكيل، ويسرد الأدوات، ويستدعيها	وصول برمجي إلى خدمات محددة ربطها المطور؛ عمل وكلاء عالي الحجم؛ طبقة اكتشاف أدوات الوكلاء المهيمنة
A2A (Google)	تنشر الوكلاء ما تقدمه في أغلفة قياسية؛ وتكتشفها وكلاء أخرى	نظم متعددة الوكلاء تحتاج فيها الوكلاء إلى العثور على وكلاء نظراء؛ طبقة الاكتشاف التي يوسعها AP2
أدلة الوكلاء (Agent.market, lobster.cash, Tenzro)	أسواق عامة تسرد واجهات API وخدمات مدفوعة؛ تستعلمها الوكلاء كفهرس	خدمات طرف ثالث تكتشف في وقت التشغيل؛ "الصفحات الصفراء" لتجارة الوكلاء
واجهات التسوق بالذكاء الاصطناعي (ChatGPT Instant Checkout, Google AI Mode, Walmart-in-ChatGPT)	منتجات ذكاء اصطناعي استهلاكية فيها اكتشاف منتجات مع checkout عبر ACP مدمج	تدفقات المستهلك حيث يتحدث المستخدم إلى الذكاء الاصطناعي وتظهر المنتجات داخله

يدعم SDK بروتوكول MCP دعما أصليا: اربط خادم MCP بوكيل في بضعة أسطر، وستصبح كل أدواته متاحة لاستدلال الوكيل. أما الاكتشاف غير القائم على MCP، فغلّفه كأداة عادية @function_tool تستعلم الدليل وترجع قوائم منظمة.

note

ربط MCPServerStreamableHttp أدناه حقيقي وقابل للاستيراد. أما استدعاء agent_market_client فهو توضيحي: يمثل عميل دليل. وطبقة @function_tool وAgent حقيقية.

from agents import Agent, function_tool
from agents.mcp import MCPServerStreamableHttp
from decimal import Decimal

# Wire an MCP discovery server: all its tools become available
research_mcp = MCPServerStreamableHttp(
    name="research-services",
    params={"url": "https://research-services.example.com/mcp"},
)

# Wire a non-MCP directory as a regular tool
@function_tool
async def search_agent_market(query: str, max_price_usdc: Decimal) -> list[dict]:
    """Search Agent.market for x402-paid services matching the query."""
    return await agent_market_client.search(query, max_price_usdc=max_price_usdc)

agent = Agent(
    name="ResearchAgent",
    instructions="Find and use research services. Prefer MCP-discovered tools; fall back to Agent.market for niche needs.",
    mcp_servers=[research_mcp],
    tools=[search_agent_market],
)

الخيار هنا ليس "MCP ضد A2A ضد الأدلة." بل السؤال: أين تعيش خدمات وكيلي فعلا؟ إذا كانت داخل مؤسستك، فاستخدم MCP. وإذا كانت عبر شبكة من وكلاء شركاء، فاستخدم A2A. وإذا كانت APIs طرف ثالث تكتشف في وقت التشغيل، فاستخدم الأدلة. وإذا كانت منتجات استهلاكية، فاستخدم واجهة تسوق بالذكاء الاصطناعي (ومعها ACP في طبقة التجارة). هذه ليست خيارات متنافية؛ الوكيل الحقيقي يستخدم عدة خيارات غالبا.

المفهوم 5: الطبقة 2، الهوية والتفويض (إثبات أن الوكيل مسموح له)

في سطر واحد: التفويض هو المكان الذي يثبت فيه الوكيل أن الإنسان سمح بهذا الإنفاق وأن الوكيل هو من يدعي أنه هو، قبل انتقال أي مال.

قبل أن يتحرك المال، يجب أن يكون شيئان صحيحين: أن الوكيل هو من يدعي أنه هو، وأن الإنسان فوّض هذا الإنفاق. هاتان مشكلتان مختلفتان بحلول مختلفة، وتحسم الطبقة 2 الاثنين قبل حدوث التسوية. إذا تجاوزت الطبقة 2، تحصل على أحد فشلين: احتيال (يستطيع وكيل أي شخص إنفاق مال أي شخص) أو شلل (تحتاج كل معاملة إلى إنسان يضغط تأكيد).

هذه هي أكثر طبقة متنازع عليها في 2026. أربعة خيارات، وأربع فلسفات:

البروتوكول	كيف يعمل	أقوى موضع له
AP2 Mandates (Google)	بيانات اعتماد موقعة: Intent Mandate ("اشتر حذاء بأقل من 120 دولارا")، وCart Mandate ("هذه السلة، هذا السعر")، وPayment Mandate ("فوّض هذه السكة")	تدفقات كثيفة التدقيق تحتاج إلى إثبات موافقة غير قابل للإنكار؛ تدفقات متعددة الوكلاء حيث لم ير التاجر وكيل المشتري من قبل
ACP SPT (OpenAI plus Stripe)	تصدر Stripe رمزا Shared Payment Token محدد النطاق لتاجر واحد ومبلغ ونافذة زمنية؛ يقدمه الوكيل؛ ويتحقق منه التاجر ويخصم	تسوق المستهلك حيث تكون Stripe هي المعالج وتحمل سكك البطاقات انضباط chargeback
TAP (Visa plus Cloudflare)	يركب توقيع هوية الوكيل في ترويسات HTTP؛ ويتحقق التجار منه عبر دليل Visa	التحقق من الهوية تحديدا (لا التفويض)؛ يضاف عادة إلى بروتوكول مصادقة آخر ولا يستخدم وحده
ERC-8004 plus on-chain reputation	سجل على السلسلة لهويات الوكلاء وتاريخ المعاملات، مع درجة سمعة من صفقات سابقة	تدفقات متعددة الوكلاء خالصة بلا ثقة سابقة؛ B2B عالي المخاطر حيث تستحق السمعة الفحص

السؤالان اللذان يجب أن تجيب عنهما الطبقة 2، وكيف يجيب كل بروتوكول:

1. "هل فوّض الإنسان هذا؟" يجيب AP2 بتفويض وقّعه المستخدم قبل التفويض. ويجيب ACP ب SPT أصدرته Stripe فقط بعد أن فوّض المستخدم على مستوى الحساب. لا يجيب TAP عن هذا؛ فهو للهوية فقط. ويجيب ERC-8004 بمعاملات موقعة على السلسلة.
2. "هل الوكيل هو من يدعي أنه هو؟" يجيب AP2 بمفتاح التوقيع (لا يستطيع التوقيع إلا الوكيل الحقيقي). ويجيب ACP بأن SPT محدد للتاجر (لا يستطيع استرداده إلا التاجر المصرح). ويجيب TAP بفحص دليل Visa. ويجيب ERC-8004 بسجل الهوية على السلسلة.

يعطيك SDK نقطتي تكامل: حاجز إدخال أداة يعمل قبل أداة الدفع، وسياق التشغيل الذي يحمل حالة كل مستخدم إليهما معا.

note

حاجز الأمان و@function_tool وربط Agent أدناه من SDK الحقيقي وتعمل (مسارات الخصائص data.context.tool_arguments وdata.context.context مؤكدة مقابل SDK المثبت). أما استدعاء stripe.PaymentTokens.create داخل الأداة فهو توضيحي؛ يستخدم ACP في الإنتاج نقاط نهاية Stripe ACP الحية.

from agents import Agent, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from decimal import Decimal
import json
import stripe

# Pattern 1: a tool input guardrail. Runs BEFORE the payment tool executes.
# This is the SDK-native way to block a payment before it happens.
@tool_input_guardrail
def block_over_user_cap(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    """Reject any payment tool call where the request would exceed the user's per-run cap."""
    args = json.loads(data.context.tool_arguments or "{}")          # raw JSON args -> dict
    requested = Decimal(str(args.get("max_amount_usd", 0)))
    ctx = data.context.context                                       # the run context (a dict)
    user_cap = Decimal(str(ctx["user_session"].per_run_spend_cap_usd))
    run_spent = Decimal(str(ctx.get("run_spend_usd", 0)))
    if run_spent + requested > user_cap:
        return ToolGuardrailFunctionOutput.reject_content(
            f"Refusing payment tool: would spend ${run_spent + requested}, exceeds run cap ${user_cap}"
        )
    return ToolGuardrailFunctionOutput.allow()

# Pattern 2: the payment tool itself, guarded at the function-tool level
@function_tool(tool_input_guardrails=[block_over_user_cap])
async def purchase_with_acp(
    ctx: RunContextWrapper,
    merchant_id: str,
    items: list,
    max_amount_usd: Decimal,
) -> PaymentToolResult:
    """Use ACP to buy items from the merchant up to max_amount_usd.
    The guardrail has already verified spend is within bounds before we reach here."""
    user_session = ctx.context["user_session"]
    spt = stripe.PaymentTokens.create(
        amount=int(max_amount_usd * 100),   # Stripe expects cents as int
        currency="usd",
        merchant_id=merchant_id,
        user_session_id=user_session.id,
        max_uses=1,
    )
    response = await acp_post(merchant_id, items, spt.token)
    return PaymentToolResult(
        status="success" if response.status == "confirmed" else "failed",
        details={"order_id": response.order_id, "merchant_status": response.status},
    )

agent = Agent(
    name="ShoppingAgent",
    instructions="Help the user shop. Always verify authorization before any purchase.",
    tools=[purchase_with_acp],
    # No output_guardrails for spend control. Those run on the agent's FINAL output,
    # not on individual tool calls. See Concept 15 for the full three-level enforcement.
)

أي حاجز أمان يوقف الدفع؟

لدى SDK ثلاثة أنواع من حواجز الأمان وتعمل في لحظات مختلفة. يعمل input_guardrail على إدخال المستخدم الأول إلى الوكيل. ويعمل output_guardrail على رد الوكيل النهائي للمستخدم. أما tool_input_guardrail وtool_output_guardrail فيعملان على كل استدعاء function-tool، قبل تنفيذه وبعده. لسلامة الدفع تحتاج إلى حاجز إدخال الأداة: يعمل قبل أن تعمل أداة الدفع ويمكنه رفض الاستدعاء. أما حاجز الإخراج فيعمل متأخرا جدا لإيقاف الدفع؛ يفيد في تنظيف الرد النهائي (مثلا حجب بيانات حساسة)، لا في حجب أداة. يعود المفهوم 15 إلى ذلك؛ فهذا هو الخطأ الأكثر شيوعا.

ينتهي اختيارك هنا إلى نموذج الثقة. إذا كان المستخدم مسجلا في تطبيقك وتستطيع إصدار SPTs عبر Stripe، يعطيك ACP القصة الأكثر جاهزية للإنتاج. وإذا كنت تحتاج إلى مسارات تدقيق غير قابلة للإنكار، كما في الصناعات المنظمة أو مشتريات B2B، تناسبك تفويضات AP2. وإذا كنت تحتاج إلى هوية تشفيرية وحدها، منفصلة عن التفويض، فأضف TAP. وفي بيئة متعددة الوكلاء خالصة بلا ثقة مشتركة، يملأ ERC-8004 الفجوة. هذه ليست بدائل قابلة للتبادل.

المفهوم 6: الطبقة 3، التجارة (دورة حياة الشراء الكاملة)

في سطر واحد: التجارة هي كل شيء في الشراء الحقيقي ليس تفويضا ولا تسوية: السلة، والطلب، والتنفيذ، والنزاع، والاسترداد؛ أما استدعاء API البسيط فيتجاوزها كليا.

يغطي التفويض والتسوية معا "ينتقل المال بإذن." أما التجارة فتغطي كل شيء آخر في الشراء: السلال المنظمة، وتأكيد الطلب، وتتبع التنفيذ، وتسوية النزاعات، والاستردادات، وchargebacks، والإرجاعات. هذه هي الطبقة التي تفصل checkout عن تحويل المال. استدعاء API بسيط بين الآلات لا يحتاج إلى طبقة تجارة؛ إنه مجرد استدعاء API. أما شراء المستهلك فيحتاج إليها بوضوح.

ثلاثة خيارات مختلفة فعليا:

البروتوكول	ما يفعله	الأنسب له
ACP (OpenAI plus Stripe)	تدفق منظم: صيغة السلة، وتأكيد الطلب، وحالة التنفيذ، وتصعيد النزاع، والاستردادات. يبقى التاجر merchant of record.	تسوق المستهلك: سلع تجزئة، واشتراكات، وتنفيذ مادي. يشغّل ChatGPT Instant Checkout.
UCP (Google)	دورة حياة مشابهة، مبنية حول واجهات التسوق في Google (Gemini وGoogle AI Mode) وGoogle Pay	تجار Google Shopping؛ ووكلاء على واجهات Google AI
Direct API (machine-to-machine)	لا بروتوكول تجارة إطلاقا، مجرد HTTP API. الدفع عبر x402 أو MPP. لا سلة ولا نزاعات ولا استردادات.	وصول API، وحوسبة، وتدفقات بيانات: مشتريات يكون الشيء المشترى فيها استجابة API عديمة الحالة

إذن يتنافس ACP وUCP؛ أما "API مباشر" فهو غياب هذه الطبقة، لا منافس ثالث، وغالبا يجلس سعيدا إلى جانب لا شيء. تختار منصة المستهلك ACP أو UCP (أو الاثنين إذا امتدت عبر ChatGPT وGemini). ولا تختار سوق API شيئا في هذه الطبقة إطلاقا، لأن مشترياتها لا تملك دورة حياة تديرها.

الجزء الذي يستخف به معظم المهندسين هو الاستردادات والنزاعات. العميل الذي يطلب قميصا بمقاس خاطئ يتوقع أن يرجعه. يجب أن يقول بروتوكول التجارة كيف يبدأ الإرجاع، وكيف يسمع الوكيل عنه، وكيف يعود الاسترداد عبر التسوية. يفعل ACP ذلك جيدا بإبقاء التاجر merchant of record: تعمل آليات النزاع القائمة (تدفقات chargeback في Stripe، وسياسة الإرجاع لدى البائع) كما هي. وتفشل الطرق المباشرة ل API غالبا لأنها تتجاهل ذلك. لا يوجد مسار استرداد غالبا، وهذا مقبول في استدعاء API بقيمة 0.0001 دولار، وخاطئ في رصيد API بقيمة 500 دولار.

تحتاج تدفقات التجارة عادة إلى عدة أدوات تعمل بالتسلسل:

note

استدعاءات acp_client أدناه توضيحية؛ تمثل خلفية تجارة ACP. نماذج Pydantic و@function_tool وربط Agent حقيقية.

from agents import Agent, function_tool
from pydantic import BaseModel
from decimal import Decimal

class CartItem(BaseModel):
    sku: str
    quantity: int
    unit_price: Decimal

class OrderResult(BaseModel):
    order_id: str
    status: str  # "confirmed", "fulfilled", "shipped", "delivered"
    tracking_url: str | None = None
    estimated_arrival: str | None = None

@function_tool
async def acp_create_cart(merchant_id: str, items: list[CartItem]) -> PaymentToolResult:
    """Create a cart at an ACP merchant. Does NOT charge yet."""
    cart = await acp_client.cart.create(merchant_id=merchant_id, items=items)
    return PaymentToolResult(
        status="success",
        details={"cart_id": cart.id, "merchant_id": merchant_id, "item_count": len(items)},
    )

@function_tool
async def acp_checkout(cart_id: str, spt_token: str) -> OrderResult:
    """Complete the checkout for a previously-created cart."""
    return await acp_client.checkout.complete(cart_id=cart_id, spt_token=spt_token)

@function_tool
async def acp_check_order_status(order_id: str) -> OrderResult:
    """Get the current status of an order. The agent calls this to follow up."""
    return await acp_client.order.status(order_id=order_id)

@function_tool
async def acp_initiate_refund(order_id: str, reason: str) -> RefundResult:
    """Start a refund for an order. Returns refund_id for follow-up."""
    response = await acp_client.refund.create(order_id=order_id, reason=reason)
    return RefundResult(
        refund_id=response.refund_id,
        order_id=order_id,
        status=response.status,
        amount_refunded_usd=response.amount_refunded_usd,
    )

shopping_agent = Agent(
    name="ShoppingAgent",
    instructions="Help the user shop. Create the cart first, confirm items with the user, then check out. Handle refund requests with an ACP refund.",
    tools=[acp_create_cart, acp_checkout, acp_check_order_status, acp_initiate_refund],
)

السؤال الذي تطرحه هنا هو هل تحتاج حالة الاستخدام إلى دورة حياة تجارة أصلا. إذا كانت تحتاجها (سلة، استرداد، نزاع)، فاختر ACP للوصول إلى ChatGPT أو UCP للوصول إلى Google، أو الاثنين. وإذا لم تكن تحتاجها، فتجاوز هذه الطبقة وانتقل مباشرة من التفويض إلى التسوية. الخطآن هما: فرض بروتوكول تجارة مستهلك على استدعاء بين آلات (أكثر مما يلزم)، أو تجاوز التجارة في شراء مستهلك حقيقي ثم إعادة تنفيذ النزاعات بطريقة رديئة لاحقا (أقل مما يلزم).

المفهوم 7: الطبقة 4، التسوية (حين يتحرك المال فعلا)

في سطر واحد: التسوية هي موضع انتقال الدولارات فعلا من حيازة إلى أخرى، والاختيار يتعلق غالبا باقتصاديات المعاملة.

انقل القيمة من المشتري إلى البائع. كل ما فوق هذه الطبقة تنسيق؛ أما التسوية فهي حيث تنتقل الدولارات (أو العملات المستقرة، أو أي وحدة أخرى) فعلا. لا يكون الوكيل قد أكمل معاملة إلا عند اكتمال التسوية.

أربعة خيارات جدية، لكل منها اقتصادياته وحدوده:

البروتوكول	كيف يعمل	الاقتصاديات	الأنسب له
x402	أصيل في HTTP؛ يسوي عبر تحويل عملة مستقرة على Base/Solana/EVM، موقّع ب EIP-3009	gas أقل من سنت، نهائية خلال 1-2 ثانية، بلا رسوم بروتوكول	مدفوعات صغيرة بين الآلات؛ تكرار عال وقيمة منخفضة (وصول API، فوترة لكل استدعاء)
MPP	جلسات: يفوض الوكيل سقفا ومدة مسبقا، ثم يبث مدفوعات مقاسة. متعدد السكك (عملة مستقرة على Tempo، وLightning، وبطاقات)	رسوم Stripe على سكك البطاقات؛ شبه صفر على العملة المستقرة؛ مناسب للاشتراكات	المؤسسات والتدفقات متعددة السكك؛ الاشتراكات المتكررة؛ حالات تحتاج fiat وcrypto في غلاف واحد
سكك البطاقات (Stripe/Adyen/Worldpay)	Visa وMastercard وAmex عبر Stripe؛ يقدم الوكيل SPT (ACP) أو Payment Mandate (AP2)؛ ويخصم المعالج من البطاقة	نحو 2.9% زائد 0.30 دولار للبطاقات؛ آليات نزاع راسخة	تدفقات المستهلك؛ معاملات فيها تعرض ل chargeback؛ قبول بطاقات دولي
تحويل بنكي / Lightning	ACH وSEPA وBitcoin Lightning	ACH نحو 0.25 دولار ثابت؛ Lightning أقل من سنت؛ SEPA نحو 0.20 يورو	تدفقات عالية القيمة حيث تؤلم رسوم البطاقة 2.9%؛ مدفوعات صغيرة عابرة للحدود عبر Lightning

يتعلق اختيار التسوية غالبا بالمال. تذهب المدفوعات دون الدولار إلى x402، حيث gas العملة المستقرة أقل من سنت. وتذهب مشتريات المستهلك حتى نحو 1,000 دولار إلى سكك البطاقات عبر ACP، حيث تستحق حماية chargeback رسوم 2.9%. وتذهب الاشتراكات المتكررة إلى جلسات MPP. وتذهب تحويلات B2B الكبيرة إلى السكك البنكية أو Lightning. كما أن الطبقات الأعلى تفرض الاختيار عادة: ACP في التجارة يعني غالبا سكك البطاقات في التسوية؛ وخادم MCP المحجوب ب x402 في الاكتشاف يعني غالبا x402 في التسوية.

انتبه لفخ الأرقام العنوانية. قد تضلل المقالات التي تستشهد بحجم معاملات x402 أو عدد تكاملات MPP. السؤال الصحيح ليس "أيهما حجمه أكبر؟" بل "أيهما يناسب اقتصاديات هذه المعاملة؟" استدعاء API بقيمة 0.001 دولار على سكك البطاقات يكلف رسوما أكثر من قيمة الاستدعاء نفسه. ومشتريات بقيمة 5,000 دولار على عملة مستقرة x402 ترمي حماية chargeback التي تمنحها البطاقة.

تعمل التسوية عادة كأثر جانبي لأداة في طبقة أعلى: نادرا ما يستدعي الوكيل settle_payment() مباشرة؛ تحدث التسوية داخل acp_checkout() أو x402_fetch(). لكن حدود الإنفاق على مستوى SDK تبقى مهمة:

note

استدعاء x402_client.get أدناه توضيحي؛ يمثل جلب x402 من جهة المشتري. ربط @function_tool وفحص الإنفاق كود Python حقيقيان.

from agents import function_tool, RunContextWrapper
from decimal import Decimal
from .models import X402PaymentResult, PaymentToolResult

@function_tool
async def x402_fetch(
    ctx: RunContextWrapper,
    url: str,
    max_payment_usdc: Decimal,
) -> X402PaymentResult | PaymentToolResult:
    """Fetch a paid URL via x402. Settlement is automatic if cost <= max_payment_usdc."""
    # Check the SDK-level spend tracker before initiating the request
    spent_so_far = Decimal(str(ctx.context.get("session_x402_spend_usdc", Decimal(0))))
    session_cap = ctx.context["user_session"].x402_session_cap_usdc
    if spent_so_far + max_payment_usdc > session_cap:
        return PaymentToolResult(
            status="rejected",
            error=f"Would exceed session spend cap (already spent ${spent_so_far})",
        )

    # Initiate the x402 flow: the server returns 402, the agent retries with a signed payment
    response = await x402_client.get(url, max_payment_usdc=max_payment_usdc)

    # Update the spend tracker, kept in ctx.context for cross-tool visibility
    ctx.context["session_x402_spend_usdc"] = spent_so_far + response.amount_paid_usdc
    return X402PaymentResult(
        content=response.content,
        amount_paid_usdc=response.amount_paid_usdc,
        tx_hash=response.tx_hash,
    )

شيء يجب توضيحه: فحص if spent_so_far + ... حارس لين داخل جسم الأداة، مفيد لفشل سريع ومفهوم للمستخدم، لكنه ليس السلامة الحقيقية. السلامة التي تحميك فعلا هي محفظة العقد الذكي للوكيل. حتى لو حذفت الفحص داخل الأداة، فإن حدود المحفظة من المفهوم 15 سترفض التحويل على السلسلة. فحص الأداة لتجربة المستخدم؛ وحدود المحفظة للسلامة.

الحركة هنا هي أن تختار السكة التي تناسب اقتصاديات المعاملة، ثم تؤكد توافقها مع اختيارك في التجارة. المدفوعات بين الآلات دون الدولار تذهب إلى x402. مشتريات المستهلك تذهب إلى سكك البطاقات عبر ACP. واشتراكات المؤسسات تذهب إلى MPP. لا تختر التسوية بمعزل؛ اخترها بوصفها قاع مكدس مركب.

---

الجزء 3: البروتوكولات الأربعة بعمق، مع تكامل OpenAI Agents SDK

وضع الجزءان 1 و2 الإطار: أربع طبقات، وعدة بروتوكولات في كل طبقة، وSDK بوصفه العميل الموحد. يمشي الجزء 3 عبر البروتوكولات الأربعة الرئيسية عن قرب. في كل واحد تحصل على ماهيته، وبدائياته الأساسية، وكود تكامله مع SDK، وملاحظات قصيرة عن كيف تنشره الفرق وتشغله.

اقرأها كأربع جولات عميقة متوازية. لكل بروتوكول الشكل نفسه، لذلك تستطيع مقارنتها جنبا إلى جنب.

كل بروتوكول في المفاهيم 8 إلى 11 يربط ب SDK عبر أحد الأنماط في هذا المخطط. يلمّح مكدس حدود الإنفاق في الأسفل إلى المفهوم 15 في الجزء 6. أبقه في زاوية عينك أثناء القراءة: انضباط السلامة الذي يعرضه هو ما يحول هذا الكود إلى شيء يمكن نشره فعلا.

🧰 Pydantic كطبقة عقد: اقرأ مرة واحدة، وينطبق على كل بروتوكول أدناه

تستخدم كل عينة كود في الجزء 3 نماذج Pydantic، لا dicts عادية في Python، لحِمل البروتوكولات، وقيم إرجاع الأدوات، وأجسام طلبات واستجابات FastAPI، وحمولات أحداث Inngest. هذا هو الجزء الذي لا يمكنك تجاوزه. إنه ما يبقي النظام كله متماسكا.

تعبر أربعة حدود في تدفق تجارة وكلاء نموذجي:

1. ترجع @function_tool في SDK قيمة إلى مستدل الوكيل.
2. تعبر تلك القيمة الشبكة إلى نقطة نهاية بروتوكول (ACP أو AP2 أو x402 أو MPP).
3. يعيد البروتوكول استجابة تعبر راجعة.
4. أحيانا يصل webhook لاحقا ويعبر إلى handler في FastAPI.

عند كل حد، يفقد dict غير المعرّف الحقول بصمت، ويسقط التحويلات، ويرسل الشكل الخطأ. تلتقط نماذج Pydantic أنواع الفشل الأربعة كلها عند الحد نفسه، بأخطاء تشير إلى الحقل المحدد.

هذا هو النمط الذي يتكرر في كل مفهوم أدناه:

from pydantic import BaseModel, Field
from decimal import Decimal
from typing import Literal
from agents import function_tool, RunContextWrapper

class CartItem(BaseModel):
    sku: str
    quantity: int = Field(ge=1)
    unit_price_usd: Decimal

class CheckoutRequest(BaseModel):
    merchant_id: str
    items: list[CartItem]
    max_total_usd: Decimal = Field(gt=0)

class CheckoutResult(BaseModel):
    order_id: str
    status: Literal["confirmed", "failed", "pending_user_confirmation"]
    total_charged_usd: Decimal
    estimated_delivery: str | None = None

@function_tool
async def acp_checkout(ctx: RunContextWrapper, request: CheckoutRequest) -> CheckoutResult:
    # Pydantic has already validated the request shape before this line runs.
    # Returning a CheckoutResult means the agent's reasoner gets typed feedback.
    ...

ثلاثة أسباب ملموسة تجعله مهما بهذا الحجم:

1. يستخدم المستدل نوع الإرجاع كتغذية راجعة. عندما ترجع acp_checkout قيمة CheckoutResult ذات نوع محدد، تصل خطوة الاستدلال التالية لدى الوكيل إلى أسماء حقول وأنواع نظيفة، لا dict محولا إلى string. ترتفع دقة اختيار الأدوات بصورة قابلة للقياس.
2. يستخدم FastAPI نماذج Pydantic أصليا. كل webhooks من Stripe، وcallbacks تفويضات AP2، وأحداث جلسات MPP تفكك إلى نماذج Pydantic في handler FastAPI: النماذج نفسها التي ترجعها أدوات الوكيل. عقد واحد، ونقطتا نهاية.
3. تحمل أحداث Inngest حمولات Pydantic. عندما يطلق handler webhook في FastAPI حدث Inngest، تكون الحمولة نموذج Pydantic. يتلقى step.wait_for_event المعلق الحمولة ذات النوع مباشرة. لا parsing ل JSON داخل workflow.

استخدم Decimal للمال دائما. يستخدم كل مبلغ نقدي في هذه الدورة Decimal، لا float. تفقد حسابات الفاصلة العائمة للمال الدقة بطرق تتراكم عبر آلاف المدفوعات الصغيرة. يقبل Stripe SDK النوعين لكنه يعيد التقارير في Decimal. وتصل مبالغ x402 كأعداد صحيحة على السلسلة (ل USDC ست خانات عشرية) تغلفها في Decimal. يبقى المال Decimal في كل موضع لا يرسل فيه إلى صيغة سلكية.

نماذج النتائج المشتركة. بدلا من إعادة تعريف أنواع النتائج في كل مفهوم، تعرّف الدورة مجموعة صغيرة من نماذج النتائج مرة واحدة. تستورد كل كتل الكود أدناه هذه النماذج بالاسم. ضعها في models.py:

from pydantic import BaseModel, Field
from decimal import Decimal
from typing import Literal
from datetime import datetime

# --- Tool-result models (returned by @function_tool functions) ---

class PaymentToolResult(BaseModel):
    """Generic envelope for any payment-related tool action."""
    status: Literal["success", "failed", "rejected", "pending"]
    error: str | None = None
    details: dict | None = None  # protocol-specific details

class MandateResult(BaseModel):
    """Result of creating an AP2 mandate (Intent / Cart / Payment)."""
    mandate_id: str | None = None
    status: Literal["signed", "declined", "pending", "failed"]
    expires_at: str | None = None  # ISO 8601
    error: str | None = None

class OrderStatusResult(BaseModel):
    """Result of fetching ACP order status."""
    order_id: str
    status: Literal["confirmed", "shipped", "delivered", "cancelled", "refunded", "pending"]
    tracking_url: str | None = None
    estimated_delivery: str | None = None

class RefundResult(BaseModel):
    """Result of initiating an ACP refund."""
    refund_id: str
    order_id: str
    status: Literal["initiated", "processing", "completed", "failed"]
    amount_refunded_usd: Decimal | None = None

class DiscoveryResult(BaseModel):
    """Result of an Agent.market or similar agent-directory search."""
    service_id: str
    name: str
    description: str
    price_per_call_usdc: Decimal
    endpoint_url: str

class X402PaymentResult(BaseModel):
    """Result of an x402-paid fetch."""
    content: str
    amount_paid_usdc: Decimal
    tx_hash: str | None = None

class MPPSessionResult(BaseModel):
    """Result of creating or closing an MPP session."""
    session_id: str
    status: Literal["active", "closed", "expired", "failed"]
    total_charged_usd: Decimal | None = None  # only populated on close
    expires_at: datetime | None = None
    rail_breakdown: dict[str, Decimal] | None = None  # only on close

class MPPMeteredCallResult(BaseModel):
    """Result of a metered call within an active MPP session."""
    session_id: str
    cost_usd: Decimal
    response_payload: dict
    accumulated_session_spend_usd: Decimal

# --- FastAPI handler models (request/response bodies for webhooks/callbacks) ---

class WebhookAck(BaseModel):
    """Generic webhook handler ack."""
    received: bool = True
    event_id: str | None = None

# --- Inngest workflow result models (return types of @inngest_client functions) ---

class WorkflowResult(BaseModel):
    """Generic workflow completion envelope."""
    status: Literal["completed", "abandoned", "failed", "partial"]
    reason: str | None = None
    output: dict | None = None

تتكرر هذه النماذج في الجزء 3 والجزء 6. تستوردها كتل الكود أدناه بالاسم ولا تعيد تعريفها. لن يكرر هذا الشريط الجانبي ذلك.

المفهوم 8: ACP (Agentic Commerce Protocol)، بروتوكول تسوق المستهلك

في سطر واحد: ACP هو الطريقة التي يكمل بها وكيل checkout حقيقيا لدى تاجر حقيقي نيابة عن شخص، مع بقاء التاجر مسؤولا عن البيع.

ما هو. ACP مواصفة مفتوحة بنتها OpenAI وStripe، وأطلقت في 29 سبتمبر 2025 مع Etsy وShopify كشريكي إطلاق. وبحلول أوائل 2026 أصبحت تشغل ChatGPT Instant Checkout عبر التجار المدمجين مع Shopify وعدة علامات تجزئة كبيرة. تختلف أعداد التجار وقوائم العلامات حسب المصدر، لذلك تحقق من دليل تكامل ACP لمعرفة التبني الحالي. البروتوكول Apache 2.0، ويحكم عبر عملية Specification Enhancement Proposal في github.com/agentic-commerce-protocol/agentic-commerce-protocol. يعلّم المستودع المواصفة ك beta، لذلك توقع انحرافا بين أمثلة الدورة والمواصفة الحية.

أين يجلس. يعيش ACP غالبا في الطبقة 3 (التجارة) ويمتد إلى الطبقة 2 (التفويض) عبر آلية رموزه. يغطي تكوين السلة، وcheckout، وإدارة الطلب، وحالة التنفيذ، وآليات الاسترداد. يبقى التاجر merchant of record: تمر chargebacks والإرجاعات وخدمة العملاء كلها عبر أنظمة التاجر القائمة. (merchant of record هو النشاط التجاري المسؤول قانونيا عن المعاملة.)

البدائيتان المهمتان.

1. رمز Shared Payment Token (SPT). رمز لمرة واحدة من معالج الدفع (Stripe في البناء المرجعي)، مقفل على تاجر واحد، وسقف مبلغ واحد، ونافذة زمنية قصيرة واحدة، وغالبا استخدام واحد. إذا حاول وكيل مسموح له بإنفاق 50 دولارا أن ينفق 1000 دولار، يفشل SPT على مستوى البروتوكول. SPT هو كيف يبقي ACP الوكيل داخل نطاق تفويض المستخدم.
2. تفويض Cart Mandate (عبر امتداد AP2). يستطيع ACP التركيب مع AP2 لمزيد من صرامة التدقيق: يوقّع المستخدم Cart Mandate قبل أن يرسل الوكيل SPT. هذا اختياري في ACP لكنه صار شائعا في التدفقات المنظمة. (التفويض إثبات موقّع بأن إنسانا فوّض نوعا محددا من الإنفاق.)

تكامل OpenAI Agents SDK. ACP هو البروتوكول الأكثر جاهزية للإنتاج مع SDK، لأن OpenAI شاركت في بنائه. يعطيك Stripe Python SDK مع غلاف عميل ACP رقيق التكامل الكامل.

note

استدعاءات acp_client وstripe.PaymentTokens.create(...) أدناه توضيحية: تبين شكل تكامل ACP. إصدار SPT الحقيقي يمر عبر نقاط نهاية Stripe ACP الحية، لا عبر طريقة stripe.PaymentTokens. ربط الوكيل وحاجز الأمان ونماذج النتائج حقيقية وتعمل اليوم. راجع الملاحظة بعد الكتلة.

from agents import Agent, Runner, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from pydantic import BaseModel
from decimal import Decimal
from typing import Literal
import json
import stripe
import time
from .models import OrderStatusResult, RefundResult

class CartItem(BaseModel):
    sku: str
    name: str
    quantity: int
    unit_price_usd: Decimal

class CheckoutResult(BaseModel):
    order_id: str
    status: Literal["confirmed", "failed", "pending_user_confirmation"]
    total_charged_usd: Decimal
    estimated_delivery: str | None = None

# Tool input guardrail: refuse the checkout if the user can't authorize the spend
@tool_input_guardrail
def verify_user_can_spend(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    args = json.loads(data.context.tool_arguments or "{}")
    max_total = Decimal(str(args.get("max_total_usd", 0)))
    merchant_id = args.get("merchant_id", "")
    user_session = data.context.context["user_session"]
    if not user_session.can_spend(max_total, merchant_id):
        return ToolGuardrailFunctionOutput.reject_content(
            f"User cannot authorize ${max_total} at merchant {merchant_id}"
        )
    return ToolGuardrailFunctionOutput.allow()

@function_tool
async def acp_browse_merchant(merchant_id: str, query: str) -> list[dict]:
    """Search a merchant's catalog via their ACP catalog endpoint."""
    response = await acp_client.catalog.search(
        merchant_id=merchant_id,
        query=query,
        limit=20,
    )
    return [item.model_dump() for item in response.items]

@function_tool(tool_input_guardrails=[verify_user_can_spend])
async def acp_create_cart_and_checkout(
    ctx: RunContextWrapper,
    merchant_id: str,
    items: list[CartItem],
    max_total_usd: Decimal,
) -> CheckoutResult:
    """Create a cart at the merchant and complete checkout in one transaction.
    Mints an SPT scoped to max_total_usd; the merchant verifies and charges.
    The verify_user_can_spend guardrail above has already validated authorization."""

    user_session = ctx.context["user_session"]

    # Mint the Shared Payment Token via Stripe (Decimal to cents via quantize)
    cents = int((max_total_usd * 100).quantize(Decimal("1")))
    spt = stripe.PaymentTokens.create(
        amount=cents,
        currency="usd",
        merchant_id=merchant_id,
        user_session_id=user_session.id,
        max_uses=1,
        expires_at=int(time.time()) + 600,  # 10-minute window
    )

    # Submit the cart and SPT to the merchant's ACP endpoint
    result = await acp_client.checkout.complete(
        merchant_id=merchant_id,
        items=[i.model_dump() for i in items],
        spt_token=spt.token,
    )

    # Update the user's spend tracker (Decimal in, Decimal out)
    user_session.record_spend(
        amount_usd=Decimal(str(result.total_charged_usd)),
        merchant_id=merchant_id,
        order_id=result.order_id,
    )

    return CheckoutResult(**result.model_dump())

@function_tool
async def acp_check_order(order_id: str) -> OrderStatusResult:
    """Get the current status of an ACP order (fulfillment, shipping, delivery)."""
    raw = await acp_client.orders.get(order_id=order_id)
    return OrderStatusResult(
        order_id=raw.order_id,
        status=raw.status,
        tracking_url=raw.tracking_url,
        estimated_delivery=raw.estimated_delivery,
    )

@function_tool
async def acp_refund(
    order_id: str,
    reason: str,
    amount_usd: Decimal | None = None,
) -> RefundResult:
    """Start a refund for an ACP order. Returns refund_id for follow-up.
    If amount_usd is None, a full refund is requested."""
    raw = await acp_client.refunds.create(
        order_id=order_id,
        reason=reason,
        amount_usd=amount_usd,
    )
    return RefundResult(
        refund_id=raw.refund_id,
        order_id=order_id,
        status=raw.status,
        amount_refunded_usd=raw.amount_refunded_usd,
    )

shopping_agent = Agent(
    name="ShoppingAgent",
    instructions="""Help the user shop at ACP-enabled merchants. Workflow:
    1. Use acp_browse_merchant to find products matching the user's request
    2. Present the matched items to the user (via reasoning, not a tool)
    3. When the user confirms, use acp_create_cart_and_checkout to complete the purchase
    4. Use acp_check_order to report order status when the user asks
    5. Use acp_refund only when the user explicitly requests a return""",
    tools=[acp_browse_merchant, acp_create_cart_and_checkout, acp_check_order, acp_refund],
    model="gpt-5.5",
)

ما هو حقيقي هنا: طبقة Agent وRunner و@function_tool وtool_input_guardrail، إضافة إلى نماذج النتائج ذات الأنواع. هذا هو الجزء الذي تعيد استخدامه لكل بروتوكول. وما هو بديل: acp_client واستدعاء stripe.PaymentTokens.create، اللذان يمثلان خلفية ACP-Stripe. في الإنتاج تبدل تلك الخلفية بنقاط نهاية Stripe ACP الحية ويبقى الباقي في مكانه. وهذا هو الشكل نفسه مع mock قابل للتشغيل حتى تراه يعمل:

class MockACPClient:
    """Illustrative backend. Real ACP uses live Stripe ACP endpoints, not stripe.PaymentTokens."""
    async def checkout(self, merchant_id: str, amount_usd) -> dict:
        return {"order_id": "ord_mock_1", "status": "confirmed", "total_charged_usd": str(amount_usd)}

acp_client = MockACPClient()   # stands in for the ACP/Stripe backend

الغلاف القياسي (نذكره مرة هنا وتشير إليه المفاهيم الثلاثة التالية). يعمل ACP على غلاف السحابة العادي بلا أجزاء إضافية. يعمل Stripe SDK داخل handler FastAPI؛ واستدعاءات ACP صادرة عبر HTTPS؛ وتعيش SPTs في سياق تشغيل الوكيل المحدد للتشغيل، وتمر عبر RunContextWrapper. القاعدة الوحيدة: ضع مفاتيح Stripe API في مخزن أسرار (key vault)، لا في متغيرات البيئة، ودوّرها حسب جدول Stripe. لا حاجة إلى sandbox (ف ACP لا يشغل كودا)، ولا حاجة إلى تخزين خاص (تستمر الطلبات في Stripe ونظام التاجر، مع سجلات ظل اختيارية للتدقيق). خط الأساس هذا للنشر هو نفسه في AP2 وx402 وMPP. لا تقول المفاهيم الثلاثة التالية إلا ما يضيفه كل واحد. يغطي درس deploying-agents المكثف نشر السحابة (قيد العمل).

تشغيله بمتانة (Inngest). معاملات ACP قصيرة، غالبا 5 إلى 30 ثانية من البداية إلى النهاية. تناسب كتل Inngest step.run بنظافة. وتظهر قيمة step.wait_for_event عندما يجب أن يتوقف الوكيل مؤقتا حتى يؤكد المستخدم السلة بين إنشاء السلة وcheckout (نمط الإنسان داخل الحلقة). تغطي الدورة المكثفة في عامل الإنتاج طبقة المتانة هذه بعمق؛ وهذا هو الشكل:

import inngest
from datetime import timedelta
from .models import WorkflowResult

@inngest_client.create_function(
    fn_id="shopping-workflow",
    trigger=inngest.TriggerEvent(event="shopping/checkout.requested"),
    concurrency=[inngest.Concurrency(limit=5, key="event.data.user_id")],
)
async def shopping_workflow(ctx: inngest.Context) -> dict:
    # Run the agent to produce the cart proposal
    cart = await ctx.step.run(
        "agent-builds-cart", build_cart_fn, ctx.event.data["user_query"],
    )

    # Wait for the user to confirm the proposed cart (human-in-the-loop)
    confirmation = await ctx.step.wait_for_event(
        "wait-for-user-confirm",
        event="shopping/cart.confirmed",
        if_exp=f"async.data.cart_id == '{cart['cart_id']}'",
        timeout=timedelta(minutes=15),
    )
    if confirmation is None:                 # timeout returns None
        return {"status": "abandoned", "reason": "user did not confirm in time"}

    # The user confirmed; complete checkout with the now-valid SPT
    result = await ctx.step.run("complete-checkout", complete_checkout_fn, cart["cart_id"])
    return {"status": "completed", "output": result}

تفصيلان يوقعان الناس في الخطأ: يطابق wait_for_event الحمولة الواردة ب if_exp (تعبير صغير)، وتعيش الحقول المنتظرة تحت بادئة async.. ويرجع timeout قيمة None، لذلك افحصها قبل المتابعة.

أين تخطئ الفرق. تتجاوز خطوة تأكيد المستخدم. يدعم ACP كلا من "يؤكد المستخدم كل سلة" و"فوّض المستخدم هذا النوع من الشراء مسبقا." تختار الفرق غالبا الثاني للسرعة، ثم تكتشف أن خطأ صغيرا في ضبط SPT يسمح للوكيل بشراء عناصر مختلفة قليلا بلا طريقة للتعافي. ابدأ بتأكيد السلة خلال أول شهر في الإنتاج. لا ترخها إلا بعد قياس معدل صحة سلال الوكيل.

المفهوم 9: AP2 (Agent Payments Protocol)، طبقة التفويض

في سطر واحد: ينتج AP2 إثباتات موقعة بأن إنسانا سمح بالإنفاق؛ لا ينقل المال بنفسه، بل يثبت أن المال كان مسموحا له أن يتحرك.

ما هو. AP2 مواصفة مفتوحة من Google مع أكثر من 60 شريكا، أطلقت في سبتمبر 2025 (آخر نسخة v0.2.0، أبريل 2026). ترخيصها Apache 2.0، وتحفظ في github.com/google-agentic-commerce/AP2، مع تنفيذات مرجعية في Python وTypeScript وKotlin وGo. AP2 هو طبقة التفويض، لا بروتوكول تجارة ولا تسوية. ينتج تفويضات موقعة تثبت أن الوكيل مخول بالإنفاق، ثم يترك التسوية الفعلية لأي سكة مناسبة (بطاقات، بنك، أو x402 عبر امتداد a2a-x402).

أين يجلس. يعيش AP2 في الطبقة 2 (الهوية والتفويض). ويبنى على بروتوكولين تحته: A2A (Agent2Agent، للمراسلة بين الوكلاء) وMCP (لتعريض الأدوات). ينتقل تفويض AP2 كبيان اعتماد موقّع عبر A2A أو مرفقا باستدعاء أداة MCP.

البدائيات الثلاث المهمة، وهي أنواع التفويض الثلاثة.

التفويض	متى ينشأ	ما الذي يثبته
Intent Mandate	عند بدء المهمة، ويوقّعه المستخدم في واجهته	أن المستخدم سمح للوكيل بالعمل داخل قواعد محددة (حدود سعر، نوافذ زمنية، تجار مسموحون)
Cart Mandate	بعد أن يبني الوكيل سلة محددة، ويوقّعه المستخدم قبل checkout (تدفقات الإنسان الحاضر)	أن المستخدم وافق على هذه السلة بالضبط وبهذا السعر بالضبط
Payment Mandate	عند لحظة الدفع، يوقّعه المستخدم أو يولد آليا مقابل Intent Mandate	أن المستخدم فوّض هذه الدفعة بالضبط على هذه السكة بالضبط

مسار التدقيق. تشكل التفويضات الثلاثة سلسلة لا يستطيع الموقّع إنكارها لاحقا: Intent ("اشتر حذاء بأقل من 120 دولارا") يؤدي إلى Cart ("هذا الحذاء بسعر 110 دولارات") يؤدي إلى Payment ("اخصم من هذه محفظة العملة المستقرة"). يشير كل تفويض إلى ما قبله. إذا طعن أحد في أي خطوة لاحقا في نزاع أو مطالبة احتيال، تكون السلسلة كلها قابلة للتدقيق. هذه الخاصية غير قابلة للإنكار: عبارة "لم أفوض هذا" لا تصمد أمام توقيع المستخدم نفسه. لهذا يناسب AP2 الصناعات المنظمة مثل الرعاية الصحية والخدمات المالية، حيث يحمل سؤال "هل فوّض المستخدم هذا فعلا؟" وزنا قانونيا.

ما الذي يتغير في SDK. لا يملك AP2 مكانا أصليا في OpenAI Agents SDK؛ تستخدم نسخه المرجعية Google Agent Development Kit. تربطه كدوال @function_tool تنشئ التفويضات وتوقعها وتتحقق منها وترسلها. الغلاف هو نفسه غلاف المفهوم 8. الشيء الوحيد الذي يضيفه AP2: واجهة توقيع يوقّع فيها المستخدم كل تفويض فعلا.

note

استيرادات from ap2 import ... وMandateSigner واستدعاءات ap2_x402 أدناه توضيحية. حزمة Python الحقيقية ل AP2 هي ap2، مع نماذج التفويض تحت ap2.types.mandate، وحقولها الحقيقية تختلف عن principal_did / agent_did / rules المبسطة هنا. لا توجد فئة MandateSigner؛ يستخدم التوقيع الحقيقي verifiable credentials عبر A2A. مفهوم سلسلة التفويضات حقيقي؛ هذا الكود نفسه بديل تعليمي. أما طبقة SDK ونماذج النتائج ذات الأنواع فحقيقية.

from agents import Agent, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from ap2 import IntentMandate, CartMandate, PaymentMandate, MandateSigner
from pydantic import BaseModel
from decimal import Decimal
from datetime import datetime
from .models import MandateResult, PaymentToolResult

class IntentRules(BaseModel):
    max_total_usd: Decimal
    allowed_merchants: list[str] | None = None
    allowed_categories: list[str] | None = None
    expires_at: str  # ISO 8601 datetime

# Guardrail: refuse any cart that has no preceding Intent Mandate
@tool_input_guardrail
def require_intent_mandate(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    intent = data.context.context.get("intent_mandate")
    if not intent:
        return ToolGuardrailFunctionOutput.reject_content(
            "No Intent Mandate found. Create one via ap2_create_intent_mandate first."
        )
    return ToolGuardrailFunctionOutput.allow()

@function_tool
async def ap2_create_intent_mandate(
    ctx: RunContextWrapper,
    task_description: str,
    rules: IntentRules,
) -> MandateResult:
    """Create an Intent Mandate at the start of a purchasing task.
    Needs the user's signature, so call this BEFORE the agent shops."""

    user_session = ctx.context["user_session"]
    mandate = IntentMandate(
        principal_did=user_session.did,  # decentralized identifier
        agent_did=ctx.context["agent_did"],
        task=task_description,
        rules=rules.model_dump(),
        issued_at=datetime.utcnow().isoformat(),
    )

    # Send to the user's signing UI; block until the user signs or rejects
    signed = await user_session.signer.request_signature(
        mandate,
        ui_prompt="Approve this shopping task?",
        timeout_seconds=300,
    )
    if not signed:
        return MandateResult(status="declined", error="User declined to sign Intent Mandate")

    # Store the mandate for later reference by Cart/Payment mandates
    ctx.context["intent_mandate"] = signed
    return MandateResult(mandate_id=signed.id, status="signed", expires_at=rules.expires_at)

@function_tool(tool_input_guardrails=[require_intent_mandate])
async def ap2_create_cart_mandate(
    ctx: RunContextWrapper,
    cart_items: list[dict],
    total_usd: Decimal,
    merchant_id: str,
) -> MandateResult:
    """Create a Cart Mandate that references the current Intent Mandate.
    Needs the user's signature in human-present flows.
    The require_intent_mandate guardrail above has verified an Intent Mandate exists."""

    intent = ctx.context["intent_mandate"]  # guaranteed present by the guardrail

    # Check that the cart fits inside the Intent Mandate's rules
    if total_usd > Decimal(str(intent.rules["max_total_usd"])):
        return MandateResult(
            status="failed",
            error=f"Cart total ${total_usd} exceeds Intent Mandate cap ${intent.rules['max_total_usd']}",
        )

    cart_mandate = CartMandate(
        parent_intent_id=intent.id,
        cart_items=cart_items,
        total_usd=str(total_usd),  # serialize Decimal as a string on the wire
        merchant_id=merchant_id,
        issued_at=datetime.utcnow().isoformat(),
    )

    user_session = ctx.context["user_session"]
    signed = await user_session.signer.request_signature(
        cart_mandate,
        ui_prompt=f"Approve this cart for ${total_usd}?",
        timeout_seconds=300,
    )
    if not signed:
        return MandateResult(status="declined", error="User declined to sign Cart Mandate")

    ctx.context["cart_mandate"] = signed
    return MandateResult(mandate_id=signed.id, status="signed")

@function_tool
async def ap2_settle_via_x402(
    ctx: RunContextWrapper,
    merchant_x402_url: str,
) -> PaymentToolResult:
    """Use the AP2 a2a-x402 extension to settle the current Cart Mandate via an x402 stablecoin payment."""
    cart = ctx.context.get("cart_mandate")
    if not cart:
        return PaymentToolResult(status="failed", error="No Cart Mandate to settle")

    # The a2a-x402 extension generates a Payment Mandate authorizing the x402 transfer
    payment_mandate = await ap2_x402.create_payment_mandate(
        cart_mandate=cart,
        rail="x402",
        chain="eip155:8453",  # Base
        asset="USDC",
    )

    # Dispatch the x402 payment with the Payment Mandate attached as proof
    result = await x402_client.pay(
        url=merchant_x402_url,
        amount_usdc=Decimal(str(cart.total_usd)),
        payment_mandate=payment_mandate,
    )
    return PaymentToolResult(status="success", details=result.model_dump())

procurement_agent = Agent(
    name="ProcurementAgent",
    instructions="""Enterprise procurement workflow:
    1. ALWAYS create an Intent Mandate first via ap2_create_intent_mandate
    2. Search merchants for matching items
    3. Build a cart and create a Cart Mandate via ap2_create_cart_mandate
    4. Settle via x402 (stablecoin) using ap2_settle_via_x402, or via card rails
    5. Record every mandate ID in the procurement system for audit""",
    tools=[ap2_create_intent_mandate, ap2_create_cart_mandate, ap2_settle_via_x402],
    model="gpt-5.5",
)

ما هو حقيقي: طبقة SDK ونماذج النتائج ذات الأنواع نفسها مثل المفهوم 8. وما هو بديل: فئات تفويض ap2، وMandateSigner، وامتداد ap2_x402. سلسلة التفويض نفسها فكرة حقيقية تستطيع بناءها؛ أما API الدقيق هنا فمبسط للتعليم. وهذا mock قابل للتشغيل حتى يعمل النمط:

class MockSignedMandate:
    def __init__(self, mid, rules=None): self.id, self.rules = mid, (rules or {})
class MockSigner:
    """Illustrative. Real AP2 signing uses verifiable credentials over A2A; no MandateSigner class exists."""
    async def request_signature(self, mandate, ui_prompt="", timeout_seconds=300):
        return MockSignedMandate(getattr(mandate, "id", "mandate_mock_1"))
ap2_x402 = type("Mock", (), {"create_payment_mandate": staticmethod(lambda **k: MockSignedMandate("pm_mock_1"))})()

ما يضيفه إلى الغلاف. شيئان فوق خط أساس المفهوم 8: واجهة توقيع ليوقّع المستخدم التفويضات (تطبيق ويب، أو تطبيق هاتف، أو أداة توقيع قائمة على الإشعارات)، وتخزين متين للتفويضات الموقعة يستمر مدة نوافذ النزاع (احتفاظ 7 سنوات معيار شائع للتفويضات المالية). واجهة التوقيع هي الفارق الحقيقي عن ACP: يستطيع ACP إعادة استخدام جلسات login القائمة، أما AP2 فيحتاج إلى تدفق توقيع مخصص.

تشغيله بمتانة (Inngest). توقيع التفويض هو المثال المدرسي لاستخدام step.wait_for_event. تطلق دالة الوكيل حدث "طلب توقيع"، وتتعلق الدالة، ويوقّع المستخدم في الواجهة فيطلق حدث "تم التوقيع"، ثم تستأنف الدالة. لا يحترق compute أثناء الانتظار، وهذا مهم لأن توقيع مؤسسة قد يستغرق ساعات. تتعمق دورة Production Worker في نمط الاستئناف هذا.

أين تخطئ الفرق. تعامل إنشاء التفويض كهم عند checkout وتوقّع التفويض بعد أن يكون الوكيل قد أنجز كثيرا من العمل. أنشئ Intent Mandate أولا، قبل أن يتسوق الوكيل أصلا. يلتقط ذلك عدم تطابق النطاق مبكرا (أراد المستخدم أحذية لكن Intent Mandate لا يسمح إلا بمستلزمات مكتبية) بدلا من أن يحدث بعد أن يحرق الوكيل compute في بناء سلة لن يستطيع تمويلها أبدا.

المفهوم 10: x402، بروتوكول التسوية الأصيل في HTTP

في سطر واحد: يتيح x402 للوكيل دفع تكلفة استدعاء API خلال ثانية إلى ثانيتين بعملة مستقرة، بإحياء رمز HTTP 402 القديم "Payment Required".

ما هو. يحول x402 رمز الحالة HTTP 402 الخامل إلى طبقة دفع عاملة لواجهات API وتجارة الآلة إلى الآلة. أنشأته Coinbase (مايو 2025)، وأطلقت V2 في ديسمبر 2025، وهو الآن تحت رعاية x402 Foundation التابعة ل Linux Foundation (أبريل 2026)، مع Cloudflare وStripe وAWS وGoogle وغيرهم كأعضاء. ترخيصه Apache 2.0. بحلول أوائل 2026، يذكر x402 أكثر من 100 مليون دفعة حتى الآن عبر Base وSolana، مع منظومة facilitators متنامية؛ تتحرك الأرقام بسرعة، لذلك تحقق من الأرقام الحالية في x402.org وصفحات إطلاق x402 من Coinbase.

أين يجلس. يعيش x402 غالبا في الطبقة 4 (التسوية) لتدفقات الآلة إلى الآلة، لكنه يمتد إلى الطبقة 1 (عبر Agent.market وأدلة مشابهة) وإلى الطبقة 3 (يعمل كطبقة تجارة كاملة للوصول إلى API بسيط حيث لا توجد دورة شراء حقيقية). في التدفقات الخالصة بين الآلات، يكون x402 غالبا البروتوكول الوحيد الذي تحتاجه.

البدائيات الأربع المهمة.

1. رمز الحالة HTTP 402. عندما يطلب عميل غير مدفوع موردا مدفوعا، يعيد الخادم 402 Payment Required مع ترويسة تحمل متطلبات الدفع: المخطط (exact لمبلغ ثابت)، والشبكة (بصيغة CAIP-2 مثل eip155:8453، ومعناها فقط "Base")، والأصل (USDC)، وعنوان المستلم، والحد الأقصى، وانتهاء الصلاحية. (CAIP-2 طريقة قياسية لتسمية blockchain حتى يبقى البروتوكول محايدا تجاه السلسلة.)
2. ترويسة تفويض الدفع. يعيد العميل المحاولة ومعه ترويسة تحمل تفويض دفع موقّعا. يوقّع التفويض خارج السلسلة، لذلك لا يدفع المشتري gas.
3. معيار EIP-3009 (transferWithAuthorization). معيار Ethereum الذي يبنى عليه x402. يسمح للمشتري بتوقيع دفعة خارج السلسلة يرسلها شخص آخر إلى السلسلة، حتى لا يلمس المشتري blockchain مباشرة ولا يدفع gas.
4. دور Facilitator. طرف ثالث اختياري يتحقق من التوقيع ويرسل الدفع على السلسلة للتاجر، فلا يشغل التاجر أي سباكة blockchain. تشغل Coinbase وCloudflare facilitators.

ملاحظة عن أسماء الترويسات: تحقق من الوثائق الحالية

مر x402 بإصداري V1 وV2، وتختلف أسماء الترويسات عبر نسخ المواصفة وال facilitators وSDKs. تستخدم وثائق Cloudflare الحالية PAYMENT-REQUIRED في استجابة 402، وPAYMENT-SIGNATURE في إعادة المحاولة من العميل، وPAYMENT-RESPONSE في رد النجاح. تستخدم بعض الأمثلة الأقدم X-PAYMENT وX-PAYMENT-PROOF. التدفق واحد؛ تختلف أسماء السلك فقط. تحقق من الأسماء الدقيقة مقابل x402.gitbook.io ووثائق facilitator الذي تدمج معه قبل كتابة الكود. يستخدم الأثر أدناه الأدوار (طلب، 402 مع المتطلبات، إعادة محاولة موقعة، نجاح مع إثبات) من دون اختيار تقليد تسمية واحد.

التدفق في أثر واحد.

1. Agent: GET https://api.example.com/data
2. Server: 402 Payment Required
   <payment-required header>: { network: "eip155:8453", asset: USDC,
                                recipient: 0xMerchant..., max_amount: 100000,
                                expiry: 1716304800 }
3. Agent (signs an EIP-3009 authorization off-chain):
   GET https://api.example.com/data
   <payment-signature header>: <base64-encoded signed authorization>
4. Server: 200 OK
   <payment-response header>: <transaction hash>
   { data: ... }

تستغرق المعاملة كلها ثانية إلى ثانيتين. لا إنشاء حساب، ولا مفتاح API، ولا جلسة، ولا إنسان داخل الحلقة.

ما الذي يتغير في SDK. لدى x402 أبسط قصة بين الأربعة. توفر Cloudflare مساعدا يغلّف عميل MCP بقدرة دفع x402؛ وللاستخدام غير MCP، تغطيه مكتبة Python من جهة المشتري مع النمط القانوني أدناه. الغلاف هو خط أساس المفهوم 8 نفسه.

note

الباني from x402_client import ... المعروض هنا، واستيراد from cloudflare_agents import withX402Client، والحقل response.amount_paid_usdc توضيحية. الحزمة الحقيقية من جهة المشتري هي x402-client (from x402_client import X402Client، والباني الحقيقي X402Client(account=...)، و.get(url) يعيد httpx.Response). باني المحفظة الأغنى هنا بديل تعليمي، والاستدعاء الحي يحتاج إلى حساب ممول ونقطة نهاية 402 حقيقية، وهو ما لا تفعله هذه الدورة. withX402Client لدى Cloudflare مساعد JavaScript؛ لا توجد حزمة Python اسمها cloudflare_agents. خادم MCP (MCPServerStreamableHttp) Python حقيقي؛ وتغليفه بالدفع توضيحي هنا. لا تتحرك أموال حقيقية.

from agents import Agent, function_tool, RunContextWrapper
from agents.mcp import MCPServerStreamableHttp
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from x402_client import X402Client, X402Wallet
from cloudflare_agents import withX402Client
from decimal import Decimal
import json
from .models import DiscoveryResult, X402PaymentResult, PaymentToolResult

# Pattern 1: wrap an MCP server's tools with x402 payment ability
research_mcp = MCPServerStreamableHttp(
    name="research-services",
    params={"url": "https://research-services.example.com/mcp"},
)
research_mcp_with_payments = withX402Client(
    research_mcp,
    wallet=agent_wallet,  # smart-contract wallet the agent controls
    max_per_call_usdc=Decimal("0.10"),
    max_per_session_usdc=Decimal("10.00"),
)

# Pattern 2: direct x402 calls as @function_tool
x402_client = X402Client(wallet=agent_wallet)

# Guardrail: refuse any x402 call that would exceed the session cap
@tool_input_guardrail
def enforce_x402_session_cap(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    args = json.loads(data.context.tool_arguments or "{}")
    max_payment = Decimal(str(args.get("max_payment_usdc", 0)))
    ctx = data.context.context
    spent = Decimal(str(ctx.get("session_x402_spend_usdc", 0)))
    cap = Decimal(str(ctx["user_session"].x402_session_cap_usdc))
    if spent + max_payment > cap:
        return ToolGuardrailFunctionOutput.reject_content(
            f"x402 session cap would be exceeded: ${spent} spent + ${max_payment} requested > ${cap}"
        )
    return ToolGuardrailFunctionOutput.allow()

@function_tool(tool_input_guardrails=[enforce_x402_session_cap])
async def x402_fetch(
    ctx: RunContextWrapper,
    url: str,
    max_payment_usdc: Decimal = Decimal("0.10"),
) -> X402PaymentResult | PaymentToolResult:
    """Fetch a URL that may require an x402 payment up to max_payment_usdc.
    Signs the EIP-3009 authorization and retries with the payment-signature header.
    The enforce_x402_session_cap guardrail above has already checked the session bound."""

    try:
        response = await x402_client.get(url, max_payment_usdc=max_payment_usdc)
        # Update the spend tracker for later guardrail checks
        current = Decimal(str(ctx.context.get("session_x402_spend_usdc", 0)))
        ctx.context["session_x402_spend_usdc"] = current + Decimal(str(response.amount_paid_usdc))
        return X402PaymentResult(
            content=response.content,
            amount_paid_usdc=Decimal(str(response.amount_paid_usdc)),
            tx_hash=response.tx_hash,
        )
    except X402PaymentRequired as e:
        return PaymentToolResult(
            status="rejected",
            error=f"Resource requires ${e.required_usdc}, exceeds max ${max_payment_usdc}",
        )

@function_tool
async def x402_search_agent_market(
    query: str,
    max_price_per_call_usdc: Decimal = Decimal("0.05"),
) -> list[DiscoveryResult]:
    """Search Agent.market for x402-paywalled services matching the query."""
    results = await agent_market_client.search(
        query=query,
        max_price_per_call_usdc=max_price_per_call_usdc,
    )
    return [
        DiscoveryResult(
            service_id=r.service_id,
            name=r.name,
            description=r.description,
            price_per_call_usdc=Decimal(str(r.price_per_call_usdc)),
            endpoint_url=r.endpoint_url,
        )
        for r in results
    ]

research_agent = Agent(
    name="ResearchAgent",
    instructions="""Research user queries by paying for data sources via x402.
    Workflow:
    1. Use x402_search_agent_market to find relevant paid services
    2. Use x402_fetch to pull data from selected services (max $0.10 per call)
    3. Synthesize the findings into a final report
    Stay under $10 per research session.""",
    tools=[x402_fetch, x402_search_agent_market],
    mcp_servers=[research_mcp_with_payments],
)

ما هو حقيقي: طبقة SDK، ونماذج النتائج ذات الأنواع، وMCPServerStreamableHttp (خادم MCP Python حقيقي). وما هو بديل: عميل x402 من جهة المشتري كما كتب هنا، وغلاف withX402Client (JavaScript فقط في الواقع)، وagent_market_client. هذه mocks قابلة للتشغيل حتى يعمل نمط الغلاف من دون أموال حقيقية:

from decimal import Decimal

class MockX402Response:
    def __init__(self, content, amount, tx):
        self.content, self.amount_paid_usdc, self.tx_hash = content, amount, tx
class MockX402Client:
    """Illustrative. Real buyer-side package: x402-client (X402Client(account=...), .get() -> httpx.Response)."""
    async def get(self, url: str, max_payment_usdc: Decimal = Decimal("0.10")):
        return MockX402Response(content="<paid data>", amount=Decimal("0.02"), tx="0xmocktx")
x402_client = MockX402Client()
agent_wallet = object()   # stands in for the wallet handle

def withX402Client(mcp_server, **kwargs):
    """Illustrative: JavaScript-only in reality. Returns the server unchanged for the Python demo."""
    return mcp_server

ما يضيفه إلى الغلاف. تقريبا لا شيء. لا يحتاج x402 إلى بنية إضافية فوق خط أساس المفهوم 8. تعيش محفظة العقد الذكي للوكيل على عنوان في السلسلة التي يتعامل عليها (Base في معظم الحالات)، ويجلس مفتاح توقيع المحفظة في key vault، وتتولى مكتبة جهة المشتري التوقيع وإعادة محاولة HTTP. (محفظة العقد الذكي هي محفظة crypto تفرض حدود إنفاقها بكود على blockchain نفسها، لذلك تصمد حتى إذا اختل كود الوكيل.)

تشغيله بمتانة (Inngest). استدعاءات x402 قصيرة (ثانية إلى ثانيتين) وidempotent: ينتج الطلب والتوقيع نفسيهما النتيجة نفسها دائما. تناسب step.run بنظافة، وتفيد memoization هنا. إذا تعطل التشغيل بعد الدفع مقابل 5 من 10 استدعاءات API، تحفظ الاستدعاءات الخمسة المدفوعة وتدفع إعادة المحاولة فقط للخمسة الباقية. ومن دون ذلك، كانت إعادة المحاولة ستدفع لكل العشرة مرة أخرى.

سقف المحفظة هو السلامة التي تحميك

الفخ هو معاملة x402 كأنك تسلم الوكيل بطاقة ائتمان. السلامة التي تحميك فعلا هي حد الإنفاق على السلسلة في المحفظة، لا max_payment_usdc لكل طلب. إذا تجاوزت حد السلسلة واستخدمت محفظة ساخنة بلا سقف، فقد تفرغها حلقة وكيل عالقة واحدة. اضبط حدود إنفاق على السلسلة لكل هوية وكيل، ولكل جلسة، ولكل تاجر، بثلاث طبقات مستقلة، حتى لا يفرغ خطأ في واحدة منها المحفظة.

المفهوم 11: MPP (Machine Payments Protocol)، بروتوكول التسوية القائم على الجلسات

في سطر واحد: يتيح MPP للوكيل فتح حساب مسبق الدفع بسقف إنفاق، ثم بث مدفوعات صغيرة كثيرة مقابله عبر عدة سكك حتى يغلق الحساب.

ما هو. تبني Stripe وTempo بروتوكول MPP، مع إعلانات إطلاق عامة في مارس 2026. Tempo هي blockchain من الطبقة 1 حضنتها Stripe وParadigm لمدفوعات الآلة عالية التكرار. أطلق MPP مع شركاء عبر Stripe وVisa وLightspark (شبكة Lightning) وغيرهم من لاعبي المنظومة؛ تنمو قائمة الشركاء، لذلك تحقق من mpp.dev ووثائق MPP لدى Cloudflare للمشاركين الحاليين. ترخيصه Apache 2.0. MPP هو جواب Stripe في طبقة التسوية على x402: حالة استخدام متداخلة، وفلسفة مختلفة.

أين يجلس. يعيش MPP في الطبقة 4 (التسوية) ويتنافس مباشرة مع x402 على مدفوعات الآلة إلى الآلة. الفرق الأساسي: MPP متعدد السكك ويدعم المصادقة لكل charge ولكل session، بينما x402 قائم على العملة المستقرة فقط ولكل طلب.

شكل HTTP. حيث يستخدم x402 رمز 402 وترويسات مخصصة، يركب MPP الدفع فوق مصادقة HTTP القياسية. يعيد الخادم WWW-Authenticate: Payment مع المتطلبات، ويعيد العميل المحاولة مع Authorization: Payment <signed-payload>، ويرد الخادم ب Payment-Receipt الذي يحمل إثبات التسوية. يجعل ذلك MPP يبدو كتدفق HTTP auth مألوف بدلا من بروتوكول مخصص.

نوعا intent.

Intent	دورة الحياة	الأنسب له
charge	تحويل لمرة واحدة، يفوّض ويسوى في round-trip واحد	مشتريات مفردة، وصول API لمرة واحدة، استبدال استدعاءات x402 الفردية عندما تحتاج إلى عدة سكك
session	يفوّض الوكيل سقفا ومدة مسبقا، ثم يبث مدفوعات صغيرة مقاسة حتى يغلق	مدفوعات صغيرة عالية التكرار حيث يكون توقيع كل طلب مكلفا؛ اشتراكات متكررة؛ نموذج "Stripe Subscription للوكلاء"

المقايضة مقابل x402.

البعد	x402	MPP
تكرار التفويض	لكل طلب (توقيع EIP-3009 في كل مرة)	لكل charge أو لكل session (تفويض واحد، عدة استدعاءات مقاسة)
شكل HTTP	402 مخصص مع ترويسات دفع	WWW-Authenticate / Authorization / Payment-Receipt القياسية
السكك	عملة مستقرة فقط (USDC على Base/Solana/EVM)	متعددة السكك (عملة Tempo المستقرة، Lightning، بطاقات، ACH)
الرسوم	بلا رسوم بروتوكول مع gas أقل من سنت	رسوم معالجة Stripe على سكك البطاقات؛ شبه صفر على عملة Tempo المستقرة
دعم التكرار	محدود (يحتاج توقيعا لكل فترة)	أصلي عبر intent من نوع session
أفضل ملاءمة	استدعاءات API لمرة واحدة، وبنية عامة	اشتراكات متكررة، وتجار مؤسسات مدمجون مع Stripe، وتدفقات تحتاج إلى fallback نقدي

ما الذي يتغير في SDK. يندمج MPP عبر Stripe Python SDK وسطح MPP في Stripe. يميل التكامل إلى إدارة الجلسات. الغلاف هو خط أساس المفهوم 8 نفسه.

note

استيراد from stripe.mpp import MPPSession واستدعاءات stripe.MPPSession أدناه توضيحية. لا توجد stripe.MPPSession ولا stripe.mpp في Stripe Python SDK؛ MPP معيار Stripe وTempo (mainnet في 18 مارس 2026)، ويدمج عبر سطح MPP لدى Stripe. مفهوم الجلسة (تفويض سقف مسبق، وبث استدعاءات مقاسة) حقيقي؛ أما API الدقيق هنا فبديل تعليمي. طبقة SDK ونماذج النتائج ذات الأنواع حقيقية.

from agents import Agent, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    ToolInputGuardrailData,
    ToolGuardrailFunctionOutput,
)
from decimal import Decimal
import json
import stripe
from stripe.mpp import MPPSession
from .models import MPPSessionResult, MPPMeteredCallResult, PaymentToolResult

@tool_input_guardrail
def verify_mpp_session_authorized(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    """Refuse session creation if the user has not pre-authorized this service."""
    args = json.loads(data.context.tool_arguments or "{}")
    service_id = args.get("service_id", "")
    max_total = Decimal(str(args.get("max_total_usd", 0)))
    user_session = data.context.context["user_session"]
    if not user_session.can_authorize_mpp_session(max_total, service_id):
        return ToolGuardrailFunctionOutput.reject_content(
            f"User has not authorized MPP sessions of ${max_total} for service {service_id}"
        )
    return ToolGuardrailFunctionOutput.allow()

@function_tool(tool_input_guardrails=[verify_mpp_session_authorized])
async def mpp_create_session(
    ctx: RunContextWrapper,
    service_id: str,
    max_total_usd: Decimal,
    duration_seconds: int = 3600,
) -> MPPSessionResult:
    """Create an MPP session for one service with a spending cap and duration.
    Returns session_id for the metered calls that follow.
    The verify_mpp_session_authorized guardrail above has already validated consent."""

    user_session = ctx.context["user_session"]
    cents = int((max_total_usd * 100).quantize(Decimal("1")))
    session = stripe.MPPSession.create(
        service_id=service_id,
        max_total_usd=cents,
        duration_seconds=duration_seconds,
        user_session_id=user_session.id,
        # The MPP server picks the rail (stablecoin/Lightning/card) by service preference
    )

    ctx.context.setdefault("mpp_sessions", {})[service_id] = session.id
    return MPPSessionResult(
        session_id=session.id,
        status="active",
        expires_at=session.expires_at,
    )

@function_tool
async def mpp_metered_call(
    ctx: RunContextWrapper,
    service_url: str,
    payload: dict,
    cost_estimate_usd: Decimal,
) -> MPPMeteredCallResult | PaymentToolResult:
    """Make a metered call inside an active MPP session.
    The session's running spend updates automatically; the cap is enforced server-side."""

    service_id = extract_service_id(service_url)
    sessions = ctx.context.get("mpp_sessions", {})
    session_id = sessions.get(service_id)
    if not session_id:
        return PaymentToolResult(
            status="failed",
            error=f"No active MPP session for service {service_id}. Create one first.",
        )

    response = await mpp_client.metered_call(
        url=service_url,
        payload=payload,
        session_id=session_id,
        cost_estimate_usd=cost_estimate_usd,
    )
    return MPPMeteredCallResult(
        session_id=session_id,
        cost_usd=Decimal(str(response.cost_usd)),
        response_payload=response.payload,
        accumulated_session_spend_usd=Decimal(str(response.accumulated_session_spend_usd)),
    )

@function_tool
async def mpp_close_session(ctx: RunContextWrapper, session_id: str) -> MPPSessionResult:
    """Close an MPP session and finalize payment. Returns the total charged and the breakdown by rail."""
    closed = await stripe.MPPSession.close(session_id)
    return MPPSessionResult(
        session_id=session_id,
        status="closed",
        total_charged_usd=Decimal(str(closed.total_charged_usd)),
        rail_breakdown={
            rail: Decimal(str(amount))
            for rail, amount in closed.rail_breakdown.items()
        },
    )

api_consumer_agent = Agent(
    name="APIConsumerAgent",
    instructions="""Consume third-party APIs efficiently using MPP sessions.
    Workflow:
    1. Identify the service to consume
    2. Create an MPP session with mpp_create_session ($X cap, Y seconds duration)
    3. Make metered calls via mpp_metered_call
    4. Close the session with mpp_close_session when done
    Sessions are cheaper than per-request payment for high-frequency calls.""",
    tools=[mpp_create_session, mpp_metered_call, mpp_close_session],
    model="gpt-5.5",
)

ما هو حقيقي: طبقة SDK ونماذج النتائج ذات الأنواع. وما هو بديل: stripe.MPPSession وmpp_client. دورة حياة الجلسة نمط حقيقي؛ وAPI الدقيق في Stripe هنا مبسط للتعليم. وهذا mock قابل للتشغيل:

from decimal import Decimal

class MockMPPSession:
    @staticmethod
    def create(**kw):
        return type("S", (), {"id": "mpp_sess_1", "expires_at": None})()
    @staticmethod
    async def close(session_id):
        return type("S", (), {"total_charged_usd": Decimal("4.20"),
                              "rail_breakdown": {"stablecoin": Decimal("4.20")}})()
mpp_client = type("Mock", (), {"metered_call": staticmethod(
    lambda **k: type("R", (), {"cost_usd": Decimal("0.05"), "payload": {},
                               "accumulated_session_spend_usd": Decimal("0.05")})())})()

ما يضيفه إلى الغلاف. خطوة إعداد واحدة: يجب أن يكون MPP مفعلا في حساب Stripe. يتولى خادم MPP لدى Stripe تكامل Tempo blockchain، لذلك لا يلمس الوكيل Tempo مباشرة. وبعد Stripe SDK وإدارة المفاتيح (كما في ACP)، لا يوجد شيء إضافي.

تشغيله بمتانة (Inngest). تطابق جلسات MPP نمط الدوال طويلة التشغيل في Inngest بنظافة. تصبح دورة حياة الجلسة (أنشئ، استخدم، أغلق) تسلسلا من كتل step.run، مع إمكانية step.sleep لانتهاء زمني. نموذج الجلسة والتنفيذ المتين يتركبان جيدا: كلاهما مبني لعمل متعدد الخطوات ذي حالة.

أين تخطئ الفرق. تنشئ جلسات كبيرة جدا أو طويلة جدا. سقف الجلسة هو حد خسارتك إذا حدث خطأ. جلسة 1,000 دولار وضعت "للراحة" تعرضك لخسارة 1,000 دولار من حلقة وكيل تسوء. اضبط الجلسات على حجم العمل المتوقع فعلا: لمهمة من 30 استدعاء، جلسة 5 دولارات مدتها 5 دقائق أكثر أمانا من جلسة 50 دولارا مدتها ساعة.

---

الجزء 4: قواعد التركيب، متى تستخدم أي بروتوكولات معا

مشى الجزء 3 عبر البروتوكولات الأربعة واحدا تلو الآخر. الآن نعيد تركيبها. قابلت الفكرة في الجزء 1: يستخدم النظام الحقيقي عدة بروتوكولات في الوقت نفسه، واحدا لكل طبقة. يعطيك الجزء 4 قواعد التركيبات التي تعمل، والتي تفشل، وكيف تبقي المكدس صغيرا بقدر ما تسمح به المهمة.

المفهوم 12: أصغر مكدس صالح لمدفوعات الوكلاء

في سطر واحد: المكدس الصحيح هو أصغر مجموعة بروتوكولات تطلق قيمة لحالة استخدام واحدة، لا البروتوكولات الأربعة في كل طبقة.

قبل أن تركب أي شيء، اسأل سؤالا واحدا: ما أصغر مكدس يطلق قيمة هنا؟ الإجابة نادرا ما تكون "كل البروتوكولات الأربعة، كل طبقة." تبدأ معظم أنظمة الإنتاج بطبقة واحدة موصولة بالكامل، ولا تضيف الباقي إلا عندما تجبرها حالة الاستخدام.

هذا هو أصغر مكدس لكل حالة استخدام شائعة. وهو يلمّح أيضا إلى القرارات الخمسة في الجزء 5.

حالة الاستخدام	الطبقة 1 (الاكتشاف)	الطبقة 2 (Auth)	الطبقة 3 (التجارة)	الطبقة 4 (التسوية)	لماذا هذا هو MVP
تسوق المستهلك (وكيل يشتري سلع تجزئة لشخص)	واجهة تسوق بالذكاء الاصطناعي، أو خادم MCP مع فهرس تاجر	ACP SPT (أو تفويض AP2 في المجالات المنظمة)	ACP	سكك البطاقات عبر Stripe	يريد معظم المشترين حماية chargeback. ACP مع Stripe هو المسار الذي يطلق اليوم.
وكيل يدفع ل API (وكيل يستدعي APIs طرف ثالث مدفوعة)	خادم MCP يدعم x402، أو دليل Agent.market	توقيع EIP-3009 (افتراضي x402)	لا شيء: استدعاء API مباشر	x402 على Base أو Solana	في الآلة إلى الآلة، تنطبق الطبقتان 2 و4 معا. لا توجد دورة حياة شراء.
مشتريات مؤسسة (وكيل يشتري من موردين معتمدين تحت قواعد)	اكتشاف A2A داخل شبكة الشركاء	AP2 Intent Mandate (مطلوب للتدقيق)	ACP أو UCP لموردي الفهارس؛ API مباشر لمشتريات الخدمات	جلسات MPP للتكرار؛ ACP SPT عبر سكك البطاقات للشراء المفرد	مسار التدقيق هو الجزء الذي لا يمكنك تجاوزه. تفويضات AP2 مطلوبة.
سوق متعدد الوكلاء (وكيل يستأجر وكلاء آخرين)	A2A أو دليل وكلاء	تفويض AP2 مع فحص سمعة ERC-8004	لا شيء: مباشر من وكيل إلى وكيل	x402 (الأكثر شيوعا)، أو MPP إذا كان Stripe موصولا بالفعل	الثقة تعمل في الاتجاهين. يحتاج كل طرف إلى هوية قابلة للتحقق وإثبات دفع.

قاعدة التركيب. اختر البروتوكول في كل طبقة التي تطلبها حالة الاستخدام. لا تضف بروتوكولات إلى طبقات لا تلمسها حالة الاستخدام أبدا. لا يحتاج وكيل يدفع ل API خالص إلى ACP. ولا ينبغي لوكيل تسوق مستهلك أن يتجه إلى x402 لقميص بقيمة 50 دولارا، لأن حماية chargeback في البطاقات تستحق رسوم 2.9%.

الفخ. تحاول بعض الفرق ربط ACP وAP2 وx402 وMPP كلها معا "للمرونة." تحصل على مساحة تكامل أكبر أربع مرات ولا جواب واضحا عن "أي بروتوكول يعمل متى." اختر مكدسا واحدا. أطلقه. أضف مكدسا ثانيا فقط عندما تطلب حالة استخدام ثانية ذلك.

المفهوم 13: متى تتركب البروتوكولات عبر الطبقات وتتنافس داخل طبقة واحدة

في سطر واحد: البروتوكولات في طبقات مختلفة مصممة لتتراكم معا؛ أما البروتوكولات في الطبقة نفسها فمصممة لتحل محل بعضها، لذلك السؤال الوحيد هو هل يقع بروتوكولان في الطبقة نفسها.

ترتبط البروتوكولات بطريقتين. إما أن تتركب لأنها تقع في طبقات مختلفة وبنيت لتتراكم، أو تتنافس لأنها تقع في الطبقة نفسها وبنيت للاستبدال. يأتي معظم الالتباس المعماري من قراءة الحالة الثانية كأنها الأولى.

عبر الطبقات، بنيت للتركيب:

التركيب	خريطة الطبقات	أين يطلق
AP2 + ACP	AP2 في الطبقة 2 (auth بدرجة تدقيق)، وACP في الطبقة 3 (التجارة)	مجالات منظمة حيث يحتاج تدفق ACP العادي إلى تدقيق إضافي
AP2 + x402	AP2 في الطبقة 2 (تفويض mandate)، وx402 في الطبقة 4 (تسوية عملة مستقرة)	تدفقات crypto-native تحتاج إلى تدقيق، عبر امتداد a2a-x402
ACP + x402	ACP في الطبقة 3 (التجارة)، وx402 في الطبقة 4 لتدفقات فرعية بين الآلات داخل شراء مستهلك	منصات هجينة حيث يتضمن شراء المستهلك بعض إنفاق API
MCP + x402 (via withX402Client)	MCP في الطبقة 1 (الاكتشاف)، وx402 في الطبقة 4 (التسوية)	نمط Cloudflare القياسي لأدوات MCP المدفوعة

داخل طبقة واحدة، بنيت للتنافس:

المنافسة	الطبقة	متى يفوز كل واحد
AP2 vs. ACP SPT vs. TAP	الطبقة 2	AP2 لتدفقات درجة التدقيق؛ وACP SPT لتدفقات المستهلك الموصولة ب Stripe؛ وTAP لفحوص الهوية فقط
ACP vs. UCP	الطبقة 3	ACP للوصول إلى ChatGPT؛ وUCP للوصول إلى Gemini؛ والاثنان للبائعين عبر الواجهات
x402 vs. MPP	الطبقة 4	x402 للمدفوعات الصغيرة لمرة واحدة وتدفقات العملة المستقرة الخالصة؛ وMPP للجلسات والاشتراكات والتدفقات متعددة السكك

الاختبار. عندما تحتار بين بروتوكولين، اسأل شيئا واحدا: هل هما في الطبقة نفسها؟ إذا كان الجواب نعم، فاختر واحدا، أو ادفع تكلفة دعم الاثنين لتدفقات فرعية مختلفة. وإذا كان الجواب لا، فهما غالبا يتركبان، والتصميم الصحيح يستخدم الاثنين كثيرا.

تركيب عملي: مكدس AP2 + x402. هذا هو النمط crypto-native، وشائع الآن في B2B وتدفقات وكيل إلى وكيل:

Layer 1 (Discovery): A2A directory inside the partner network
Layer 2 (Auth):      AP2 Intent + Cart + Payment Mandates
Layer 3 (Commerce):  Often none (direct service request), or ACP for a catalog
Layer 4 (Settlement): x402 via the a2a-x402 extension

يركب تجميع SDK أدوات البروتوكولات من الجزء 3. هذه الأدوات (a2a_discover_partners وap2_create_intent_mandate والباقي) عرّفت في المفاهيم 8 إلى 11؛ وهنا لا تفعل إلا ربطها بوكيل واحد.

agent = Agent(
    name="EnterpriseB2BAgent",
    instructions="...",
    tools=[
        # Layer 1: Discovery
        a2a_discover_partners,
        # Layer 2: Authorization
        ap2_create_intent_mandate,
        ap2_create_cart_mandate,
        # Layer 4: Settlement (a2a-x402 composes Layers 2 and 4)
        ap2_settle_via_x402,
    ],
    model="gpt-5.5",
    # No commerce tools: direct B2B procurement.
)

المفهوم 14: الكلفة والكمون، ما الذي يفرض الاختيار

في سطر واحد: يحدد حجم المعاملة والزمن الذي يمكنك انتظاره بروتوكول التسوية، لأن رسوم البطاقات تسحق المدفوعات الصغيرة وcheckout البطيء يكسر الحلقات الضيقة.

لكل تركيب سعر وسرعة. يكلف تدفق تسوق مستهلك على ACP مع سكك بطاقات نحو 2.9% + 0.30 دولار لكل معاملة ويستغرق 5 إلى 30 ثانية من البداية إلى النهاية. ويكلف وكيل API على x402 أقل من سنت ويستغرق 1 إلى 2 ثانية. يحدد التركيب الصحيح جزئيا مقدار الكلفة والانتظار الذي تستطيع حالة الاستخدام تحمله.

الكلفة لكل معاملة حسب التركيب.

التركيب	الكلفة النموذجية لكل معاملة	الكمون النموذجي
ACP + card rails (تسوق المستهلك)	2.9% + 0.30 دولار (سعر Stripe)	5 إلى 30 ثانية
ACP + MPP sessions (اشتراكات)	2.9% على البطاقات، أو نحو 0.5% على عملة Tempo المستقرة، لكل جلسة	1 إلى 3 ثوان لكل استدعاء مقاس
AP2 + x402 (B2B stablecoin)	gas أقل من سنت، بلا رسوم بروتوكول	2 إلى 5 ثوان (توقيع التفويض يضيف 1 إلى 3)
x402 only (API-paying)	gas أقل من سنت، بلا رسوم بروتوكول	1 إلى 2 ثانية
MPP sessions only (recurring API)	شبه صفر على عملة Tempo المستقرة؛ سعر Stripe على البطاقات	50 إلى 500 ms لكل استدعاء مقاس داخل جلسة نشطة

ما يفرضه المال. الرسوم التي تتجاوز نحو 5% من المعاملة مشكلة. استدعاء API بقيمة 0.05 دولار يدفع 2.9% + 0.30 دولار في رسوم بطاقة يكلف رسوما أكثر من قيمة الاستدعاء نفسه، وهذه إشارة واضحة لاستخدام x402 أو MPP بعملة مستقرة بدلا من ذلك. أما قميص بقيمة 50 دولارا يدفع النسبة نفسها ومعها 0.30 دولار فلا مشكلة. الخط بالدولار حيث تتوقف سكك البطاقات عن المنطقية يقع حول 5 إلى 10 دولارات. تحت ذلك تفوز سكك مدفوعات الآلة؛ وفوق ذلك تكون حماية chargeback في البطاقات عادة مستحقة للرسوم.

ما يفرضه الكمون. الانتظار فوق 5 ثوان لكل خطوة تواجه المستخدم مشكلة. يمكن لتوقيع تفويض AP2 أن يضيف 1 إلى 3 ثوان، وأكثر بكثير إذا انتظر توقيع إنسان؛ ويضيف checkout عبر ACP 5 إلى 30 ثانية. في تدفقات وكيل إلى وكيل بلا إنسان في الحلقة، تكون الميزانية أضيق غالبا، أحيانا دون الثانية، مما يجعل x402 على Base وMPP على Tempo الخيارين الافتراضيين.

شجرة القرار مضغوطة.

What is the transaction value?
├── Sub-dollar (per-call API, per-token billing)
│   → x402 only, or MPP sessions
├── $1 to $10 (small, low-stakes buys)
│   → x402 or MPP, with AP2 audit if you need it
├── $10 to $1,000 (consumer purchases)
│   → ACP + card rails (chargeback cover is worth the fee)
└── $1,000+ (B2B, enterprise procurement)
    → AP2 + ACP/UCP + MPP sessions, or bank rails

What is the latency budget?
├── Sub-second (multi-agent loops)
│   → MPP sessions on Tempo, or x402 on Base
├── 1 to 5 seconds (interactive)
│   → x402 or MPP; AP2 only if mandates are pre-signed
└── 5+ seconds (an acceptable user wait)
    → Full ACP checkout works

---

الجزء 5: مختبر القرار، خمسة أمثلة عملية

أعطتك الأجزاء 2 إلى 4 الإطار: أربع طبقات، وبضعة بروتوكولات لكل طبقة، وتركيب عبر الطبقات. يمشي الجزء 5 عبر خمسة قرارات حقيقية. يوضح كل واحد المنطق الكامل: ما الطبقات التي تلمسها حالة الاستخدام، وأي بروتوكول في كل طبقة، ولماذا، وكيف يبدو كود الوكيل.

اقرأ عبر الصف لترى مكدس حالة استخدام كامل. واقرأ أسفل العمود لترى ما يتغير في طبقة واحدة مع تغير حالة الاستخدام. تمشي القرارات الخمسة أدناه عبر الصفوف بالتفصيل. تغطي الأربعة الأولى المصفوفة؛ ويعيد الخامس بناء واحد منها في toolchain مختلفة تماما ليثبت أن الإطار غير مربوط بأي مورد.

القرار 1: وكيل تسوق للمستهلك (نمط ChatGPT Instant Checkout)

حالة الاستخدام. تبني وكيلا يساعد الناس على التسوق لدى تجار يدعمون ACP مثل Walmart وEtsy وبائعي Shopify. يقول المستخدم ما يريد؛ يبحث الوكيل في الفهارس، ويعرض الخيارات، وينفذ checkout عند التأكيد. مشتري واحد إلى تاجر واحد، 5 إلى 500 دولار لكل طلب، مع ضرورة الاستردادات وchargebacks.

المشي عبر الطبقات الأربع.

• الطبقة 1 (الاكتشاف). يجب أن يعثر الوكيل على منتجات عبر تجار كثيرين. يمكنك دمج فهرس MCP لكل تاجر، أو استخدام واجهة ChatGPT Shopping التي تجمع تجار ACP بالفعل. الاختيار: واجهة التسوق بالذكاء الاصطناعي، لأن عمل الاكتشاف منجز بالفعل وربط مليون فهرس بنفسك غير قابل للحياة.
• الطبقة 2 (التفويض). المستخدم مسجل في الواجهة ويؤكد كل عملية شراء، لذلك التفويض بسيط. الاختيار: ACP SPT، تصدره Stripe لكل شراء، ومحدد لتاجر واحد ومبلغ واحد ونافذة 10 دقائق.
• الطبقة 3 (التجارة). دورة الحياة الكاملة مهمة: السلة، وcheckout، والتنفيذ، والنزاعات، والاستردادات. الاختيار: ACP، البروتوكول المبني لهذا بالضبط.
• الطبقة 4 (التسوية). تقع قيم 5 إلى 500 دولار في منطقة سكك البطاقات المناسبة، وتستحق حماية chargeback الرسوم. الاختيار: سكك البطاقات عبر Stripe، وهي السكة التي يفترضها ACP افتراضيا.

التنفيذ. هذا هو كود المفهوم 8؛ الأدوات تأتي من المفهوم 8.

shopping_agent = Agent(
    name="ShoppingAgent",
    instructions="""Help the user shop. Workflow:
    1. Use acp_browse_merchant to find products matching the request
    2. Show matched items; wait for the user to confirm
    3. On confirm, use acp_create_cart_and_checkout to buy
    4. Use acp_check_order for status
    5. Use acp_refund only when the user asks""",
    tools=[acp_browse_merchant, acp_create_cart_and_checkout, acp_check_order, acp_refund],
    model="gpt-5.5",
)

أكثر فشل محتمل في الإنتاج. عدم تطابق السلة: يبني الوكيل سلة لا تطابق الطلب ("طلبت الأحمر، فوصل الوردي"). الإصلاح: اطلب من المستخدم تأكيد السلة قبل إصدار SPT، وسجل دقة السلة، واضبط التعليمات عندما تهبط الدقة تحت 95%.

تشغيله بمتانة (Inngest). استخدم نمط step.wait_for_event من المفهوم 8 لبوابة تأكيد السلة، مع سقف concurrency لكل مستخدم.

اختر هذا أولا. هذه هي حالة الاستخدام الحية اليوم: ChatGPT Instant Checkout، ومنظومة ACP، وكل تاجر Shopify لديه ACP مفعلا. إذا كنت تستطيع إطلاق تركيب واحد فقط، فأطلق هذا.

القرار 2: وكيل بحث يدفع لواجهات API (نمط x402 فقط)

حالة الاستخدام. تبني وكيل بحث يدفع لواجهات API طرف ثالث مقابل بيانات: تغذيات مالية، وأخبار، وبحث متخصص. يكتشف APIs المدفوعة في وقت التشغيل عبر دليل مثل Agent.market، ويوازن الكلفة مقابل القيمة، ويدفع لكل استدعاء. مدفوعات صغيرة عالية التكرار من 0.001 إلى 0.50 دولار، بلا إنسان في الحلقة بعد بدء المهمة، ولا دورة حياة تجارة.

المشي عبر الطبقات الأربع.

• الطبقة 1 (الاكتشاف). تتيح Agent.market وأدلة مشابهة محجوبة ب x402 للوكيل العثور على خدمات في وقت التشغيل؛ وتعالج خوادم MCP التي تدعم x402 الخدمات المدمجة مسبقا. الاختيار: Agent.market مع MCP عبر Cloudflare، اكتشاف وقت تشغيل مع مجموعة fallback موصلة مسبقا.
• الطبقتان 2 و4 (التفويض والتسوية، منطبقتان). لا إنسان بعد بدء المهمة. تحد حدود المحفظة على السلسلة من إنفاق كل معاملة؛ وتعمل حدود المستخدم عبر tool_input_guardrail في SDK على كل أداة دفع. توقيع EIP-3009 هو التفويض والتسوية معا. الاختيار: x402 على Base (USDC)، بلا بروتوكول تفويض منفصل.
• الطبقة 3 (التجارة). "الشراء" مجرد استدعاء API. الاختيار: لا شيء، وصول API مباشر عبر x402.

التنفيذ. هذا هو كود المفهوم 10؛ الأدوات تأتي من المفهوم 10.

research_agent = Agent(
    name="ResearchAgent",
    instructions="""Research the user's query by paying for data via x402.
    1. Use x402_search_agent_market to find relevant paid services
    2. Use x402_fetch to pull data (max $0.10 per call)
    3. Write up the findings
    Stay under $10 per session.""",
    tools=[x402_fetch, x402_search_agent_market],
    mcp_servers=[research_mcp_with_payments],
    model="gpt-5.5",
    # Spend caps live on x402_fetch via tool_input_guardrails
    # (Concept 10's enforce_x402_session_cap), not on the agent.
)

أكثر فشل محتمل في الإنتاج. إنفاق منفلت من حلقة عالقة: يعيد الوكيل جلب البيانات نفسها ويحرق الميزانية. الإصلاح: حدود المحفظة على السلسلة (السلامة التي تحميك فعلا)، وحواجز إنفاق الجلسة في SDK، وذاكرة dedup حتى لا تدفع عمليات الجلب المتطابقة مرة أخرى.

تشغيله بمتانة (Inngest). تفيد memoization في step.run هنا. إذا وقع crash وسط الجلسة، تستأنف إعادة المحاولة مع البيانات المدفوعة بالفعل محفوظة. ويمنع سقف concurrency لكل مستخدم دفعة واحدة من السيطرة.

آلة إلى آلة خالصة. تنطبق الطبقتان 2 و4 في توقيع واحد، والطبقة 3 فارغة. هذا المكدس أبسط بنيويا من القرار 1: بروتوكولات أقل، ونقاط تكامل أقل، وكلفة أقل لكل استدعاء. المقابل هو غياب حماية chargeback ودلالات التجارة، وهذا مناسب لأن حالة الاستخدام لا تحتاج أيهما.

القرار 3: وكيل مشتريات مؤسسة (نمط AP2 مع مكدس مركب)

حالة الاستخدام. تبني وكيل مشتريات لمؤسسة منظمة في الخدمات المالية. يفوّض المشترون مهاما: "اشتر 50 لوحة مفاتيح مريحة من موردينا المعتمدين، بأقل من 5,000 دولار، قبل الجمعة." مسار التدقيق مطلوب قانونيا، وحدود الإنفاق تعمل على عدة مستويات، والموردون معتمدون مسبقا، لذلك لا يوجد اكتشاف وقت تشغيل.

المشي عبر الطبقات الأربع.

• الطبقة 1 (الاكتشاف). قائمة الموردين معروفة مسبقا. الاختيار: خادم MCP داخلي مع فهارس الموردين، لأن نطاق الاكتشاف محدود.
• الطبقة 2 (التفويض). مسار تدقيق غير قابل للإنكار هو الجزء الذي لا يمكنك تجاوزه؛ والسجل غير القابل للإنكار هو سجل لا يستطيع الموقّع إنكاره لاحقا. تعطي تفويضات AP2 ذلك بالضبط. الاختيار: AP2، مع Intent Mandate عند إنشاء المهمة، وCart Mandate قبل checkout، وPayment Mandate عند التسوية، وكل واحد يوقّعه مسؤول المشتريات.
• الطبقة 3 (التجارة). يعرّض الموردون الكبار ACP أو UCP؛ ويعرض الأصغر APIs B2B مباشرة. الاختيار: ACP للموردين الداعمين ل ACP، وAPI مباشر للباقي؛ ويتعامل الوكيل مع الاثنين.
• الطبقة 4 (التسوية). يفضل الموردون المتكررون جلسات MPP؛ وتفضل المشتريات المفردة سكك البطاقات لحماية chargeback عند القيم الأعلى. الاختيار: جلسات MPP للتكرار، وACP SPT مع سكك البطاقات للمشتريات المفردة، ويختار ذلك حسب تاريخ المورد.

التنفيذ. يركب هذا أدوات المفاهيم 8 و9 و11.

procurement_agent = Agent(
    name="ProcurementAgent",
    instructions="""Run procurement under audit-grade compliance:
    1. ALWAYS create an Intent Mandate first via ap2_create_intent_mandate
    2. Search approved suppliers via the internal MCP server
    3. Build the cart and create a Cart Mandate via ap2_create_cart_mandate
    4. Recurring suppliers: use an MPP session.
       One-off buys: use ACP plus card rails via ap2_settle_via_acp
    5. Record every mandate ID in the procurement audit log""",
    mcp_servers=[approved_suppliers_mcp],
    tools=[
        ap2_create_intent_mandate,
        ap2_create_cart_mandate,
        ap2_settle_via_acp,
        mpp_create_session,
        mpp_metered_call,
        mpp_close_session,
    ],
    model="gpt-5.5",
    # Caps and mandate rules run via tool_input_guardrails on the paying tools
    # (Concepts 9, 11, 15). require_intent_mandate on ap2_create_cart_mandate blocks
    # any cart with no prior Intent Mandate; enforce_per_run_spend_cap blocks any
    # payment over the user's run cap.
)

أكثر فشل محتمل في الإنتاج. عدم تطابق نطاق Intent Mandate: يوقّع المسؤول تفويضا، يعمل الوكيل، ثم لا تناسب السلة التفويض. الإصلاح: تحقق من نطاق التفويض قبل أن يبدأ الوكيل التسوق (قاعدة المفهوم 9 "أنشئ Intent Mandates أولا")، وارفض المهام التي يتجاوز نطاقها ما يستطيع المستخدم تفويضه.

تشغيله بمتانة (Inngest). توقيع تفويضات AP2 يناسب step.wait_for_event بطبيعته. تستخدم المهام متعددة المراحل step.run لكل مرحلة، مع سقف concurrency لكل مستخدم.

صناعة منظمة. تفرض قواعد التدقيق AP2 في الطبقة 2؛ ويفرض الموردون المتنوعون ACP وAPIs مباشرة معا في الطبقة 3؛ ويفرض التكرار مقابل الشراء المفرد MPP والبطاقات معا في الطبقة 4. هذا التركيب أثقل من القرارين 1 و2، ومطلب التدقيق هو ما يبرر الوزن.

القرار 4: سوق متعدد الوكلاء (نمط AP2 مع x402 ومع ERC-8004)

حالة الاستخدام. تبني منصة تستأجر فيها الوكلاء وكلاء آخرين. يحتاج Agent A إلى بحث؛ ويبيع Agent B البحث كخدمة x402. لا يثق أحدهما بالآخر بعد، ويجب أن تكون المعاملات قابلة للتحقق، والدفع crypto-native خالص بلا بطاقات. وكيل إلى وكيل، 0.10 إلى 100 دولار لكل معاملة، وكل طرف يحتاج إلى هوية قابلة للتحقق.

المشي عبر الطبقات الأربع.

• الطبقة 1 (الاكتشاف). ينشر Agent B قدرته عبر A2A؛ ويعثر عليه Agent A. الاختيار: A2A، البروتوكول المبني لهذا.
• الطبقة 2 (التفويض). لا إنسان عند وقت المعاملة، لكن يجب أن تكون الثقة قابلة للتحقق في الاتجاهين. تثبت تفويضات AP2 موافقة المستخدم؛ ويمنح ERC-8004 ل Agent B سمعة على السلسلة يستطيع Agent A فحصها أولا. الاختيار: AP2 مع ERC-8004، مركبان للتحقق الثنائي الكامل.
• الطبقة 3 (التجارة). لا سلال، ولا استردادات، فقط "نفذ هذه المهمة وسلم التقرير." الاختيار: لا شيء، طلب واستجابة مباشران عبر A2A.
• الطبقة 4 (التسوية). Crypto-native، دون ثانية، ولا حاجة إلى حماية chargeback. الاختيار: x402 عبر امتداد a2a-x402 إلى AP2.

التنفيذ. يركب هذا أدوات المفاهيم 9 و10 مع اكتشاف A2A.

researcher_hiring_agent = Agent(
    name="ResearcherHiringAgent",
    instructions="""Hire research-specialist agents to work for you:
    1. Use a2a_discover_researchers to find available agents
    2. Check ERC-8004 reputation (>50 successful jobs, no flagged disputes)
    3. Create an Intent Mandate scoped by amount and recipient
    4. Submit the task via A2A with the mandate attached
    5. Receive the result; settle via ap2_settle_via_x402""",
    tools=[
        a2a_discover_researchers,
        erc8004_check_reputation,
        ap2_create_intent_mandate,
        a2a_submit_task_with_mandate,
        ap2_settle_via_x402,
    ],
    model="gpt-5.5",
)

أكثر فشل محتمل في الإنتاج. الثقة في سمعة تم التلاعب بها: درجات ERC-8004 قابلة للتدقيق، لكن يستطيع مشغل تضخيم واحدة بوظائف صغيرة ناجحة. الإصلاح: اجمع السمعة مع إشارات أخرى (هوية المشغل، عتبات حجم المعاملة، تاريخ النزاعات)، وأضف مراجعة بشرية فوق مبلغ محدد للأطراف المقابلة لأول مرة.

تشغيله بمتانة (Inngest). استخدم fan out عند استئجار عدة متخصصين معا؛ واستخدم step.wait_for_event لكل نتيجة وstep.run لكل مرحلة. يلمس هذا القرار كل بدائيات Inngest.

اقتصاد متعدد الوكلاء خالص. لا إنسان داخل الحلقة عند وقت المعاملة؛ يعمل الوكيلان داخل نطاقات مفوضة مسبقا. هذا هو الشكل الذي يبنى حوله اقتصاد الوكلاء، ولديه أكثر أنماط الفشل، لأن الثقة الثنائية بلا علاقة سابقة صعبة، ومكدس البروتوكولات لا يحل إلا جزءا منها.

القرار 5: مكدس بلا Stripe ولا OpenAI (إثبات أن الإطار ينتقل)

استخدمت كل عينة كود حتى الآن stripe.PaymentTokens.create(...) وOpenAI Agents SDK. هذه هي أكثر التكاملات نضجا ل ACP، وSDK هو وقت تشغيل هذه الدورة، لكن المعمارية ذات الطبقات الأربع محايدة تجاه المكدس بالتصميم، والدورة التي تعرض مكدسا واحدا فقط لم تثبت ذلك. لذلك يعيد هذا القرار بناء القرار 2، وكيل البحث الذي يدفع ل API، على toolchain مختلفة تماما: Google Agent Development Kit (ADK) لوقت التشغيل، ومحفظة عقد ذكي من Coinbase للهوية على السلسلة، وتفويضات AP2 للتفويض، وx402 مباشر للتسوية. صفر Stripe، صفر OpenAI.

تبقى حالة الاستخدام هي حالة القرار 2. وكيل بحث يدفع 0.001 إلى 0.10 دولار لكل استدعاء، بسقف 10 دولارات لكل جلسة. دون الدولار، بلا إنسان بعد التفويض، وتسوية آلة إلى آلة. وتبقى المعمارية أيضا معمارية القرار 2: يطبق x402 الطبقتين 2 و4 معا، ولا طبقة تجارة، وMCP للاكتشاف. تتغير المكتبة فقط.

note

الكتلة أدناه توضيحية. بنية Google ADK (Agent ومزخرف الأداة) حقيقية وgoogle-adk حزمة حقيقية. قطع الدفع بدائل: حزمة AP2 الحقيقية هي ap2 بحقوق تفويض مختلفة جدا وبلا فئة MandateSigner، وتضبط محفظة Coinbase عبر coinbase_agentkit وفئته SmartWalletProvider، والاستدعاء الحي ل x402 يحتاج إلى حساب ممول ونقطة نهاية 402 حقيقية. لا يتحرك مال حقيقي هنا. المهم هو الشكل، لا مشتري قابل للتشغيل.

from google.adk import Agent
from google.adk.tools import function_tool  # ADK's tool decorator
from coinbase_agentkit import AgentKit, SmartWalletProvider
from decimal import Decimal
from datetime import datetime, timedelta

# Shared result models come from the Pydantic sidebar in Part 3.
from .models import X402PaymentResult, PaymentToolResult, DiscoveryResult

# --- Illustrative stand-ins for the payment rails (real APIs differ) ---
class MockMandate:
    def __init__(self, mid, rules=None): self.id, self.rules = mid, (rules or {})
class MockSigner:
    async def sign(self, mandate): return mandate          # real AP2 signs over A2A
agent_market_client = type("Mock", (), {"search": staticmethod(lambda **k: [])})()
x402_client = type("Mock", (), {})()                       # real client: x402-client
# -----------------------------------------------------------------------

# Layer 2 (Authorization): a Coinbase smart-contract wallet gives the agent its
# on-chain identity and its spend caps. This is the analog of a Stripe customer
# plus per-customer caps, but enforced by the chain.
wallet_provider = SmartWalletProvider(
    config={
        "chain": "base-mainnet",
        "spend_limits": {
            "per_transaction_usdc": Decimal("0.50"),
            "per_session_usdc": Decimal("10.00"),
            "per_day_usdc": Decimal("100.00"),
        },
    },
)
agent_kit = AgentKit(wallet_provider=wallet_provider)

# Layer 2 (Authorization): an AP2 Intent Mandate, signed by the user at session start.
# The analog of a signed, revocable Stripe authorization, but declarative.
async def create_research_intent_mandate(user_did, user_signer, session_cap_usdc):
    mandate = MockMandate(
        "intent_mock_1",
        rules={
            "max_total_usd": str(session_cap_usdc),
            "allowed_categories": ["data-api", "research-service"],
            "expires_at": (datetime.utcnow() + timedelta(hours=1)).isoformat(),
        },
    )
    return await user_signer.sign(mandate)

# Layer 1 (Discovery) + Layer 4 (Settlement) as one tool.
# ADK's @function_tool is the analog of the SDK's @function_tool.
@function_tool
async def x402_paid_fetch(url: str, max_payment_usdc: Decimal) -> X402PaymentResult | PaymentToolResult:
    """Fetch a URL that may need x402 payment up to max_payment_usdc.
    The wallet handles the signature; the on-chain cap is the safety that protects you."""
    # ADK has no tool_input_guardrail, so the check runs in-tool,
    # backed by the wallet's on-chain cap (the layer nothing can bypass).
    resp = await x402_client.get(url, max_payment_usdc=max_payment_usdc)
    return X402PaymentResult(
        content=resp.content,
        amount_paid_usdc=Decimal(str(resp.amount_paid_usdc)),
        tx_hash=resp.tx_hash,
    )

@function_tool
async def search_agent_directory(query: str, max_price_per_call_usdc: Decimal) -> list[DiscoveryResult]:
    """Search Agent.market for x402-paywalled services."""
    results = await agent_market_client.search(query=query, max_price_per_call_usdc=max_price_per_call_usdc)
    return [
        DiscoveryResult(
            service_id=r.service_id,
            name=r.name,
            description=r.description,
            price_per_call_usdc=Decimal(str(r.price_per_call_usdc)),
            endpoint_url=r.endpoint_url,
        )
        for r in results
    ]

# The agent itself: a Google ADK Agent, the direct analog of the SDK's Agent.
research_agent = Agent(
    name="research-agent",
    description="Research the user's query by paying for data via x402.",
    instructions="""Research the query.
    1. Use search_agent_directory to find relevant paid services
    2. Use x402_paid_fetch to pull data ($0.50 max per call)
    3. Write up the findings
    Stay under $10 per session.""",
    model="deepseek-v4-flash",  # illustrative; any ADK-compatible model id works
    tools=[search_agent_directory, x402_paid_fetch],
)

# Driver: the user signs the Intent Mandate once at session start;
# the agent then runs on its own inside the mandate's scope until the session ends.
async def run_research_session(user_did, user_signer, query):
    intent = await create_research_intent_mandate(
        user_did=user_did,
        user_signer=user_signer,
        session_cap_usdc=Decimal("10.00"),
    )
    return await research_agent.run_async(
        query,
        context={"intent_mandate": intent, "wallet": agent_kit},
    )

الترجمة سطرا بسطر. كود القرار 2 باستخدام OpenAI مع Stripe على اليسار، وكود القرار 5 باستخدام Google ADK مع Coinbase على اليمين. هذا الجدول هو الخلاصة: كل صف هو المفهوم نفسه باسم مختلف.

القرار 2 (OpenAI + Stripe)	القرار 5 (Google ADK + Coinbase)	المفهوم نفسه، مكتبة مختلفة
from agents import Agent, function_tool	from google.adk import Agent + from google.adk.tools import function_tool	وقت تشغيل الوكيل
@function_tool (OpenAI Agents SDK)	@function_tool (Google ADK)	مزخرف الأداة
RunContextWrapper	وسيط context={...} إلى run_async	حالة كل تشغيل
stripe.Customer.modify(...) for caps	SmartWalletProvider(spend_limits={...})	حدود الإنفاق، أصلية للسلسلة
tool_input_guardrail decorator	فحص داخل الأداة + حدود المحفظة	تحقق قبل التنفيذ
Runner.run(agent, ...)	agent.run_async(...)	تنفيذ الوكيل

ما يبقى متطابقا. المعمارية: الطبقة 1 هي MCP أو دليل، والطبقة 2 تفويض مع حدود محفظة، والطبقة 3 لا شيء (تطبق الآلة إلى الآلة التجارة معا)، والطبقة 4 x402. البدائيات: Intent Mandate، وتوقيعات EIP-3009، واستجابات 402، وترويسات payment-signature، وحدود الإنفاق على السلسلة. الإطار هو ما يبقى بعد تبديل المكتبة.

فرقان تشغيليان حقيقيان.

1. لا يملك Google ADK حاجز إدخال أداة أصليا من الدرجة الأولى (حتى منتصف 2026). إن tool_input_guardrail في SDK مفيد فعلا للفحوص قبل التنفيذ؛ ولا يملك مزخرف الأداة في ADK مقابلا مباشرا بعد. workaround هو تحقق داخل الأداة مدعوم بحدود المحفظة على السلسلة. ما زالت الحدود تحميك؛ الفحص داخل الأداة يفشل أسرع فقط. إذا اخترت ADK، فأنت تقايض راحة حاجز الأمان بقصة مختلفة متعددة الوكلاء.
2. توقيع تفويضات AP2 أكثر أصلية هنا. بنت Google AP2، لذلك تدمج منظومة ADK تدفقات واجهة توقيع التفويض بسلاسة أكبر. إذا كانت حالة استخدامك تعتمد بقوة على تفويض صارم بال mandating (القرارات 3 و4)، فإن ADK مع AP2 ملاءمة حقيقية، لا مجرد بديل.

خلاصة القرار 5: معمارية الطبقات الأربع ليست قصة Stripe وOpenAI متنكرة. الاكتشاف، والتفويض، والتجارة، والتسوية؛ اختر بروتوكولا واحدا لكل طبقة، وبرره مقابل حالة الاستخدام: هذا يبقى بعد أي تبديل مكتبة. استخدم القرار 5 Google ADK ومحفظة Coinbase لبناء التركيب نفسه تماما مثل القرار 2؛ تغيرت الاستيرادات فقط. إذا لم يمكن التعبير عن إطار في مكدس مختلف، فهو درس مكتبة يرتدي زيا تنكريا. هذا الإطار ليس كذلك.

---

الجزء 6: مخاوف الإنتاج، ما الذي يقتلك عندما يعمل النظام فعلا

بنت الأجزاء 1 إلى 5 الإطار ومشت عبر قرارات حقيقية. يغطي الجزء 6 الأشياء التي تحدد هل يصمد مكدسك أمام مستخدمين حقيقيين. هذه هي الإخفاقات التي لا تظهر في demo وتظهر في الساعة 2 صباحا.

مسار القارئ

إذا كنت تقرأ للفهم لا للإطلاق، يمكنك قراءة الجزء 6 بسرعة. وإذا كنت تبني أيا من هذا بجدية، فهنا يصبح الأمر حقيقيا. المفاهيم الأربعة هنا هي الفرق بين نظام يعمل ونظام يستنزف محفظة.

المفهوم 15: فرض حدود الإنفاق على ثلاثة مستويات معمارية

في سطر واحد: أوقف الوكيل عن تجاوز الإنفاق بفرض الحد في ثلاثة أماكن مستقلة، حتى يلتقط المكانان الآخران خطأ أحدها.

أكبر طريقة يفشل بها نظام تجارة وكلاء فشلا سيئا بسيطة: ينفق الوكيل أكثر مما سمح له. يمكن لحلقة وكيل عالقة أن تستنزف محفظة في ثوان. لكل بروتوكول سقفه الخاص (مبلغ SPT في ACP، وسقف جلسة MPP، وحد x402 لكل طلب)، لكن لا يكفي أي واحد منها وحده. تفرض أنظمة الإنتاج حدود الإنفاق على ثلاثة مستويات منفصلة.

المستوى 1: حدود المحفظة وطريقة الدفع. هذا هو السقف الذي يحميك فعلا. تحمل محفظة العقد الذكي للوكيل (ل x402) أو حساب Stripe الخاص به (ل ACP وMPP) حدود إنفاق مضبوطة على مستوى البنية التحتية. تفرض السلسلة أو Stripe هذه الحدود مهما فعل كود الوكيل. هذا هو المستوى الوحيد الذي يصمد عندما تفشل حلقة الوكيل تماما.

ل x402 مع محفظة عقد ذكي:

note

استدعاء SmartContractWallet.deploy(...) أدناه توضيحي. يبين أين تعيش حدود المستوى 1، ولا يمثل حزمة pip حقيقية. عمليا تضبط هذه الحدود على محفظة عقد ذكي حقيقية (مثلا عبر SmartWalletProvider في Coinbase AgentKit). الدرس الحقيقي هو انضباط المستويات الثلاثة.

from decimal import Decimal

# Set ONCE when the wallet is deployed. The agent cannot change this.
wallet_spend_limits = {
    "max_per_transaction_usdc": Decimal("10.00"),    # cap per single transfer
    "max_per_day_usdc": Decimal("100.00"),           # rolling 24-hour cap
    "max_per_merchant_usdc": Decimal("50.00"),       # cap to any single recipient
}
agent_wallet = SmartContractWallet.deploy(
    owner=user_did,
    spend_limits=wallet_spend_limits,
    chain="eip155:8453",  # Base
)

في ACP أو MPP مع Stripe، يعيش السقف على عميل Stripe:

stripe.Customer.modify(...) استدعاء Stripe API حقيقي. يعمل ضد حساب Stripe الخاص بك.

# Set once via the Stripe Dashboard or API. The caps live in Stripe's infrastructure.
stripe.Customer.modify(
    user_session.stripe_customer_id,
    metadata={
        "max_per_session_usd": "500",
        "max_per_day_usd": "2000",
    },
)
# When an SPT or MPP session is minted above these, Stripe rejects it at the API level.

المستوى 2: حواجز أدوات SDK. يعمل tool_input_guardrail في OpenAI Agents SDK قبل تنفيذ كل أداة دفع ويمكنه رفض الاستدعاء. قابلته في المفهوم 5: إنه طريقة SDK الأصلية لإيقاف الدفع قبل حدوثه، والعائلة التي تعمل في الوقت المناسب. هنا يأخذ المعالجة الكاملة، لأن هذه كتلة guardrail القانونية للدورة كلها. الكود أدناه حقيقي ويعمل.

import json
from decimal import Decimal
from agents import Agent, function_tool, RunContextWrapper
from agents.tool_guardrails import (
    tool_input_guardrail,
    tool_output_guardrail,
    ToolInputGuardrailData,
    ToolOutputGuardrailData,
    ToolGuardrailFunctionOutput,
)

# Tool INPUT guardrail: pre-payment check. Runs BEFORE the tool executes.
@tool_input_guardrail
def enforce_per_run_spend_cap(data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput:
    """Reject any payment tool call where the run's total spend would exceed the user's cap.
    Runs before the tool executes, the only guardrail family that can stop a payment in time."""
    args = json.loads(data.context.tool_arguments or "{}")          # raw JSON args string -> dict
    requested = Decimal(str(args.get("max_total_usd") or args.get("max_payment_usdc") or 0))
    ctx = data.context.context                                       # the run context (a dict)
    cap = Decimal(str(ctx["user_session"].per_run_spend_cap_usd))
    spent = Decimal(str(ctx.get("run_spend_usd", 0)))
    if spent + requested > cap:
        return ToolGuardrailFunctionOutput.reject_content(
            f"Refused: would spend ${spent + requested}, run cap is ${cap}"
        )
    return ToolGuardrailFunctionOutput.allow()

# Tool OUTPUT guardrail: post-payment check. Runs AFTER the tool executes.
# Useful for verifying receipts (paid more than expected, wrong amount, etc.).
@tool_output_guardrail
def verify_receipt_integrity(data: ToolOutputGuardrailData) -> ToolGuardrailFunctionOutput:
    output = data.output or {}
    if isinstance(output, dict) and "amount_paid_usdc" in output:
        # Cross-check the receipt against what we asked for.
        pass
    return ToolGuardrailFunctionOutput.allow()

# Attach BOTH guardrails to every payment-authorizing tool.
@function_tool(
    tool_input_guardrails=[enforce_per_run_spend_cap],
    tool_output_guardrails=[verify_receipt_integrity],
)
async def x402_fetch(
    ctx: RunContextWrapper,
    url: str,
    max_payment_usdc: Decimal,
) -> "X402PaymentResult":
    ...

# An agent-level output_guardrail is fine for final-reply safety
# (like redacting PII in the agent's answer), but it does NOT prevent payments.
agent = Agent(
    name="ShoppingAgent",
    tools=[x402_fetch],
    # output_guardrails=[response_safety_guardrail],  # different job, not payment safety
)

استخدم عائلة guardrail الصحيحة

لدى SDK ثلاث عائلات حواجز أمان. تعمل حواجز الإدخال على أول رسالة من المستخدم إلى الوكيل. وتعمل حواجز الإخراج على الرد النهائي للوكيل. وتعمل حواجز الأدوات (tool_input_guardrail وtool_output_guardrail) على كل استدعاء أداة مخصصة. لسلامة الدفع تريد tool_input_guardrail تحديدا: إنه العائلة الوحيدة التي تعمل قبل أداة الدفع ويمكنها حجبها. اللجوء إلى output_guardrail للتحكم في الإنفاق هو الخطأ الأكثر شيوعا في كود تجارة الوكلاء. عندما يعمل، يكون المال قد ذهب.

المستوى 3: حدود التطبيق ومنطق العمل. يفرض كودك قواعد المستخدم الخاصة: سقوف يومية لكل مستخدم، وسقوف لكل فئة، وتجار مسموحون. هنا تعيش قواعد العمل. "يمكن لهذا المستخدم إنفاق 500 دولار يوميا لدى أي تاجر، لكن 50 دولارا فقط يوميا لدى غير الموثقين." هذه القاعدة تنتمي هنا، لا إلى بروتوكول. هذا كود Python عادي وحقيقي.

from decimal import Decimal

class UserSession:
    def can_spend(self, amount_usd: Decimal, merchant_id: str) -> bool:
        # Per-day cap
        if self.today_spend_usd + amount_usd > self.daily_cap_usd:
            return False
        # Per-merchant cap
        merchant_cap = self._merchant_cap_for(merchant_id)
        if self.merchant_spend_usd[merchant_id] + amount_usd > merchant_cap:
            return False
        # Per-category cap (for example, "office supplies" vs "personal")
        category = self._category_for(merchant_id)
        if self.category_spend_usd[category] + amount_usd > self.category_caps[category]:
            return False
        return True

يعيش كل مستوى في بنية مختلفة. المستوى 1 في السلسلة أو Stripe. والمستوى 2 في SDK الوكيل. والمستوى 3 في كود تطبيقك. يلتقط خطأ في واحد منهما المستويان الآخران. إذا تجاوزت المستوى 1، فقد يفرغ خطأ حلقة وكيل واحد المحفظة كلها. وإذا تجاوزت المستوى 2، تفقد القدرة على إجهاض تشغيل في منتصف الطريق. وإذا تجاوزت المستوى 3، لا تستطيع فرض سياسة كل مستخدم أو كل فئة.

الفخ هو الثقة في سقوف البروتوكولات وحدها. سقف SPT في ACP، وسقف جلسة MPP، والحد لكل طلب في x402 كلها حدود على مستوى البروتوكول. توقف إساءة استخدام بروتوكول محددة، لكنها لا تجمع عبر البروتوكولات. فريق يستخدم SPTs من ACP فقط بسقف 50 دولارا لكل واحد لا يملك حماية من وكيل يصدر 100 منها متتالية، بإجمالي 5,000 دولار. توجد المستويات الثلاثة أعلاه بالضبط لأن سقوف البروتوكولات لا تتراكم.

المفهوم 16: نظافة هوية الوكيل: المفاتيح والمحافظ وسجلات التدقيق

في سطر واحد: مفتاح توقيع الوكيل هو الشيء الوحيد الفاصل بين الإنفاق المصرح والاحتيال، لذلك تحمي المفتاح، وتفصله لكل وكيل، وتدوره، وتسجل كل إنفاق في تخزين متين.

تجلب تجارة الوكلاء فشلا فريدا للأنظمة المستقلة: هوية الوكيل التشفيرية هي كل ما يفصل الإنفاق الحقيقي عن الاحتيال. إذا تسرب مفتاح التوقيع، يمكن استنزاف المحفظة (أو عميل Stripe، أو موقّع تفويض AP2) أو انتحالها حتى تدور المفتاح. نظافة الهوية هي مجموعة العادات التي تمنع ذلك. هناك أربع عادات.

1. فصل المحفظة لكل وكيل. يحصل كل وكيل، أو كل فئة وكلاء، على محفظته أو مقبض دفعه الخاص. لا تشارك مفاتيح التوقيع بين وكلاء لهم وظائف مختلفة. إذا شارك وكيل التسوق ووكيل المشتريات محفظة واحدة، فإن اختراق أيهما يستنزف الاثنين. تكلف المحافظ المنفصلة قليلا جدا (كلفة نشر لمرة واحدة) والمكسب الأمني حقيقي.

SmartContractWallet.deploy(...) توضيحي كما في المفهوم 15. النمط هو المهم: محفظة واحدة لكل فئة وكيل، لا مشاركة.

# Wrong: one wallet shared across agents
shared_wallet = SmartContractWallet.deploy(...)
shopping_agent.wallet = shared_wallet
procurement_agent.wallet = shared_wallet
research_agent.wallet = shared_wallet  # one compromise drains all three

# Right: a separate wallet per agent class
shopping_agent.wallet = SmartContractWallet.deploy(
    spend_limits={"max_per_day_usdc": 100},
)
procurement_agent.wallet = SmartContractWallet.deploy(
    spend_limits={"max_per_day_usdc": 1000, "allowed_recipients": [...]},
)
research_agent.wallet = SmartContractWallet.deploy(
    spend_limits={"max_per_day_usdc": 50, "max_per_call_usdc": 0.50},
)

2. تدوير المفاتيح، بجدول وعند الطلب. دوّر مفاتيح التوقيع كل 90 يوما كخط أساس (مطابقا لنصيحة Stripe لمفاتيح API). ودوّرها فورا عندما يترك مشغل وكيل الفريق، أو عندما يمس نشر سطح التوقيع، أو عندما يبدو شيء غريبا. عادة التدوير أهم من عدد الأيام الدقيق.

note

النمط أدناه حقيقي. اسم العميل azure_key_vault توضيحي؛ استخدم SDK خزانة المفاتيح لدى مزودك. النقطة هي أن المفتاح يعيش في vault وتقرأ نسخته الحالية عند الاستخدام.

# Read the current key version from the vault. The version changes when the key rotates.
def get_signing_key(agent_class: str) -> SigningKey:
    return azure_key_vault.get_latest_version(
        secret_name=f"agent-wallet-signing-key-{agent_class}",
    )

# Old transactions, signed with the previous version, stay valid until they expire.
# New transactions use the current version.

3. سجلات تدقيق تصمد بعد crash. يسجل كل قرار تفويض إلى تخزين متين يعيش منفصلا عن وقت تشغيل الوكيل: كل SPT صدر، وكل تفويض وقع، وكل توقيع x402، وكل جلسة MPP فتحت. إذا تعطل الوكيل، يجب أن يبقى سجل التدقيق. تمنحك Neon Postgres مع memoization في Inngest ذلك؛ ويمكنك أيضا الكتابة مباشرة إلى تخزين كائنات (S3 أو ما يعادله) لأقصى متانة.

note

نمط سجل التدقيق هو الدرس. اسم العميل neon_client توضيحي؛ استخدم عميل قاعدة بياناتك. القاعدة هي تسجيل القرار قبل حدوث الدفع.

# Every payment-authorizing action logs to durable storage BEFORE the action completes.
@function_tool
async def acp_create_cart_and_checkout(
    ctx: RunContextWrapper,
    merchant_id: str,
    items: list["CartItem"],
    max_total_usd: Decimal,
) -> "CheckoutResult":
    audit_id = str(uuid4())

    # Log the authorization decision FIRST, before any payment happens.
    await neon_client.audit_log.insert({
        "audit_id": audit_id,
        "agent_class": ctx.context["agent_class"],
        "user_did": ctx.context["user_session"].did,
        "action": "acp_create_cart_and_checkout",
        "merchant_id": merchant_id,
        "max_total_usd": max_total_usd,
        "timestamp": datetime.utcnow().isoformat(),
        "status": "initiated",
    })

    try:
        result = await _actually_complete_checkout(merchant_id, items, max_total_usd)
        await neon_client.audit_log.update(audit_id, {
            "status": "completed",
            "actual_total_usd": result.total_charged_usd,
            "order_id": result.order_id,
        })
        return result
    except Exception as e:
        await neon_client.audit_log.update(audit_id, {"status": "failed", "error": str(e)})
        raise

4. تتبعات موزعة عبر المعاملة كلها. يخبرك سجل التدقيق ما نجح. وتخبرك traces بما حدث، بما في ذلك الاستدعاءات التي فشلت أو أعيدت أو توقفت. في تجارة الوكلاء، يكون trace الكامل غالبا الطريقة الوحيدة لتنقيح معاملة فاشلة، لأن طلب مستخدم واحد يمكن أن يتفرع إلى تشغيل SDK، و5 إلى 10 استدعاءات أداة، و2 أو 3 طلبات HTTP بروتوكول، وwebhook من Stripe يصل لاحقا، ودالة Inngest تستأنف بعد ساعات. ومن دون trace ID واحد يربط ذلك كله، تصبح post-mortem مستحيلة. كود OpenTelemetry أدناه هو API OTel الحقيقي والمستقر.

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

tracer = trace.get_tracer("agent-commerce")

@function_tool
async def acp_create_cart_and_checkout(
    ctx: RunContextWrapper,
    merchant_id: str,
    items: list["CartItem"],
    max_total_usd: Decimal,
) -> "CheckoutResult":
    # The span name is the protocol action; attributes capture what you filter by
    # in your observability tool (Datadog, Honeycomb, Grafana, and so on).
    with tracer.start_as_current_span(
        "acp.checkout",
        attributes={
            "agent.class": ctx.context["agent_class"],
            "user.did": ctx.context["user_session"].did,
            "acp.merchant_id": merchant_id,
            "acp.max_total_usd": float(max_total_usd),
            "acp.item_count": len(items),
        },
    ) as span:
        try:
            result = await _actually_complete_checkout(merchant_id, items, max_total_usd)
            span.set_attribute("acp.order_id", result.order_id)
            span.set_attribute("acp.actual_total_usd", float(result.total_charged_usd))
            span.set_status(Status(StatusCode.OK))
            return result
        except Exception as e:
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.record_exception(e)
            raise

يتدفق سياق trace تلقائيا عبر httpx وopenai-agents وInngest عند وجود instrumentation الصحيح. عندما يصل webhook من Stripe بعد 20 دقيقة لنزاع على هذا الطلب، ينضم handler ال webhook إلى trace نفسه عبر trace_id المحمول في metadata الطلب. يغطي trace ID واحد عمر المعاملة كله، من أول طلب إلى حل النزاع.

تجيب سجلات التدقيق وtraces عن أسئلة مختلفة، وتحتاج إلى الاثنين. تجيب سجلات التدقيق عن أسئلة العمل ("كم أنفق هذا المستخدم يوم الثلاثاء؟"). وتجيب traces عن أسئلة التنقيح ("لماذا فشل checkout للطلب abc123؟"). يسجل سجل التدقيق المسار الناجح فقط؛ ولا يلتقط أبدا الاستدعاء الذي أعيد خمس مرات قبل أن يعمل، أو استدعاء البروتوكول الذي توقف 30 ثانية قبل timeout. لا تظهر أشكال الفشل تلك إلا في traces. تجاوز traces لأن لديك سجلات تدقيق خطأ في فهم وظيفة كل أداة.

أكثر خطأ شائع في الهوية هو معاملة عنوان المحفظة كهوية الوكيل. العنوان عام: يستطيع أي أحد الإرسال إليه وأي أحد التحقق من أن معاملة صدرت منه. مفتاح التوقيع الخاص هو الهوية، وتحميه على هذا الأساس. الفرق التي تحفظ مفاتيح التوقيع في متغيرات البيئة (أو أسوأ، في الكود المصدري) سلمت هوية الوكيل لكل من يستطيع الوصول إلى تلك الأسرار. تذهب المفاتيح إلى vault، لا إلى env vars.

المفهوم 17: آليات النزاع والاسترداد عبر البروتوكولات الأربعة

في سطر واحد: يتعامل كل بروتوكول مع النزاعات والاستردادات بشكل مختلف، وغالبا يكون نموذج النزاع الذي تحتاجه حالة الاستخدام أقوى ما يحدد البروتوكولات التي تركبها.

يعامل كل واحد من البروتوكولات الأربعة النزاعات والاستردادات بطريقته، وغالبا يحسم هذا الاختلاف اختيار البروتوكول لحالة استخدام. حالة استخدام تحتاج إلى حماية chargeback لا يمكن أن تعمل على x402 خالص. وحالة يكون فيها البائع بلا إعداد خدمة عملاء لا يمكن أن تستخدم ACP. هذه طريقة كل بروتوكول في التعامل مع النزاع، حتى تطابق البروتوكول مع نموذج النزاع الذي تحتاجه فعلا.

بروتوكول ACP: النزاعات عبر شبكة البطاقات. بما أن التاجر يبقى merchant of record في ACP، فإن كل مسار نزاع قياسي في شبكات البطاقات يعمل. يبدأ بنك المشتري chargeback؛ وتتولى Stripe (أو أي معالج) دفاع التاجر؛ ويتبع التاجر سياسة الاسترداد القائمة. هذه أكبر ميزة عملية ل ACP. كل توقعات مشتري التجزئة حول الإرجاع تعمل ببساطة.

لاسترداد ACP يبدأه الوكيل:

acp_client.refunds.create(...) توضيحي، مثل استدعاءات عميل ACP الأخرى في هذه الدورة. نموذج النتيجة وربط الأداة حقيقيان.

@function_tool
async def acp_refund(
    order_id: str,
    reason: str,
    amount_usd: Decimal | None = None,
) -> "RefundResult":
    """Start a refund through ACP. The merchant's standard refund policy applies."""
    raw = await acp_client.refunds.create(
        order_id=order_id,
        reason=reason,
        amount_usd=amount_usd,  # None means full refund
    )
    return RefundResult(
        refund_id=raw.refund_id,
        order_id=order_id,
        status=raw.status,
        amount_refunded_usd=raw.amount_refunded_usd,
    )

بروتوكول AP2: تسوى النزاعات عبر مسار التدقيق. مساهمة AP2 في النزاع هي سلسلة التفويضات (Intent ثم Cart ثم Payment). عندما يظهر نزاع، تكون تلك السلسلة دليلا على ما فوّضه المستخدم فعلا، وتملك قوة قانونية. لا تستبدل مسار النزاع في السكة الأساسية: إذا فوّض AP2 دفعة بطاقة عبر Stripe، فإن عملية نزاع Stripe تبقى مطبقة. يضيف AP2 إثبات ما وافق عليه المستخدم.

تدفق النزاع:

1. User claims: "I never authorized this purchase."
2. Merchant retrieves the signed AP2 Cart Mandate from the transaction record.
3. Merchant presents the Cart Mandate (with the user's signature) to the
   payment processor as part of the dispute defense.
4. The card network or processor checks the signature against the user's
   registered public key. If it is valid, the dispute is resolved for the merchant.

بروتوكول x402: لا آلية نزاع رسمية. مدفوعات x402 الخالصة غير قابلة للاسترداد بالتصميم. تستقر الدفعة على السلسلة خلال ثانية إلى ثانيتين؛ ولا chargeback. هذا أكبر حد عملي ل x402. يناسب استدعاء API بقيمة 0.001 دولار (النزاع يكلف أكثر من الدفعة) وخاطئ لأي شيء قد يريد فيه المشتري استردادا عادلا.

ثلاث طرق لتليين خاصية x402 غير القابلة للاسترداد:

• ضمان Escrow. للمدفوعات الأعلى قيمة في x402، استخدم smart-contract escrow يحفظ الأموال حتى يرسل المشتري إشارة القبول. يتضمن ERC-8004 بدائيات escrow للمعاملات متعددة الوكلاء.
• التركيب مع AP2 وسكة مختلفة. إذا احتجت إلى سرعة x402 ومعها دعم نزاعات، فإن تركيب AP2 مع x402 يعطيك سلسلة التفويضات كدليل ويبقي x402 التسوية سريعة. تبقى التسوية نفسها غير قابلة للعكس؛ يثبت التفويض فقط ما اتفق عليه.
• ضمانات البائع. في APIs المدفوعة، تتضمن شروط البائع غالبا قواعد استرداد تنفذ خارج السلسلة (يرسل البائع USDC طوعا إذا فشلت الخدمة). يعمل ذلك مع بائعين ذوي سمعة، وينهار مع مجهولين.

بروتوكول MPP: النزاعات عبر Stripe. ترث جلسات MPP المسواة على سكك البطاقات آليات النزاع القياسية في Stripe، كما في ACP. أما جلسات MPP المسواة بعملة مستقرة على Tempo أو عبر Lightning فتسير عبر تسوية النزاعات من جهة البائع لدى Stripe. تحمل Stripe التاجر مسؤولية النتيجة بغض النظر عن السكة.

غالبا يقود نموذج النزاع الذي تحتاجه التركيب أكثر من الكلفة أو الكمون. منصة تسوق مستهلك تحتاج chargebacks، لذلك يناسب ACP. سوق API خالص بين الآلات لا يحتاج نزاعات إطلاقا، لذلك يناسب x402. منصة مشتريات مؤسسة تحتاج دليلا بدرجة تدقيق، لذلك يناسب AP2 مع أي سكة تسوية.

المفهوم 18: سباكة FastAPI وInngest webhook لإغلاق حلقة الطلب والاستجابة

في سطر واحد: تصل بعض أحداث الدفع وفق جدولها الخاص (نزاعات، توقيعات تفويض، طلبات دفع من جهة البائع)، لذلك تحتاج إلى handler رقيق في FastAPI يلتقطها وحدث Inngest يحمل العمل إلى workflow متين.

تعاملت الأجزاء 1 إلى 5، والمفاهيم 15 إلى 17، مع الوكيل كمشتر: يرسل طلبا، يرد البروتوكول، ويفكر SDK في النتيجة. لكن تجارة الوكلاء تعمل في الاتجاهين. ترسل Stripe webhooks من نوع charge.dispute.created. ويحدث توقيع تفويض AP2 خارج خادمك، على جهاز المستخدم، ويرجع لاحقا. ويحتاج بائعو x402 إلى middleware من جهة الخادم يعيد 402 Payment Required ويفحص ترويسات X-PAYMENT. لا يناسب أي من ذلك استدعاء Runner.run() واحدا. تحتاج هذه الأحداث إلى handlers في FastAPI كحد HTTP، وأحداث Inngest كجسر راجع إلى workflows متينة.

يمشي هذا المفهوم عبر الأنماط الثلاثة التي ستحتاجها في الإنتاج. من دونها، ستكون في النظام شقوق تسقط منها الأحداث غير المتزامنة.

النمط 1: webhook من Stripe يتدفق إلى دالة Inngest معلقة. عندما يرفع مستخدم chargeback على طلب ACP، ترسل Stripe charge.dispute.created إلى نقطة نهايتك. قد يصل بعد خمس دقائق من الطلب أو بعد 60 يوما. workflow الذي وضع الطلب انتهى منذ زمن، لكن الوكيل ما زال يحتاج إلى الرد: إخطار التاجر، التسجيل في التدقيق، وربما بناء دفاع. يحول handler FastAPI ال webhook إلى حدث Inngest، وتلتقطه دالة Inngest وتشغل وكيلا للتعامل مع النزاع.

stripe.Webhook.construct_event(...) وstripe.error.SignatureVerificationError واجهات Stripe API حقيقية. ربط Inngest حقيقي. لاحظ API الإرسال: تخرج الأحداث ك send(events=[inngest.Event(...)])، لا dict عار.

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
from decimal import Decimal
import stripe, inngest

app = FastAPI()
inngest_client = inngest.Inngest(app_id="agent-commerce", is_production=False)

class StripeDisputeEventPayload(BaseModel):
    order_id: str
    dispute_id: str
    amount_usd: Decimal
    reason: str
    raw_event_id: str

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
    # 1. Verify the Stripe signature. This security gate is required.
    signature = request.headers.get("Stripe-Signature")
    payload = await request.body()
    try:
        event = stripe.Webhook.construct_event(
            payload=payload, sig_header=signature, secret=settings.stripe_webhook_secret,
        )
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    # 2. Route by event type. This handler does NO business logic.
    #    It only fires Inngest events so the durable workflow does the work.
    if event.type == "charge.dispute.created":
        await inngest_client.send(events=[
            inngest.Event(
                name="stripe/dispute.created",
                data=StripeDisputeEventPayload(
                    order_id=event.data.object.metadata.get("order_id"),
                    dispute_id=event.data.object.id,
                    amount_usd=Decimal(event.data.object.amount) / 100,
                    reason=event.data.object.reason,
                    raw_event_id=event.id,
                ).model_dump(),
                id=event.id,            # idempotency seed
            ),
        ])

    # 3. ACK Stripe right away. The real work runs in Inngest, durably.
    return {"received": True, "event_id": event.id}

# The Inngest function that handles the dispute: fully durable and retryable.
@inngest_client.create_function(
    fn_id="handle-stripe-dispute",
    trigger=inngest.TriggerEvent(event="stripe/dispute.created"),
    # Idempotency by raw_event_id makes sure Stripe retries do not process twice.
    idempotency="event.data.raw_event_id",
)
async def handle_stripe_dispute(ctx: inngest.Context) -> dict:
    payload = StripeDisputeEventPayload(**ctx.event.data)

    # Log to audit immediately.
    await ctx.step.run("audit-dispute-received", log_dispute_to_neon, payload)

    # Run an agent to assemble the dispute defense.
    defense_agent = Agent(
        name="DisputeDefenseAgent",
        instructions="Assemble dispute defense materials: order receipt, AP2 mandate if any, "
                     "delivery confirmation, customer communication history.",
        tools=[fetch_order_details, fetch_mandate_chain, fetch_delivery_proof, submit_dispute_response],
    )
    defense = await ctx.step.run(
        "build-and-submit-defense",
        Runner.run, defense_agent, f"Build defense for dispute {payload.dispute_id}",
    )
    return {"status": "completed", "output": {"defense_submitted": defense.final_output.model_dump()}}

يبقى handler FastAPI رقيقا (تحقق، ثم أطلق حدثا). وتكون دالة Inngest متينة (idempotency، وretries، وstep memoization). يحدث استدلال الوكيل داخل دالة Inngest، لا داخل handler ال webhook. يهم هذا الفصل لأن Stripe تتوقع استجابة 2xx خلال نحو خمس ثوان، وقد يستغرق تشغيل وكيل 30 ثانية أو أكثر.

النمط 2: callback لتوقيع AP2 يستأنف step.wait_for_event معلقا. أظهر المفهوم 9 أداة توقيع AP2 تطلب توقيعا وتنتظر. في الإنتاج يحدث ذلك التوقيع خارج خادمك: يفتح المستخدم تطبيق الهاتف، يرى التفويض، يضغط Approve، ويرسل التفويض الموقّع عائدا. كان workflow في Inngest معلقا على step.wait_for_event؛ يطلق callback في FastAPI الحدث الذي يوقظه.

note

إن callback التوقيع route خاص بك في FastAPI. يستخدم ارتباط Inngest if_exp= (تعبير CEL)، والحمولة المنتظرة تحت بادئة async.. يرجع wait_for_event قيمة None عند timeout. أما ap2_verify_signature وعميل التخزين فهما توضيحيان؛ شكل workflow حقيقي.

from datetime import timedelta
from pydantic import BaseModel

class MandateSignedPayload(BaseModel):
    mandate_id: str
    user_did: str
    signature: str  # the user's signature over the mandate hash
    signed_at: str

@app.post("/callbacks/ap2/mandate-signed")
async def mandate_signed_callback(payload: MandateSignedPayload, request: Request):
    # 1. Verify the signature against this user's registered public key.
    is_valid = await ap2_verify_signature(
        mandate_id=payload.mandate_id,
        user_did=payload.user_did,
        signature=payload.signature,
    )
    if not is_valid:
        raise HTTPException(status_code=400, detail="Invalid mandate signature")

    # 2. Persist the signed mandate. Mandates have a 7-year retention requirement.
    await neon_client.mandates.insert({
        "mandate_id": payload.mandate_id,
        "user_did": payload.user_did,
        "signature": payload.signature,
        "signed_at": payload.signed_at,
        "status": "signed",
    })

    # 3. Fire the Inngest event that resumes the agent's workflow.
    await inngest_client.send(events=[
        inngest.Event(name="ap2/mandate.signed", data=payload.model_dump(), id=payload.mandate_id),
    ])
    return {"received": True, "event_id": payload.mandate_id}

# The Inngest function that was waiting. if_exp correlates the wait to this mandate.
@inngest_client.create_function(
    fn_id="agent-procurement-workflow",
    trigger=inngest.TriggerEvent(event="procurement/task.created"),
)
async def procurement_workflow(ctx: inngest.Context) -> dict:
    # ... agent creates the Intent Mandate, fires "ap2/mandate.signing.requested" ...

    # Suspend until the user signs. Zero compute is used during the wait.
    signed = await ctx.step.wait_for_event(
        "wait-for-intent-mandate-signature",
        event="ap2/mandate.signed",
        if_exp=f"async.data.mandate_id == '{ctx.event.data['mandate_id']}'",
        timeout=timedelta(hours=24),   # users can take real time
    )

    if signed is None:                 # timeout returns None
        return {"status": "abandoned", "reason": "user did not sign within 24h"}

    # Resume with the signed mandate. The agent continues from exactly where it left off.
    cont = await ctx.step.run("continue-procurement", continue_with_signed_mandate, signed)
    return {"status": "completed", "output": {"procurement_continuation": cont}}

بني step.wait_for_event في Inngest مع if_exp لهذا بالضبط. handler FastAPI جسر باتجاه واحد من HTTP إلى bus أحداث Inngest. يستأنف workflow الذي علق قبل ساعات مع الحمولة الموقعة، ويواصل الوكيل من حيث توقف.

النمط 3: middleware من جهة بائع x402 (عندما تعرض API مدفوعة، لا تستهلك واحدة فقط). في سوق متعدد الوكلاء (القرار 4)، يكون وكيلك أحيانا مشتريا وأحيانا بائعا، حيث تدفع له وكلاء أخرى مقابل البحث أو التحليل أو الكود. تحتاج جهة البائع إلى middleware في FastAPI يعيد 402 Payment Required، ويفحص ترويسات X-PAYMENT، ولا يقدم المورد إلا بعد أن يتحقق facilitator من الدفع.

note

X402Middleware من جهة البائع أدناه توضيحي؛ لا توجد حزمة PyPI اسمها x402_server. يأتي x402 الحقيقي من جهة الخادم عبر حزمة x402 ومساعدات الخادم وال facilitator مع تكامل framework. فكرة التماثل بين المشتري والبائع حقيقية، وroute FastAPI حقيقي.

from fastapi import FastAPI, Request, Response
from decimal import Decimal
# Illustrative seller-side imports (see note above).
from x402_server import X402Middleware, PaymentRequirement

# Configure the middleware once at app startup.
app = FastAPI()
app.add_middleware(
    X402Middleware,
    payment_requirements_by_route={
        "/api/research": PaymentRequirement(
            scheme="exact",
            network="eip155:8453",  # Base
            asset=USDC_BASE_CONTRACT,
            recipient=settings.merchant_wallet_address,
            max_amount_usdc=Decimal("0.50"),
            expiry_seconds=300,
        ),
        "/api/code-review": PaymentRequirement(
            scheme="exact",
            network="eip155:8453",
            asset=USDC_BASE_CONTRACT,
            recipient=settings.merchant_wallet_address,
            max_amount_usdc=Decimal("2.00"),
            expiry_seconds=300,
        ),
    },
    facilitator_url="https://facilitator.cloudflare.com/x402",
)

# Your business logic. The middleware enforces payment before this runs.
@app.post("/api/research")
async def research_endpoint(request: Request):
    # By the time we reach here, the X-PAYMENT header has been verified and settled.
    # request.state.x402_proof carries the on-chain transaction hash for audit.
    query = (await request.json())["query"]

    research_agent = Agent(
        name="ResearchAgent",
        instructions="Conduct deep research on the query and return a structured report.",
        tools=[search_web, fetch_papers, summarize],
    )
    result = await Runner.run(research_agent, query)

    return Response(
        content=result.final_output,
        headers={"X-PAYMENT-PROOF": request.state.x402_proof.tx_hash},
    )

إن x402 متناظر. كود جهة المشتري من المفهوم 10 وmiddleware جهة البائع هنا هما نصفا البروتوكول نفسه. يشغل السوق متعدد الوكلاء الاثنين: تشتري وكلاؤه من خدمات خارجية وتبيع لوكلاء خارجية.

تتكرر ثلاثة إخفاقات في الإنتاج:

1. تنفيذ handlers ال webhook لمنطق العمل inline. handler في FastAPI يشغل الوكيل داخل استجابة webhook سيتجاوز timeout ذي الخمس ثوان لدى Stripe. تعيد Stripe المحاولة، فيعمل الوكيل مرتين، وتخصم من المستخدم مرتين. يبقى handler رقيقا؛ وتكون دالة Inngest متينة.
2. نسيان idempotency في webhooks. تعيد Stripe تسليمات فاشلة بالمعرف نفسه event.id. من دون مفتاح idempotency على دالة Inngest، تنشئ كل إعادة محاولة نسخة مكررة. استخدم "event.data.raw_event_id" كمفتاح idempotency.
3. لا فحص توقيع على callbacks. يجب أن تتحقق callbacks الخاصة ب AP2 mandate-signed من توقيع المستخدم مقابل المفتاح العام المسجل. وإلا يستطيع أي caller تزوير أحداث mandate-signed. الفحص الفاشل يرجع 400، لا تحذيرا مسجلا.

---

الجزء 7: الخاتمة، ما كانت الدورة تعلمه فعلا

المفهوم 19: انضباط التركيب الطبقي

في سطر واحد: يتلخص كل شيء في هذه الدورة في مهمة واحدة: اقرأ حالة الاستخدام، قسمها إلى أربع طبقات، واختر البروتوكول الصحيح في كل طبقة.

تحتوي هذه الدورة على 19 مفهوما و5 قرارات. كلها scaffolding لادعاء واحد: تجارة الوكلاء في 2026 ليست بروتوكولا واحدا بل معمارية طبقية، ومهمتك أن تختار البروتوكول الصحيح في كل طبقة لحالة الاستخدام أمامك. يتبع كل شيء آخر ذلك.

يظهر الشكل نفسه على ثلاثة مقاييس.

على مقياس البروتوكول، تحل البروتوكولات الأربعة الرئيسية مشكلات مختلفة في طبقات مختلفة: ACP في الطبقة 3، وAP2 في الطبقة 2، وx402 وMPP في الطبقة 4. معاملتها كمتنافسين في الطبقة نفسها هو الخطأ المعماري الأكثر شيوعا. لا تتنافس إلا حيث تتداخل طبقاتها (x402 ضد MPP في التسوية؛ وAP2 ضد SPT في ACP للتفويض في بعض التدفقات). وفي كل مكان آخر تتركب.

على مقياس النظام، يملك نظام الإنتاج بروتوكولا واحدا من كل طبقة، موصولا عبر OpenAI Agents SDK بوصفه العميل الموحد. تتطابق @function_tool وRunContextWrapper وtool_input_guardrail في SDK بنظافة مع قضايا كل طبقة. SDK ليس بروتوكولا. إنه المنسق الذي يسمح لك بتركيب البروتوكولات بنظافة.

على مقياس الانضباط، عملك هو قراءة حالة استخدام، وتقسيمها إلى الطبقات الأربع، واختيار البروتوكول الصحيح في كل واحدة، وتبرير كل اختيار مقابل القيود الحقيقية للحالة: قيمة المعاملة، وميزانية الكمون، ونموذج النزاع، ومتطلبات التدقيق. العمل ليس "اختر بروتوكولا مفضلا." بل "في حالة الاستخدام هذه، ماذا تطلب كل طبقة؟"

مشت القرارات الخمسة في الجزء 5 عبر حالات استخدام حقيقية بهذا الانضباط. وصل القرار 1 (تسوق المستهلك) إلى ACP مع سكك بطاقات Stripe لأن حالة الاستخدام احتاجت حماية chargeback. ووصل القرار 2 (وكيل يدفع ل API) إلى x402 وحده، لأن الطبقتين 2 و4 تنطبقان في تدفقات الآلة إلى الآلة. ووصل القرار 3 (مشتريات المؤسسة) إلى أعقد تركيب (AP2 مع ACP مع MPP)، لأن حاجتي التدقيق والتكرار فرضتا ذلك. ووصل القرار 4 (سوق متعدد الوكلاء) إلى AP2 مع ERC-8004 ومع x402، لأن حاجة الثقة الثنائية لا يمكن تلبيتها بطريقة أخرى. وأعاد القرار 5 بناء القرار 2 على مكدس مختلف تماما (Google ADK مع محفظة Coinbase مع AP2 مع x402)، ووصل إلى الشكل ذي الطبقات الأربع نفسه، مثبتا أن المعمارية ليست غلافا حول Stripe وOpenAI.

التركيب الذي تختاره تحدده حالة الاستخدام، لا الذوق. فريق يختار x402 لأن "العملات المستقرة هي المستقبل" لكنه يبني تجربة تسوق للمستهلك يملك التركيب الخطأ. وفريق يختار ACP لأن "Stripe enterprise-grade" لكنه يدفع لوصول API بسعر 0.001 دولار لكل استدعاء يملك التركيب الخطأ. دع حالة الاستخدام تقود الاختيار.

---

ورقة مختصرة: الإطار في صفحة واحدة

اطبعها. علّقها على الحائط. استخدمها عند مراجعة أي تصميم لتجارة الوكلاء.

الطبقات الأربع (احفظ هذا المكدس)

Layer 1: DISCOVERY      ->  "What's available to buy?"
Layer 2: AUTHORIZATION  ->  "Am I allowed to spend this?"
Layer 3: COMMERCE       ->  "What's the full purchase lifecycle?"
Layer 4: SETTLEMENT     ->  "Where does the money actually move?"

البروتوكولات في كل طبقة (أفضل الاختيارات في 2026)

الطبقة	أفضل الاختيارات	اختر بناء على
الاكتشاف	MCP، وA2A، وأدلة الوكلاء، وواجهات التسوق بالذكاء الاصطناعي	أين تعيش الخدمات التي يحتاجها الوكيل
التفويض	تفويضات AP2، وACP SPT، وTAP، وERC-8004	نموذج الثقة: تدقيق صارم، أصلي ل Stripe، هوية فقط، أو متعدد الوكلاء
التجارة	ACP، وUCP، وAPI مباشر (لا شيء)	هل تحتاج حالة الاستخدام إلى دورة حياة تجارة؟
التسوية	x402، وMPP، وسكك البطاقات، والبنك/Lightning	الاقتصاديات: قيمة المعاملة تقود السكة

التركيبات القانونية الأربعة

حالة الاستخدام	المكدس
تسوق المستهلك	واجهة AI + ACP SPT + ACP + بطاقات Stripe
وكيل يدفع ل API	MCP/دليل + EIP-3009 + (لا شيء) + x402
مشتريات مؤسسة	A2A/MCP + AP2 + ACP/UCP + MPP/cards
سوق متعدد الوكلاء	A2A + AP2 + ERC-8004 + (لا شيء) + x402

فرض حدود الإنفاق على ثلاثة مستويات (مطلوب)

Level 1: Wallet / payment-method limits  (smart-contract caps OR Stripe customer caps)
Level 2: SDK tool guardrails             (tool_input_guardrail on each payment tool)
Level 3: Application business logic      (per-user, per-category, per-merchant policies)

إذا تجاوزت أيا من هذه، فأنت على بعد bug واحد من استنزاف كامل. الخطأ الأكثر شيوعا هو استخدام output_guardrail على مستوى الوكيل للمستوى 2. يعمل على الرد النهائي للوكيل، متأخرا جدا لإيقاف الدفع. استخدم tool_input_guardrail بدلا منه.

العتبة الاقتصادية (احفظها)

Transaction value
├── < $5         → x402 or MPP stablecoin (card fees exceed the transaction)
├── $5 - $1,000  → ACP + card rails (chargeback protection worth the 2.9%)
└── > $1,000     → AP2 + composed stack (audit + dispute defense + multi-rail)

نموذج النزاع (غالبا أقوى قيد)

Use case needs chargeback protection?
├── Yes → ACP + card rails (or MPP card mode)
└── No  → x402 acceptable (faster, cheaper, no refunds)

Use case needs audit evidence?
├── Yes → AP2 at Layer 2 (mandates as legally-admissible evidence)
└── No  → SPT or EIP-3009 sufficient

قائمة تحقق الإنتاج قبل الإطلاق

• إعداد حدود إنفاق المحفظة/طريقة الدفع في المستوى 1
• ربط tool_input_guardrail بكل أداة تفوض دفعا (المستوى 2)
• يفرض التطبيق سقوفا لكل مستخدم، ولكل فئة، ولكل تاجر (المستوى 3)
• مفاتيح التوقيع في key vault (Azure Key Vault أو ما يعادله)، لا في env vars
• فصل المحفظة لكل وكيل (لا مفاتيح مشتركة عبر فئات الوكلاء)
• تعريف جدول تدوير المفاتيح (خط أساس 90 يوما)
• سجلات التدقيق إلى تخزين متين مستقل عن runtime الوكيل
• تمتد traces من OpenTelemetry عبر SDK وhttpx وInngest وFastAPI؛ trace ID واحد لكل معاملة
• نماذج Pydantic عند كل حد (إرجاع الأدوات، وحمولات البروتوكولات، وأجسام FastAPI، وأحداث Inngest)
• Decimal للمال في كل مكان (لا float أبدا)
• يتحقق handler webhook من Stripe من التوقيع ويطلق حدث Inngest (handler رقيق، دالة متينة)
• يتحقق callback تفويض AP2 الموقّع من التوقيع ويستأنف step.wait_for_event
• إذا كنت تعرض APIs مدفوعة: اضبط x402 seller-side middleware لكل route
• مفتاح idempotency في كل دالة Inngest يحركها webhook
• فهم وتوثيق آلية النزاع والاسترداد لكل بروتوكول
• بوابة تأكيد human-in-the-loop لأول 30 يوما في الإنتاج
• قياس دقة السلة قبل تخفيف بوابة التأكيد

---

مرجع سريع: شجرة القرار مضغوطة

عندما تكون لديك حالة استخدام جديدة، امش هذه الشجرة.

1. What's the agent buying?
   ├── Retail goods for a user → Decision 1 pattern (ACP + cards)
   ├── API access for itself → Decision 2 pattern (x402-only)
   ├── Supplier goods/services for an org → Decision 3 pattern (AP2 + composed)
   └── Work from another agent → Decision 4 pattern (AP2 + ERC-8004 + x402)

2. What's the transaction value?
   ├── < $5 → settlement = x402 or MPP stablecoin
   ├── $5-$1,000 → settlement = card rails via ACP/MPP
   └── > $1,000 → settlement = MPP sessions OR bank rails

3. What's the dispute model?
   ├── Need chargeback protection → must include card rails somewhere
   ├── Need audit evidence → must include AP2 at Layer 2
   └── Neither → x402 / direct rail sufficient

4. What's the latency budget?
   ├── < 1 sec → only stablecoin rails work
   ├── 1-5 sec → x402, MPP, or pre-signed AP2 mandates
   └── > 5 sec → full ACP checkout works

5. Compose: pick one protocol from each layer, justified against the constraints above.

---

قالب مراجعة التصميم: أسئلة تطرحها عند مراجعة أي معمارية لتجارة الوكلاء

عندما تراجع تصميم زميل (أو مسودتك أنت بعد يوم من الابتعاد)، امش هذه الأسئلة بالترتيب. يرتبط كل سؤال بقسم من هذه الدورة؛ إذا لم ترض الإجابة السؤال، فارجع إلى ذلك القسم.

أسئلة المعمارية

1. ما حالة الاستخدام التي يخدمها هذا؟ إذا كانت "عدة حالات"، فما الأساسية؟ التصميم الذي يحاول خدمة الحالات القانونية الأربع كلها هو anti-pattern من المفهوم 12؛ علّمه.
2. امش الطبقات الأربع صراحة. الاكتشاف؟ التفويض؟ التجارة؟ التسوية؟ أي بروتوكول في كل طبقة؟ اجعل تركيب البروتوكولات الأربعة يقال بصوت واضح.
3. برر اختيار كل طبقة مقابل حالة الاستخدام. لماذا هذا البروتوكول في هذه الطبقة؟ ما الذي كان سيتغير مع بروتوكول آخر؟ ينطبق اختبار عبر/داخل من المفهوم 13 هنا.
4. هل يوجد تداخل بروتوكولات في أي طبقة؟ بروتوكولان يتنافسان في الطبقة 4؟ اثنان في الطبقة 2؟ إذا نعم، فهذه تعقيد يحتاج إلى حالة استخدام تبرره.

الأسئلة الاقتصادية

5. ما توزيع قيمة المعاملات؟ دون الدولار؟ من 10 إلى 100 دولار؟ أكثر من 1,000 دولار؟ يجب أن يقود عتبة المفهوم 14 الاقتصادية اختيار التسوية.
6. ما ميزانية الكمون؟ دون الثانية؟ 5 إلى 30 ثانية؟ يجب أن تقيد ميزانية الكمون اختيارات التسوية والتفويض.
7. ما الكلفة لكل معاملة عند الحجم المتوقع؟ اجمع رسوم البروتوكول مع رسوم المعالج مع كلفة البنية لكل معاملة. استخدمها لفحص أن التركيب مجد اقتصاديا.

أسئلة التشغيل

8. هل فرض حدود الإنفاق موجود على المستويات الثلاثة؟ المحفظة/طريقة الدفع، SDK، منطق عمل التطبيق. المفهوم 15: تجاوز أي واحد يعرّضك.
9. هل نظافة الهوية موجودة؟ فصل المحفظة لكل وكيل، تدوير المفاتيح، سجلات التدقيق إلى تخزين متين، traces موزعة ب trace ID واحد لكل معاملة. عادات المفهوم 16 الأربع.
10. آليات النزاع والاسترداد؟ هل يتعامل التركيب مع النزاعات التي ستحصل في حالة الاستخدام فعلا؟ المفهوم 17: غالبا هذا أقوى قيد.
11. الغلاف التشغيلي؟ أين يتولى Inngest (أو تنفيذ متين مكافئ) التدفقات طويلة التشغيل وretries وidempotency؟ مغطى في الدورة المكثفة في عامل الإنتاج.
12. بوابات human-in-the-loop؟ أين يؤكد المستخدم أو المشغل داخل التدفق؟ ابدأ ببوابة تأكيد لأول 30 يوما في الإنتاج.

أسئلة webhooks وcallbacks غير المتزامنة (المفهوم 18)

13. هل handler webhook من Stripe موجود ورقيق؟ يتحقق من التوقيع، ويطلق حدث Inngest، ويرد ACK في أقل من خمس ثوان. يعيش منطق العمل في دالة Inngest، لا في handler.
14. هل callback توقيع تفويض AP2 موجود ويتحقق من توقيعات المستخدم؟ من دونه لا يستأنف step.wait_for_event لدى الوكيل أبدا. ومن دون التحقق من التوقيع يستطيع أي شخص تزوير أحداث mandate-signed.
15. إذا كنت تعرض APIs مدفوعة: هل x402 seller-side middleware مضبوط لكل route؟ تشغل الأسواق متعددة الوكلاء كود جهة المشتري وجهة البائع معا؛ وتحتاج جهة البائع إلى middleware.
16. هل مفاتيح idempotency في Inngest موجودة على كل دالة يحركها webhook؟ تعيد Stripe المحاولة ب event.id نفسه؛ ومن دون idempotency تخصم أو تسترد مرتين.
17. طبقة العقد: هل توجد نماذج Pydantic عند كل حد؟ إرجاع الأدوات، وحمولات البروتوكولات، وأجسام طلب واستجابة FastAPI، وحمولات أحداث Inngest، كلها ذات أنواع. Decimal للمال، لا float أبدا.

أسئلة أنماط الفشل

18. ما أكثر نمط فشل احتمالا في الإنتاج؟ كان لكل قرار قانوني واحد: عدم تطابق السلة (القرار 1)، وإنفاق منفلت (القرار 2)، وعدم تطابق نطاق Intent Mandate (القرار 3)، وسمعة متلاعب بها (القرار 4). ما نمطك؟
19. ما التخفيف؟ كان لكل نمط فشل في الجزء 5 تخفيف صريح. هل يتضمنه التصميم، أم يعتمد على "البروتوكول سيتعامل معه"؟

أسئلة جاهزية المسار (لنشر الإنتاج)

20. من أي مسار تعلم يعمل الفريق؟ قارئ، مبتدئ، متوسط، متقدم. كن صريحا. يحتاج نشر الإنتاج إلى عمق المسار المتقدم، لا معرفة مسار القارئ.
21. ما خطة rollback إذا أساء الوكيل التصرف؟ kill switch للمحفظة؟ إلغاء SPT؟ تعطيل الوكيل على مستوى SDK؟ عرّف ذلك قبل الإطلاق، لا بعد الحادث.

---

المراجع

مصادر أولية للبروتوكولات الأربعة. فضّلها على التعليقات الثانوية؛ فالمواصفات تتطور أسرع من الكتابات عنها.

بروتوكول ACP (Agentic Commerce Protocol)

• مستودع المواصفة: github.com/agentic-commerce-protocol/agentic-commerce-protocol (Apache 2.0)
• موقع المطورين: agenticcommerce.dev
• القائمان على الصيانة: OpenAI وStripe
• وثائق تكامل Stripe: stripe.com/docs/agentic-commerce
• آخر نسخة مواصفة مذكورة: 2026-04-17

بروتوكول AP2 (Agent Payments Protocol)

• مستودع المواصفة: github.com/google-agentic-commerce/AP2 (Apache 2.0)
• موقع الوثائق: ap2-protocol.org
• أعضاء التحالف: Google مع أكثر من 60 شريكا (Salesforce وServiceNow وAdobe وShopee وEtsy وAdyen وAmerican Express وJCB وUnionPay International وPayPal وMastercard وCoinbase وغيرهم)
• مستودع امتداد a2a-x402: github.com/google-a2a/a2a-x402
• آخر نسخة مذكورة: v0.2.0 (أبريل 2026)

بروتوكول x402

• مستودع المواصفة: github.com/coinbase/x402 (Apache 2.0)
• موقع الوثائق: x402.gitbook.io (وأيضا x402.org)
• أنشأته Coinbase وتديره الآن x402 Foundation تحت Linux Foundation (أبريل 2026)، مع Cloudflare وStripe وAWS وGoogle وغيرهم.
• تكامل Cloudflare OpenAI Agents SDK: developers.cloudflare.com/agents/x402
• دليل التبني: agent.market
• من أعضاء x402 Foundation: Adyen وAWS وAmerican Express وBase وCircle وCloudflare وCoinbase وGoogle وMastercard وMicrosoft وPolygon Labs وShopify وSolana Foundation وStripe وVisa

بروتوكول MPP (Machine Payments Protocol)

• موقع المواصفة: mpp.dev
• المطوران المشاركان: Stripe وTempo (blockchain L1 من Stripe، محضونة مع Paradigm)
• إعلان Stripe: stripe.com/blog/machine-payments-protocol
• تاريخ الإطلاق: 18 مارس 2026 (mainnet)
• الشركاء عند الإطلاق: Stripe وVisa وLightspark وAnthropic وOpenAI وShopify وأكثر من 100 خدمة

بروتوكولات مجاورة

• بروتوكول A2A (Agent2Agent، Google): github.com/google-a2a/A2A، البروتوكول الذي يوسعه AP2
• معيار MCP (Model Context Protocol، Anthropic): modelcontextprotocol.io، طبقة الأدوات والسياق التي تكتشف الوكلاء الخدمات من خلالها
• بروتوكول UCP (Universal Commerce Protocol، Google): أعلن عنه كنظير ACP لواجهات تسوق Google
• بروتوكول TAP (Trusted Agent Protocol، Visa وCloudflare): بروتوكول تحقق الهوية أطلق في 14 أكتوبر 2025
• معيار ERC-8004: معيار ثقة على السلسلة للمعاملات متعددة الوكلاء

عدة OpenAI Agents SDK

• حزمة Python SDK: pip install openai-agents (آخر إصدار 19 مايو 2026)
• الوثائق: openai.github.io/openai-agents-python
• حزمة JavaScript/TypeScript SDK: github.com/openai/openai-agents-js

دورات Agent Factory ذات الصلة

• الدورة المكثفة في بناء وكلاء الذكاء الاصطناعي: أساسيات OpenAI Agents SDK (Agent وRunner.run() و@function_tool)
• الدورة المكثفة في عامل الإنتاج: Inngest بوصفه الغلاف التشغيلي لتدفقات وكلاء متينة
• الدورة المكثفة في التطوير المدفوع بالتقييم: اختبار الوكلاء وتقييمها
• الدورة المكثفة في اختيار المعماريات الوكيلة: أي نمط وكيلي يناسب أي حالة استخدام
• الدورة المكثفة في نشر الوكلاء (cloud deployment): غلاف السحابة (Azure Container Apps مع Neon وCloudflare). قيد العمل؛ أضف الرابط عند إطلاقها.

أدوات التشغيل وطبقة العقد (تستخدم بكثافة في المفهومين 16 و18)

• منصة Inngest: inngest.com، منصة التنفيذ المتين التي تشغل workflows الوكلاء في هذه الدورة
• إطار FastAPI: fastapi.tiangolo.com، طبقة HTTP غير المتزامنة حيث تعيش نقاط نهاية webhook وmiddleware x402 من جهة البائع
• مكتبة Pydantic: pydantic.dev، عقد النوع عند كل حد (إرجاع الأدوات، وحمولات البروتوكولات، وأجسام FastAPI، وأحداث Inngest)
• نظام OpenTelemetry: opentelemetry.io، tracing موزع مع auto-instrumentation ل httpx وopenai-agents وFastAPI وInngest؛ trace ID واحد يغطي المعاملة كلها
• عميل httpx: python-httpx.org، عميل HTTP غير المتزامن تحت x402-client وعميل ACP وامتداد AP2 a2a-x402
• خدمة Azure Key Vault: حيث تعيش مفاتيح التوقيع في الإنتاج، وتدور وفق انضباط المفهوم 16
• وثائق Stripe webhooks: stripe.com/docs/webhooks، التحقق من التوقيع، وأنواع الأحداث، ودلالات retry

---

أصبحت تملك الإطار كله: الطبقات الأربع، والبروتوكولات الأربعة، وقواعد تركيبها، وخمسة قرارات عملية، ومخاوف الإنتاج التي تحدد هل يصمد النظام، ومرجعا من صفحة واحدة للاحتفاظ به. ستواصل البروتوكولات التحرك. أما انضباط الطبقات الأربع فلن يتحرك. ابن من الطبقات، وستبقى على صواب حتى عندما تتغير أسماء البروتوكولات.