MCP To Agent Helper

MCP 工具助手

MCP 工具助手桥接器 FastMCP它与OpenAI的Agent框架结合，构建了一个工具生态系统。该系统能够发现MCP服务器上暴露的工具，并将这些工具封装为OpenAI的（工具/接口） FunctionTools，并将FastMCP的进度和日志事件向前传递，以便代理用户界面（UI）能够显示实时状态更新。这填补了代理框架中的一个空白，因为该框架本身并不原生地将工具状态流式传输回客户端。

特点/功能

• 自动发现MCP工具并将其作为OpenAI函数工具暴露出来
• 将 FastMCP 的进度和日志消息流式传输回您的代理运行时
• 提供工具元数据的可选缓存和缓存刷新功能
• 默认使用SSE传输，并且仍然支持Streamable HTTP或任何其他传输方式 fastmcp.infer_transport 理解
• 支持通过环境变量或显式头部传递承载令牌
• 为常见部署目标提供现成的传输工厂助手

安装

pip install fastmcp openai

克隆或复制 utility/mcp_tool_helper.py 将其集成到你的项目中。（计划发布一个打包版本。）

快速入门

from utility.mcp_tool_helper import MCPToolkit, UserContext

# Option A: connect to an SSE-enabled FastMCP server
toolkit = MCPToolkit(host="127.0.0.1", port=8010, sse_path="/sse/")

# Option B: launch tools via stdio (Python process)
# toolkit = MCPToolkit(
#     server_cmd=".venv/bin/python",
#     server_script="path/to/mcp_tools.py",
#     server_args=("--flag",),
# )

# Discover tools and hand them to an OpenAI agent
tools = toolkit.get_function_tools()

agent = Agent(
    name="Generate Course Content",
    instructions="You are a course builder...",
    tools=tools,
    model="gpt-5",
)

context = UserContext(
    emit_progress=lambda tool, progress, total, msg: print(f"[{tool}] {progress}/{total}: {msg}"),
    emit_log=lambda tool, level, data: print(f"[{tool}] {level}: {data}"),
)

result = Runner.run_streamed(
    agent,
    input="Create a lesson plan for AI ethics.",
    context=context,
)

进度和日志流

UserContext 为进度和日志消息提供回调函数。在每次调用工具时 _build_function_tool 将 FastMCP 的原生回调绑定到这些处理程序中。提供您自己的函数以在用户界面中显示状态；省略这些函数则会回退到标准输出日志记录。

def handle_progress(tool, progress, total, message):
    update_ui_progress(tool, progress, total, message)

def handle_log(tool, level, data):
    append_log_entry(tool, level, data)

context = UserContext(
    emit_progress=handle_progress,
    emit_log=handle_log,
)

选择交通工具

Stdio（通常指标准输入输出库，如C语言中的\）

toolkit = MCPToolkit(
    server_cmd=".venv/bin/python",
    server_script="path/to/mcp_tools.py",
    server_args=("--flag",),
)

SSE（默认）

toolkit = MCPToolkit()  # reads MCP_* env vars or falls back to http://127.0.0.1:8010/sse/

根据需要覆盖连接的部分内容：

toolkit = MCPToolkit(
    base_url="https://mcp.example.com",
    sse_path="/custom/sse/",
    auth_token_env="MCP_AUTH_TOKEN",
)

# Tip: pass headers per request when your app already has the bearer token.
toolkit = MCPToolkit(
    base_url="https://mcp.example.com",
    headers={"Authorization": f"Bearer {token}"},
)

可流式传输的HTTP

from utility.mcp_tool_helper import build_streamable_http_transport_factory

toolkit = MCPToolkit(
    transport_factory=build_streamable_http_transport_factory(
        url="https://mcp.example.com/mcp",
        headers={"Authorization": "Bearer TOKEN"},
    ),
)

自定义SSE传输

from utility.mcp_tool_helper import build_sse_transport_factory

toolkit = MCPToolkit(
    transport_factory=build_sse_transport_factory(
        url="https://mcp.example.com/mcp/sse/",
        headers={"X-Custom": "value"},
    ),
)

自动推断传输

# Accepts URLs, paths, ClientTransport instances, or MCP config dicts
toolkit = MCPToolkit(transport="https://mcp.example.com/mcp")

缓存与刷新

MCPToolkit 发现的缓存 FunctionTool首次使用时请致电 refresh_cache() 强制重新加载，或者使用工具包构建 preload=False 将发现过程推迟到首次查找时进行。

toolkit = MCPToolkit(..., preload=False)
tools = toolkit.get_function_tools()  # loads on demand
toolkit.refresh_cache()               # refreshes immediately

认证

设置 MCP_TEST_TOKEN （或提供 auth_token_env=）当你已经拥有一个静态承载令牌时。\ 或者通过（某种方式）提供明确的头部信息 headers= 就……争论/辩论 MCPToolkit 或者 get_mcp_function_tools或者确保您的自定义传输工厂附加了您的部署所期望的任何认证方案。

实用的环境变量

• MCP_HOST （默认 127.0.0.1；也接受旧版（或传统方式） MCP_HOST)
• MCP_PORT （默认 8010; 遗产（或传统） MCP_PORT)
• MCP_SSE_PATH （默认 /sse/; 传统（或遗留问题） MCP_SSE_PATH)
• MCP_SCHEME （默认 http; 遗产（或传统） MCP_SCHEME)
• MCP_AUTH_TOKEN (承载令牌；旧版/遗留系统) MCP_TEST_TOKEN）

提供明确的论据以 MCPToolkit / get_mcp_function_tools 要覆盖这些设置中的任何一个，或者提供如stdio设置等 server_cmd 和 server_script。

典型的SSE部署流程

1. 将您的FastMCP服务器部署在HTTPS后端，并公开其SSE端点（例如。， /sse/）。
2. 颁发有时限的访问令牌（例如，Azure AD）并将其交给客户端。
3. 在客户端，调用 MCPToolkit(base_url="https://...", headers={"Authorization": f"Bearer {token}"})。
4. 可选地在多个请求中重用相同的工具包；调用 refresh_cache() 如果工具在服务器端发生变化。

本地开发（从stdio迁移到SSE）

1. 从stdio开始： MCPToolkit(server_cmd="python", server_script="local_server.py")。
2. 通过提供主机/端口/路径来切换到SSE，而无需修改调用点：

toolkit = MCPToolkit(
    base_url="http://127.0.0.1:8010",
    sse_path="/sse/",
)

1. 如果同时支持stdio和SSE，则实例化两个工具包，并根据环境进行选择。

便捷助手

顶级辅助函数（或称为顶级帮助器）模拟了构造函数，并返回一个列表，其中包含 FunctionTool在一个通话中。

from utility.mcp_tool_helper import get_mcp_function_tools

tools = get_mcp_function_tools(
    host="localhost",
    port=8010,
    auth_token_env="MCP_TEST_TOKEN",
    transport="https://mcp.example.com/mcp",  # optional override
)

# or via stdio
# tools = get_mcp_function_tools(
#     server_cmd=".venv/bin/python",
#     server_script="path/to/mcp_tools.py",
# )

开发说明

• 需要FastMCP 2.9或更高版本以及OpenAI的Agent SDK。
• 工具调用时传入的JSON输入无效会引发（错误） ModelBehaviorError 以指示代理端的问题。
• 支持同步调用者：助手在内部执行异步发现 asyncio.run如果事件循环已经在运行，则会回退到一个工作线程。

许可证

GPL-3.0（通用公共许可证第3版）。