Pydantic AI भुगतान एकीकरण।
कोई Pydantic AI पैकेज नहीं है, और आपको इसकी आवश्यकता नहीं है। @agent.tool_plain के साथ एक टाइप की गई फ़ंक्शन पंजीकृत करें, असली blockchain0x क्लाइंट को कॉल करें, और आपका एजेंट Base पर USDC को स्थानांतरित कर सकता है।
कोई Pydantic AI-specific package नहीं है, और आपको इसकी आवश्यकता नहीं है। Pydantic AI एक function को के साथ validated tool में बदल देता है, इसलिए आप वास्तविक Python client को एक typed function में wrap करते हैं और agent USDC भेज सकता है, invoices settle कर सकता है, और wallets पढ़ सकता है। यह OpenAI, Anthropic, Google, Mistral, Groq, और हर अन्य Pydantic AI provider के साथ काम करता है, क्योंकि tool model को नहीं, बल्कि HTTP API को call करता है। Payments Base पर settle होते हैं।
Pydantic AI की टाइपिंग जीतती है। इसका उपयोग भुगतान के लिए भी करें।
Pydantic AI का पूरा बिंदु मजबूत टाइपिंग है: एजेंट इनपुट टाइप किए गए हैं, आउटपुट टाइप किए गए हैं, निर्भरताएँ टाइप की गई हैं, पुनः प्रयास टाइप किए गए हैं। ढांचा सीमा पर गलत डेटा को अस्वीकार करता है बजाय इसके कि इसे LLM कॉल के माध्यम से फैलने दिया जाए। आपका वॉलेट उपकरण ठीक इसी पर निर्भर करता है - तर्कों को प्रकारों के साथ घोषित करें, और Pydantic AI हस्ताक्षर और डॉकस्ट्रिंग से एक मान्य स्कीमा बनाता है इससे पहले कि मॉडल इसे कॉल करने की अनुमति दी जाए।
व्यावहारिक लाभ: यदि मॉडल send_usdc को एक फ्लोट के साथ कॉल करने की कोशिश करता है बजाय एक मूल-इकाई स्ट्रिंग के, तो विफलता Pydantic के सत्यापन परत पर होती है, किसी भी HTTP कॉल से पहले। आप एक स्पष्ट ValidationError देखते हैं जिसमें एक उपयोगी संदेश होता है बजाय API से 422 के जिसमें रहस्यमय फ़ील्ड पथ होते हैं। यही कारण है कि आपको यहाँ स्वयं क्लाइंट को लपेटने की आवश्यकता है बजाय एक सामान्य एडेप्टर के लिए पहुंचने के - आप वायर तक टाइपिंग बनाए रखते हैं।
Pydantic AI और कोर SDK इंस्टॉल करें। दो कुंजी।
कोई blockchain0x Pydantic AI पैकेज जोड़ने के लिए नहीं है। आप Pydantic AI (Python 3.10+) और असली blockchain0x कोर SDK स्थापित करते हैं, फिर नीचे दी गई फ़ंक्शन लिखते हैं। यही पूरी निर्भरता सूची है।
pip install pydantic-ai blockchain0xexport OPENAI_API_KEY=sk-... export BLOCKCHAIN0X_API_KEY=sk_test_... # sk_test_ = Base Sepolia, sk_live_ = Base mainnet
OPENAI_API_KEY (या आपके द्वारा उपयोग किए जाने वाले किसी भी provider का समकक्ष). BLOCKCHAIN0X_API_KEY आपके dashboard से एक sk_test_ testnet या sk_live_ mainnet key है; client इसे environment से पढ़ता है. अगर आपका agent पैसा भी प्राप्त करता है, तो webhook handler को अतिरिक्त रूप से BLOCKCHAIN0X_WEBHOOK_SECRET चाहिए.
एक टाइप की गई फ़ंक्शन जो भुगतान करती है, एक उपकरण के रूप में पंजीकृत।
नीचे पूरी एकीकरण है। send_usdc वास्तविक blockchain0x क्लाइंट को कॉल करता है; @agent.tool_plain इसे प्रकार संकेत और डॉकस्ट्रिंग से निर्मित स्कीमा के साथ पंजीकृत करता है। इसे चलाएँ और एजेंट बेस पर USDC को स्थानांतरित करता है, तर्कों को कॉल से पहले मान्य किया जाता है।
from pydantic_ai import Agent from blockchain0x import Client blockchain0x = Client() # reads BLOCKCHAIN0X_API_KEY from the environment agent = Agent( "openai:gpt-4o", system_prompt="You pay vendor invoices in USDC within owner-set limits.", ) # Register a plain function as a tool. No dedicated package needed. @agent.tool_plain def send_usdc(agent_id: str, to: str, amount_wei: str) -> str: """Send a USDC payment from an agent wallet. amount_wei is USDC base units (6 decimals), so "10000" is 0.01 USDC. """ return str( blockchain0x.payments.create(body={"agentId": agent_id, "to": to, "amountWei": amount_wei}) ) result = agent.run_sync( "Pay 0.01 USDC from agent agt_123 to 0xVendor for the dataset." ) print(result.output)
When the agent decides to pay, it calls send_usdc, the SDK submits the transfer, and you get a transaction hash back. amount_wei is base units, so 0.01 USDC is "10000". A sk_test_ key keeps it on Base Sepolia until you switch to sk_live_. Want a typed result object instead of a string? Set the agent's output_type to your own Pydantic model and return it from the tool. Add read and settle functions the same way.
एक हस्ताक्षरित वेबहुक के साथ इनबाउंड भुगतानों की पुष्टि करें।
यदि आपका एजेंट भी USDC प्राप्त करता है, तो इसे वेबहुक के साथ पुष्टि करें न कि पोलिंग द्वारा। सत्यापन सहायक Node SDK में भेजा जाता है; एक Python सेवा में आप दस्तावेजित HMAC के खिलाफ हाथ से सत्यापित करते हैं। यदि आप टाइप किए गए घटनाएँ चाहते हैं, तो अपना खुद का Pydantic मॉडल परिभाषित करें और हस्ताक्षर की जांच के बाद कच्चे शरीर को model_validate_json करें - टाइपिंग जोड़ने के लिए आपकी है। FastAPI उदाहरण नीचे।
import hmac, hashlib, os, time from fastapi import FastAPI, Request, HTTPException app = FastAPI() SECRET = os.environ["BLOCKCHAIN0X_WEBHOOK_SECRET"].encode() @app.post("/webhooks/payment") async def receive(request: Request): raw = await request.body() # RAW bytes - do not parse first sig = request.headers.get("X-Blockchain0x-Signature", "") ts = request.headers.get("X-Blockchain0x-Timestamp", "") parts = dict(p.split("=", 1) for p in sig.split(",") if "=" in p) t, v1 = parts.get("t", ts), parts.get("v1", sig) want = hmac.new(SECRET, t.encode() + b"." + raw, hashlib.sha256).hexdigest() if not hmac.compare_digest(want, v1) or abs(time.time() - int(t)) > 300: raise HTTPException(status_code=401) if request.headers.get("X-Blockchain0x-Event-Type") == "payment.received": await trigger_followup() # USDC landed - run the next step return {"ok": True}
एल्गोरिदम t.rawBody पर HMAC-SHA256 है, एक स्थायी-समय तुलना, और एक 300-सेकंड पुनःप्रसारण विंडो। कच्चे शरीर को await request.body() के माध्यम से पढ़ें, कभी भी request.json() को फिर से serialize न करें, क्योंकि यह सिग्नेचर द्वारा कवर किए गए बाइट्स को बदलता है। शिप किए गए इवेंट्स हैं payment.received, payment.sent, wallet.deployed, और webhook.test; शाखा के लिए X-Blockchain0x-Event-Type हेडर पर संकीर्ण करें।
आप जिस क्लाइंट को लपेट रहे हैं वह खुला है। इसे पढ़ें।
clone करने के लिए कोई Pydantic AI starter package नहीं है - ऊपर दिया गया recipe ही integration है। blockchain0x SDKs GitHub पर open source हैं; यह recipe Python SDK (blockchain0x-python) को wrap करता है, और पूरा method surface docs में है। function bodies के reference के लिए इसे पढ़ें।
github.com/tosh-labs/blockchain0x-pythonपूरी SDK method surface और scopes the docs पर documented हैं. Base Sepolia के against एक sk_test_ key से शुरुआत करें, फिर जब function अपेक्षा के अनुसार काम करे तो sk_live_ पर switch करें.
पांच Pydantic AI-विशिष्ट जाल से बचें।
Pydantic AI की मजबूत टाइपिंग अधिकांश एकीकरण बग को सीमा पर पकड़ लेती है; ये कुछ हैं जो फिसल जाते हैं।
कोई Pydantic AI पैकेज नहीं है - आप एक उपकरण पंजीकृत करते हैं
Blockchain0x LangChain और CrewAI के लिए एडाप्टर और MCP सर्वर भेजता है; कोई समर्पित Pydantic AI पैकेज नहीं है। ऊपर दिया गया नुस्खा यह है: एक साधारण प्रकारित फ़ंक्शन जिसे @agent.tool_plain के साथ सजाया गया है जो असली blockchain0x क्लाइंट को कॉल करता है। Pydantic AI हस्ताक्षर और docstring को पढ़ता है ताकि एक मान्य उपकरण स्कीमा बनाया जा सके, जो कि वही टाइपिंग जीत है जिसके लिए आप Pydantic AI में आए थे।
राशि को एक स्ट्रिंग रखें; Pydantic को रेखा पकड़ने दें
payments.create takes amountWei: a string of USDC base units (6 decimals), so 0.01 USDC is "10000" and 5 USDC is "5000000". Type the tool argument as str and Pydantic AI rejects a float at the boundary with a clear ValidationError, before any HTTP call. That is the whole point of doing this in Pydantic AI - the malformed value never reaches the API.
send_payment प्रारंभ में 503 का उत्तर दे सकता है।
payments.create डिफ़ॉल्ट रूप से पुनः प्रयास नहीं करता है और आपके नेटवर्क के लिए चेन एडेप्टर के कनेक्ट होने तक 503 वापस कर सकता है। अपने उपकरण के अंदर त्रुटि को पकड़ें और एक स्पष्ट संदेश लौटाएं जिस पर मॉडल कार्य कर सके, बजाय इसके कि एजेंट लूप करने दें। स्वचालित रूप से निर्मित idempotency कुंजी का अर्थ है कि एक मैनुअल पुनः प्रयास डबल-पेमेंट नहीं करेगा।
प्रदाता-पूर्ववर्ती मॉडल स्ट्रिंग्स
Pydantic AI प्रदाता-पूर्ववर्ती मॉडल स्ट्रिंग्स का उपयोग करता है: 'openai:gpt-4o', 'anthropic:claude-3-5-sonnet-latest', 'google-gla:gemini-1.5-pro'. पूर्ववर्ती को छोड़ने पर एक गुप्त 'unknown model' त्रुटि मिलती है। आपका उपकरण प्रदाता-निष्पक्ष है क्योंकि यह HTTP API को कॉल करता है, मॉडल को नहीं - पूर्ववर्ती को स्वतंत्र रूप से बदलें और वॉलेट फ़ंक्शन अपरिवर्तित है।
tool_plain बनाम tool (RunContext)
जब फ़ंक्शन को रन से कुछ भी नहीं चाहिए, तो @agent.tool_plain का उपयोग करें, जैसा कि ऊपर है। यदि आप एजेंट की निर्भरताएँ उपकरण के अंदर (एक प्रति-अनुरोध क्लाइंट आईडी, एक खर्च सीमा) चाहते हैं, तो @agent.tool का उपयोग करें और पहले तर्क को ctx: RunContext[Deps] के रूप में घोषित करें ताकि ctx.deps को पढ़ा जा सके। दोनों को मिलाना - tool_plain पर ctx तर्क घोषित करना - पंजीकरण पर एक TypeError उठाता है।