将 MCP 服务器与 Yi-AI Router 结合使用

MCP (Model Context Protocol) 服务器为 LLM 提供工具调用能力。通过 Yi-AI Router，您可以连接 MCP 服务器到任何支持函数/工具调用的模型。

工作原理

MCP 协议定义了 LLM 可以调用的工具。流程如下：

1. MCP 客户端连接到本地 MCP 服务器，发现可用工具
2. 客户端将 MCP 工具定义转换为 OpenAI 函数调用格式
3. 工具随请求一起发送到 Yi-AI Router
4. 模型决定调用哪个工具，在响应中返回 tool_calls
5. 客户端通过 MCP 服务器执行工具，返回结果
6. 模型生成最终回复

快速开始

前提条件

• Python 3.10+
• 已注册 Yi-AI 账户并获取 API Key（sk-yi-...）
• 一个可连接的 MCP 服务器（如文件系统、数据库等）

示例：文件系统 MCP 服务器

创建 .env 文件：

OPENAI_API_KEY=sk-yi-your-key

创建 mcp-client.py：

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 = "claude-sonnet-4-20250514"

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


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


class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.openai = OpenAI(
            base_url="https://www.yiairouter.com/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(
            "\nConnected to server, tools:",
            [tool.name for tool 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(tool) for tool 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 = content.tool_calls[0].function.arguments
            tool_args = json.loads(tool_args) if tool_args else {}
            try:
                result = await self.session.call_tool(tool_name, tool_args)
                final_text.append(
                    f"[Called 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!")
        print("Enter your queries or type 'quit' to exit.")
        while True:
            try:
                query = input("\nQuery: ").strip()
                result = await self.process_query(query)
                print("Result:")
                print(result)
            except Exception as e:
                print(f"Error: {str(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())

运行

pip install mcp openai python-dotenv
python mcp-client.py

客户端将连接到文件系统 MCP 服务器并启动交互式会话。您可以通过 Yi-AI Router 查询文件和目录。

支持的模型

任何支持函数/工具调用的 Yi-AI Router 模型都可以与 MCP 配合使用，包括：

• Claude Sonnet 4 / Haiku 4
• GPT-4o / GPT-4o-mini
• Gemini 2.5 Flash / Pro
• DeepSeek V3 / R1
• 以及其他所有支持工具调用的模型

其他 MCP 服务器

同样的模式适用于任何 MCP 服务器。替换 SERVER_CONFIG 即可：

# 数据库 MCP
SERVER_CONFIG = {
    "command": "npx",
    "args": ["-y", "@anthropic/mcp-server-db", "postgresql://..."],
    "env": None,
}

# 网页抓取 MCP
SERVER_CONFIG = {
    "command": "npx",
    "args": ["-y", "@anthropic/mcp-server-fetch"],
    "env": None,
}

# GitHub MCP
SERVER_CONFIG = {
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-github"],
    "env": {"GITHUB_TOKEN": "..."},
}

Yi-AI Router 的处理方式

Yi-AI Router 的 /api/v1/chat/completions 端点透明透传所有 OpenAI 格式字段 — 包括 tools、tool_choice 和 tool_calls。无需特殊配置。平台自动处理模型映射、路由和计费。