企业官网建设流程全解析

1. 为什么需要一个 MCP Host：从 Server 到对话闭环

上一章我们把 MCP Server 跑起来了，它能列出目录、读文件、统计行数，但如果你只是单独运行它，会发现它安安静静地待在那里，什么也不做。原因很简单：MCP Server 是“能力提供者”，它只负责响应请求，不会主动发起对话。真正驱动整个链路跑起来的，是坐在另一端的 MCP Host。

MCP Host 是什么？用一句话说，它是“能力消费者”，代表最终用户的意图去连接 Server、发现工具、发起调用、处理结果。你可以把它理解成一个翻译官：用户说“帮我看看 welcome.txt 里写了啥”，Host 把这句话翻译成 JSON-RPC 请求发给 Server，Server 执行完把结果返回，Host 再把结果翻译成人能看懂的内容。适合谁？适合已经写完 Server、想验证端到端链路是否通的开发者，也适合想理解 MCP 协议握手细节的同学。

这一章我会带你用 Python + mcp-sdk 从零搭一个最小可用的 MCP Host，通过 stdio 启动本地 Server 子进程，走 JSON-RPC 完成初始化、工具列表查询和对话调用。同时把模型调用通道统一到 TaoToken 的 API 上，用一个 Key 打通连接与对话，省去到处配不同厂商 Base URL 的麻烦。整个流程跑通后，你会亲眼看到 Host 和 Server 之间是怎么“对话”的。

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

在写 Host 代码之前，先把模型调用这一层理顺。MCP Host 本身负责协议通信，但如果你想让 Host 具备“对话”能力——比如把工具调用结果交给大模型做总结、或者让模型决定调用哪个工具——就需要一个稳定的模型 API 通道。TaoToken 在这里扮演的角色就是统一入口：一个 Key、一个 Base URL，兼容主流模型调用格式，不用为每个模型单独维护配置。

先拿到你的 API Key。访问 https://taotoken.net/api-keys 创建，建议按项目命名，方便后续区分。拿到后不要硬编码进代码，用环境变量管理：

export TAOTOKEN_API_KEY="sk-你的实际Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"

如果你用的是 Claude Code 这类工具，配置方式略有不同。Claude Code 的 settings 文件通常放在~/.claude/settings.json，需要写全三件套：Base URL、Key、Model ID。下面是一个可复制的片段，路径和字段名保持原样：

{ "env": { "ANTHROPIC_BASE_URL": "https://taotoken.net/api", "ANTHROPIC_AUTH_TOKEN": "sk-你的实际Key", "ANTHROPIC_MODEL": "claude-sonnet-4-20250514" } }

注意这里 Base URL 用的是https://taotoken.net/api，不要多加路径后缀，SDK 会自己拼接。Key 放在ANTHROPIC_AUTH_TOKEN字段里，Model ID 按你实际要用的模型填。如果你用的是 Cline 或 Roo Code 这类支持 MCP 的编辑器插件，配置项名称可能是baseUrl、apiKey、model，逻辑一样，把这三个值填对就行。

为什么要统一到 TaoToken？我试过在多个项目里分别配不同厂商的 Key，时间一长自己都记不清哪个 Key 对应哪个服务，换机器还要重新翻记录。统一到一个 Base URL 后，切换模型只需要改 Model ID 一个字段，Base URL 和 Key 不动，维护成本低很多。而且 TaoToken 的 API 通道兼容 OpenAI 和 Anthropic 两种调用格式，Host 里想用哪种 SDK 都行。

配置完成后，建议先用一个最简单的 curl 验证 Key 是否有效：

curl https://taotoken.net/api/v1/models \ -H "Authorization: Bearer $TAOTOKEN_API_KEY"

如果返回模型列表，说明 Key 和 Base URL 都没问题。这一步别跳过，后面 Host 报 401 的时候你会感谢自己提前验证过。

3. 可复制配置：Host 骨架代码与 stdio 启动

现在进入正题，写 Host 代码。核心思路是：Host 用asyncio.create_subprocess_shell启动 Server 子进程，接管它的 stdin/stdout/stderr，然后通过 stdin 写 JSON-RPC 请求、从 stdout 读响应。请求和响应靠id字段匹配，用一个字典存Future对象来实现异步等待。

先建文件。假设你上一章的 Server 在my-mcp-server/main.py，在同一个目录下创建host.py：

cd my-mcp-server touch host.py

下面是完整的 Host 骨架代码，可以直接复制。我把它拆成几个关键部分讲，但代码是完整的：

import asyncio import json import logging from typing import Dict, Any, Optional logging.basicConfig( level=logging.INFO, format='%(asctime)s - HOST - %(levelname)s - %(message)s' ) class StdioMcpHost: def __init__(self, server_command: str): self.server_command = server_command self.process: Optional[asyncio.subprocess.Process] = None self.reader: Optional[asyncio.StreamReader] = None self.writer: Optional[asyncio.StreamWriter] = None self._request_id_counter = 0 self._pending_requests: Dict[int, asyncio.Future] = {} async def start(self): logging.info(f"启动 Server 命令: '{self.server_command}'") self.process = await asyncio.create_subprocess_shell( self.server_command, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) self.reader = self.process.stdout self.writer = self.process.stdin asyncio.create_task(self._listen_to_server()) asyncio.create_task(self._listen_to_stderr()) logging.info("Host 已连接到 Server 的 stdio 管道") async def stop(self): if self.process and self.process.returncode is None: logging.info("正在关闭 Server...") self.writer.close() await self.writer.wait_closed() self.process.terminate() await self.process.wait() logging.info("Server 已停止") async def _listen_to_server(self): while self.process.returncode is None: try: line = await self.reader.readline() if not line: break response = json.loads(line.decode('utf-8')) request_id = response.get('id') if request_id in self._pending_requests: self._pending_requests.pop(request_id).set_result(response) else: logging.info(f"收到通知: {response}") except (json.JSONDecodeError, UnicodeDecodeError) as e: logging.error(f"解析响应失败: {line.strip()}, 错误: {e}") except Exception as e: logging.error(f"监听 Server 时出错: {e}", exc_info=True) break async def _listen_to_stderr(self): while self.process.returncode is None: line = await self.process.stderr.readline() if not line: break logging.warning(f"SERVER_STDERR: {line.decode('utf-8').strip()}") async def send_request(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]: self._request_id_counter += 1 request_id = self._request_id_counter request_obj = { "jsonrpc": "2.0", "id": request_id, "method": method, "params": params } future = asyncio.get_running_loop().create_future() self._pending_requests[request_id] = future request_line = json.dumps(request_obj) + '\n' self.writer.write(request_line.encode('utf-8')) await self.writer.drain() logging.info(f"发送请求: {request_line.strip()}") response = await asyncio.wait_for(future, timeout=10.0) return response

这段代码里最关键的是send_request和_listen_to_server的配合。send_request在发送前先创建一个Future并以id为键存进_pending_requests，相当于立了个凭证；_listen_to_server收到响应后按id找到对应的Future并set_result，唤醒等待的协程。这就是异步请求-响应匹配的核心机制。

接下来是交互式 CLI 和入口函数：

async def interactive_cli(host: StdioMcpHost): print("\n--- MCP Host 交互式 CLI ---") print("命令: ls <path>, cat <path>, countlines <path>, tools, exit") while True: try: user_input = await asyncio.to_thread(input, "> ") parts = user_input.strip().split() if not parts: continue command = parts[0].lower() if command == 'exit': break elif command == 'tools': response = await host.send_request("project/listTools", {}) print(json.dumps(response, indent=2)) elif command == 'ls' and len(parts) > 1: response = await host.send_request("fs/listDirectory", {"path": parts[1]}) print(json.dumps(response, indent=2)) elif command == 'cat' and len(parts) > 1: response = await host.send_request("fs/readFile", {"path": parts[1]}) if 'result' in response and 'content' in response['result']: import base64 decoded = base64.b64decode(response['result']['content']).decode('utf-8') print("--- 文件内容 ---") print(decoded) print("----------------") else: print(json.dumps(response, indent=2)) elif command == 'countlines' and len(parts) > 1: response = await host.send_request( "project/executeTool", {"name": "project/countLines", "parameters": {"path": parts[1]}} ) print(json.dumps(response, indent=2)) else: print(f"未知命令: '{user_input}'") except Exception as e: logging.error(f"CLI 出错: {e}", exc_info=True) async def main(): server_command = "python3 main.py" host = StdioMcpHost(server_command) try: await host.start() await asyncio.sleep(1) await interactive_cli(host) finally: await host.stop() if __name__ == "__main__": asyncio.run(main())

注意server_command这里写的是python3 main.py，前提是你的虚拟环境已激活，python3指向虚拟环境里的解释器。如果你用的是 Windows，改成python main.py。另外await asyncio.sleep(1)是给 Server 一点启动时间，避免 Host 发请求时 Server 还没准备好。

4. 验证请求：一次完整的连接与对话

代码写完了，现在跑起来验证。确保你在my-mcp-server目录下，虚拟环境已激活，然后执行：

python3 host.py

你应该会看到类似下面的输出：

2025-01-15 10:00:00,123 - HOST - INFO - 启动 Server 命令: 'python3 main.py' 2025-01-15 10:00:00,125 - HOST - WARNING - SERVER_STDERR: Server initialized with workspace root: /path/to/workspace 2025-01-15 10:00:00,127 - HOST - INFO - Host 已连接到 Server 的 stdio 管道 --- MCP Host 交互式 CLI --- 命令: ls <path>, cat <path>, countlines <path>, tools, exit >

看到SERVER_STDERR那行说明 Host 成功捕获到了 Server 的日志，通信管道是通的。现在依次输入命令验证。

先查工具列表：

> tools

响应应该包含project/countLines工具的定义，说明 JSON-RPC 的project/listTools方法调用成功。再列目录：

> ls .

响应里会列出welcome.txt和code_example.py。然后读文件：

> cat welcome.txt

你会看到文件内容被 Base64 解码后打印出来。最后调用自定义工具：

> countlines code_example.py

响应里lineCount字段会显示行数。到这里，一次完整的连接与对话就验证完了。整个过程走的是 stdio + JSON-RPC，Host 发请求、Server 响应，id字段一一对应。

如果你想让 Host 具备模型对话能力，可以在send_request拿到工具结果后，把结果拼进 prompt 发给 TaoToken 的 API。比如用 OpenAI 兼容格式：

import os from openai import OpenAI client = OpenAI( base_url=os.environ["TAOTOKEN_BASE_URL"], api_key=os.environ["TAOTOKEN_API_KEY"] ) def summarize_tool_result(tool_result: str) -> str: resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "你是一个工具结果总结助手。"}, {"role": "user", "content": f"请总结以下工具调用结果：\n{tool_result}"} ] ) return resp.choices[0].message.content

这样 Host 就不只是转发请求，还能对结果做二次加工，形成“调用工具 → 模型总结 → 返回用户”的完整闭环。Base URL 和 Key 都从环境变量读，换模型只改model字段。

5. 本篇常见错排查：401、local proxy failed 与 reading choices

跑通之后，我把踩过的坑整理一下，你遇到报错可以对照排查。

401 Unauthorized：最常见的原因是 Key 没传对或 Base URL 写错。检查TAOTOKEN_API_KEY环境变量是否真的导出成功，可以用echo $TAOTOKEN_API_KEY确认。如果用的是 Claude Code 的 settings.json，确认字段名是ANTHROPIC_AUTH_TOKEN而不是ANTHROPIC_API_KEY，这两个字段在不同版本里容易搞混。Base URL 必须是https://taotoken.net/api，不要写成https://taotoken.net/api/v1，SDK 会自己拼/v1，多写一层就 404 了。

local proxy failed：这个报错通常出现在 Host 启动 Server 子进程时，环境变量没继承过去。asyncio.create_subprocess_shell默认继承当前进程的环境变量，但如果你在代码里手动清了env参数，或者用了env={}，子进程就拿不到 Key。解决办法是显式传递：

import os self.process = await asyncio.create_subprocess_shell( self.server_command, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=os.environ.copy() )

reading choices 报错：如果你在 Host 里调模型 API，报'NoneType' object has no attribute 'choices'或者reading 'choices'，说明 API 返回体结构不对。先打印原始响应看看：

resp = client.chat.completions.create(...) print(resp.model_dump_json(indent=2))

常见原因是 Model ID 填错了，或者 Base URL 指向了不兼容的端点。确认 Model ID 是 TaoToken 支持的模型名，Base URL 用https://taotoken.net/api。

OAuth 相关报错：如果你用的是 Claude Code 或 Codex 这类带 OAuth 流程的工具，报 OAuth 错误通常是因为同时配了 OAuth 和 API Key，两者冲突。解决办法是只保留一种鉴权方式，用 API Key 就把 OAuth 相关配置删掉。Codex 的auth.json里如果同时有api_key和oauth_token，删掉oauth_token字段。

Server 启动超时：Host 发请求后 10 秒没响应，asyncio.wait_for会抛TimeoutError。先看SERVER_STDERR有没有报错，常见原因是 Server 脚本路径不对，或者虚拟环境没激活导致python3找不到依赖。把server_command改成绝对路径试试，比如/path/to/.venv/bin/python /path/to/main.py。

排查的时候记住一个原则：先确认 Server 单独能跑，再确认 Host 能连上，最后确认请求能通。分层排查比一股脑看日志快得多。

6. 继续往下走：把 Host 接进你的工作流

到这里，最小可用的 MCP Host 已经跑通了。你可以在这个骨架上继续加东西：比如把交互式 CLI 换成 Web 界面，或者把 Host 集成进你的编码助手，让它在需要读文件、统计代码行数时自动调用 MCP Server。核心的 stdio + JSON-RPC 通信机制不变，变的只是上层怎么用。

如果你想让 Host 具备更完整的模型对话能力，建议把模型调用统一到 TaoToken 的 API 通道上。一个 Key 管所有模型，Base URL 固定，切换模型只改 Model ID。配置文档在 https://taotoken.net/doc 有详细说明，API Key 在 https://taotoken.net/api-keys 创建。想先试试模型对话效果，可以直接用 https://taotoken.net/models 的在线对话验证。如果你打算长期做编码类 Agent，Coding Plan 会更划算，详情看 https://taotoken.net/coding-plan。

下一步我建议你试着把 Host 的send_request封装成一个工具注册表，让模型自己决定调用哪个工具。这样 Host 就从“手动敲命令”进化成“模型自主决策”，离真正的 Agent 就不远了。