跳转到主要内容
CREWAI 集成

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-crewai
环境变量
export 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支付数据供应商,研究员代理使用结果。财务人员持有钱包工具;研究员则没有。这是一个支出的团队的标准模式。

CREW.PY
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.

WEBHOOK 处理

当付款到账时触发下一个团队。

当 USDC 结算到您的代理时,Blockchain0x 会将签名事件 POST 到您的 webhook URL。验证助手包含在 Node SDK 中;Python 服务手动根据文档中的 HMAC 进行验证,这正是助手的全部功能。下面是 Flask 示例;相同的代码在 FastAPI 或任何 Python Web 框架中都能工作。

WEBHOOK_HANDLER.PY
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 只需要支付,给它一个单一的工厂。

SINGLE_TOOL.PY
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的基于角色的设计使支付集成比其他框架更清晰。它也有自己的陷阱。提前了解这些可以节省时间。

PITFALL 1

将钱包工具放在错误的代理上

CrewAI 代理有角色。将钱包工具交给研究人员,研究人员应该进行研究,而 LLM 开始尝试在分析过程中移动资金。干净的模式是一个专门的财务主管代理,持有工具并作为自己的任务运行;工作代理根本没有钱包访问权限。上面的示例遵循该形状。复制它。

PITFALL 2

顺序与层次过程

CrewAI 支持顺序(默认)和层次执行。在层次结构中,管理代理协调团队。如果允许经理委派钱包工具,它可以将其交给任何下属,这正是你不希望发生的支出权限。要么将工具保留在一个专用代理上并顺序运行,要么限制经理的委派规则。

PITFALL 3

从任务上下文读取结算状态

入站结算是异步的。如果研究任务在其上下文中列出了支付任务(很好),但LLM随后尝试从该上下文中读取最终支付状态(不好,确认稍后通过Webhook到达),您会得到混乱的输出。将下游工作限制在Webhook触发的信号上,而不是任务上下文。CrewAI的任务回调是正确的钩子。

PITFALL 4

金额是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.

PITFALL 5

Webhook处理程序在团队进程之外

您的 webhook 处理程序通常是一个单独的 Flask 或 FastAPI 进程,而不是 Crew 运行时的一部分。Crew 完成其 kickoff() 并返回;webhook 在支付结算时稍后到达。通过队列或 webhook 写入的数据库行与 Crew 在下次运行时读取的行连接这两者。不要让 Crew 等待 webhook 处于活动状态 - CrewAI 运行的设计是短暂的。

常见问题

三个CrewAI特定问题。

这与CrewAI Flows(更新的基于图的执行)兼容吗?

是的。create_blockchain0x_toolset 返回标准 CrewAI 工具,因此它们可以在经典的 Crew + Task 结构和 CrewAI Flow 的 @start/@listen 模式中工作。在 Flow 中,将支付或结算步骤放在专用财务主管步骤之前,然后让工作步骤监听在您的 webhook 处理程序更新共享状态时触发的 Flow 事件。

同一团队可以同时运行多个付费任务吗?

是的。payments.create(即 blockchain0x_send_payment 包装的内容)会自动生成一个稳定的 Idempotency-Key,因此重试的提交会合并为一次链上转账,而不是重复支付。如果您同时启动多个并发操作,每个操作都移动资金,请为每个操作提供自己的意图(不同的接收者或元数据),以便它们看起来不会彼此重复。除非您想在进程之间去重,否则您无需自己管理幂等性头。

团队可以结算发票吗?是否有退款工具?

它可以结算:blockchain0x_settle_invoice结算您在仪表板中创建的支付请求,并记录链上证明(交易哈希、付款人、验证的USDC金额)。没有退款助手 - 该标识符不存在。要退还资金,财务人员只需使用blockchain0x_send_payment向付款人发送新的USDC支付。将其保留在财务角色上,以便支出权限保持在一个地方。

为您的团队添加计费角色。

从pip安装到您第一次付费团队启动只需几分钟。免费开始。