Pydantic AI支付集成。
没有 Pydantic AI 包,您不需要一个。使用 @agent.tool_plain 注册一个类型化函数,调用真实的 blockchain0x 客户端,您的代理可以在 Base 上移动 USDC。
没有 Pydantic AI-specific package,而且你也不需要。Pydantic AI 会用 将函数变成经过校验的工具,因此你只需把真实的 Python client 封装成 typed function,代理就能发送 USDC、结算 invoice、读取钱包。它适用于 OpenAI、Anthropic、Google、Mistral、Groq 以及所有其他 Pydantic AI provider,因为这个工具调用的是 HTTP API,而不是模型。支付在 Base 上结算。
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 会从环境变量读取它。如果你的 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.
通过签名的 webhook 确认入站支付。
如果您的代理还接收USDC,请通过Webhook确认,而不是轮询。验证助手包含在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 就是集成方式。blockchain0x SDK 在 GitHub 上开源;这个 recipe 封装了 Python SDK (blockchain0x-python),完整的方法表面已记录在 docs 中。可将其作为 function bodies 的参考。
github.com/tosh-labs/blockchain0x-python完整的 SDK method surface 和 scopes 已记录在 the docs 中。先使用 sk_test_ key 对接 Base Sepolia 进行测试,然后在 function 符合预期后切换到 sk_live_。
五个需要避免的 Pydantic AI 特定陷阱。
Pydantic AI的强类型在边界捕捉到大多数集成错误;这些是少数几个漏网之鱼。
没有 Pydantic AI 包 - 您注册一个工具
Blockchain0x 为 LangChain 和 CrewAI 以及 MCP 服务器提供适配器;没有专门的 Pydantic AI 包。上面的配方是路径:一个用 @agent.tool_plain 装饰的普通类型化函数,调用真实的 blockchain0x 客户端。Pydantic AI 读取签名和文档字符串以构建经过验证的工具架构,这正是你来 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。在您的工具内部捕获错误并返回模型可以采取的清晰消息,而不是让代理循环。自动铸造的幂等性密钥意味着手动重试不会重复支付。
提供者前缀的模型字符串
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。混合这两者 - 在 tool_plain 上声明 ctx 参数 - 在注册时会引发 TypeError。