インストール
要件: 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(環境設定をオーバーライド) |
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")
Webhook
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())