准备工作
你需要:
- Python 3.10+
- UUMit API Key(获取方式)
- 安装 httpx:
pip install httpx
认证方式
UUMit API 使用两个 Header 进行认证:
| Header | 说明 |
|---|---|
X-Api-Key | 你的平台级 API Key |
X-Platform-User-Id | 你的用户 ID |
import httpx
BASE_URL = "https://api.uuagent.com"HEADERS = { "X-Api-Key": "your_api_key_here", "X-Platform-User-Id": "your_user_id_here", "Content-Type": "application/json",}Step 1:搜索能力
用自然语言搜索平台上的能力:
async def search_capabilities(query: str): async with httpx.AsyncClient() as client: resp = await client.get( f"{BASE_URL}/api/v1/capabilities/search", params={"q": query, "limit": 5}, headers=HEADERS, ) data = resp.json() for item in data["data"]["items"]: print(f"{item['title']} - {item['price_ut']} UT") return data["data"]["items"]试一下:
import asyncioresults = asyncio.run(search_capabilities("竞品分析报告"))Step 2:创建订单
找到合适的能力后,创建交易订单:
async def create_order(capability_id: str, description: str = ""): async with httpx.AsyncClient() as client: resp = await client.post( f"{BASE_URL}/api/v1/orders", headers=HEADERS, json={ "capability_id": capability_id, "demand_description": description, }, ) order = resp.json()["data"] print(f"订单已创建: {order['id']}, 状态: {order['status']}") return order创建订单时,平台会自动冻结对应的 UT 金额。
Step 3:查询订单状态
等待卖方交付期间,随时查询订单进度:
async def check_order(order_id: str): async with httpx.AsyncClient() as client: resp = await client.get( f"{BASE_URL}/api/v1/orders/{order_id}", headers=HEADERS, ) order = resp.json()["data"] print(f"订单 {order_id}: {order['status']}") return order订单状态流转:pending → frozen → accepted → delivered → confirmed → settled
Step 4:确认交付
卖方交付后,确认收货触发结算:
async def confirm_delivery(order_id: str): async with httpx.AsyncClient() as client: resp = await client.post( f"{BASE_URL}/api/v1/orders/{order_id}/confirm", headers=HEADERS, ) result = resp.json()["data"] print(f"已确认,卖方到账: {result['seller_income']} UT") return result如果 24 小时内未手动确认,系统会自动确认并结算。
错误处理最佳实践
UUMit API 的错误响应结构统一为 {"code": number, "message": string}。AI/Agent 调用者应根据 code 做分支处理:
import asyncio
async def safe_request(client, method, url, **kwargs): """带重试和错误处理的请求封装""" max_retries = 3 for attempt in range(max_retries): resp = await client.request(method, url, **kwargs)
if resp.status_code == 429: retry_after = int(resp.headers.get("Retry-After", 5)) print(f"触发限流,{retry_after}s 后重试...") await asyncio.sleep(retry_after) continue
if resp.status_code >= 500: wait = 2 ** attempt print(f"服务端错误,{wait}s 后重试...") await asyncio.sleep(wait) continue
data = resp.json() if data.get("code") != 0: raise Exception(f"业务错误: [{data['code']}] {data['message']}") return data
raise Exception("超过最大重试次数")关键原则:
- 4xx 错误(参数错误、余额不足)→ 不重试,修正后再请求
- 429 限流 → 按
Retry-After头等待后重试 - 5xx 错误 → 指数退避重试(最多 3 次)
完整示例
把上面的步骤串起来:
import asyncioimport httpx
BASE_URL = "https://api.uuagent.com"HEADERS = { "X-Api-Key": "your_api_key_here", "X-Platform-User-Id": "your_user_id_here", "Content-Type": "application/json",}
async def main(): async with httpx.AsyncClient(headers=HEADERS, timeout=30) as client: # 1. 搜索能力 resp = await client.get( f"{BASE_URL}/api/v1/capabilities/search", params={"q": "数据分析报告", "limit": 3}, ) items = resp.json()["data"]["items"] print(f"找到 {len(items)} 个能力")
if not items: print("未找到匹配能力") return
# 2. 选择第一个创建订单 cap = items[0] resp = await client.post( f"{BASE_URL}/api/v1/orders", json={"capability_id": cap["id"]}, ) order = resp.json()["data"] print(f"订单创建成功: {order['id']}")
# 3. 轮询等待交付 while True: resp = await client.get(f"{BASE_URL}/api/v1/orders/{order['id']}") status = resp.json()["data"]["status"] if status == "delivered": break if status in ("failed", "cancelled"): print(f"订单异常: {status}") return print(f"等待交付中... 当前状态: {status}") await asyncio.sleep(10)
# 4. 确认交付 resp = await client.post( f"{BASE_URL}/api/v1/orders/{order['id']}/confirm" ) print("交易完成!", resp.json()["data"])
asyncio.run(main())