企业官网建设流程全解析

1. 从一次工具调用超时说起：MCP 的 stdio 与 SSE 到底怎么选

MCP（Model Context Protocol）是让大模型调用外部工具的协议层，你可以把它理解成"给模型装插件的 USB 接口"。它规定了模型怎么发现工具、怎么传参、怎么拿回结果。而 stdio 和 SSE 是 MCP 客户端连接服务端的两种通信方式：stdio 走标准输入输出，把服务端当子进程拉起；SSE 走 HTTP 长连接，服务端是独立进程，客户端通过 URL 连过去。

我最早做多工具 Agent 时，三个服务端全用 stdio，本地跑得好好的，一放到内网服务器上就出问题——客户端进程一挂，所有工具全断，日志还散在四个终端里。后来把天气、SQL、Python 执行三个服务端改成 SSE 独立部署，客户端只负责连 URL，稳定性立刻不一样。但反过来，纯本地做代码补全、文件读写这类工具，stdio 又比 SSE 省事得多，不用管端口、不用管进程存活。

所以这篇不空谈概念，直接给你两套可复制的客户端骨架（client_stdio.py和client_sse.py），配上三个真实服务端（天气、SQL、Python 执行），再演示启动、连通性验证、以及怎么在两种方式之间切换。适合正在搭 MCP 工具链、纠结通信方式、或者被"连不上/工具列表为空"卡住的开发者。核心检索词就三个：MCP、stdio、SSE，下面全部围绕它们展开。

2. 前置准备：TaoToken 统一 Key 与 API 通道

不管用哪种通信方式，客户端最终都要调 LLM 来做 Function Calling。这里我用 TaoToken 作为统一的模型接入通道，好处是一个 Key 走通所有模型，不用在多个平台之间切来切去。

先去控制台拿 Key：

• 控制台入口：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
• API Keys 管理：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite

拿到 Key 之后，API 基地址用https://taotoken.net/api（注意这个地址不加 UTM 参数，直接写进代码）。如果你要长期跑编码类 Agent，可以看下 Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite

环境依赖用requirements.txt固定版本，避免mcp和openai版本打架：

mcp==1.6.0 openai==1.74.0 httpx==0.28.1 httpx-sse==0.4.0 pymysql==1.1.1 pandas==2.2.3 python-dotenv==1.1.0 uvicorn==0.34.1 sse-starlette==2.2.1

.env文件统一放模型配置，两种通信方式共用同一份：

BASE_URL=https://taotoken.net/api MODEL=qwen2.5-coder:14b OPENAI_API_KEY=你的TaoTokenKey

注意：BASE_URL结尾不要多加/v1，OpenAI SDK 会自己拼路径。我试过写成https://taotoken.net/api/v1，结果 404，排查了半小时。

3. 可复制配置：stdio 与 SSE 两套客户端骨架

3.1 stdio 客户端：把服务端当子进程拉起

stdio 的核心是StdioServerParameters，它告诉客户端用什么命令启动服务端脚本。Python 脚本用python，JS 脚本用node。

import asyncio import os import json from typing import Dict from contextlib import AsyncExitStack from openai import OpenAI from dotenv import load_dotenv from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client load_dotenv() class MultiServerMCPClient: def __init__(self): self.exit_stack = AsyncExitStack() self.client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("BASE_URL") ) self.model = os.getenv("MODEL") self.sessions: Dict[str, ClientSession] = {} self.all_tools = [] async def _start_one_server(self, script_path: str) -> ClientSession: is_python = script_path.endswith(".py") is_js = script_path.endswith(".js") if not (is_python or is_js): raise ValueError("服务端脚本必须是 .py 或 .js 文件") command = "python" if is_python else "node" server_params = StdioServerParameters( command=command, args=[script_path], env=None ) stdio_transport = await self.exit_stack.enter_async_context( stdio_client(server_params) ) read_stream, write_stream = stdio_transport session = await self.exit_stack.enter_async_context( ClientSession(read_stream, write_stream) ) await session.initialize() return session async def connect_to_servers(self, servers: dict): for server_name, script_path in servers.items(): session = await self._start_one_server(script_path) self.sessions[server_name] = session resp = await session.list_tools() for tool in resp.tools: function_name = f"{server_name}_{tool.name}" self.all_tools.append({ "type": "function", "function": { "name": function_name, "description": tool.description, "parameters": tool.inputSchema } }) print("\n已连接到下列服务端:") for name in servers: print(f" - {name}: {servers[name]}") print("\n汇总的工具:") for t in self.all_tools: print(f" - {t['function']['name']}") async def _call_mcp_tool(self, tool_full_name: str, tool_args: dict) -> str: parts = tool_full_name.split("_", 1) if len(parts) != 2: return f"无效的工具名称: {tool_full_name}" server_name, tool_name = parts session = self.sessions.get(server_name) if not session: return f"找不到服务端: {server_name}" resp = await session.call_tool(tool_name, tool_args) return resp.content if resp.content else "工具执行无输出" async def chat_loop(self): print("\n多服务端 MCP 客户端已启动，输入 quit 退出。") messages = [] while True: query = input("\n你: ").strip() if query.lower() == "quit": break messages.append({"role": "user", "content": query}) messages = messages[-20:] response = self.client.chat.completions.create( model=self.model, messages=messages, tools=self.all_tools ) if response.choices[0].finish_reason == "tool_calls": tool_call = response.choices[0].message.tool_calls[0] tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) print(f"\n[调用工具: {tool_name}, 参数: {tool_args}]\n") result = await self._call_mcp_tool(tool_name, tool_args) messages.append(response.choices[0].message.model_dump()) messages.append({ "role": "tool", "content": result, "tool_call_id": tool_call.id }) response = self.client.chat.completions.create( model=self.model, messages=messages ) print(f"\nAI: {response.choices[0].message.content}") async def cleanup(self): await self.exit_stack.aclose() async def main(): servers = { "weather": "weather_server.py", "SQLServer": "sql_server.py", "PythonServer": "python_server.py" } client = MultiServerMCPClient() try: await client.connect_to_servers(servers) await client.chat_loop() finally: await client.cleanup() if __name__ == "__main__": asyncio.run(main())

3.2 SSE 客户端：连 URL 而不是拉进程

SSE 客户端把stdio_client换成sse_client，参数从脚本路径变成 URL。其余逻辑完全一致。

from mcp.client.sse import sse_client async def _start_one_server(self, server_url: str) -> ClientSession: sse_transport = await self.exit_stack.enter_async_context( sse_client(url=server_url) ) read_stream, write_stream = sse_transport session = await self.exit_stack.enter_async_context( ClientSession(read_stream, write_stream) ) await session.initialize() return session async def main(): servers = { "weather": "http://127.0.0.1:8001/sse", "PythonServer": "http://127.0.0.1:8002/sse", "SQLServer": "http://127.0.0.1:8003/sse" } client = MultiServerMCPClient() try: await client.connect_to_servers(servers) await client.chat_loop() finally: await client.cleanup()

3.3 服务端切换：一行注释决定通信方式

三个服务端脚本（weather_server.py、sql_server.py、python_server.py）的末尾都长这样，切换通信方式只改一行：

if __name__ == "__main__": # 以标准 I/O 方式运行 MCP 服务端 # mcp.run(transport='stdio') # 使用 SSE 方式运行 MCP 服务端 mcp.run(transport='sse')

用FastMCP初始化时指定端口，SSE 模式才会监听对应端口：

from mcp.server.fastmcp import FastMCP mcp = FastMCP("WeatherServer", port=8001)

4. 启动与连通性验证：两种方式完整动作

4.1 SSE 模式启动

每个服务端单独开一个终端，用uv run拉起：

uv run weather_server.py uv run python_server.py uv run sql_server.py

看到类似Uvicorn running on http://127.0.0.1:8001就说明 SSE 服务端起来了。然后另开终端跑客户端：

uv run client_sse.py

客户端启动后会打印已连接的服务端和汇总工具列表。如果工具列表为空，说明 SSE 连接没建立成功，先检查端口和 URL 路径（/sse不能少）。

4.2 stdio 模式启动

stdio 不需要单独启动服务端，客户端会自己拉起子进程。直接跑：

uv run client_stdio.py

启动日志里会看到三个服务端被依次拉起，工具列表同样会打印出来。这里有个坑：stdio 模式下服务端脚本里的mcp.run(transport='sse')必须注释掉，否则子进程会去抢端口，客户端反而连不上。

4.3 连通性验证：发一条真实请求

不管哪种模式，启动后输入一句需要调工具的话，比如：

你: 帮我查一下南昌的天气

正常输出会先打印[调用工具: weather_query_weather, 参数: {...}]，然后返回格式化后的天气文本。如果只返回模型自己的话、没有工具调用日志，说明all_tools是空的，回到上一步检查工具列表。

再验证 SQL 工具：

你: 查一下 school 数据库里 students 表有多少行

Python 执行工具：

你: 用 Python 算一下 1 到 100 的和

三个工具都能触发，说明 MCP 链路完全通了。

5. 本篇常见错排查

5.1 工具列表为空 / 连接超时

SSE 模式下最常见。先确认服务端真的在监听：curl http://127.0.0.1:8001/sse应该返回事件流而不是拒绝连接。如果服务端在另一台机器上，把127.0.0.1换成实际内网 IP，并确认防火墙放行。stdio 模式下工具为空，多半是command写错，比如系统里只有python3没有python，改成python3即可。

5.2 端口冲突

SSE 模式三个服务端端口必须不同（8001/8002/8003）。如果报Address already in use，用lsof -i:8001找到占用进程杀掉，或者改FastMCP初始化时的port参数。

5.3 Function Calling 参数格式报错

MCP 返回的inputSchema字段名和 OpenAI 要求的parameters不一致。如果你直接透传，OpenAI 会报Invalid schema。上面代码里我做了字段映射，把inputSchema转成parameters，并保留type、properties、required三个字段。这一步不能省。

5.4 模型不触发工具调用

有些模型对 Function Calling 支持不完整，会直接把工具描述当普通文本回复。换一个明确支持 tools 参数的模型即可。在 TaoToken 的模型对话页可以先手动验证模型是否支持工具调用：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite

5.5 stdio 子进程残留

客户端异常退出时，stdio 拉起的子进程可能没被回收。AsyncExitStack的aclose()负责清理，确保finally里调用了cleanup()。如果还是残留，手动pkill -f server.py。

6. 选型结论与接入入口

回到最初的问题：本地工具链、单机开发、工具数量少且生命周期跟客户端绑定，用 stdio，省去端口和进程管理；远程服务、多客户端共享、需要独立部署和监控，用 SSE，服务端和客户端解耦。两者可以混用——比如天气服务走 SSE 远程部署，文件读写走 stdio 本地拉起，客户端里同时维护两种 session 即可。

接入所需的 Key 和文档入口：

• API Keys：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
• 接入文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
• 模型对话验证：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite

最后留一个实操建议：先把三个服务端都用 SSE 跑起来，确认工具列表和调用链路没问题，再把其中不需要远程共享的改成 stdio。这样排障时变量最少，不会一上来就被"到底是通信方式问题还是工具本身问题"绕进去。