[具身智能-357]：示例代码：MCP Client与用户通过CLI交互 + Deekseek大模型决策 + MCP Server计算加法、减法等运算

这是一个完整的、可运行的示例项目基于MCP (Model Context Protocol) 官方 Python SDKDeepSeek API实现。项目分为服务端MCP Server和客户端CLI LLM Tool Calling 循环两部分。 环境准备pip install mcp openai # 确保已安装 Python 3.10在终端设置 DeepSeek API Key替换为你的真实 Keyexport DEEPSEEK_API_KEYsk-xxxxxxxxxxxxxxxxxxxxxxxx️ 1. MCP Server 端 (math_server.py)提供计算能力通过 MCP 协议暴露calculate工具。# math_server.py from mcp.server.fastmcp import FastMCP mcp FastMCP(MathCalculator) mcp.tool() def calculate(operation: str, a: float, b: float) - str: 执行基础数学运算。operation 可选值: add, subtract, multiply, divide try: if operation add: return f{a} {b} {a b} elif operation subtract: return f{a} - {b} {a - b} elif operation multiply: return f{a} * {b} {a * b} elif operation divide: if b 0: return ❌ 错误除数不能为 0 return f{a} / {b} {a / b} else: return f❌ 不支持的运算类型: {operation} except Exception as e: return f❌ 计算异常: {str(e)} if __name__ __main__: print( MCP Math Server 启动中...) mcp.run() 启动一个Web Service 2. MCP Client CLI DeepSeek (client.py)通过 CLI 接收用户输入调用DeepSeek 进行意图识别与工具调度DeepSeek进行调度指示并不直接驱动MCP Server执行拿到结果驱动MCP Server执行是MCP Client的职责并与 MCP Server 交互。# client.py import asyncio import json import os from mcp.client.stdio import StdioServerParameters, stdio_client from mcp.client.session import ClientSession from openai import AsyncOpenAI # 配置 DeepSeek API DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) if not DEEPSEEK_API_KEY: raise ValueError(请设置环境变量 DEEPSEEK_API_KEY) llm_client AsyncOpenAI( api_keyDEEPSEEK_API_KEY, base_urlhttps://api.deepseek.com ) SYSTEM_PROMPT 你是一个数学计算助手。当用户提出计算需求时请使用提供的 calculate 工具。 工具参数说明 - operation: 运算类型 (add / subtract / multiply / divide) - a, b: 参与计算的数字 请直接返回工具调用结果不要自行心算。 async def run_mcp_math_cli(): # 1. 配置 MCP Server 启动参数 server_params StdioServerParameters(commandpython, args[math_server.py]) print(*50) print( MCP Math Client (Powered by DeepSeek)) print( 输入数学问题如: 帮我算一下 23.5 加 47输入 quit 退出。) print(*50) # 2. 建立 stdio 连接并初始化 MCP Session async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 3. 获取并转换 MCP 工具格式为 OpenAI/DeepSeek 兼容格式 tools_res await session.list_tools() openai_tools [ { type: function, function: { name: tool.name, description: tool.description or , parameters: tool.inputSchema # MCP 原生即为 JSON Schema } } for tool in tools_res.tools ] messages [{role: system, content: SYSTEM_PROMPT}] # 4. CLI 交互主循环 while True: try: user_input input(\n 你: ).strip() except EOFError: break if user_input.lower() in (quit, exit, q): print( 感谢使用再见) break if not user_input: continue messages.append({role: user, content: user_input}) # 5. LLM Tool Calling 循环 while True: response await llm_client.chat.completions.create( modeldeepseek-chat, messagesmessages, toolsopenai_tools, tool_choiceauto ) choice response.choices[0] msg choice.message # 情况ALLM 返回最终文本答案 if msg.content: messages.append({role: assistant, content: msg.content}) print(f DeepSeek: {msg.content}) break # 情况BLLM 请求调用工具 if msg.tool_calls: for tool_call in msg.tool_calls: func_name tool_call.function.name func_args json.loads(tool_call.function.arguments) print(f [调度] 调用 {func_name}({func_args})) try: # 调用 MCP Server 工具 result await session.call_tool(func_name, func_args) # MCP 返回结果通常是 TextContent 列表 tool_output result.content[0].text if result.content else 无输出 except Exception as e: tool_output f❌ 工具执行失败: {str(e)} # 将工具结果追加到对话上下文 messages.append({ role: tool, tool_call_id: tool_call.id, content: tool_output }) # 继续循环让 LLM 基于工具结果生成最终回复 if __name__ __main__: asyncio.run(run_mcp_math_cli()) 运行方式打开两个终端窗口或在一个终端后台运行 Server终端 1启动 MCP Serverpython math_server.py终端 2启动 Client 并与用户交互python client.py交互示例 你: 帮我把 150 减去 45.5然后再乘以 2 [调度] 调用 calculate({operation: subtract, a: 150.0, b: 45.5}) [调度] 调用 calculate({operation: multiply, a: 104.5, b: 2.0}) DeepSeek: 150 减去 45.5 等于 104.5再乘以 2 的结果是 209.0。 架构与关键设计说明模块职责技术实现要点MCP Server暴露计算能力无状态、跨语言可调用使用FastMCP快速声明工具自动将函数签名转为 JSON SchemaStdio 通信客户端与子进程 Server 通信stdio_client通过标准输入/输出传递 JSON-RPC 消息无需网络端口LLM 决策层意图识别、参数提取、多轮工具调用DeepSeekdeepseek-chat原生支持 Function Calling通过tool_choiceauto自动触发上下文管理维护system - user - assistant - tool消息链严格遵循 OpenAI Chat Completion 规范确保 LLM 能接收工具返回结果并生成自然语言容错与安全防止除零错误、异常捕获、无限循环防护Server 端校验b0Client 端捕获Exception并注入错误提示给 LLM 扩展建议支持复杂表达式当前为二元运算可改造为calculate(expression: str)并调用sympy或numexpr解析。并发请求MCP Server 默认单线程生产环境可改用mcp.server.sse.SseServerTransport暴露 HTTP/SSE 端点支持多 Client 并发。本地模型替换将AsyncOpenAI的base_url指向Ollama或vLLM即可无缝切换至本地部署的 Qwen/GLM/DeepSeek 开源模型。工具权限控制在session.call_tool前加入参数白名单校验或频率限制防止恶意注入。