Pydantic AI支払い統合。
Pydantic AIパッケージはなく、必要ありません。@agent.tool_plainで型付き関数を登録し、実際のblockchain0xクライアントを呼び出せば、エージェントはBase上でUSDCを移動できます。
Pydantic AI 専用パッケージはありませんし、必要もありません。Pydantic AI は で関数を検証済みの tool に変換するため、実際の Python client を型付き関数でラップすれば、エージェントは USDC を送信し、請求書を決済し、wallet を読み取れます。tool は model ではなく HTTP API を呼び出すため、OpenAI、Anthropic、Google、Mistral、Groq、その他あらゆる Pydantic AI provider で動作します。支払いは Base で決済されます。
Pydantic AIの型付けが勝ちます。支払いにも使用してください。
Pydantic AIの全体のポイントは強い型付けです: エージェントの入力は型付けされ、出力も型付けされ、依存関係も型付けされ、再試行も型付けされます。このフレームワークは、LLM呼び出しを通じてデータが伝播するのを許可するのではなく、境界で不正なデータを拒否します。あなたのウォレットツールはまさにそれに基づいています - 引数を型で宣言し、Pydantic AIはモデルが呼び出すことを許可される前に、シグネチャとドキュメント文字列から検証されたスキーマを構築します。
実際の利点：モデルがベースユニットの文字列の代わりに浮動小数点数でsend_usdcを呼び出そうとすると、失敗はHTTP呼び出しの前にPydanticの検証レイヤーで発生します。暗号的なフィールドパスを持つAPIからの422ではなく、有用なメッセージを伴った明確なValidationErrorが表示されます。これが、一般的なアダプタを使うのではなく、ここでクライアントを自分でラップする理由です - ワイヤまでの型を維持します。
Pydantic AIとコアSDKをインストールします。2つのキー。
追加する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 キーで、client は環境変数から読み込みます。agent が受け取る側の支払いもある場合、webhook handler には追加で BLOCKCHAIN0X_WEBHOOK_SECRET が必要です。
支払いを行う型付き関数、ツールとして登録されます。
以下は全体の統合です。send_usdcは実際のblockchain0xクライアントを呼び出します；@agent.tool_plainは型ヒントとドキュメント文字列から構築されたスキーマでそれを登録します。実行すると、エージェントはBase上で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を受け取る場合、ポーリングではなくWebhookで確認します。verifyヘルパーは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()を再シリアライズしてはいけません。なぜなら、それは署名がカバーするバイトを変更するからです。出荷されたイベントはpayment.received、payment.sent、wallet.deployed、およびwebhook.testです; X-Blockchain0x-Event-Typeヘッダーを絞り込んで分岐します。
ラップしているクライアントはオープンです。読んでください。
クローンする Pydantic AI の starter package はありません - 上の recipe がそのまま integration です。blockchain0x の SDK は GitHub で open source です。この recipe は Python SDK (blockchain0x-python) をラップしており、完全なメソッド一覧は docs にあります。function body の参考として読んでください。
github.com/tosh-labs/blockchain0x-pythonSDK のメソッド一覧とスコープは the docs に記載されています。まずは Base Sepolia で sk_test_ key を使い、関数が想定どおり動いたら sk_live_ に切り替えてください。
Pydantic AIに特有の避けるべき5つの罠。
Pydantic AIの強い型付けは、境界でほとんどの統合バグをキャッチします。これらは、わずかに漏れるものです。
Pydantic AIパッケージはありません - ツールを登録します。
Blockchain0xは、LangChainとCrewAI用のアダプタとMCPサーバーを出荷します。専用のPydantic AIパッケージはありません。上記のレシピはパスです：実際のblockchain0xクライアントを呼び出す@agent.tool_plainで装飾されたプレーンな型付き関数です。Pydantic AIはシグネチャとドキュメンテーション文字列を読み取って検証されたツールスキーマを構築します。これは、あなたがPydantic AIに求めた型付けの利点です。
amountを文字列のままにしておきます; 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を返すことがあります。ツール内でエラーをキャッチし、モデルがアクションを実行できる明確なメッセージを返してください。エージェントがループしないようにします。自動生成された冪等性キーにより、手動での再試行は二重支払いを引き起こしません。
プロバイダー接頭辞のモデル文字列
Pydantic AIは、プロバイダー接頭辞付きのモデル文字列を使用します: 'openai:gpt-4o', 'anthropic:claude-3-5-sonnet-latest', 'google-gla:gemini-1.5-pro'。接頭辞を省略すると、暗号的な「不明なモデル」エラーが発生します。あなたのツールはHTTP APIを呼び出すため、プロバイダーに依存しません - 接頭辞を自由に入れ替えてもウォレット機能は変更されません。
tool_plain vs tool (RunContext)
@agent.tool_plainを使用して、関数が実行から何も必要ない場合、上記のようにします。ツール内にエージェントの依存関係（リクエストごとのクライアントID、支出上限）が必要な場合は、@agent.toolを使用し、最初の引数をctx: RunContext[Deps]として宣言してctx.depsを読み取ります。2つを混合すること - tool_plainでctx引数を宣言することは、登録時にTypeErrorを引き起こします。