1. 一条请求里要经过哪些「形态」 #

可以把整条链路看成多段管道，每一段输入输出类型不同：

阶段	输入形态	输出形态	主要代码位置
HTTP 入参	JSON 字节流	Pydantic 模型	FastAPI + schemas.AgentChatSendRequest
智能体绑定的 MCP	ORM / DB 行	纯 dict 列表	_mcp_service_dicts
历史消息	ORM 消息行列表	OpenAI Chat messages 列表	build_llm_messages_from_history
MCP 工具清单	SDK 对象 list_tools	内部统一 dict + OpenAI tools	_list_tools_for_service → build_openai_tools
OpenAI工具参数	MCP 工具里的 inputSchema	OpenAI Chat Completions 里 tools[].function.parameters	§5.0、build_openai_tools、_normalize_tool_schema
模型服务地址	配置字符串	OpenAI SDK 要求的 base_url	_openai_base_url
单次请求体	内存中的 req_payload dict	HTTP 请求（SDK 封装）	_openai_chat_kwargs + AsyncOpenAI
模型流式响应	SSE/JSON 流 chunk 对象	应用内事件 dict	_iter_chat_completion_events
工具调用增量	流式 tool_calls 片段	完整的 tool_calls 数组	_merge_stream_tool_calls
工具参数	模型输出的 JSON 字符串	dict 传给 MCP	_tool_call_args_dict
MCP 执行结果	call_tool 返回结构	单行/多行文本	_call_tool_for_service
推送给浏览器	Python dict	SSE 文本帧	_sse
会话标题（首条）	用户长文本	截断后的标题字符串	_build_session_title_from_question

下面按时间顺序展开每一段为什么要转换、规则是什么、常见坑在哪。

2. HTTP 请求体：JSON → Pydantic #

形态：客户端发送 Content-Type: application/json 的字节流（UTF-8）。

转换：FastAPI 将 body 解析为 Python 对象，再校验为 AgentChatSendRequest（字段含 content、enable_mcp_tools 等）。

要点：

• 校验通过后，路由里拿到的已是结构化对象；后续 payload.content.strip() 是在字符串层再规范化。
• 若网关/终端编码不是 UTF-8，解析可能失败；见项目里若存在的请求体编码中间件说明。

3. 数据库与 ORM：MCP 服务 → 纯字典列表 #

形态：Agent 上挂的是 MCP 服务 ID 列表；库中每条 MCP 服务是 ORM 行。

转换：_mcp_service_dicts 对每个 ID 调用 mcp_repository.get_mcp_service，拼成：

{"id": <int>, "name": "<str>", "protocol": "<str>", "config": { ... }}

为什么要变成 dict：

• 下层 generate_with_tools / _mcp_transport_streams 只依赖可序列化、可透传的配置（protocol、config），与 SQLAlchemy 会话解耦，避免把 ORM 实体传到异步 MCP 客户端里引发会话/线程问题。

细节：

• name/protocol 统一转成 str，config 缺省为 {}，减少 None 分支。

4. 历史消息：ORM 行 → OpenAI Chat messages #

函数：build_llm_messages_from_history(agent, message_rows)。

输入：

• agent：取 system_prompt。
• message_rows：库里的多行 AgentChatMessage（role、content、meta）。

输出：形如 OpenAI Chat Completions 所需的列表：

[
  {"role": "system", "content": "..."},
  {"role": "user", "content": "..."},
  {"role": "assistant", "content": "..."},
  {"role": "tool", "content": "...", "tool_call_id": "..."},  # 若 meta 里有
  ...
]

转换规则：

1. 首条固定为 system，内容来自 agent.system_prompt（strip）。
2. 只保留 role ∈ {user, assistant, tool}，其它跳过。
3. role == "tool" 时，从 meta 读 tool_call_id 写入消息体，供下一轮模型把工具结果对应到哪一次 function call。

语义：这是把「业务库里的聊天表」映射到「大模型厂商认识的对话格式」，是多轮 + 工具调用能否连贯的关键。

5. MCP 工具：SDK 对象 → 内部列表 → OpenAI tools #

5.0 对应总览表「OpenAI 工具参数」一行：inputSchema → parameters #

总览表里这一行概括的是：把 MCP 声明的工具入参格式，变成 Chat Completions 里 function.parameters 能接受的 JSON Schema。

一侧	字段 / 位置	含义
MCP（list_tools 返回的每个 Tool）	inputSchema（规范里常用驼峰；对象上也可能出现 input_schema）	描述该工具调用时参数对象的 JSON Schema（多为 type: "object" + properties + required 等）。
OpenAI Chat Completions	tools[].function.parameters	同样是一份 JSON Schema，描述模型应生成的 function.arguments（JSON 对象字符串）需符合的结构。

是否「同一种格式」：在协议意图上，两边都是 JSON Schema 描述参数对象，因此你的代码把 MCP 的 schema 直接塞进 OpenAI 的 parameters 槽位。命名上 MCP 叫 inputSchema，OpenAI 叫 parameters，只是 API 字段名不同，不是两套无关的类型。

在本项目里的落点（app/services/agent_chat.py）：

1. _list_tools_for_service 把 SDK 对象的 inputSchema / input_schema 读出来，放进内部统一结构的 input_schema 键。
2. build_openai_tools 构建每条 function 时写："parameters": _normalize_tool_schema(t.get("input_schema"))。

_normalize_tool_schema 做什么（与总览表「主要代码位置」一致）：

• 若 input_schema 已是 dict：原样返回，即对 OpenAI 透传 MCP 给出的 JSON Schema。
• 否则（None、非 dict 等）：退回 {"type": "object", "properties": {}}，避免向大模型 API 提交非法类型导致整次请求失败。

兼容性与坑：

• OpenAI 兼容的 JSON Schema 只是 子集；MCP 若使用复杂 $ref、不支持的关键字或过深嵌套，可能在请求阶段报错，需在服务端或 MCP 侧收紧 schema。
• MCP 未声明 schema 时可能得到 {}；透传后是否被 API 接受取决于供应商实现，必要时应在归一化里强制补全最小 object schema。

更细的函数级说明见下文 5.1～5.4（其中 5.4 对 _normalize_tool_schema 作实现层摘要）。

5.1 list_tools 结果 → 统一 dict #

函数：_list_tools_for_service(service)。

1. 按 protocol 建立传输（stdio / sse / streamable-http），ClientSession.initialize() 后 list_tools()。
2. 取 tools_result.tools（缺省 []）。
3. 每个 tool 压成：

{
  "name": str,
  "description": str,
  "input_schema": ...,  # inputSchema 或 input_schema，缺省可按后续归一化
}

形态意义：MCP SDK 可能返回带 inputSchema（驼峰）的对象；这里统一成后续构建 OpenAI 可用的字段名思路（input_schema 键名在本层 dict 中）。

5.2 内部 dict → OpenAI Tools 数组 #

函数：build_openai_tools(services)。

对每个工具生成一项：

{
  "type": "function",
  "function": {
    "name": "<暴露名>",
    "description": "<str>",
    "parameters": <JSON Schema dict>,
  },
}

同时维护 tool_mapping：暴露名 → {service, tool_name, service_name}，供真正 call_tool 时用真实 MCP 工具名（未归一化的原名）。

5.3 工具「暴露名」归一化 #

函数：_normalize_tool_name(service_name, tool_name)。

• 服务名与工具名转小写，非 [a-z0-9_-] 的字符压成 _，合并为 服务__工具，最长 64 字符。
• 原因：多个 MCP 服务可能有同名工具；OpenAI 侧 function.name 必须唯一，且只使用安全字符集。

5.4 parameters JSON Schema 归一化（实现摘要） #

函数：_normalize_tool_schema(schema)。

• 已是 dict → 原样作为 parameters（即 5.0 所述透传）。
• 否则 → {"type": "object", "properties": {}}，避免向模型提交非法 schema 导致请求失败。

概念与字段名对应关系见 5.0。

6. 模型服务端点：配置 URL → OpenAI SDK base_url #

函数：_openai_base_url(api_base_url)。

• 去掉末尾 /。
• 若尚不以 /v1 结尾（大小写不敏感），则追加 /v1。

原因：与 OpenAI 官方 Python SDK 约定一致：base_url 指向 API 根（例如 https://api.openai.com/v1），具体路径由 SDK 拼接 chat/completions，而不是把完整路径写进 base。

7. 单次补全请求：内存 dict → SDK 参数 → 网络协议 #

组装：generate_with_tools 里每轮构造：

req_payload = {"model": model_name, "messages": messages, "temperature": 0.3}
# 若有工具：
req_payload["tools"] = tools
req_payload["tool_choice"] = "auto"

转 SDK：_openai_chat_kwargs(payload, stream=True) 产出 chat.completions.create(**kwargs) 的关键字参数，包含 stream=True。

网络侧：HTTP +（通常）SSE 或 chunked 流；对应用代码而言已是 AsyncOpenAI 的异步流对象。

8. 流式响应：SDK chunk → 两种应用内事件 #

函数：_iter_chat_completion_events。

8.1 文本增量：chunk → {"type": "delta", "text": "..."} #

• 从每个 chunk 的 choices[0].delta.content 取字符串片段。
• 有内容则 yield 一条 delta，供路由包成 SSE 推给前端做打字机效果。

8.2 工具调用增量：chunk → 内存里累加 #

• delta.tool_calls 可能分多段到达（先 id、再 name、再 arguments 片段）。
• _merge_stream_tool_calls 按 index 槽位把字符串 += 拼到 function.name 和 function.arguments 上，得到与「非流式一整条 message」同构的 tool_calls 列表。

8.3 一轮结束：伪 Chat Completions 结构 #

流结束后组装：

msg = {"role": "assistant", "content": raw_content or None}
if tool_calls_by_index:
    msg["tool_calls"] = [...]
yield {"type": "__agent_chat_round_done__", "data": {"choices": [{"message": msg}]}}

为什么要伪装成 choices[0].message 形状：与 REST 非流式响应字段对齐，generate_with_tools 可以用同一套解析逻辑读「本轮助手消息 + 是否要带工具」。

9. 模型输出的 arguments 字符串 → Python dict（MCP 入参） #

函数：_tool_call_args_dict(args_text)。

• args_text 来自合并后的 function.arguments，应当是 JSON 对象字符串。
• json.loads 成功且为 dict → 作为 call_tool 的参数；失败或非 dict → {}。

注意：模型有时会输出残缺 JSON；此时 MCP 侧可能报错或得到空参，需要在产品层考虑重试或提示。

10. MCP call_tool 结果 → 单条 tool 角色消息文本 #

函数：_call_tool_for_service。

• 再次建立 transport + ClientSession，call_tool(tool_name, args)。
• 遍历返回的 content 列表，取每项的 text，非空则收集，最后用 \n 拼成一个字符串。

形态转换：MCP 多段内容（列表）→ Chat API 要求的 tool 消息 content 字符串（OpenAI 风格里 tool 消息内容是字符串）。

去向：generate_with_tools 把该字符串放入：

{"role": "tool", "tool_call_id": call_id, "content": result_text}

再进入下一轮 messages，让模型基于工具结果继续生成。

11. 应用内事件 dict → SSE 文本帧 #

函数：agent_chat.py 路由中的 _sse(data)。

• json.dumps(data, ensure_ascii=False)：中文不转义为 \uXXXX，便于前端与调试阅读。
• 前缀 data: + 负载 + \n\n：符合 Server-Sent Events 一帧格式。

响应头：StreamingResponse(..., media_type="text/event-stream")。

11.1 不向客户端转发的内部/收尾类型 #

• AGENT_CHAT_STREAM_RUN_COMPLETE（agent_chat_stream_run_complete）：路由用于取出 final_text 写库，不 yield _sse。
• 内部还有 __agent_chat_round_done__、__agent_chat_tool_done__：在 generate_with_tools 与 _iter_tool_run 中消费，不把每一种都暴露给前端（暴露的是 delta、tool_start、tool_end 等）。

12. 落库：字符串助手回复 #

流正常结束后，路由用累积的 reply（字符串）调用 create_message(..., "assistant", reply)，再走 touch_session。

形态：内存中的整段最终文本 → 数据库一条 assistant 消息（与流式传输中的多段 delta 不同；delta 仅用于在线展示）。

13. 会话标题（首条用户消息时）：长文本 → 短标题 #

函数：_build_session_title_from_question。

• 空白压缩为单词间单空格。
• 空 → "新对话"；否则最长 24 字符，超出加 ...。

这是对展示用字符串的规范化，不参与模型推理格式。

14. 多轮工具循环中的「消息列表」演变（概念） #

messages 在内存中的形态始终是 OpenAI Chat 消息数组，但每轮会追加不同结构：

1. 模型无工具：追加 {"role":"assistant","content": final_text}，结束。
2. 模型有工具：追加带 tool_calls 的 assistant，再对每个 call 追加 role: tool 行，然后进入下一轮请求。

格式约束：顺序上须满足厂商约定——先有带 tool_calls 的 assistant，再跟对应 tool_call_id 的 tool 消息。

15. 调试建议：从哪里打断点看「格式是否对」 #

1. build_llm_messages_from_history 返回值：system 是否在首条；tool 是否带对 tool_call_id。
2. build_openai_tools 的 tools 与 tool_mapping：暴露名是否唯一；parameters 是否为合法 JSON Schema。
3. _iter_chat_completion_events 最后一 yield：合并后的 msg 是否含完整 tool_calls。
4. _sse 输出：浏览器 Network 里是否连续收到 data: {...}\n\n。
5. 数据库：assistant 一条是否等于前端拼接 delta 后的全文（逻辑上应一致）。

16. 相关源文件索引 #

文件	职责
app/routers/agent_chat.py	SSE、会话/用户消息落库、组装 MCP dict、调用 generate_with_tools
app/services/agent_chat.py	OpenAI 流式、工具注册、MCP transport、工具执行、消息轮次
app/schemas.py	AgentChatSendRequest 等请求体形态