File size: 3,314 Bytes
bde2f3a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """x402 AutoGen adapter — agentic payment flows for multi-agent conversations.
AutoGen agents negotiate who pays for tool calls. x402 handles the
payment transparently — agents request, verify, and deduct.
Usage:
from x402_frameworks import x402AutoGen
from autogen import AssistantAgent, UserProxyAgent
x402 = x402AutoGen(api_key="your-key")
analyst = AssistantAgent(
name="TokenAnalyst",
system_message="You analyze tokens using x402 tools.",
functions=[x402.as_function("token_scan")],
)
# When analyst calls token_scan, x402 auto-pays from credits
"""
from __future__ import annotations
import functools
import json
import logging
from typing import Any, Callable, Optional
from .base import x402Adapter
logger = logging.getLogger("x402.autogen")
class x402AutoGen(x402Adapter):
"""AutoGen adapter — handles payment for tool calls in multi-agent convos.
Features:
- AutoGen function format: `as_function()` returns AutoGen-compatible spec
- Multi-agent: each agent has its own budget or shares a pool
- Transparent: agents just call functions, x402 handles payment
"""
def as_function(self, tool_name: str, description: str = "") -> dict:
"""Return an AutoGen-compatible function specification.
AutoGen agents use this to discover and call x402 tools.
Payment is handled transparently when the function is called.
"""
return {
"name": f"x402_{tool_name}",
"description": description or f"x402 paid tool: {tool_name}. Pay per call via x402 protocol.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": f"Input for {tool_name}. On-chain payment verified before execution.",
}
},
"required": ["query"],
},
}
def wrap_tool(self, tool_name: str, tool_func: Optional[Callable] = None) -> Callable:
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
available, price, source = self.check_available(tool_name)
if not available:
invoice = self.get_invoice(tool_name)
from .langchain_adapter import PaymentRequiredError
raise PaymentRequiredError(
tool=tool_name,
price_usd=price,
pay_to=invoice.get("pay_to", ""),
chain=invoice.get("chain", "base"),
invoice_id=invoice.get("id", ""),
)
result = func(*args, **kwargs)
if source == "paid":
self.record_usage(tool_name, price)
return result
return wrapper
if tool_func is not None:
return decorator(tool_func)
return decorator
def register_agent(self, agent_name: str, budget_usd: float = 10.0):
"""Register an AutoGen agent with a personal budget."""
logger.info(f"x402 AutoGen: agent '{agent_name}' registered with ${budget_usd} budget")
|