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

这是一个完整的、可运行的示例项目，基于MCP (Model Context Protocol) 官方 Python SDK+DeepSeek API实现。项目分为服务端（MCP Server）和客户端（CLI + LLM Tool Calling 循环）两部分。

📦 环境准备

pip install mcp openai # 确保已安装 Python 3.10+

在终端设置 DeepSeek API Key（替换为你的真实 Key）：

export DEEPSEEK_API_KEY="sk-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_key=DEEPSEEK_API_KEY, base_url="https://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(command="python", 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( model="deepseek-chat", messages=messages, tools=openai_tools, tool_choice="auto" ) choice = response.choices[0] msg = choice.message # 情况A：LLM 返回最终文本答案 if msg.content: messages.append({"role": "assistant", "content": msg.content}) print(f"🤖 DeepSeek: {msg.content}") break # 情况B：LLM 请求调用工具 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 Server

python 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。

🔍 架构与关键设计说明

🛠 扩展建议

1. 支持复杂表达式：当前为二元运算，可改造为calculate(expression: str)并调用sympy或numexpr解析。
2. 并发请求：MCP Server 默认单线程，生产环境可改用mcp.server.sse.SseServerTransport暴露 HTTP/SSE 端点，支持多 Client 并发。
3. 本地模型替换：将AsyncOpenAI的base_url指向Ollama或vLLM，即可无缝切换至本地部署的 Qwen/GLM/DeepSeek 开源模型。
4. 工具权限控制：在session.call_tool前加入参数白名单校验或频率限制，防止恶意注入。