MCP 网络研讨会演示

这个项目展示了 模型上下文协议（MCP） 一种将大型语言模型（如GPT）与多个工具服务器连接的实际实现架构。

架构概述

                         [USER]
                            |
                            | (1. chat input/output)
                            |
              +-------------v-------------+
              |   MAIN.PY (Entry Point)   |
              +-----------+---------------+
                          |
                          | (2. orchestrates)
                          |
         +----------------+----------------+
         |                                 |
         v                                 v
  +-------------+                  +---------------+
  |  LLM AGENT  |  (3. OpenAI) -> |   OPENAI API  |
  | (agent.py)  |   str:
        # Add user message to history
        self.messages.append({"role": "user", "content": user_input})

        # Call GPT with available tools
        response = self.client.chat.completions.create(
            model=self.model,
            messages=self.messages,
            tools=openai_tools,  # MCP tools in OpenAI format
            tool_choice="auto"
        )

        # Multi-turn orchestration (up to 10 iterations)
        while response_message.tool_calls:
            # Execute each tool call via MCP Host
            result = await self.mcp_host.call_tool(function_name, function_args)

            # Add result and get next response
            next_response = self.client.chat.completions.create(...)

主要特点：

多轮编排在多个迭代中自动串联工具调用
对话记忆保持对话的完整上下文
智能工具选择GPT根据用户意图决定使用哪个工具

把它看作是大脑负责思考需要做什么，并协调执行过程。

______________________________________________________________________

2. MCP 主机（`mcp_host.py`)

它是什么连接管理器，用于在LLM Agent与MCP服务器之间建立桥梁。

职责:

管理与多个MCP服务器的连接 ClientSession
从（指定位置）加载配置 mcp_config.json
生成MCP服务器进程（使用stdio传输）
发现并聚合来自所有连接服务器的工具
将工具执行请求路由到适当的服务器
将MCP工具转换为OpenAI函数调用格式

密钥代码 (mcp_host.py:16-26): （这行代码表示的是文件 mcp_host.py 中的第16到26行）

class MCPHost:
    def __init__(self):
        self.sessions = {}           # server_name -> ClientSession
        self.exit_stack = AsyncExitStack()  # manage connections for proper closure/cleanup
        self.tools = []              # All available tools
        self.tool_to_server = {}     # Maps tools to their servers

关键方法：

connect_to_servers()连接到配置中所有已启用的服务器
call_tool()将路由工具调用转发到正确的服务器
get_tools_for_openai()将MCP工具转换为OpenAI格式

可以将其视为管理所有MCP服务器连接并路由请求的连接中心。

______________________________________________________________________

3. MCP客户端（位于MCP主机内部）

这是什么连接该组件的部件 TO MCP服务器（非独立服务）。

它栖息的地方里面 mcp_host.py as ClientSession 物体。

职责:

为每个MCP服务器创建一个会话
使用（某种方法/工具）生成服务器进程 command + args 来自 mcp_config.json
管理标准输入/输出通信通道
发送工具执行请求
接收并返回工具结果

密钥代码 (mcp_host.py:88-102)：

async def connect_to_server(self, server_name: str, command: str, args: list[str], env: dict = None):
    # Create transport connection to server
    stdio_transport = await self.exit_stack.enter_async_context(
        stdio_client(server_params)
    )

    # Create client session for this server
    stdio, write = stdio_transport
    session = await self.exit_stack.enter_async_context(
        ClientSession(stdio, write)
    )

    await session.initialize()  # Handshake with server

配置 (来自 mcp_config.json):

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/directory"],
      "env": {},
      "description": "Access files on your local system",
      "enabled": true
    },
    "task-manager": {
      "command": "python",
      "args": ["./task_manager_server.py"],
      "env": {},
      "description": "Our custom task manager - built by us!",
      "enabled": true
    }
  }
}

可以将其视为为每个工具提供商提供独立的电话线路，负责连接和沟通管理。

在这个项目中，所有服务器都使用 stdio 传输 (stdin/stdout)，但MCP支持如HTTP等不同的传输机制，用于远程服务器。

______________________________________________________________________

4. MCP服务器（例如。， `task_manager_server.py`）

它是什么一项提供特定工具、资源和功能的独立服务。

职责：

实现了MCP协议（标准化通信）
公开具有定义模式的工具、资源和提示
当客户端请求时执行工具逻辑
通过资源提供上下文数据
返回MCP格式的结果

密钥代码 (task_manager_server.py:31-71): （文件 task_manager_server.py 的第31行到第71行）

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """List available tools"""
    return [
        types.Tool(
            name="create_task",
            description="Create a new task with a description and priority",
            inputSchema={
                "type": "object",
                "properties": {
                    "description": {"type": "string"},
                    "priority": {"type": "string", "enum": ["low", "medium", "high"]}
                },
                "required": ["description", "priority"]
            }
        ),
        types.Tool(
            name="list_tasks",
            description="List all tasks"
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict | None = None) -> list[types.TextContent]:
    """Execute a tool"""
    if name == "create_task":
        task = Task(
            description=arguments["description"],
            priority=arguments["priority"]
        )
        tasks.append(task)
        return [types.TextContent(type="text", text=f"Task created: {task}")]

关键特性：

模式定义工具具有明确定义的输入/输出模式
类型安全MCP强制执行结构化通信
隔离每台服务器独立运行

把它想象成一种提供特定功能并遵循MCP通信协议的专业工作者。

______________________________________________________________________

它们如何协同工作

1. 初始化 (main.py):

# Load config, connect to servers, initialize agent
host = MCPHost()
await host.connect_to_servers(config, os.environ)
agent = LLMAgent(mcp_host=host, api_key=api_key)

2. 用户输入 （main.py → agent.py):

user_input = input("You: ")
response = await agent.chat(user_input)

3. 工具发现 (agent.py ← mcp_host.py):

# Agent gets all available tools from MCP Host
openai_tools = self.mcp_host.get_tools_for_openai()

# Calls GPT with tools as function calling options
response = self.client.chat.completions.create(
    model=self.model,
    messages=self.messages,
    tools=openai_tools,
    tool_choice="auto"
)

4. 工具执行 (agent.py → mcp_host.py → MCP 服务器)：

# Agent executes tool via MCP Host
result = await self.mcp_host.call_tool(function_name, function_args)

# MCP Host routes to correct server
session = self.sessions[server_name]
result = await session.call_tool(tool_name, arguments)

5. 多轮协调（或“多回合编排”） (agent.py)：

# Continue until no more tool calls needed
for iteration in range(max_iterations):
    if not response_message.tool_calls:
        break
    # Execute tools and get next response

示例流程：

User: "Create a task to buy groceries with high priority"
  ↓
LLM Agent: Analyzes input, decides to use "create_task" tool
  ↓
MCP Host: Routes to task-manager server
  ↓
Task Manager Server: Creates task, returns confirmation
  ↓
LLM Agent: Formats response for user
  ↓
Output: "I've created a high priority task: 'buy groceries'"

______________________________________________________________________

关键概念

传输层

MCP支持两种主要的传输机制：

stdio（标准输入/输出）：

- 用于本地服务器（如我们的Python任务管理器） - 通过标准输入/输出流进行通信 - 非常适合生成本地进程

HTTP（超文本传输协议）:

- 用于远程服务器 - 标准的HTTP请求/响应 - 可以包含用于流传输的服务器发送事件（Server-Sent Events）

标准IO配置示例：

{
  "command": "python",
  "args": ["./task_manager_server.py"]
}

传输与协议：

交通数据如何传输（stdio，HTTP）
协议；规程；议定书使用的是什么数据格式（MCP消息）

可以把这想象成邮件投递：

交通卡车、飞机、自行车（信件传递的方式）
协议信封格式，地址书写规范（信件的外观）

MCP可以使用不同的传输方式，但始终采用相同的协议格式：

stdio 传输本地服务器通过标准输入/输出使用JSON-RPC
HTTP传输通过HTTP的JSON-RPC用于远程服务器
SSE（服务器发送事件）通过HTTP实现的具有流处理能力的JSON-RPC
可流式传输的HTTP传输使用可选的服务器发送事件（Server-Sent Events）进行HTTP POST请求，适用于远程服务器

这种分离使得MCP能够在不同的通信通道上工作，同时保持协议的一致性。

______________________________________________________________________

MCP术语：主机、客户端和会话

核心定义：

MCP 主机您的申请（该 MCPHost 班级里 mcp_host.py)
MCP 客户端客户端库/包（mcp) 负责处理MCP协议通信的
客户端会话与特定MCP服务器的会话连接

建筑：

MCP Host (your application)
└── MCP Client (the mcp package/library)
    ├── ClientSession #1 → MCP Server #1 (filesystem)
    ├── ClientSession #2 → MCP Server #2 (task manager)
    └── ClientSession #3 → MCP Server #3 (airbnb)

在实践中：

MCP客户端是 mcp 你导入的包：

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# Create transport and session for each server
stdio_transport = await stdio_client(server_params)
session = ClientSession(stdio, write)

# Store in your host
self.sessions = {}  # server_name -> ClientSession

每个 ClientSession 连接到一个特定的MCP服务器。

______________________________________________________________________

连接管理

MCP 主机管理所有连接，但将职责委派出去：

MCP 主机管理：

OpenAI API连接（用于LLM决策）
用户输入输出（聊天界面）
通过清理生命周期 AsyncExitStack

MCP 客户端（ClientSession）管理：

通过stdio进行的单个服务器连接
工具向服务器发送执行请求

class MCPHost:
    def __init__(self):
        self.exit_stack = AsyncExitStack()  # Manages cleanup
        self.sessions = {}                   # Stores ClientSessions

______________________________________________________________________

AsyncExitStack：连接生命周期管理

它是什么： AsyncExitStack() 管理着 生命周期和清理 异步连接的。

它的功能/作用:

建立连接 - 在异步上下文管理器创建时进行注册
关闭连接 - 完成时确保妥善清理

代码示例（mcp_host.py:48-50, 109）：

self.exit_stack = AsyncExitStack()

# Register the transport for cleanup
stdio_transport = await self.exit_stack.enter_async_context(
    stdio_client(server_params)
)

# Register the session for cleanup
session = await self.exit_stack.enter_async_context(
    ClientSession(stdio, write)
)

# Later: Close ALL registered resources at once
async def cleanup(self):
    await self.exit_stack.aclose()  # ← Closes everything

为什么需要：

没有它，你将需要手动关闭每一个 stdio_client 运输并且每个 ClientSession
即使发生异常，也保证清理工作
防止资源泄漏（僵尸进程、打开的文件描述符）

重要区别:

exit_stack 管理连接 在清理/关闭方面
self.sessions 管理连接 在路由和使用方面

______________________________________________________________________

MCP 客户端-服务器通信模型

一个MCP客户端 ↔ 一个MCP服务器（直接1:1通信）

每个MCP客户端都会建立一个 直接、专用的连接 使用单一的MCP服务器。这在配对组件之间创建了一个直接的通信通道。

在这个项目中：

# Each ClientSession connects to exactly ONE server
self.sessions = {
    "filesystem": ClientSession(...),  # 1:1 with filesystem server
    "todo": ClientSession(...),        # 1:1 with task manager server
    "airbnb": ClientSession(...)       # 1:1 with airbnb server
}

要点：

每个 ClientSession 管理与唯一一个MCP服务器的通信
MCP主机协调多个1:1客户端-服务器对
服务器之间无法直接通信
所有的协调工作都通过主持人进行

MCP协议原语：

MCP定义了两类基本元素：

服务器暴露的原始数据/基本功能 （服务器提供给客户端的内容）：

工具 - AI应用程序可以调用的可执行函数以执行操作
资源 - 提供上下文信息的数据源（例如，文件内容、API数据）
提示 - 常用任务的可重用交互模板

客户端暴露的原语 （客户端提供给服务器端的内容）：

抽样 - 服务器可以通过其连接的客户端（经用户批准后）请求大型语言模型（LLM）完成任务
诱导（或引出） - 服务器可以通过客户端按需向用户请求特定信息
记录（日志） - 服务器可以向客户端发送诊断信息，用于调试和监控

所有功能均保持人工监督和安全控制。

______________________________________________________________________

MCP 对战 OpenAI 客户端

方面	MCP 客户端	OpenAI 客户端
目的连接到工具服务器	连接到LLM API
协议根据提供的信息，翻译如下：
**模型上下文协议（MCP）	OpenAI API**
角色 **	“手” - 执行工具	“脑” - 做出决策	**	`from mcp import ClientSession` 进口 `from openai import OpenAI`
**	**	`session.call_tool("read_file", {...})` 示例 `client.chat.completions.create(...)`

______________________________________________________________________

|

运行演示程序

pip install mcp openai python-dotenv
npm install -g @modelcontextprotocol/server-filesystem @openbnb/mcp-server-airbnb

先决条件

export OPENAI_API_KEY='your-key-here'

安装依赖项： mcp_config.json设置OpenAI API密钥：

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/your/directory"],
      "env": {},
      "description": "Access files on your local system",
      "enabled": true
    },
    "task-manager": {
      "command": "python",
      "args": ["./task_manager_server.py"],
      "env": {},
      "description": "Our custom task manager - built by us!",
      "enabled": true
    },
    "airbnb": {
      "command": "npx",
      "args": ["-y", "@openbnb/mcp-server-airbnb@latest", "--ignore-robots-txt"],
      "env": {},
      "description": "Search Airbnb listings and properties",
      "enabled": true
    }
  }
}

在服务器中进行配置

python main.py

:

You: Give me the best hotels in Paris on airbnb
You: Create a text file and ingest the summary inside with the hotels sorted by rating
You: Create a task with high priority to visit the best place
You: Show me the list of tasks available

______________________________________________________________________

跑

.
├── main.py                  # Entry point and orchestration
├── agent.py                 # LLM Agent implementation
├── mcp_host.py              # MCP Host + Client implementation
├── task_manager_server.py   # Custom MCP Server (todo)
├── mcp_config.json          # Server configuration
└── README.md                # This file

______________________________________________________________________

演示查询

项目结构

摘要这个演示展示了具有多层结构的完整MCP（多层控制协议/模型等，具体含义根据上下文确定）实现
main.py（主程序文件）协调整个应用程序的入口点
LLM Agent（agent.py）使用GPT-4o-mini进行决策并协调多轮工具执行的AI推理组件
MCP 主机（mcp_host.py）连接管理器，用于在LLM代理与MCP服务器之间建立桥梁
MCP 客户端（ClientSession）主机内部的协议处理器，通过标准输入输出（stdio）与服务器通信

MCP 服务器提供功能暴露的工具提供商（文件系统、任务管理器、Airbnb）

关键架构要点：这个（或“该”）
大型语言模型代理（LLM Agent） 是大脑决定使用哪种工具 这个（或“它”）
MCP 主机 管理与工具服务器的连接 这个（或：该）
MCP 客户端
(ClientSession) 处理协议通信

______________________________________________________________________

每个服务器作为一个独立的进程运行，通过标准输入/输出进行通信

多轮编排使AI能够自动串联多个工具调用

额外学习资源要了解更多关于模型上下文协议的信息并探索社区实现情况：官方MCP文档：
https://modelcontextprotocol.io/docs/入门指南/简介- 全面指南：理解和实施MCP（多协议转换/管理控制协议/等，具体含义需根据上下文确定）：

MCP Webinar Demo

快速接入

详细介绍

MCP 网络研讨会演示

目录

架构概述

2. MCP 主机（`mcp_host.py`)

3. MCP客户端（位于MCP主机内部）

4. MCP服务器（例如。， `task_manager_server.py`）

它们如何协同工作

关键概念

传输层

MCP术语：主机、客户端和会话

连接管理

AsyncExitStack：连接生命周期管理

MCP 客户端-服务器通信模型

MCP 对战 OpenAI 客户端

|

|

在服务器中进行配置

:

跑

演示查询

每个服务器作为一个独立的进程运行，通过标准输入/输出进行通信

目录标签

接入字段

权限和风险

来源信息

继续浏览同类 MCP

MCP Webinar Demo

快速接入

详细介绍

MCP 网络研讨会演示

目录

架构概述

2. MCP 主机（mcp_host.py)

3. MCP客户端（位于MCP主机内部）

4. MCP服务器（例如。， task_manager_server.py）

它们如何协同工作

关键概念

传输层

MCP术语：主机、客户端和会话

连接管理

AsyncExitStack：连接生命周期管理

MCP 客户端-服务器通信模型

MCP 对战 OpenAI 客户端

|

|

在服务器中进行配置

:

跑

演示查询

每个服务器作为一个独立的进程运行，通过标准输入/输出进行通信

目录标签

接入字段

权限和风险

来源信息

继续浏览同类 MCP

2. MCP 主机（`mcp_host.py`)

4. MCP服务器（例如。， `task_manager_server.py`）