Skip to main content
Glama
README.md
# Python MCP + Multi-Agent 完整入门与实战指南

这份教程通过一个“待办事项 MCP Server”,从零演示如何使用 Python 构建 MCP,
并进一步展示如何让 Manager、Reader、Writer 三个 Agent 分工协作:

- 注册 Tool、Resource 和 Prompt;
- 编写 MCP Client 并调用 Server;
- 使用 STDIO 运行本地 Server;
- 把 Server 注册到 Codex;
- 使用 `Agent.as_tool()` 组织 Multi-Agent;
- 给不同 Agent 分配不同的 MCP 工具;
- 排查连接、Schema 和工具调用问题。

项目使用官方 MCP Python SDK v2。

---

## 0. 三条命令跑通

```bash
cd /Users/eric_zcz/Desktop/MCP
uv sync
uv run python src/client.py
```

成功后会依次看到:

```text
1. connected / negotiated
2. tools/list
3. tools/call add_todo
4. tools/call list_todos
5. tools/call complete_todo
6. resources/list
7. resources/read
8. prompts/list
9. prompts/get
```

这说明协议协商、工具发现、工具调用、资源读取和 Prompt 获取都已完成。

接着运行不需要 API Key 的 Multi-Agent 权限检查:

```bash
uv run python src/multiagent_demo.py --check
```

应该看到:

```text
Todo Manager 可调用: ask_todo_reader, ask_todo_writer
Todo Reader 可调用: list_todos
Todo Writer 可调用: add_todo, complete_todo
权限隔离检查通过
```

完整的独立运行说明也可以查看 [`MULTI_AGENT_DEMO.md`](MULTI_AGENT_DEMO.md)。

---

## 1. 项目结构

```text
MCP/
├── .codex/
│   └── config.toml       # 向 Codex 注册 MCP Server
├── src/
│   ├── server.py         # 注册并实现 MCP 能力
│   ├── client.py         # 发现并调用 MCP 能力
│   └── multiagent_demo.py # 不同 Agent 使用不同工具
├── pyproject.toml        # Python 和依赖声明
├── uv.lock               # 精确依赖版本
├── MULTI_AGENT_DEMO.md   # Multi-Agent 运行指南
└── README.md             # MCP 基础教程
```

主要文件:

- [`src/server.py`](src/server.py):MCP Server;
- [`src/client.py`](src/client.py):独立测试 Client;
- [`src/multiagent_demo.py`](src/multiagent_demo.py):Manager、Reader、Writer 工具隔离;
- [`MULTI_AGENT_DEMO.md`](MULTI_AGENT_DEMO.md):Multi-Agent 使用指南;
- [`.codex/config.toml`](.codex/config.toml):Codex 项目级配置。

---

## Multi-Agent 架构速览

这个示例使用一个 Manager 和两个 Specialist:

```text
用户自然语言
    │
    ▼
Todo Manager
    ├── ask_todo_reader ──> Todo Reader
    │                         └── list_todos
    │
    └── ask_todo_writer ──> Todo Writer
                              ├── add_todo
                              └── complete_todo
                                      │
                                      ▼
                               共享的 MCP Client
                                      │ STDIO
                                      ▼
                                src/server.py
```

Manager 不直接调用待办工具。它先理解用户意图,再把查询任务交给 Reader,
把添加或完成任务交给 Writer。

### 每个 Agent 只获得需要的工具

Reader 只有查询工具:

```python
reader_agent = Agent(
    name="Todo Reader",
    instructions="你是只读待办查询 Agent。你只能使用 list_todos。",
    tools=[list_todos],
)
```

Writer 只有修改工具:

```python
writer_agent = Agent(
    name="Todo Writer",
    instructions="你负责添加和完成待办。",
    tools=[add_todo, complete_todo],
)
```

真正限制能力的是 `tools=[...]`,而不只是提示词。Reader 收到的工具 Schema 中没有
`add_todo` 和 `complete_todo`,因此无法生成可执行的写入调用。

### 使用 `as_tool()` 让 Manager 调用 Specialist

`as_tool()` 是 OpenAI Agents SDK 中 `Agent` 对象的方法。它把一个 Agent 包装成另一个
Agent 可以调用的工具:

```python
reader_as_tool = reader_agent.as_tool(
    tool_name="ask_todo_reader",
    tool_description="查询待办事项,只能进行只读操作。",
)

writer_as_tool = writer_agent.as_tool(
    tool_name="ask_todo_writer",
    tool_description="添加或完成待办事项。",
)

manager_agent = Agent(
    name="Todo Manager",
    tools=[writer_as_tool, reader_as_tool],
)
```

因此 Manager 能看到的是两个 Specialist 工具,而不是底层的三个待办工具。

### Agent Tool 和 MCP Tool 的关系

示例中的包装函数把 Agent SDK 与 MCP 连接起来:

```python
@function_tool
async def list_todos(status: str = "all") -> str:
    result = await client.call_tool(
        "list_todos",
        {"status": status},
    )
    return mcp_result_to_text(result)
```

完整调用链是:

```text
LLM 选择 list_todos
        ↓
Agent SDK 执行 @function_tool 包装函数
        ↓
client.call_tool("list_todos", ...)
        ↓
MCP Client 通过 STDIO 发送 tools/call
        ↓
Todo MCP Server 执行真正的 list_todos
        ↓
结果返回给 Specialist,再返回给 Manager
```

这里有三层不同的“工具”:

| 所在层 | 能看到的工具 |
|---|---|
| Manager | `ask_todo_reader`、`ask_todo_writer` |
| Reader / Writer | 按角色分配的 Agent SDK Function Tool |
| MCP Server | `list_todos`、`add_todo`、`complete_todo` 的真正实现 |

### 运行完整 Multi-Agent

先设置 OpenAI API Key:

```bash
export OPENAI_API_KEY="你的 API Key"
```

然后输入自然语言任务:

```bash
uv run python src/multiagent_demo.py \
  "添加两个待办:学习 MCP、测试 Multi-Agent,然后列出所有未完成事项"
```

执行这个复合任务时,Manager 会先调用 Writer 完成两次写入,再调用 Reader 查询,
最后由 Manager 汇总答案。Reader 和 Writer 共享同一个 MCP Client 和同一个 STDIO
Server 进程,因此本次运行中的待办数据也是共享的。

Multi-Agent 解决“谁负责什么”,MCP 解决“Agent 如何用统一协议调用外部能力”:

```text
Agent / as_tool()  ── 负责编排和分工
MCP                ── 负责工具发现、Schema 和调用协议
STDIO              ── 负责本项目 Client 与 Server 之间的消息传输
```

生产环境仍然必须在 MCP Server 内校验用户身份、租户和数据权限;Agent 的工具列表是
能力隔离的一部分,不能代替服务端授权。

---

## 2. MCP 的整体结构

MCP(Model Context Protocol)是让 AI 应用连接外部工具和上下文的标准协议。

```text
用户
  │
  ▼
Codex(Host,内部包含 MCP Client)
  │
  │ STDIO / MCP 消息
  ▼
src/server.py(MCP Server)
  ├── Tools
  ├── Resource
  └── Prompt
```

四个关键角色:

- **Host**:承载模型和用户界面,例如 Codex;
- **Client**:连接 Server,发现和调用它的能力;
- **Server**:执行真实业务逻辑;
- **Transport**:传输 MCP 消息,本项目使用 STDIO。

模型不会直接执行 Python 函数。完整过程是:

1. Client 从 Server 获取工具名称、描述和 JSON Schema;
2. 模型判断是否需要调用工具并生成参数;
3. Client 把工具名和参数发送给 Server;
4. Server 验证参数并执行 Python 函数;
5. 执行结果返回给模型;
6. 模型根据结果回答用户。

---

## 3. “注册”有两层含义

这是 MCP 最容易混淆的概念。

### 3.1 向 Server 注册能力

在 `src/server.py` 中:

```python
@mcp.tool()
def add_todo(...):
    ...


@mcp.resource("todo://all")
def all_todos():
    ...


@mcp.prompt()
def plan_my_day(...):
    ...
```

这些装饰器告诉 MCP Server 自己能提供什么。

### 3.2 向 Codex 注册 Server

在 `.codex/config.toml` 中:

```toml
[mcp_servers.todo_demo]
command = "/Users/eric_zcz/Desktop/MCP/.venv/bin/python"
args = ["/Users/eric_zcz/Desktop/MCP/src/server.py"]
cwd = "/Users/eric_zcz/Desktop/MCP"
enabled = true
```

这告诉 Codex 如何启动并连接整个 Server。

简单记忆:

```text
@mcp.tool()               工具 → Server
[mcp_servers.todo_demo]   Server → Codex
```

---

## 4. 创建 Python MCP Server

代码位于 [`src/server.py`](src/server.py)。

### 4.1 导入 SDK

```python
from mcp import types
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
```

- `MCPServer` 创建高层 MCP Server;
- `types` 提供 MCP 标准数据类型;
- `ToolError` 表示可安全返回给模型的业务错误。

很多旧教程使用 `FastMCP`,这是早期 SDK 写法。本项目使用 Python SDK v2,因此使用 `MCPServer`。

### 4.2 用 Python 类型生成 Schema

```python
TodoTitle = Annotated[
    str,
    StringConstraints(
        strip_whitespace=True,
        min_length=1,
        max_length=200,
    ),
]

TodoId = Annotated[int, Field(gt=0)]
TodoFilter = Literal["all", "open", "done"]
```

SDK 会根据类型生成 JSON Schema:

- 标题必须是 1~200 个字符;
- ID 必须是正整数;
- `status` 只能是 `all`、`open` 或 `done`。

非法参数会在业务函数运行前被拒绝。

### 4.3 创建 Server

```python
mcp = MCPServer(
    name="todo-demo",
    version="1.0.0",
    instructions=(
        "Use list_todos before completing an item "
        "when its id is unknown."
    ),
)
```

- `name`:Server 名称;
- `version`:Server 实现版本;
- `instructions`:跨工具工作流和约束。

例如“不知道 ID 时先调用 list_todos”属于跨工具规则,适合写进 `instructions`。

---

## 5. 注册 Tool

Tool 是模型能够选择调用的动作。

```python
@mcp.tool(
    title="Add todo",
    annotations=types.ToolAnnotations(
        read_only_hint=False,
        destructive_hint=False,
        idempotent_hint=False,
        open_world_hint=False,
    ),
)
def add_todo(title: TodoTitle) -> TodoResult:
    """Create a new todo item and return it."""
    ...
```

SDK 自动提取:

- 函数名 → 工具名 `add_todo`;
- docstring → 工具描述;
- 参数类型 → 输入 Schema;
- 返回类型 → 输出 Schema。

工具注解的含义:

- `read_only_hint=False`:会修改状态;
- `destructive_hint=False`:不会删除或覆盖重要数据;
- `idempotent_hint=False`:重复调用会创建多个事项;
- `open_world_hint=False`:不会访问开放的外部世界。

这些只是给 Host 的行为提示,不是安全授权。真实权限检查必须由 Server 完成。

### 5.1 执行业务逻辑

```python
todo = {
    "id": next_id,
    "title": title,
    "status": "open",
    "created_at": datetime.now(UTC).isoformat(),
}

todos.append(todo)
return {"todo": todo}
```

结构化返回值会变成:

- `content`:模型可读取的内容块;
- `structuredContent`:程序可读取的结构化结果。

### 5.2 返回业务错误

```python
raise ToolError(f"Todo {id} does not exist.")
```

这表示工具已经正常执行,但业务条件不满足。模型可以读取错误并调整下一步操作。

---

## 6. 注册 Resource

Resource 是由 URI 标识的只读上下文:

```python
@mcp.resource(
    "todo://all",
    name="all-todos",
    title="All todos",
    mime_type="application/json",
)
def all_todos() -> str:
    return json.dumps(todos, ensure_ascii=False, indent=2)
```

Client 读取:

```python
snapshot = await client.read_resource("todo://all")
```

Resource 适合提供文件内容、数据库 Schema、项目文档、配置或状态快照。

---

## 7. 注册 Prompt

Prompt 是可复用的消息模板:

```python
@mcp.prompt(
    name="plan_my_day",
    title="Plan my day",
)
def plan_my_day(focus: str = "") -> str:
    instruction = "Call list_todos with status=open, then make a plan"
    return f"{instruction} prioritizing {focus}."
```

Client 获取填充后的 Prompt:

```python
prompt = await client.get_prompt(
    "plan_my_day",
    {"focus": "先理解协议,再配置客户端"},
)
```

Prompt 只生成消息,本身不会执行 `list_todos`。

| 能力 | 用途 | 谁决定使用 |
|---|---|---|
| Tool | 执行查询或操作 | 模型 |
| Resource | 提供只读上下文 | Host/应用 |
| Prompt | 生成可复用消息 | 用户或 Host |

---

## 8. 使用 STDIO 启动 Server

Server 文件最后是:

```python
if __name__ == "__main__":
    mcp.run()
```

`mcp.run()` 默认使用 STDIO:

```text
Host → Server stdin:MCP 请求
Server stdout → Host:MCP 响应
```

单独运行:

```bash
.venv/bin/python src/server.py
```

程序看起来会一直等待,这是正常的:它正在等 Client 发送请求。按 `Ctrl+C` 退出。

STDIO 的 stdout 是协议通道。日志应使用 Python `logging` 并写到 stderr,避免破坏协议数据。

---

## 9. MCP Client 如何调用

代码位于 [`src/client.py`](src/client.py)。

### 9.1 声明 Server 启动参数

```python
server = StdioServerParameters(
    command=sys.executable,
    args=[str(SERVER_PATH)],
    cwd=PROJECT_ROOT,
)
```

`sys.executable` 确保子进程使用当前虚拟环境的 Python。

### 9.2 连接

```python
async with Client(server) as client:
    ...
```

进入 `async with` 时,Client 会启动 Server、建立 STDIO 通道并协商协议版本和能力;退出时自动关闭连接和子进程。

### 9.3 发现工具

```python
tools = await client.list_tools()
```

返回内容包括名称、描述、输入 Schema、输出 Schema 和工具注解。模型正是根据这些信息决定如何调用。

### 9.4 调用工具

```python
result = await client.call_tool(
    "add_todo",
    {"title": "学习 MCP"},
)
```

调用流程:

```text
Client 发送 name + arguments
          ↓
Server 验证 inputSchema
          ↓
执行 Python 函数
          ↓
验证 outputSchema
          ↓
返回 content + structuredContent
```

`src/client.py` 不调用大模型,它用于独立验证 MCP Server。接入 Codex 后,发现和调用过程由 Codex 内部完成。

---

## 10. 注册到 Codex

项目已经配置 [`.codex/config.toml`](.codex/config.toml):

```toml
[mcp_servers.todo_demo]
command = "/Users/eric_zcz/Desktop/MCP/.venv/bin/python"
args = ["/Users/eric_zcz/Desktop/MCP/src/server.py"]
cwd = "/Users/eric_zcz/Desktop/MCP"
enabled = true
startup_timeout_sec = 20
tool_timeout_sec = 30
```

操作步骤:

1. 用 Codex 打开 `/Users/eric_zcz/Desktop/MCP`;
2. 信任项目;
3. 执行一次 `uv sync`,确保 `.venv` 存在;
4. 重启 Codex 或 IDE extension;
5. 在 Codex TUI 输入 `/mcp`;
6. 确认 `todo_demo` 已连接并显示三个工具。

也可以使用 CLI 注册:

```bash
codex mcp add todo_demo -- /Users/eric_zcz/Desktop/MCP/.venv/bin/python /Users/eric_zcz/Desktop/MCP/src/server.py
codex mcp list
```

项目配置和 CLI 注册选择一种即可,避免同名重复注册。

---

## 11. 在 Codex 中触发工具

连接成功后输入:

```text
请使用 todo_demo 添加两个待办事项:
1. 学习 MCP
2. 测试 Python MCP Server
然后列出全部未完成事项。
```

预期调用:

```text
add_todo(title="学习 MCP")
add_todo(title="测试 Python MCP Server")
list_todos(status="open")
```

再尝试:

```text
请先查询当前待办事项,然后把“学习 MCP”标记为完成。
```

预期先调用 `list_todos` 找到 ID,再调用 `complete_todo`。

---

## 12. 使用 MCP Inspector

Inspector 可以绕过 Codex,直接查看和调用 Server:

```bash
uv run mcp dev src/server.py
```

打开页面后:

1. 连接 Server;
2. 打开 Tools;
3. 选择 `add_todo`;
4. 输入:

```json
{
  "title": "从 Inspector 添加"
}
```

还可以检查自动生成的 Schema、读取 `todo://all`、获取 `plan_my_day`,以及测试非法参数。

---

## 13. 常见问题

### 13.1 找不到 `mcp` 模块

说明没有使用项目虚拟环境:

```bash
uv sync
uv run python src/client.py
```

### 13.2 Server 启动后没有输出

这是正常的。STDIO Server 正在等待 Client。运行 `uv run python src/client.py` 查看完整流程。

### 13.3 Codex 看不到 `todo_demo`

依次检查:

1. 项目是否被信任;
2. 是否执行过 `uv sync`;
3. `.venv/bin/python` 是否存在;
4. `.codex/config.toml` 是否使用正确的绝对路径;
5. 修改配置后是否重启 Codex;
6. `/mcp` 是否显示连接错误。

### 13.4 Server 一连接就断开

先执行:

```bash
.venv/bin/python -m compileall -q src
.venv/bin/python src/client.py
```

如果 Client 能运行,通常是 Codex 配置或项目信任问题;如果 Client 也失败,则先处理 Python 异常。

### 13.5 工具存在但模型不调用

检查 docstring、参数名称、Schema 和 Server instructions 是否清晰。开发阶段可以明确要求:

```text
请使用 todo_demo 的 list_todos 工具查询未完成事项。
```

### 13.6 重启后数据消失

本例的数据保存在内存列表中。Server 进程结束后数据会重置。真实项目可替换成 SQLite、PostgreSQL、Redis 或外部 API,MCP 注册方式无需改变。

### 13.7 `FastMCP` 和 `MCPServer` 混用

旧版教程经常使用 `FastMCP`。本项目使用 SDK v2:

```python
from mcp.server import MCPServer
```

不要混用不同大版本的导入路径和 Client API。

---

## 14. 从 STDIO 改成远程 HTTP

本地运行使用:

```python
mcp.run()
```

远程开发测试可以改成:

```python
mcp.run(
    transport="streamable-http",
    host="127.0.0.1",
    port=8000,
)
```

Codex 配置改为:

```toml
[mcp_servers.todo_remote]
url = "https://example.com/mcp"
bearer_token_env_var = "TODO_MCP_TOKEN"
```

不要把 Token 直接提交到配置文件。生产环境还需考虑认证授权、TLS、限流、审计、持久化、超时和多用户隔离。

---

## 15. 推荐练习顺序

1. 修改 `add_todo` 的 docstring,观察 `tools/list`;
2. 修改 `TodoTitle` 长度限制,观察 JSON Schema;
3. 新增 `delete_todo` 工具;
4. 新增 `todo://open` Resource;
5. 给 Prompt 增加 `tone` 参数;
6. 把内存列表替换为 SQLite;
7. 最后改为 Streamable HTTP。

每次修改后验证:

```bash
.venv/bin/python -m compileall -q src
.venv/bin/python src/client.py
```

---

## 16. 最终心智模型

```text
Python 类型 + docstring
          ↓
MCP SDK 生成描述和 JSON Schema
          ↓
Client 发现 Tool / Resource / Prompt
          ↓
模型选择工具并生成参数
          ↓
Server 验证参数并调用 Python 函数
          ↓
结果返回给模型
```

核心边界:

- 模型负责判断和生成参数;
- Host/Client 负责协议连接;
- Server 负责验证、权限和真实业务;
- MCP 让三者使用统一格式通信。

## 参考资料

- [OpenAI 官方 Codex MCP 文档](https://learn.chatgpt.com/docs/extend/mcp)
- [官方 MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk)

TDQS

A4.2/5.0

Scored across 3 tools

Disambiguation5/5

Each tool maps to a distinct action on todos: create, list, and mark complete. There is no overlap or ambiguity between them.

Naming Consistency5/5

All tool names follow a clear verb_noun snake_case pattern (add_todo, list_todos, complete_todo). The singular/plural variation is natural and does not affect consistency.

Tool Count5/5

Three tools is a well-scoped size for a simple todo demo server. Each tool serves a clear purpose with no filler.

Completeness4/5

Core todo workflow is covered: create, list/filter, and complete. Minor gaps like deleting or editing todos exist, but they are not fatal for a demo-focused server.

Maintenance

ActivityMaintained
ResponsivenessNo issues