手搓Manus？MCP 原理解析与MCP Client实践

基础概念

interface Request {  method: string;  params?: { ... };}

interface Result {  [key: string]: unknown;}

interface Error {  code: number;  message: string;  data?: unknown;}

interface Notification {  method: string;  params?: { ... };}

实践

available_tools is: [{'type': 'function', 'function': {'name': 'get_offers', 'description': 'Get product offers from API', 'parameters': {'type': 'object', 'properties': {'keywords': {'type': 'string', 'description': 'Keywords to search for products', 'default': ''}, 'pageSize': {'type': 'number', 'description': 'Number of items per page', 'minimum': 1, 'maximum': 100, 'default': 10}}}}}]

[    {        "role": "user",        "content": "你好"    },    {        "role": "system",        "content": "You're a helpful digital assistant who can answer questions and support users in completing tasks."    }]

ChatCompletionMessage(content='你好！有什么可以帮助你的吗？', refusal=None, role='assistant', annotations=None, audio=None, function_call=None, tool_calls=None)

[{'role': 'user', 'content': '帮我找一些手表'}, {'role': 'system', 'content': 'You are a helpful assistant that can answer questions and help with tasks.'}]

AI response is ChatCompletionMessage(content='', refusal=None, role='assistant', annotations=None, audio=None, function_call=None, tool_calls=[ChatCompletionMessageToolCall(id='0195c8c06aaf3ea050e6d8eed17380ec', function=Function(arguments='{"keywords": "手表", "pageSize": 10}', name='get_offers'), type='function')])

AI response is ChatCompletionMessage(content='leton\n{{"name": "get_offers", "arguments": {"keywords": "手表", "pageSize": 10}}}\n</tool_call>', refusal=None, role='assistant', annotations=None, audio=None, function_call=None, tool_calls=None)

result = await self.session.call_tool(tool_name, tool_args)

meta=None content=[TextContent(type='text', text='some...product...info', annotations=None)] isError=False

{  "messages": [    {      "role": "user",      "content": "帮我找一些手表"    },    {      "role": "system",      "content": "You are a helpful assistant that can answer questions and help with tasks."    },    {      "role": "assistant",      "content": "",      "tool_calls": [        {          "id": "0195c8c06aaf3ea050e6d8eed17380ec",          "type": "function",          "function": {            "name": "get_offers",            "arguments": "{\"keywords\": \"手表\", \"pageSize\": 10}"          }        }      ]    },    {      "role": "tool",      "tool_call_id": "0195c8c06aaf3ea050e6d8eed17380ec",      "content": {        "type": "text",        "text": "some...product...info"      }    }  ]}

?AI: [?调用工具 get_offers ?参数是 {'keywords': '手表', 'pageSize': 10}]根据您的搜索，这里有几款手表供您参考：
1. 款式 ID: ididid2   价格: $0.24   供应商: Guangzhou Huapan Cosmetics Co., Ltd.   评分: **   收评次数: **   供应商年限: **年   推荐指数: ★★
2. 款式 ID: ididid   价格: $3.99   供应商: Shenzhen Top Quality Electronics Technology Co., Ltd.   评分: **   收评次数: **   供应商年限: **年   推荐指数: ★★
这两款手表的评价和销售情况都还不错，您可以根据自己的需求选择合适的款式。如果还有其他问题或需要更多信息，请随时告诉我。

附录

import asyncioimport jsonimport osimport tracebackfrom typing import Optionalfrom contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom openai import OpenAIfrom dotenv import load_dotenvload_dotenv()  # load environment variables from .envclass MCPClient:    def __init__(self):        # Initialize session and client objects        self.session: Optional[ClientSession] = None        self.exit_stack = AsyncExitStack()        self.client = OpenAI(            api_key=os.getenv("OPENAI_API_KEY"),            base_url=os.getenv("OPENAI_BASE_URL")        )        self.model = os.getenv("OPENAI_MODEL")        self.messages = [            {                "role": "system",                "content": "You are a versatile assistant capable of answering questions, completing tasks, and intelligently invoking specialized tools to deliver optimal results."            }        ]        self.available_tools = []        @staticmethod    def convert_custom_object(obj):        """        将自定义对象转换为字典        """        if hasattr(obj, "__dict__"):  # 如果对象有 __dict__ 属性，直接使用            return obj.__dict__        elif isinstance(obj, (list, tuple)):  # 如果是列表或元组，递归处理            return [MCPClient.convert_custom_object(item) for item in obj]        elif isinstance(obj, dict):  # 如果是字典，递归处理值            return {key: MCPClient.convert_custom_object(value) for key, value in obj.items()}        else:  # 其他类型（如字符串、数字等）直接返回            return obj            async def connect_to_server(self, server_script_path: str):        """Connect to an MCP server                Args:            server_script_path: Path to the server script (.py or .js)        """        is_python = server_script_path.endswith('.py')        is_js = server_script_path.endswith('.js')        if not (is_python or is_js):            raise ValueError("Server script must be a .py or .js file")                    command = "python" if is_python else "node"        server_params = StdioServerParameters(            command=command,            args=[server_script_path],            env=None        )                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()                # List available tools        response = await self.session.list_tools()        tools = response.tools        print("\nConnected to server with tools:", [tool.name for tool in tools])    async def process_query(self, query: str) -> str:        """Process a query with multi-turn tool calling support"""        # Add user query to message history        self.messages.append({            "role": "user",            "content": query        })        # Get available tools if not already set        if not self.available_tools:            response = await self.session.list_tools()            self.available_tools = [{                "type": "function",                "function": {                    "name": tool.name,                    "description": tool.description,                    "parameters": tool.inputSchema                }            } for tool in response.tools]        current_response = self.client.chat.completions.create(            model=self.model,            messages=self.messages,            tools=self.available_tools,            stream=False        )        # Print initial response if exists        if current_response.choices[0].message.content:            print("\n? AI:", current_response.choices[0].message.content)        # 直到下一次交互 AI 没有选择调用工具时退出循环        while current_response.choices[0].message.tool_calls:            # AI 一次交互中可能会调用多个工具            for tool_call in current_response.choices[0].message.tool_calls:                tool_name = tool_call.function.name                try:                    tool_args = json.loads(tool_call.function.arguments)                except json.JSONDecodeError:                    tool_args = {}                print(f"\n? 调用工具 {tool_name}")                print(f"? 参数: {tool_args}")                # Execute tool call                result = await self.session.call_tool(tool_name, tool_args)                print(f"\n工具结果: {result}")                # Add AI message and tool result to history                self.messages.append(current_response.choices[0].message)                self.messages.append({                    "role": "tool",                    "tool_call_id": tool_call.id,                    "content": json.dumps(result.content)                })            # Get next response            current_response = self.client.chat.completions.create(                model=self.model,                messages=self.messages,                tools=self.available_tools,                stream=False            )        # Add final response to history        self.messages.append(current_response.choices[0].message)        return current_response.choices[0].message.content or ""    async def chat_loop(self):        """Run an interactive chat loop"""        print("\nMCP Client Started!")        print("Type your queries or 'quit' to exit.")                while True:            try:                query = input("\nCommend: ").strip()                                if query.lower() == 'quit':                    break                                    response = await self.process_query(query)                print("\n?AI: " + response)                                except Exception as e:                print(f"\nError occurs: {e}")                traceback.print_exc()        async def cleanup(self):        """Clean up resources"""        await self.exit_stack.aclose()async def main():    if len(sys.argv) < 2:        print("Usage: python client.py <path_to_server_script>")        sys.exit(1)            client = MCPClient()    try:        await client.connect_to_server(sys.argv[1])        await client.chat_loop()    finally:        await client.cleanup()if __name__ == "__main__":    import sys    asyncio.run(main())

/** * MCP客户端实现 *  * 提供与MCP服务器的连接、工具调用和聊天交互功能 *  * 主要功能： * 1. 连接Python或JavaScript实现的MCP服务器 * 2. 获取服务器提供的工具列表 * 3. 通过OpenAI API处理用户查询 * 4. 自动处理工具调用链 * 5. 提供交互式命令行界面 *  * 使用说明： * 1. 确保设置OPENAI_API_KEY环境变量 * 2. 通过命令行参数指定MCP服务器脚本路径 * 3. 启动后输入查询或'quit'退出 *  * 依赖： * - @modelcontextprotocol/sdk: MCP协议SDK * - openai: OpenAI API客户端 * - dotenv: 环境变量加载 */import { Client } from "@modelcontextprotocol/sdk/client/index.js";import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";import OpenAI from "openai";import type { ChatCompletionMessageParam } from "openai/resources/chat/completions";import type { Tool } from "@modelcontextprotocol/sdk/types.js";import * as dotenv from "dotenv";import * as readline from 'readline';// 加载环境变量配置dotenv.config();/** * MCP客户端类，封装与MCP服务器的交互逻辑 */class MCPClient {    private openai: OpenAI; // OpenAI API客户端实例    private client: Client; // MCP协议客户端实例    private messages: ChatCompletionMessageParam[] = [        {            role: "system",            content: "You are a versatile assistant capable of answering questions, completing tasks, and intelligently invoking specialized tools to deliver optimal results."        },    ]; // 聊天消息历史记录，用于维护对话上下文    private availableTools: any[] = []; // 服务器提供的可用工具列表，格式化为OpenAI工具格式    /**     * 构造函数，初始化OpenAI和MCP客户端     *      * @throws {Error} 如果OPENAI_API_KEY环境变量未设置     *      * 初始化过程：     * 1. 检查必要的环境变量     * 2. 创建OpenAI客户端实例     * 3. 创建MCP客户端实例     * 4. 初始化消息历史记录     */    constructor() {        if (!process.env.OPENAI_API_KEY) {            throw new Error("OPENAI_API_KEY环境变量未设置");        }        this.openai = new OpenAI({            apiKey: process.env.OPENAI_API_KEY,            baseURL: process.env.OPENAI_BASE_URL,        });        this.client = new Client(            {                name: "my-mcp-client",                version: "1.0.0",            },        );    }    /**     * 连接到MCP服务器     *      * @param {string} serverScriptPath - 服务器脚本路径(.py或.js)     * @returns {Promise<void>} 连接成功时解析     * @throws {Error} 如果服务器脚本不是.py或.js文件，或连接失败     *      * 连接过程：     * 1. 检查脚本文件扩展名     * 2. 根据扩展名决定使用python或node执行     * 3. 通过stdio建立连接     * 4. 获取服务器工具列表并转换为OpenAI工具格式     *      * 注意事项：     * - 服务器脚本必须具有可执行权限     * - 连接成功后会自动获取工具列表     */    async connectToServer(serverScriptPath: string) {        const isPython = serverScriptPath.endsWith('.py');        const isJs = serverScriptPath.endsWith('.js');        if (!isPython && !isJs) {            throw new Error("Server script must be a .py or .js file");        }        const command = isPython ? "python" : "node";        const transport = new StdioClientTransport({            command,            args: [serverScriptPath],        });        await this.client.connect(transport);        // 获取并转换可用工具列表        const tools = (await this.client.listTools()).tools as unknown as Tool[];        this.availableTools = tools.map(tool => ({            type: "function" as const,            function: {                name: tool.name as string,                description: tool.description as string,                parameters: {                    type: "object",                    properties: tool.inputSchema.properties as Record<string, unknown>,                    required: tool.inputSchema.required as string[],                },            }        }));        console.log("\n已连接到服务器，可用工具:", tools.map(tool => tool.name));    }    /**     * 处理工具调用链     *      * @param {OpenAI.Chat.Completions.ChatCompletion} response - 初始OpenAI响应，包含工具调用     * @param {ChatCompletionMessageParam[]} messages - 当前消息历史记录     * @returns {Promise<OpenAI.Chat.Completions.ChatCompletion>} 最终OpenAI响应     *      * 处理流程：     * 1. 检查响应中是否包含工具调用     * 2. 循环处理所有工具调用     * 3. 解析每个工具调用的参数     * 4. 执行工具调用     * 5. 将工具结果添加到消息历史     * 6. 获取下一个OpenAI响应     *      * 错误处理：     * - 参数解析失败时使用空对象继续执行     * - 工具调用失败会抛出异常     *      * 注意事项：     * - 此方法会修改传入的messages数组     * - 可能多次调用OpenAI API     */    private async toolCalls(response: OpenAI.Chat.Completions.ChatCompletion, messages: ChatCompletionMessageParam[]) {        let currentResponse = response;        // 直到下一次交互 AI 没有选择调用工具时退出循环        while (currentResponse.choices[0].message.tool_calls) {            if (currentResponse.choices[0].message.content) {                console.log("\n? AI: tool_calls", JSON.stringify(currentResponse.choices[0].message));            }            // AI 一次交互中可能会调用多个工具            for (const toolCall of currentResponse.choices[0].message.tool_calls) {                const toolName = toolCall.function.name;                const rawArgs = toolCall.function.arguments;                let toolArgs;                try {                    console.log(`rawArgs is ===== ${rawArgs}`)                    toolArgs = "{}" == JSON.parse(rawArgs) ? {} : JSON.parse(rawArgs);                    if (typeof toolArgs === "string") {                        toolArgs = JSON.parse(toolArgs);                    }                } catch (error) {                    console.error('⚠️ 参数解析失败，使用空对象替代');                    toolArgs = {};                }                console.log(`\n? 调用工具 ${toolName}`);                console.log(`? 参数:`, toolArgs);                // 调用工具获取结果                const result = await this.client.callTool({                    name: toolName,                    arguments: toolArgs                });                console.log(`\n result is ${JSON.stringify(result)}`);                // 添加 AI 的响应和工具调用结果到消息历史                // console.log(`? currentResponse.choices[0].message:`, currentResponse.choices[0].message);                messages.push(currentResponse.choices[0].message);                messages.push({                    role: "tool",                    tool_call_id: toolCall.id,                    content: JSON.stringify(result.content),                } as ChatCompletionMessageParam);            }            // console.log(`? messages: `, messages);            // 获取下一个响应            currentResponse = await this.openai.chat.completions.create({                model: process.env.OPENAI_MODEL as string,                messages: messages,                tools: this.availableTools,            });        }        return currentResponse;    }    /**     * 处理用户查询     *      * @param {string} query - 用户输入的查询字符串     * @returns {Promise<string>} AI生成的响应内容     *      * 处理流程：     * 1. 将用户查询添加到消息历史     * 2. 调用OpenAI API获取初始响应     * 3. 如果有工具调用，处理工具调用链     * 4. 返回最终响应内容     *      * 错误处理：     * - OpenAI API调用失败会抛出异常     * - 工具调用链中的错误会被捕获并记录     *      * 注意事项：     * - 此方法会更新内部消息历史     * - 可能触发多个工具调用     */    async processQuery(query: string): Promise<string> {        // 添加用户查询到消息历史        this.messages.push({            role: "user",            content: query,        });        // 初始OpenAI API调用        let response = await this.openai.chat.completions.create({            model: process.env.OPENAI_MODEL as string,            messages: this.messages,            tools: this.availableTools,        });        // 打印初始响应        if (response.choices[0].message.content) {            console.log("\n? AI:", response.choices[0].message);        }        // 处理工具调用链        if (response.choices[0].message.tool_calls) {            response = await this.toolCalls(response, this.messages);        }        // 更新消息历史        this.messages.push(response.choices[0].message);        return response.choices[0].message.content || "";    }    /**     * 启动交互式聊天循环     *      * @returns {Promise<void>} 当用户退出时解析     *      * 功能：     * 1. 持续接收用户输入     * 2. 处理用户查询     * 3. 显示AI响应     * 4. 输入'quit'退出     *      * 实现细节：     * - 使用readline模块实现交互式输入输出     * - 循环处理直到用户输入退出命令     * - 捕获并显示处理过程中的错误     *      * 注意事项：     * - 此方法是阻塞调用，会一直运行直到用户退出     * - 确保在调用前已连接服务器     */    async chatLoop() {        console.log("\nMCP Client Started!");        console.log("Type your queries or 'quit' to exit.");        const rl = readline.createInterface({            input: process.stdin,            output: process.stdout,        });        while (true) {            const query = await new Promise<string>((resolve) => {                rl.question("\nQuery: ", resolve);            });            if (query.toLowerCase() === 'quit') {                break;            }            try {                const response = await this.processQuery(query);                console.log("\n" + response);            } catch (e) {                console.error("\nError:", e instanceof Error ? e.message : String(e));            }        }        rl.close();    }    /**     * 清理资源     *      * @returns {Promise<void>} 资源清理完成后解析     *      * 关闭以下资源：     * 1. MCP客户端连接     * 2. 任何打开的句柄     *      * 最佳实践：     * - 应在程序退出前调用     * - 建议在finally块中调用以确保执行     *      * 注意事项：     * - 多次调用是安全的     * - 清理后实例不可再用     */    async cleanup() {        if (this.client) {            await this.client.close();        }    }}/** * 主函数 *  * 程序入口点，执行流程： * 1. 检查命令行参数 * 2. 创建MCP客户端实例 * 3. 连接到指定服务器脚本 * 4. 启动交互式聊天循环 * 5. 退出时清理资源 *  * @throws {Error} 如果缺少命令行参数或连接失败 *  * 使用示例： * ```bash * node index.js /path/to/server.js * ``` *  * 退出码： * - 0: 正常退出 * - 1: 参数错误或运行时错误 */async function main() {    if (process.argv.length < 3) {        console.log("Usage: node dist/index.js <path_to_server_script>");        process.exit(1);    }    const client = new MCPClient();    try {        await client.connectToServer(process.argv[2]);        await client.chatLoop();    } finally {        await client.cleanup();    }}main().catch((error) => {    console.error("Error:", error);    process.exit(1);});