File size: 3,202 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 CrewAI adapter — multi-agent payment management.
CrewAI runs multiple agents that call tools in parallel. x402 manages
a shared credit pool across all agents in a crew.
Usage:
from x402_frameworks import x402CrewAI
from crewai import Agent, Task, Crew
x402 = x402CrewAI(api_key="your-key", max_budget_usd=50)
researcher = Agent(
role="Token Researcher",
goal="Research crypto tokens",
tools=[x402.wrap("token_scan", scan_func)],
)
# Crew runs, x402 manages payments across all agents
"""
from __future__ import annotations
import functools
import logging
from typing import Any, Callable, Optional
from .base import x402Adapter
logger = logging.getLogger("x402.crewai")
class x402CrewAI(x402Adapter):
"""CrewAI adapter — shared payment pool across all agents in a crew.
Features:
- Shared credit pool (all agents in a crew share a budget)
- Per-crew budget tracking
- Auto-suspend when budget exhausted
- Detailed billing report per crew run
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._crew_budget: dict[str, float] = {}
def set_crew_budget(self, crew_name: str, budget_usd: float):
"""Set a budget for a specific crew. Agents in this crew share the budget."""
self._crew_budget[crew_name] = budget_usd
def wrap(self, tool_name: str, tool_func: Callable) -> Callable:
"""Wrap a tool function with x402 payment for CrewAI agents.
Unlike LangChain's decorator, this returns the wrapped function
directly for use in CrewAI's Agent tool list.
"""
return self.wrap_tool(tool_name)(tool_func)
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)
logger.info(f"CrewAI x402: {tool_name} = ${price:.4f}")
return result
return wrapper
if tool_func is not None:
return decorator(tool_func)
return decorator
def get_crew_report(self) -> dict:
"""Get billing report for the current session."""
return {
"total_spent": round(self._spent_this_month, 2),
"budget_remaining": round(max(0, self.max_budget_usd - self._spent_this_month), 2),
"tools_called": 0, # Tracked internally
}
|