UUMit Logo
返回博客

用 Python 3 分钟接入 UUMit API

· UUMit 团队

准备工作

你需要:

  1. Python 3.10+
  2. UUMit API Key获取方式
  3. 安装 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 asyncio
results = 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

订单状态流转:pendingfrozenaccepteddeliveredconfirmedsettled

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 asyncio
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",
}
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())

了解更多

分享: Twitter