MCPHero-MCP作为工具/MCP作为功能
库将MCP用作本地AI库中的工具/函数
灵感
现在每个人都使用MCP,但许多人仍然使用没有MCP支持的老派AI客户端。这些客户端库类似 openai 或 google-genai 仅支持工具/函数调用。 创建此项目是为了将MCP服务器作为工具轻松连接到这些库。
概念
两个主要流程:
list_tools-通过http调用MCP服务器以获取工具定义,然后将它们映射到AI库工具定义process_tool_calls-获取AI库的tool_calls,解析它们,将请求发送到mcp服务器,返回结果
安装
基础(无LLM SDK依赖关系):
pip install mcphero对于OpenAI支持:
pip install "mcphero[openai]"对于Google Gemini支持:
pip install "mcphero[google-genai]"快速开始
通用(与提供者无关)
使用 MCPToolAdapter 当您的框架有自己的工具调用循环时,或者当您只需要执行原始MCP工具而不需要任何LLM SDK依赖时。
import asyncio
from mcphero import MCPToolAdapter, GenericToolCall
async def main():
adapter = MCPToolAdapter("https://api.mcphero.app/mcp/your-server-id")
# Discover available tools
tools = await adapter.discover_tools()
for tool in tools:
print(tool.name, tool.description)
# Execute tool calls directly
results = await adapter.process_tool_calls([
GenericToolCall(name="get_weather", arguments={"city": "London"}, id="1"),
])
for result in results:
print(result.content)
asyncio.run(main())或者直接调用单个工具:
result = await adapter.call_tool("get_weather", {"city": "London"})开放人工智能
import asyncio
from openai import OpenAI
from mcphero import MCPToolAdapterOpenAI
async def main():
adapter = MCPToolAdapterOpenAI("https://api.mcphero.app/mcp/your-server-id")
client = OpenAI()
# Get tool definitions
tools = await adapter.get_tool_definitions()
# Make request with tools
messages = [{"role": "user", "content": "What's the weather in London?"}]
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
)
# Process tool calls if present
if response.choices[0].message.tool_calls:
tool_results = await adapter.process_tool_calls(
response.choices[0].message.tool_calls
)
# Continue conversation with results
messages.append(response.choices[0].message)
messages.extend(tool_results)
final_response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
)
print(final_response.choices[0].message.content)
asyncio.run(main())谷歌双子星
import asyncio
from google import genai
from google.genai import types
from mcphero import MCPToolAdapterGemini
async def main():
adapter = MCPToolAdapterGemini("https://api.mcphero.app/mcp/your-server-id")
client = genai.Client(api_key="your-api-key")
# Get tool definitions
tool = await adapter.get_tool()
# Make request with tools
response = client.models.generate_content(
model="gemini-2.5-flash",
contents="What's the weather in London?",
config=types.GenerateContentConfig(
tools=[tool],
automatic_function_calling=types.AutomaticFunctionCallingConfig(
disable=True
),
),
)
# Process function calls if present
if response.function_calls:
results = await adapter.process_function_calls(response.function_calls)
# Continue conversation with results
contents = [
types.Content(role="user", parts=[types.Part.from_text("What's the weather in London?")]),
response.candidates[0].content,
*results,
]
final_response = client.models.generate_content(
model="gemini-2.5-flash",
contents=contents,
config=types.GenerateContentConfig(tools=[tool]),
)
print(final_response.text)
asyncio.run(main())多个MCP服务器
适配器本身支持一次连接到多个MCP服务器。使用 MCPServerConfig 要配置每个服务器,请将它们作为列表传递。
MCPServer配置
from mcphero import MCPServerConfig
config = MCPServerConfig(
url="https://api.mcphero.app/mcp/your-server-id", # required
name="weather", # optional, auto-derived from URL if omitted
timeout=30.0, # optional, default 30s
headers={ # optional, auth headers for the server
"Authorization": "Bearer your-token",
},
init_mode="auto", # "auto" | "on_fail" | "none"
tool_prefix="wx", # optional, prefix for tool names from this server
)| 字段 | 类型 | 默认值 | 描述 | ||
|---|---|---|---|---|---|
url | str | *必需的* | MCP服务器的HTTP端点 | ||
name | `str \ | None` | 来源于URL | 服务器的标识符(例如最后一个路径段) | |
timeout | float | 30.0 | 请求超时(秒) | ||
headers | `dict[str, str] \ | None` | None | 随每个请求发送的标头(对身份验证有用) | |
init_mode | `"auto" \ | "on_fail" \ | "none"` | "auto" | 何时运行MCP初始化握手 |
tool_prefix | `str \ | None` | None | 应用于此服务器中所有工具名称的前缀 |
init_mode 选项:
"auto"-在每次请求之前初始化连接(默认,最安全)"on_fail"-跳过初始化,但如果请求失败,则重试初始化"none"-从不初始化(对于不需要初始化的服务器)
多服务器示例
import asyncio
from openai import OpenAI
from mcphero import MCPToolAdapterOpenAI, MCPServerConfig
async def main():
adapter = MCPToolAdapterOpenAI([
MCPServerConfig(
url="https://api.mcphero.app/mcp/weather",
name="weather",
headers={"Authorization": "Bearer weather-token"},
),
MCPServerConfig(
url="https://api.mcphero.app/mcp/calendar",
name="calendar",
headers={"Authorization": "Bearer calendar-token"},
),
])
client = OpenAI()
# Tools from ALL servers are fetched in parallel and merged
tools = await adapter.get_tool_definitions()
messages = [{"role": "user", "content": "What's the weather today and what's on my calendar?"}]
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
)
# Tool calls are automatically routed to the correct server
if response.choices[0].message.tool_calls:
results = await adapter.process_tool_calls(
response.choices[0].message.tool_calls
)
messages.append(response.choices[0].message)
messages.extend(results)
final_response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
)
print(final_response.choices[0].message.content)
asyncio.run(main())工具名称冲突
当多个服务器公开同名工具时,适配器会自动在它们前面加上服务器名称以避免冲突:
# Both servers have a "search" tool
adapter = MCPToolAdapterOpenAI([
MCPServerConfig(url="https://example.com/mcp/weather", name="weather"),
MCPServerConfig(url="https://example.com/mcp/calendar", name="calendar"),
])
tools = await adapter.get_tool_definitions()
# "search" becomes "weather__search" and "calendar__search"您可以控制此行为:
# Custom separator
adapter = MCPToolAdapterOpenAI(configs, prefix_separator="-")
# "weather-search", "calendar-search"
# Disable auto-prefixing (will raise on collision)
adapter = MCPToolAdapterOpenAI(configs, auto_prefix_on_collision=False)
# Manual prefix via config (always applied, regardless of collisions)
MCPServerConfig(url="...", tool_prefix="wx")
# "wx__search"API 参考
MCPToolAdapter
from mcphero import MCPToolAdapter, GenericToolCall
adapter = MCPToolAdapter("https://api.mcphero.app/mcp/your-server-id")方法
| 方法 | 返回 | 描述 | |
|---|---|---|---|
discover_tools() | list[MCPToolDefinition] | 发现具有路由元数据的工具 | |
process_tool_calls(tool_calls, return_errors=True) | list[GenericToolResult] | 执行工具调用并返回通用结果 | |
call_tool(name, arguments) | JsonRpcResponse | 按名称调用单个工具 | |
initialize_all() | `dict[str, JsonRpcResponse \ | Exception]` | 预初始化所有服务器连接 |
MCPToolAdapterOpenAI
from mcphero import MCPToolAdapterOpenAI, MCPServerConfig
# Single server (URL string)
adapter = MCPToolAdapterOpenAI("https://api.mcphero.app/mcp/your-server-id")
# Single server (config)
adapter = MCPToolAdapterOpenAI(
MCPServerConfig(
url="https://api.mcphero.app/mcp/your-server-id",
headers={"Authorization": "Bearer ..."},
)
)
# Multiple servers
adapter = MCPToolAdapterOpenAI([
MCPServerConfig(url="https://server-a.com/mcp", name="a"),
MCPServerConfig(url="https://server-b.com/mcp", name="b"),
])方法
| 方法 | 返回 | 描述 | |
|---|---|---|---|
get_tool_definitions() | list[ChatCompletionToolParam] | 从MCP服务器获取作为OpenAI工具模式的工具 | |
process_tool_calls(tool_calls, return_errors=True) | list[ChatCompletionToolMessageParam] | 执行工具调用并返回对话结果 | |
discover_tools() | list[MCPToolDefinition] | 低级:使用路由元数据发现工具 | |
call_tool(name, arguments) | JsonRpcResponse | 低级:按名称调用单个工具 | |
initialize_all() | `dict[str, JsonRpcResponse \ | Exception]` | 预初始化所有服务器连接 |
MCPToolAdapterGemini
from mcphero import MCPToolAdapterGemini, MCPServerConfig
# Same constructor options as OpenAI adapter
adapter = MCPToolAdapterGemini("https://api.mcphero.app/mcp/your-server-id")方法
| 方法 | 返回 | 描述 |
|---|---|---|
get_function_declarations() | list[types.FunctionDeclaration] | 将工具作为Gemini FunctionDeclaration对象获取 |
get_tool() | types.Tool | 将工具作为Gemini Tool对象获取 |
process_function_calls(function_calls, return_errors=True) | list[types.Content] | 执行函数调用并返回Content对象 |
process_function_calls_as_parts(function_calls, return_errors=True) | list[types.Part] | 执行函数调用并返回Part对象 |
discover_tools() | list[MCPToolDefinition] | 低级:使用路由元数据发现工具 |
call_tool(name, arguments) | JsonRpcResponse | 低级:按名称调用单个工具 |
错误处理
所有适配器都能优雅地处理错误。当 return_errors=True (默认),失败的工具调用会返回错误消息,这些消息可以发送回模型:
# Tool call fails -> returns error in result
results = await adapter.process_tool_calls(tool_calls, return_errors=True)
# [{"role": "tool", "tool_call_id": "...", "content": "{\"error\": \"HTTP error...\"}"}]
# Skip failed calls
results = await adapter.process_tool_calls(tool_calls, return_errors=False)链接
许可证
麻省理工学院
