概念
AutoGPT 及类似的自主智能体会循环执行任务并调用工具。l402-kit 可将任何付费 API 包装为可调用工具——智能体自动完成支付、重试并继续运行,无需人工干预。安装
pip install l402kit
将 L402Client 用作 AutoGPT 兼容命令
import os
import asyncio
from l402kit import L402Client
from l402kit.wallets import BlinkWallet
wallet = BlinkWallet(api_key=os.environ["BLINK_API_KEY"])
client = L402Client(wallet=wallet, budget_sats=2000)
# AutoGPT 风格的命令定义
COMMANDS = {
"l402_fetch": {
"description": "Fetch data from a paid API using Bitcoin Lightning micropayment",
"signature": '("url": "<string>")',
"example": '("url": "https://api.example.com/data")',
}
}
def execute_command(command: str, args: dict) -> str:
if command == "l402_fetch":
url = args.get("url", "")
try:
response = asyncio.run(client.get(url))
return f"Success: {response.text[:500]}"
except Exception as e:
return f"Error: {str(e)}"
return f"Unknown command: {command}"
系统提示词补充
将以下内容添加到 AutoGPT 智能体的系统提示词中:COMMANDS:
- l402_fetch: Fetch data from a paid API using Bitcoin Lightning
Args: url (string) — the API endpoint
Cost: varies per endpoint (typically 1-100 sats)
Use when: the API returns HTTP 402 Payment Required
BUDGET: {BUDGET_SATS} sats total for this session.
Track spending and stop if budget is exceeded.
完整自主循环示例
import asyncio
from l402kit import L402Client, BudgetExceededError
from l402kit.wallets import BlinkWallet
wallet = BlinkWallet(api_key=os.environ["BLINK_API_KEY"])
client = L402Client(wallet=wallet, budget_sats=1000)
async def autonomous_data_collector(urls: list[str]) -> list[dict]:
results = []
for url in urls:
try:
response = await client.get(url)
results.append({
"url": url,
"status": "success",
"data": response.json(),
"cost_sats": client.last_payment_sats
})
except BudgetExceededError:
results.append({"url": url, "status": "budget_exceeded"})
break
except Exception as e:
results.append({"url": url, "status": "error", "error": str(e)})
print(f"Total spent: {client.spent_sats} sats")
return results
# 运行
results = asyncio.run(autonomous_data_collector([
"https://l402kit.com/api/demo",
"https://api.example.com/data",
]))
预算追踪
# 每次调用前检查剩余预算
if client.spent_sats + 100 > client.budget_sats:
print("Approaching budget limit, stopping.")
else:
response = await client.get(url)