要求: Python 3.9+
客户端初始化
import os
from agentwallex import AgentWallex
aw = AgentWallex(
api_key=os.environ["AGENTWALLEX_API_KEY"],
environment="sandbox", # "sandbox" or "production"
timeout=30, # Request timeout in seconds (default: 30)
)
| 参数 | 类型 | 默认值 | 说明 |
|---|
api_key | str | 必填 | 您的 API 密钥(awx_...) |
environment | str | "sandbox" | "sandbox" 或 "production" |
base_url | str | None | 自定义基础 URL(覆盖 environment) |
timeout | int | 30 | 请求超时时间(秒) |
Python SDK 是异步优先的。所有方法返回协程,需要使用 await。
aw.agents.create()
创建一个带有 MPC 安全钱包的新代理。
agent = await aw.agents.create(
name="my-agent",
chain="eip155:8453",
policies={
"max_transaction_amount": "100",
"daily_limit": "1000",
"allowed_addresses": ["0x..."],
"allowed_tokens": ["USDC"],
},
metadata={"team": "growth"},
)
print(f"Agent ID: {agent.id}")
print(f"Wallet: {agent.wallet.address}")
aw.agents.get(agent_id)
agent = await aw.agents.get("agent_abc123")
aw.agents.list()
result = await aw.agents.list(
limit=20,
offset=0,
status="active",
)
for agent in result.agents:
print(f"{agent.name}: {agent.wallet.address}")
aw.agents.freeze(agent_id) / aw.agents.unfreeze(agent_id)
# Freeze — immediate
await aw.agents.freeze("agent_abc123")
# Unfreeze — subject to 10-minute cool-down
await aw.agents.unfreeze("agent_abc123")
aw.payments.send()
发送直接链上支付。
tx = await aw.payments.send(
agent_id="agent_abc123",
to="0xRecipientAddress",
amount="50.00",
token="USDC",
memo="Payment for API access",
)
print(f"Hash: {tx.hash}")
print(f"Status: {tx.status}")
print(f"Fee: {tx.fee}")
aw.payments.get(transaction_id)
tx = await aw.payments.get("tx_xyz789")
aw.payments.list()
result = await aw.payments.list(
agent_id="agent_abc123",
limit=50,
status="confirmed",
)
for tx in result.transactions:
print(f"{tx.amount} {tx.token} -> {tx.status}")
x402 微支付
aw.x402.create_session()
session = await aw.x402.create_session(
agent_id="agent_abc123",
budget_limit="100.00",
chain="eip155:84532",
ttl_seconds=3600,
allowed_urls=["https://paid-api.example.com/v1/data"],
)
aw.x402.check()
info = await aw.x402.check(url="https://paid-api.example.com/v1/data")
aw.x402.pay()
result = await aw.x402.pay(
agent_id="agent_abc123",
target_url="https://paid-api.example.com/v1/data",
session_id=session.id,
chain="eip155:84532",
)
aw.x402.session_pay(session_id, target_url)
await aw.x402.session_pay(
session.id,
target_url="https://paid-api.example.com/v1/data",
)
aw.x402.http_interceptor()
创建一个拦截器,自动处理 x402 v2 挑战:
fetch_with_payment = aw.x402.http_interceptor(
agent_id="agent_abc123",
chain="eip155:84532",
)
# Automatically handles 402 -> pay -> retry
response = await fetch_with_payment("https://paid-api.example.com/v1/data")
data = response.json()
aw.policies.update()
await aw.policies.update(
"agent_abc123",
max_transaction_amount="200",
daily_limit="2000",
allowed_addresses=["0xAddr1", "0xAddr2"],
velocity_limit={"max_count": 100, "window_seconds": 3600},
)
aw.policies.get(agent_id)
policies = await aw.policies.get("agent_abc123")
aw.policies.apply_template()
await aw.policies.apply_template("agent_abc123", "conservative")
Webhooks
aw.webhooks.create()
webhook = await aw.webhooks.create(
url="https://your-app.com/webhooks/agentwallex",
events=[
"payment.completed",
"payment.failed",
"agent.frozen",
"policy.violated",
],
secret="whsec_your_signing_secret",
)
错误处理
from agentwallex import AgentWallexError, PolicyViolationError
try:
tx = await aw.payments.send(
agent_id="agent_abc123",
to="0xRecipientAddress",
amount="50.00",
token="USDC",
)
except PolicyViolationError as e:
print(f"Policy violated: {e.rule}")
print(f"Details: {e.message}")
except AgentWallexError as e:
print(f"API error [{e.code}]: {e.message}")
except Exception as e:
print(f"Unexpected error: {e}")
| 错误类 | 触发条件 |
|---|
AgentWallexError | 所有 API 错误的基类 |
PolicyViolationError | 交易被策略引擎阻止 |
AuthenticationError | API 密钥无效或缺失 |
NotFoundError | 资源不存在 |
RateLimitError | 超过速率限制 |
类型提示
Python SDK 包含完整的类型提示,支持 IDE 补全和静态分析:
from agentwallex.types import (
Agent,
Payment,
Transaction,
Policy,
X402Session,
WebhookEvent,
)
与 asyncio 配合使用
SDK 是异步优先的。在异步上下文中使用:
import asyncio
from agentwallex import AgentWallex
async def main():
aw = AgentWallex(api_key="awx_your_api_key", environment="sandbox")
agent = await aw.agents.create(
name="my-agent",
chain="eip155:84532",
)
print(f"Created agent: {agent.id}")
asyncio.run(main())