CrewAI支付集成。
给CrewAI团队一个钱包。一个专门的财务代理发送USDC,结算发票,并读取余额,而工作代理则专注于工作。
Blockchain0x 提供了一个真正的 Python 包,。安装它,设置 ,调用 ,然后把这些工具交给一个专门的 treasurer agent。该 agent 可以发送 USDC、用链上证明结算发票并读取钱包,而团队中的工作 agent 不需要任何钱包访问权限。USDC 在 Base 上结算。
CrewAI 的优势在于角色分离。也将其用于钱包。
其他框架将每个工具堆叠到单个代理上。CrewAI的设计鼓励角色分离:研究员进行研究,作家进行写作,审阅者进行审阅。资金也应得到同样的对待 - 一个专门的财务代理持有钱包工具,而工作代理则不持有。支出权限集中在一个地方,而不是在每个偶然出现在团队中的代理之间泄露。
这种形状还使审计轨迹更加清晰。当 USDC 移动时,您确切知道是哪个代理移动了它(财务主管)以及是为了哪个任务(将其移交给它的任务)。单代理集成模糊了这一界限,使得每个任务的支出分析变得更加困难。
一次pip安装。一个环境变量。
该包需要 Python 3.10 或更高版本,并与 poetry 或 pip 一起使用。它将核心 blockchain0x SDK 和 crewai 作为依赖项;无需添加其他内容。
pip install blockchain0x-crewaiexport BLOCKCHAIN0X_API_KEY=sk_test_... # sk_test_ = Base Sepolia, sk_live_ = Base mainnet
BLOCKCHAIN0X_API_KEY 是你 Blockchain0x dashboard 中的 sk_test_ testnet 或 sk_live_ mainnet key,也是 adapter 唯一需要的变量 - 它只会把 key 转发到后端,本身不保存任何 secret。下面的 webhook handler 还需要设置 BLOCKCHAIN0X_WEBHOOK_SECRET,该值会在你创建或轮换 webhook 时由 dashboard 仅返回一次。
一个具有计费和研究角色的工作双代理团队。
以下是一个完整的CrewAI团队,其中财务代理用USDC支付数据供应商,研究员代理使用结果。财务人员持有钱包工具;研究员则没有。这是一个支出的团队的标准模式。
from crewai import Agent, Task, Crew from blockchain0x import Client from blockchain0x_crewai import create_blockchain0x_toolset client = Client() # reads BLOCKCHAIN0X_API_KEY from the environment tools = create_blockchain0x_toolset(client) # A dedicated treasurer holds the wallet tools; the work agents do not. treasurer = Agent( role="Treasurer", goal="Pay vendor invoices in USDC within owner-set limits", backstory="Operates a Blockchain0x agent wallet on Base.", tools=tools, allow_delegation=False, ) researcher = Agent( role="Senior Market Analyst", goal="Produce a thorough Q4 LLM market analysis", backstory="A specialist analyst with 10 years in tech-market research.", tools=[], allow_delegation=False, ) pay_task = Task( description="Pay 0.01 USDC from our agent wallet to the data vendor at 0xVendor.", expected_output="The transaction hash of the settled payment.", agent=treasurer, ) research_task = Task( description="Using the purchased dataset, write the Q4 LLM market analysis.", expected_output="A 1,500-word market analysis.", agent=researcher, context=[pay_task], ) crew = Crew(agents=[treasurer, researcher], tasks=[pay_task, research_task]) result = crew.kickoff()
When you call crew.kickoff(), the pay_task runs first: the treasurer calls blockchain0x_send_payment, the SDK submits the transfer on Base, and the transaction hash is the task output. The research_task runs next and uses the purchased result. amount_wei is USDC base units, so 0.01 USDC is the string "10000". On a sk_test_ key the whole flow runs on Base Sepolia, and payments.create can answer 503 until the chain adapter is wired for your network.
当付款到账时触发下一个团队。
当 USDC 结算到您的代理时,Blockchain0x 会将签名事件 POST 到您的 webhook URL。验证助手包含在 Node SDK 中;Python 服务手动根据文档中的 HMAC 进行验证,这正是助手的全部功能。下面是 Flask 示例;相同的代码在 FastAPI 或任何 Python Web 框架中都能工作。
import hmac, hashlib, os, time from flask import Flask, request, abort app = Flask(__name__) SECRET = os.environ["BLOCKCHAIN0X_WEBHOOK_SECRET"].encode() @app.post("/webhooks/payment") def receive(): raw = request.get_data() # RAW bytes - do not parse first sig = request.headers.get("X-Blockchain0x-Signature", "") ts = request.headers.get("X-Blockchain0x-Timestamp", "") # Header is "t=<unix>,v1=<hex>"; some proxies send bare hex + the ts header. 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: abort(401) if request.headers.get("X-Blockchain0x-Event-Type") == "payment.received": run_followup_for(request.get_json()) # USDC landed - do the work return "ok", 200
这是真正的算法:对字符串 t.rawBody 进行 HMAC-SHA256,进行恒定时间比较,以及 300 秒的重放窗口。读取原始主体(request.get_data()),绝不要重新序列化 request.get_json(),因为那会改变签名覆盖的字节。架构:让处理程序将作业排入队列(Celery、RQ、Redis 列表),以触发实际的 CrewAI 运行。在处理程序内部调用 crew.kickoff() 可能需要几分钟,并使 HTTP 请求超时。
将整个工具集或仅一个工具交给团队。
create_blockchain0x_toolset 返回所有五个工具:blockchain0x_send_payment、blockchain0x_settle_invoice、blockchain0x_get_wallet、blockchain0x_list_wallets 和 blockchain0x_get_transaction。如果一个 Crew 只需要支付,给它一个单一的工厂。
from blockchain0x_crewai import create_payment_tool pay = create_payment_tool(client) # amount_wei is USDC base units: 6 decimals, so 10000 = 0.01 USDC. pay.run(agent_id="agt_123", to="0xRecipient", amount_wei="10000")
按 capability 划分的 factories 包括 create_payment_tool、create_invoice_tool、create_wallet_tool、create_wallets_list_tool 和 create_transaction_tool。源代码和 testnet 示例位于 the Python SDK repo;CrewAI support matrix 位于 the docs。
在将计费集成到团队时需要关注的五件事。
CrewAI的基于角色的设计使支付集成比其他框架更清晰。它也有自己的陷阱。提前了解这些可以节省时间。
将钱包工具放在错误的代理上
CrewAI 代理有角色。将钱包工具交给研究人员,研究人员应该进行研究,而 LLM 开始尝试在分析过程中移动资金。干净的模式是一个专门的财务主管代理,持有工具并作为自己的任务运行;工作代理根本没有钱包访问权限。上面的示例遵循该形状。复制它。
顺序与层次过程
CrewAI 支持顺序(默认)和层次执行。在层次结构中,管理代理协调团队。如果允许经理委派钱包工具,它可以将其交给任何下属,这正是你不希望发生的支出权限。要么将工具保留在一个专用代理上并顺序运行,要么限制经理的委派规则。
从任务上下文读取结算状态
入站结算是异步的。如果研究任务在其上下文中列出了支付任务(很好),但LLM随后尝试从该上下文中读取最终支付状态(不好,确认稍后通过Webhook到达),您会得到混乱的输出。将下游工作限制在Webhook触发的信号上,而不是任务上下文。CrewAI的任务回调是正确的钩子。
金额是USDC基本单位,以字符串形式表示
blockchain0x_send_payment takes amount_wei: a string of USDC base units, not a float and not a dollar figure. USDC has 6 decimals, so 0.01 USDC is "10000". CrewAI passes tool inputs through as-is, so a Python float arrives as a float and the SDK rejects it - but the error reads like a CrewAI tool error, which throws off first integrators. Pass a string.
Webhook处理程序在团队进程之外
您的 webhook 处理程序通常是一个单独的 Flask 或 FastAPI 进程,而不是 Crew 运行时的一部分。Crew 完成其 kickoff() 并返回;webhook 在支付结算时稍后到达。通过队列或 webhook 写入的数据库行与 Crew 在下次运行时读取的行连接这两者。不要让 Crew 等待 webhook 处于活动状态 - CrewAI 运行的设计是短暂的。