Agent API的调用本质上是客户端与服务端之间的一次HTTP交互，只不过相比传统的CRUD接口，它多了会话管理、流式输出和长任务处理这些特性。很多人拿到API文档后直接复制示例代码就开跑，结果在线上环境里接连踩坑：鉴权失败、响应超时、上下文丢失。这篇文章就把Agent API的RESTful交互拆开来讲清楚，帮助你从只会跑示例进阶到能独立设计一条稳定的调用链路。
一、RESTful基础：方法、URL与请求结构
RESTful风格的核心在于用URL表示资源、用HTTP方法表示操作意图。Agent类API通常会把「会话」「消息」「任务」抽象为资源，因此你会看到类似/v1/sessions、/v1/sessions/{session_id}/messages这样的路径设计。创建会话用POST，查询状态用GET，取消任务用DELETE，这种语义化的设计让接口行为一目了然。
一次完整的请求包含三个部分：请求行（方法与URL）、请求头（Content-Type、Authorization等）以及请求体。下面是一个典型的创建会话请求：
import requests
url = "https://api.ipipp.com/v1/sessions"
headers = {
"Content-Type": "application/json",
"Authorization": "Bearer sk-your-api-key"
}
payload = {
"agent_id": "agent_001",
"metadata": {
"user_id": "u_10086",
"channel": "web"
}
}
resp = requests.post(url, json=payload, headers=headers, timeout=30)
print(resp.status_code)
print(resp.json())这里有个细节值得注意：timeout=30千万不要省略。requests库默认不设置超时，一旦服务端响应缓慢，你的调用线程会无限期挂起，进而拖垮整个应用的连接池。建议连接超时设为5秒，读取超时设为30到60秒，可以用元组形式timeout=(5, 30)分别指定。
另外，Content-Type务必与实际请求体匹配。如果你用data=payload发送的是表单，却声明了application/json，服务端解析会直接失败并返回400错误。用json=payload参数可以让requests自动完成序列化和头设置，是最稳妥的做法。
二、鉴权与会话管理：让上下文不丢失
绝大多数Agent API采用Bearer Token鉴权，也就是在Authorization头中携带Bearer <你的密钥>。密钥管理上有两条铁律：第一，密钥绝不能硬编码进代码或提交到代码仓库，应从环境变量或密钥管理服务读取；第二，密钥要定期轮换，一旦泄露立即吊销。
Agent调用与普通接口最大的区别在于会话状态。Agent是有记忆的，多轮对话的上下文依赖于session_id。常见做法是：首次调用时创建会话拿到session_id，后续每次请求都携带它。但要注意，有些服务还要求显式传递conversation历史，避免服务端存储压力过大。两种模式的区别如下：
import os
import requests
BASE_URL = "https://api.ipipp.com/v1"
API_KEY = os.environ.get("AGENT_API_KEY")
session = requests.Session()
session.headers.update({
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
})
# 创建会话
sess = session.post(f"{BASE_URL}/sessions", json={"agent_id": "agent_001"}).json()
session_id = sess["id"]
# 基于同一会话发送消息，上下文由服务端维护
result = session.post(
f"{BASE_URL}/sessions/{session_id}/messages",
json={"role": "user", "content": "帮我总结一下上一轮的问题"}
).json()
print(result["reply"])使用requests.Session可以复用TCP连接，减少每次握手的开销，在高频调用场景下能明显降低延迟。如果接口返回401，先检查token是否过期；返回403则通常是权限不足，比如调用了未开通的Agent能力。排查时可以打印响应体，大多数服务会在body里给出具体的错误说明，比状态码本身有用得多。
三、流式响应处理：边生成边消费
Agent生成回复往往需要数秒甚至更久，如果等到全部生成完再返回，用户体验会非常差。主流Agent API都支持SSE（Server-Sent Events）流式返回，即设置Accept: text/event-stream后，服务端会分块推送增量内容。处理流式响应的关键是用stream=True并逐行迭代解析：
resp = session.post(
f"{BASE_URL}/sessions/{session_id}/messages",
json={"role": "user", "content": "写一段冒泡排序的讲解", "stream": True},
headers={"Accept": "text/event-stream"},
stream=True,
timeout=(5, 120)
)
full_text = ""
for line in resp.iter_lines(decode_unicode=True):
if not line or not line.startswith("data:"):
continue
data = line[len("data:"):].strip()
if data == "[DONE]":
break
# 每个data块是一个JSON片段，包含增量文本
import json
chunk = json.loads(data)
delta = chunk.get("delta", "")
full_text += delta
print(delta, end="", flush=True)
print()
print("完整回复：", full_text)流式处理中有两个易错点。一是必须手动处理[DONE]结束标记，否则循环可能一直阻塞到超时；二是断流处理，网络抖动会导致连接中断，生产环境应该记录已收到的内容，通过重试接口或消息续传机制恢复，而不是简单地从头再来。
四、错误处理与重试策略
再稳定的接口也会出错，关键是让调用方在错误发生时行为可控。首先要按状态码分类处理：4xx类错误（如400参数错误、401鉴权失败、429限流）属于客户端问题，重试没有意义，应该修正请求或退避等待；5xx类错误和超时属于服务端偶发问题，可以安全重试。429限流要特别对待，认真读取Retry-After响应头，按服务端要求的时间等待。
下面是一个带指数退避的通用重试封装：
import time
def call_with_retry(func, max_retries=3, base_delay=1):
for attempt in range(max_retries + 1):
try:
resp = func()
if resp.status_code < 500:
return resp
except requests.exceptions.Timeout:
if attempt == max_retries:
raise
except requests.exceptions.ConnectionError:
if attempt == max_retries:
raise
delay = base_delay * (2 ** attempt)
time.sleep(delay)
return resp
# 使用方式
resp = call_with_retry(lambda: session.get(f"{BASE_URL}/sessions/{session_id}"))
print(resp.status_code)重试次数不宜过多，三次左右是常见配置。对于创建会话这类非幂等操作，盲目重试可能产生重复会话，建议服务端提供幂等键（如请求头Idempotency-Key），调用方生成唯一ID随请求发送，服务端据此去重。同时别忘了记录完整的请求与响应日志（脱敏后），线上排障时这些日志是唯一的线索来源。把鉴权、会话、流式和重试这四块都做扎实，你的Agent API调用链路就具备了生产可用的基础素质。
Agent API调用RESTful接口API交互修改时间：2026-09-15 17:10:35