Agents/API_DOC.md

慧遇 Agent API 接口文档

基于 Google ADK api_server 提供的 REST API，供外部项目集成使用。

API 文档（Swagger UI）：启动服务后访问 http://<host>:<port>/docs

---

目录

• 快速开始
• 启动服务
• 核心概念
• 系统接口
• 会话管理
• Agent 对话
• SSE 流式对话
• WebSocket 实时对话
• 错误处理
• 代码示例

---

快速开始

# 1. 启动服务
cd agents && adk api_server . --auto_create_session --port 8000

# 2. 发送对话
curl -X POST http://localhost:8000/run \
  -H "Content-Type: application/json" \
  -d '{
    "app_name": "root_agent",
    "user_id": "user_001",
    "session_id": "my_session_001",
    "new_message": {
      "role": "user",
      "parts": [{"text": "你好"}]
    }
  }'

---

启动服务

adk api_server <agents_dir> [选项]

常用参数

参数	说明	默认值
--host	监听地址	127.0.0.1
--port	监听端口	8000
--auto_create_session	调用 /run 时自动创建不存在的会话	关闭
--session_service_uri	会话存储后端	本地 SQLite
--allow_origins	CORS 允许的源	*
--url_prefix	URL 路径前缀（如 /api/v1）	无

会话存储配置

# 内存存储（重启丢失）
adk api_server . --session_service_uri memory://

# SQLite 持久化
adk api_server . --session_service_uri sqlite:///path/to/sessions.db

# PostgreSQL
adk api_server . --session_service_uri postgresql://user:pass@host:5432/db

---

核心概念

三维会话标识

每次 API 调用需要通过三个维度定位会话：

维度	说明	示例
app_name	Agent 应用名称（目录名）	root_agent
user_id	用户唯一标识（由调用方管理）	user_001
session_id	会话唯一标识	sess_abc123

消息格式（Content）

所有对话消息使用统一的 Content 格式：

{
  "role": "user",
  "parts": [
    {"text": "消息内容"}
  ]
}

role 可选值："user" | "model"

---

系统接口

健康检查

GET /health

响应：

{"status": "ok"}

版本信息

GET /version

响应：

{
  "version": "1.28.1",
  "language": "python",
  "language_version": "3.10.12"
}

列出可用应用

GET /list-apps?detailed=true

查询参数：

参数	类型	说明
detailed	bool	是否返回详细信息，默认 false

响应（detailed=false）：

["root_agent", "note_agent"]

响应（detailed=true）：

{
  "apps": [
    {
      "name": "root_agent",
      "root_agent_name": "huiyu_agent",
      "description": "一个智能助手，能够回答用户的各种问题。",
      "language": "python",
      "is_computer_use": false
    }
  ]
}

---

会话管理

所有会话接口的基础路径：

/apps/{app_name}/users/{user_id}/sessions

创建会话

POST /apps/{app_name}/users/{user_id}/sessions

请求体（均可选）：

{
  "session_id": "my_custom_session_id",
  "state": {"key": "value"}
}

字段	类型	必填	说明
session_id	string	否	自定义会话 ID，不传则自动生成 UUID
state	object	否	会话初始状态
events	array	否	初始化事件列表

响应：

{
  "id": "c470eb5f-6a66-4005-8427-7d3d9d178171",
  "appName": "root_agent",
  "userId": "user_001",
  "state": {},
  "events": [],
  "lastUpdateTime": 1775556907.896
}

获取会话详情

GET /apps/{app_name}/users/{user_id}/sessions/{session_id}

响应： 同创建会话的响应格式。

错误： 404 会话不存在

列出用户的所有会话

GET /apps/{app_name}/users/{user_id}/sessions

响应：

[
  {
    "id": "c470eb5f-6a66-4005-8427-7d3d9d178171",
    "appName": "root_agent",
    "userId": "user_001",
    "state": {},
    "events": [],
    "lastUpdateTime": 1775556907.896
  }
]

删除会话

DELETE /apps/{app_name}/users/{user_id}/sessions/{session_id}

响应： 204 No Content

更新会话状态

PATCH /apps/{app_name}/users/{user_id}/sessions/{session_id}

请求体：

{
  "state_delta": {
    "key_to_update": "new_value"
  }
}

字段	类型	必填	说明
state_delta	object	是	要合并到会话状态的变更

响应： 更新后的 Session 对象。

---

Agent 对话

运行 Agent（同步）

POST /run

请求体：

{
  "app_name": "root_agent",
  "user_id": "user_001",
  "session_id": "c470eb5f-6a66-4005-8427-7d3d9d178171",
  "new_message": {
    "role": "user",
    "parts": [{"text": "你好，请介绍一下你自己"}]
  }
}

字段	类型	必填	说明
app_name	string	是	应用名称
user_id	string	是	用户 ID
session_id	string	是	会话 ID
new_message	Content	否*	新消息（与 invocation_id 至少提供一个）
streaming	bool	否	是否流式，默认 false
state_delta	object	否	状态变更
invocation_id	string	否	用于恢复被中断的函数调用
function_call_event_id	string	否	用于恢复长时间运行的函数

*如果未启用 --auto_create_session，session_id 对应的会话必须已存在，否则返回 404。

响应： Event[] 事件数组

[
  {
    "id": "852208bf-c883-4c44-99ef-68303e0ef4be",
    "author": "huiyu_agent",
    "content": {
      "role": "model",
      "parts": [
        {"text": "思考过程...", "thought": true},
        {"text": "你好！我是小慧，一个乐于助人的智能助手。"}
      ]
    },
    "partial": false,
    "finishReason": "STOP",
    "usageMetadata": {
      "promptTokenCount": 113,
      "candidatesTokenCount": 44,
      "totalTokenCount": 157
    },
    "timestamp": 1775556943.259,
    "invocationId": "e-90d3bd10-59e4-4403-a5d4-2a822dff0bd0"
  }
]

Event 关键字段说明：

字段	说明
content.parts[].thought	为 true 时表示该 part 是思考过程，非最终回复
content.parts[].text	文本内容
finishReason	结束原因：STOP（正常完成）、SAFETY（安全拦截）等
usageMetadata	Token 用量统计

提取最终回复的逻辑：

for event in response:
    parts = event.get("content", {}).get("parts", [])
    for part in parts:
        if part.get("text") and not part.get("thought"):
            final_reply = part["text"]

---

SSE 流式对话

运行 Agent（SSE 流式）

POST /run_sse

请求体： 与 /run 完全相同。

响应： text/event-stream

每个 SSE 事件格式：

data: {"id": "...", "content": {"role": "model", "parts": [{"text": "你"}]}, "partial": true}

data: {"id": "...", "content": {"role": "model", "parts": [{"text": "好"}]}, "partial": true}

data: {"id": "...", "content": {"role": "model", "parts": [{"text": "！"}]}, "partial": false, "finishReason": "STOP"}

错误事件格式：

data: {"error": "Session not found"}

客户端示例（Python）：

import requests

response = requests.post(
    "http://localhost:8000/run_sse",
    json={
        "app_name": "root_agent",
        "user_id": "user_001",
        "session_id": "my_session",
        "new_message": {"role": "user", "parts": [{"text": "你好"}]},
        "streaming": True,
    },
    stream=True,
)

for line in response.iter_lines():
    if line.startswith(b"data: "):
        data = json.loads(line[6:])
        if "error" in data:
            print(f"错误: {data['error']}")
            break
        parts = data.get("content", {}).get("parts", [])
        for part in parts:
            if part.get("text") and not part.get("thought"):
                print(part["text"], end="", flush=True)

客户端示例（JavaScript）：

const response = await fetch("http://localhost:8000/run_sse", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    app_name: "root_agent",
    user_id: "user_001",
    session_id: "my_session",
    new_message: { role: "user", parts: [{ text: "你好" }] },
    streaming: true,
  }),
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  const text = decoder.decode(value);
  for (const line of text.split("\n")) {
    if (line.startsWith("data: ")) {
      const data = JSON.parse(line.slice(6));
      if (data.error) { console.error(data.error); break; }
      const parts = data.content?.parts || [];
      for (const part of parts) {
        if (part.text && !part.thought) {
          process.stdout.write(part.text);
        }
      }
    }
  }
}

---

WebSocket 实时对话

连接地址

ws://<host>:<port>/run_live?app_name=root_agent&user_id=user_001&session_id=my_session

查询参数：

参数	类型	必填	说明
app_name	string	是	应用名称
user_id	string	是	用户 ID
session_id	string	是	会话 ID
modalities	string[]	否	模态类型，默认 ["AUDIO"]，可选 TEXT、AUDIO
proactive_audio	bool	否	是否启用主动音频
enable_affective_dialog	bool	否	是否启用情感对话
enable_session_resumption	bool	否	是否启用会话恢复

通信协议： 双向发送 JSON 文本消息，每条消息为 Event 对象的 JSON 序列化。

WebSocket 关闭码：

关闭码	说明
1002	会话不存在
1008	Origin 不被允许
1011	服务器内部错误

---

错误处理

所有接口在出错时返回统一的错误格式：

{"detail": "错误描述信息"}

HTTP 状态码	说明
400	请求参数错误
404	会话 / 应用不存在
422	请求体验证失败（Pydantic 校验错误）
500	服务器内部错误

---

代码示例

Python 完整集成示例

import requests

BASE_URL = "http://localhost:8000"
APP_NAME = "root_agent"


class HuiYuAgentClient:
    """慧遇 Agent API 客户端"""

    def __init__(self, base_url: str = BASE_URL, app_name: str = APP_NAME):
        self.base_url = base_url
        self.app_name = app_name

    def health_check(self) -> bool:
        """健康检查"""
        resp = requests.get(f"{self.base_url}/health")
        return resp.json().get("status") == "ok"

    def create_session(self, user_id: str, session_id: str = None) -> dict:
        """创建会话"""
        body = {}
        if session_id:
            body["session_id"] = session_id
        resp = requests.post(
            f"{self.base_url}/apps/{self.app_name}/users/{user_id}/sessions",
            json=body,
        )
        resp.raise_for_status()
        return resp.json()

    def list_sessions(self, user_id: str) -> list:
        """列出用户的所有会话"""
        resp = requests.get(
            f"{self.base_url}/apps/{self.app_name}/users/{user_id}/sessions",
        )
        resp.raise_for_status()
        return resp.json()

    def delete_session(self, user_id: str, session_id: str):
        """删除会话"""
        resp = requests.delete(
            f"{self.base_url}/apps/{self.app_name}/users/{user_id}/sessions/{session_id}",
        )
        resp.raise_for_status()

    def chat(self, user_id: str, session_id: str, message: str) -> str:
        """
        发送消息并获取回复（同步）

        Args:
            user_id: 用户 ID
            session_id: 会话 ID
            message: 用户消息文本

        Returns:
            Agent 的最终回复文本
        """
        resp = requests.post(
            f"{self.base_url}/run",
            json={
                "app_name": self.app_name,
                "user_id": user_id,
                "session_id": session_id,
                "new_message": {
                    "role": "user",
                    "parts": [{"text": message}],
                },
            },
        )
        resp.raise_for_status()
        events = resp.json()

        # 提取最终回复（跳过 thought 部分）
        for event in events:
            parts = event.get("content", {}).get("parts", [])
            for part in parts:
                if part.get("text") and not part.get("thought"):
                    return part["text"]

        return ""

    def chat_stream(self, user_id: str, session_id: str, message: str):
        """
        发送消息并流式获取回复（SSE）

        Yields:
            每次 yield 一个文本片段
        """
        resp = requests.post(
            f"{self.base_url}/run_sse",
            json={
                "app_name": self.app_name,
                "user_id": user_id,
                "session_id": session_id,
                "new_message": {
                    "role": "user",
                    "parts": [{"text": message}],
                },
                "streaming": True,
            },
            stream=True,
        )
        resp.raise_for_status()

        for line in resp.iter_lines():
            if line and line.startswith(b"data: "):
                import json
                data = json.loads(line[6:])
                if "error" in data:
                    raise Exception(data["error"])
                parts = data.get("content", {}).get("parts", [])
                for part in parts:
                    if part.get("text") and not part.get("thought"):
                        yield part["text"]


# ===== 使用示例 =====
if __name__ == "__main__":
    client = HuiYuAgentClient()

    # 1. 健康检查
    assert client.health_check(), "服务未就绪"
    print("✅ 服务正常")

    # 2. 创建会话
    session = client.create_session("user_001")
    session_id = session["id"]
    print(f"✅ 会话已创建: {session_id}")

    # 3. 同步对话
    reply = client.chat("user_001", session_id, "你好，我叫小明")
    print(f"👤 用户: 你好，我叫小明")
    print(f"🤖 Agent: {reply}")

    # 4. 上下文记忆测试
    reply = client.chat("user_001", session_id, "我叫什么名字？")
    print(f"👤 用户: 我叫什么名字？")
    print(f"🤖 Agent: {reply}")

    # 5. 流式对话
    print("👤 用户: 请用一句话介绍你自己")
    print("🤖 Agent: ", end="", flush=True)
    for chunk in client.chat_stream("user_001", session_id, "请用一句话介绍你自己"):
        print(chunk, end="", flush=True)
    print()

cURL 示例

# 创建会话
curl -s -X POST http://localhost:8000/apps/root_agent/users/user_001/sessions \
  -H "Content-Type: application/json" \
  -d '{}'

# 发送对话（使用返回的 session_id）
curl -s -X POST http://localhost:8000/run \
  -H "Content-Type: application/json" \
  -d '{
    "app_name": "root_agent",
    "user_id": "user_001",
    "session_id": "<替换为实际session_id>",
    "new_message": {"role": "user", "parts": [{"text": "你好"}]}
  }'

# 查看会话列表
curl -s http://localhost:8000/apps/root_agent/users/user_001/sessions

# 删除会话
curl -s -X DELETE http://localhost:8000/apps/root_agent/users/user_001/sessions/<session_id>