MCP（Model Context Protocol）是 Anthropic 推出的工具调用标准，与 OpenAI 兼容的 function calling 并存。通过将 MCP 工具定义转换为 OpenAI 格式，可以在 OpenRouter 上使用任意 MCP 服务器。本页展示完整的 Python 实现：使用 mcp 客户端 SDK 连接文件系统 MCP 服务器，调用 OpenRouter 处理模型请求，实现多轮工具调用对话。

MCP（Model Context Protocol）是提供 LLM 工具调用能力的流行方式，是 OpenAI 兼容工具调用的替代方案。

通过将 MCP 工具定义转换为 OpenAI 兼容格式，即可在 OpenRouter 上使用 MCP 服务器。

注意：MCP 协议有状态，需要会话管理，比直接调用 REST 接口更复杂。以下示例使用官方 MCP 客户端 SDK。

前置要求

1. 安装依赖包：

pip install mcp openai python-dotenv

2. 创建 .env 文件，设置 OPENAI_API_KEY 为你的 OpenRouter API key

3. 本示例假设系统中存在 /Applications 目录（macOS）

初始化代码

import asyncio
from typing import Optional
from contextlib import AsyncExitStack

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

from openai import OpenAI
from dotenv import load_dotenv
import json

load_dotenv()

MODEL = "anthropic/claude-3-7-sonnet"

SERVER_CONFIG = {
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Applications/"],
    "env": None
}

工具格式转换函数

MCP 和 OpenAI 的工具定义格式不同，需要做一次转换：

def convert_tool_format(tool):
    return {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description,
            "parameters": {
                "type": "object",
                "properties": tool.inputSchema["properties"],
                "required": tool.inputSchema["required"]
            }
        }
    }

MCP 客户端实现

class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.openai = OpenAI(
            base_url="https://openrouter.ai/api/v1"
        )

    async def connect_to_server(self, server_config):
        server_params = StdioServerParameters(**server_config)
        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(self.stdio, self.write)
        )
        await self.session.initialize()

        response = await self.session.list_tools()
        print("Connected to server with tools:", [t.name for t in response.tools])
        self.messages = []

    async def process_query(self, query: str) -> str:
        self.messages.append({"role": "user", "content": query})

        # 获取可用工具并转换格式
        response = await self.session.list_tools()
        available_tools = [convert_tool_format(t) for t in response.tools]

        # 第一轮：模型调用（可能触发工具调用）
        response = self.openai.chat.completions.create(
            model=MODEL,
            tools=available_tools,
            messages=self.messages
        )
        self.messages.append(response.choices[0].message.model_dump())

        final_text = []
        content = response.choices[0].message

        if content.tool_calls is not None:
            tool_name = content.tool_calls[0].function.name
            tool_args = json.loads(content.tool_calls[0].function.arguments or "{}")

            try:
                result = await self.session.call_tool(tool_name, tool_args)
                final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
            except Exception as e:
                print(f"Error calling tool {tool_name}: {e}")
                result = None

            # 将工具结果注入消息历史
            self.messages.append({
                "role": "tool",
                "tool_call_id": content.tool_calls[0].id,
                "name": tool_name,
                "content": result.content
            })

            # 第二轮：模型根据工具结果生成最终回复
            response = self.openai.chat.completions.create(
                model=MODEL,
                max_tokens=1000,
                messages=self.messages,
            )
            final_text.append(response.choices[0].message.content)
        else:
            final_text.append(content.content)

        return "\n".join(final_text)

    async def chat_loop(self):
        print("\nMCP Client Started! Type 'quit' to exit.")
        while True:
            try:
                query = input("\nQuery: ").strip()
                if query.lower() == "quit":
                    break
                result = await self.process_query(query)
                print("Result:", result)
            except Exception as e:
                print(f"Error: {e}")

    async def cleanup(self):
        await self.exit_stack.aclose()

async def main():
    client = MCPClient()
    try:
        await client.connect_to_server(SERVER_CONFIG)
        await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

运行效果

$ python mcp-client.py

Secure MCP Filesystem Server running on stdio
Allowed directories: [ '/Applications' ]

Connected to server with tools: ['read_file', 'read_multiple_files', 'write_file'...]

MCP Client Started! Type 'quit' to exit.

Query: Do I have microsoft office installed?
Result:
[Calling tool search_files with args {'path': '/Applications', 'pattern': 'Microsoft'}]
I can see from the search results that Microsoft Office is indeed installed. Found:
1. Microsoft Excel - /Applications/Microsoft Excel.app
2. Microsoft PowerPoint - /Applications/Microsoft PowerPoint.app
3. Microsoft Word - /Applications/Microsoft Word.app

架构说明

组件	作用
MCP 客户端 SDK	管理与 MCP 服务器的有状态连接
convert_tool_format()	将 MCP 工具定义转换为 OpenAI 格式
OpenAI SDK	通过 base_url 指向 OpenRouter
多轮循环	调用模型 → 执行工具 → 再次调用模型

常见问题

Q: 除了文件系统 MCP，还支持其他 MCP 服务器吗？

A: 支持。只需修改 SERVER_CONFIG 中的 command 和 args 指向不同的 MCP 服务器即可，转换逻辑和循环结构都是通用的。可在 MCP 服务器列表 查找可用服务器。

Q: 为什么不直接用 OpenRouter 的 function calling，要多绕一层 MCP？

A: 如果你的工具已经是 MCP 服务器格式（许多 IDE 插件、自动化工具都采用此格式），用这种方式可以直接复用而无需重写工具定义。如果从零开始，直接用 OpenAI 兼容的 function calling 更简单。

Q: MCP 客户端连接是否会影响性能？

A: MCP 使用 stdio 通信，启动开销较小，但长会话中保持连接需要管理 AsyncExitStack。上面的代码已包含正确的 cleanup() 处理。