MCP Host

一个生产级 MCP (Model Context Protocol) Host 实现，基于 Python，支持 Function Calling 协议。

📋 目录

项目简介
快速开始
目录结构
代码模块说明
如何使用
MCP 配置详解
嵌入到您的应用
可视化演示
测试
常见问题

项目简介

什么是 MCP Host？

MCP Host 是一个中间件，它连接 LLM（如 GPT、Claude 等）和 MCP Server（工具服务器）。

简单来说：

您的应用 → LLM（判断需要调用什么工具）→ MCP Host（执行工具调用）→ MCP Server（实际执行）

架构图

┌─────────────────────────────────────────────────────────────┐
│                     您的应用程序                              │
│  ┌───────────────────────────────────────────────────────┐  │
│  │                   LLM (GPT等)                       │  │
│  │  接收用户消息 → 判断需要调用什么工具                    │  │
│  └───────────────────────┬───────────────────────────────┘  │
│                          │                                   │
│  ┌───────────────────────▼───────────────────────────────┐  │
│  │              MCPHost (本项目)                          │  │
│  │  • 管理多个 MCP Server 连接                            │  │
│  │  • 接收 LLM 的工具调用请求                              │  │
│  │  • 路由到正确的 Server 执行                             │  │
│  │  • 返回结果给 LLM                                      │  │
│  └───┬──────────┬──────────┬──────────────────────────────┘  │
│      │          │          │                                  │
│  ┌───▼───┐  ┌──▼───┐  ┌──▼────┐                              │
│  │Server1│  │Server2│  │Server3│  ← MCP Server (外部工具)    │
│  └───────┘  └───────┘  └───────┘                              │
└─────────────────────────────────────────────────────────────┘

核心功能

✅ 多服务器管理 - 同时连接和管理多个 MCP Server
✅ Function Calling - 无缝对接 LLM 的工具调用协议
✅ 异步优先 - 全异步实现，支持并发调用
✅ 事件系统 - 连接状态变更、工具变更等事件通知
✅ 错误处理 - 完善的错误处理和超时控制
✅ 零核心依赖 - 仅使用 Python 标准库
✅ 可视化演示 - 内置 Streamlit Web 界面

快速开始

1. 克隆项目

git clone https://github.com/beidaomitu233/MCPHost-python.git
cd MCPHost-python

2. 安装依赖

# 基础安装
pip install -e .

# 包含演示应用
pip install -e ".[demo]"

# 包含开发工具
pip install -e ".[dev]"

3. 验证安装

# 运行测试
python -m mcp_host.demo.test_mcp_connection

目录结构

mcp-host/
├── mcp_host/               # 核心库
│   ├── __init__.py         # 包入口，导出常用类
│   ├── host.py             # MCPHost 主类
│   ├── models.py           # 数据模型定义
│   ├── protocol.py         # 协议处理层
│   ├── transport.py        # 传输层 (stdio/sse)
│   └── client.py           # 简化客户端
│   └── demo/               # 演示和测试
│       ├── demo_app.py     # Streamlit Web界面
│       ├── demo_server.py  # 示例MCP服务器
│       ├── example.py      # 使用示例
│       ├── test_mcp_connection.py  # 连接测试
│       ├── test_mcp_host.py        # 完整测试
│       └── mcp_settings.json       # 配置文件
├── README.md               # 中文文档
├── README_EN.md            # 英文文档
└── pyproject.toml          # 项目配置

代码模块说明

各模块详细说明

模块	用途	使用场景
`host.py`	核心 Host 类	管理所有服务器连接的主要入口
`models.py`	数据模型	配置、类型、结果等数据结构
`transport.py`	传输层	与 MCP Server 的底层通信
`protocol.py`	协议层	MCP 协议的具体实现
`client.py`	简化客户端	封装 MCPHost 的便捷使用方式

1. `models.py` - 数据模型

定义了所有数据结构：

类名	用途	使用场景
`MCPServerConfig`	服务器配置	配置如何连接一个 MCP Server
`ToolDefinition`	工具定义	描述一个工具的名称、参数、描述
`ToolCallResult`	工具调用结果	封装工具调用的返回结果
`ConnectionStatus`	连接状态枚举	表示服务器的连接状态

2. `transport.py` - 传输层

负责与 MCP Server 的底层通信：

类名	用途
`Transport`	传输层抽象基类
`StdioTransport`	通过标准输入输出通信（最常用）
`SSETransport`	通过 Server-Sent Events 通信

3. `protocol.py` - 协议层

实现 MCP 协议的逻辑：

类名	用途
`MCPServer`	封装单个服务器的完整交互流程
`FunctionCallHandler`	处理 Function Calling 协议

4. `host.py` - 核心 Host 类

这是您主要使用的类：

类/方法	用途
`MCPHost`	主入口，管理所有服务器连接
`add_server(config)`	添加并连接一个服务器
`remove_server(name)`	移除一个服务器
`get_function_definitions()`	获取工具定义（发送给 LLM）
`handle_function_call(name, args)`	执行工具调用
`shutdown()`	关闭所有连接

如何使用

推荐方式：使用 MCPClient（只需配置 JSON 文件）

这是最简单的方式。你只需要编写一个 mcp_settings.json 配置文件，其余全部自动处理。

第 1 步：创建配置文件 mcp_settings.json

{
    "mcpServers": {
        "python-os-monitor": {
            "command": "python",
            "args": ["-m", "mcp_host.demo_server"],
            "timeout": 60.0,
            "env": {"PYTHONIOENCODING": "utf-8"}
        },
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
        }
    }
}

第 2 步：在代码中使用

from mcp_host import MCPClient

# 一行代码初始化（自动读取配置、连接所有服务器）
client = MCPClient.from_settings("mcp_settings.json")

# 获取工具定义（发送给 LLM）
tools = client.get_tools()
print(f"可用工具: {[t['function']['name'] for t in tools]}")

# 调用工具
result = client.call_tool("get_system_info")
print(result.to_text())

# 用完关闭
client.close()

与 LLM 集成的完整示例

import json
from mcp_host import MCPClient
from openai import OpenAI

# 1. 初始化 MCP Client（自动从配置文件加载所有服务器）
mcp = MCPClient.from_settings("mcp_settings.json")

# 2. 获取工具定义
tools = mcp.get_tools()

# 3. 调用 LLM
client = OpenAI()
messages = [{"role": "user", "content": "帮我查看系统信息"}]

response = client.chat.completions.create(
    model="gpt-4",
    messages=messages,
    tools=tools  # 传入工具定义
)

# 4. 处理 LLM 返回的工具调用
if response.choices[0].message.tool_calls:
    for tc in response.choices[0].message.tool_calls:
        # 执行工具调用
        result = mcp.call_tool(
            tc.function.name,
            json.loads(tc.function.arguments)
        )
        
        # 将结果返回给 LLM
        messages.append({
            "role": "tool",
            "tool_call_id": tc.id,
            "content": result.to_text()
        })
    
    # 5. 让 LLM 根据工具结果生成最终回复
    final_response = client.chat.completions.create(
        model="gpt-4",
        messages=messages
    )
    print(final_response.choices[0].message.content)

mcp.close()

直接使用 MCPHost

import asyncio
from mcp_host import MCPHost
from mcp_host.models import MCPServerConfig

async def main():
    # 创建 Host
    host = MCPHost(default_timeout=60.0)
    
    # 添加服务器
    config = MCPServerConfig(
        name="python-os-monitor",
        command="python",
        args=["-m", "mcp_host.demo_server"],
        env={"PYTHONIOENCODING": "utf-8"},
    )
    await host.add_server(config)
    
    # 获取工具定义
    tools = host.get_function_definitions()
    
    # 处理工具调用
    result = await host.handle_function_call("get_system_info", {})
    print(result.to_text())
    
    # 清理
    await host.shutdown()

asyncio.run(main())

MCP 配置详解

配置格式

{
  "mcpServers": {
    "python-os-monitor": {
      "command": "python",
      "args": ["-m", "mcp_host.demo_server"],
      "disabled": false,
      "type": "stdio",
      "timeout": 60.0,
      "env": {
        "PYTHONIOENCODING": "utf-8"
      }
    }
  }
}

每个字段的含义

字段	类型	说明	示例
`mcpServers`	object	顶层对象，包含所有服务器配置	-
`python-os-monitor`	string	服务器唯一标识名，可自定义	`"my-server"`
`command`	string	启动命令，指定如何运行服务器	`"python"`, `"node"`, `"npx"`
`args`	array	命令行参数，传递给command的参数列表	`["-m", "mcp_host.demo_server"]`
`disabled`	boolean	是否禁用此服务器	`false`
`type`	string	传输类型，`stdio`（标准输入输出）或 `sse`（服务器发送事件）	`"stdio"`
`timeout`	number	请求超时时间（秒）	`60.0`
`env`	object	环境变量，传递给子进程	`{"PYTHONIOENCODING": "utf-8"}`

为什么MCP Tools这样设计？

Stdio传输模式：通过标准输入输出通信，服务器作为子进程运行，安全性高，不需要额外的网络端口
JSON-RPC 2.0协议：标准化的请求响应格式，与各种语言库兼容
多服务器支持：可同时连接多个MCP服务器，工具自动路由到正确的服务器
超时控制：防止单点故障影响整体服务，避免无限等待

嵌入到您的应用

方式一：直接嵌入

将 mcp_host/ 文件夹复制到您的项目中：

YourProject/
├── your_app/
│   ├── __init__.py
│   └── main.py          # 您的主应用
├── mcp_host/           # 直接复制过来的文件夹
│   ├── __init__.py
│   ├── models.py
│   ├── host.py
│   └── ...
└── requirements.txt

然后在您的代码中直接导入使用：

# your_app/main.py
from mcp_host import MCPHost
from mcp_host.models import MCPServerConfig

方式二：作为依赖安装

# 在您的项目目录下
pip install /path/to/MCPHost-python

然后在 requirements.txt 中添加：

git+https://github.com/beidaomitu233/MCPHost-python.git

可视化演示

启动演示

pip install streamlit
cd demo
streamlit run demo_app.py

演示功能

连接配置
- 内置经过测试的配置，一键连接测试
- 编辑自定义配置
- 查看服务器状态
工具测试
- 选择可用工具
- 输入参数
- 查看调用结果
- 查看执行时间
AI 对话
- 配置自定义API
- 与AI进行对话
- AI自动调用工具
事件日志
- 实时显示操作日志
- 按级别过滤日志

测试

运行连接测试

python -m mcp_host.demo.test_mcp_connection

运行完整测试

python -m pytest mcp_host/demo/test_mcp_host.py -v

测试覆盖

测试项	测试数量	状态
初始化 Host	5	✅
添加服务器	3	✅
注册后状态验证	4	✅
Function Calling 定义	5	✅
工具调用	4	✅
错误处理	1	✅
获取工具定义	2	✅
获取服务器实例	3	✅
服务器状态	2	✅
移除服务器	3	✅
重新添加服务器	2	✅
事件系统	1	✅
资源访问	1	✅
清理资源	2	✅

常见问题

Q: MCP Server 是什么？

A: MCP Server 是提供工具的程序，它遵循 MCP 协议。例如 weather-server 提供天气查询工具，filesystem-server 提供文件操作工具。

Q: 我怎么知道 MCP Server 提供哪些工具？

A: 连接后调用 host.get_tool_definitions() 即可获取所有可用工具列表。

Q: 支持哪些 LLM？

A: 任何支持 Function Calling / Tool Use 的 LLM 都可以对接，包括 OpenAI GPT-4、Anthropic Claude 等。

Q: 可以同时连接多个 Server 吗？

A: 可以！使用 host.add_server(config) 添加任意多个服务器。

Q: Streamlit 工具调用失败怎么办？

A: Demo 应用采用每次调用重新连接的设计，确保稳定性。如果仍有问题，请先在「连接配置」选项卡中测试连接。

License

MIT License

Name		Name	Last commit message	Last commit date
Latest commit History 4 Commits
mcp_host		mcp_host
tests		tests
.gitignore		.gitignore
README.md		README.md
README_EN.md		README_EN.md
pyproject.toml		pyproject.toml

Folders and files

Latest commit

History

Repository files navigation

MCP Host

📋 目录

项目简介

什么是 MCP Host？

架构图

核心功能

快速开始

1. 克隆项目

2. 安装依赖

3. 验证安装

目录结构

代码模块说明

各模块详细说明

1. models.py - 数据模型

2. transport.py - 传输层

3. protocol.py - 协议层

4. host.py - 核心 Host 类

如何使用

推荐方式：使用 MCPClient（只需配置 JSON 文件）

与 LLM 集成的完整示例

直接使用 MCPHost

MCP 配置详解

配置格式

每个字段的含义

为什么MCP Tools这样设计？

嵌入到您的应用

方式一：直接嵌入

方式二：作为依赖安装

可视化演示

启动演示

演示功能

测试

运行连接测试

运行完整测试

测试覆盖

常见问题

Q: MCP Server 是什么？

Q: 我怎么知道 MCP Server 提供哪些工具？

Q: 支持哪些 LLM？

Q: 可以同时连接多个 Server 吗？

Q: Streamlit 工具调用失败怎么办？

License

About

Resources

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

1. `models.py` - 数据模型

2. `transport.py` - 传输层

3. `protocol.py` - 协议层

4. `host.py` - 核心 Host 类

Packages