在经过FastMCP的入门学习后在原有服务器的基础上进行了工具的开发除了简单的打招呼工具和简单的加法工具示例还增加了一个网页资源摘取生成文档和调用了有道翻译API编写了一个可以对用户输入进行翻译的工具该工具支持中文、英文、俄文等超过80种语言之间的互相翻译用户的输入会被LLM自动检测然后调用MCP服务器上的翻译工具根据用户的需求进行相关翻译。以下是网页资源摘取工具代码mcp.tool async def summarize(url: str, ctx: Context): 获取任意 HTTP/HTTPS 网页的内容并返回前 3000 字符摘要。 支持所有公开可访问的网页 URL如 https://example.com 或 http://example.com/page await ctx.info(fFetching content from {url}) headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } try: response requests.get(url, timeout10, headersheaders) response.raise_for_status() content response.text[:3000] return f网页内容摘要前 3000 字符:\n{content} except requests.exceptions.HTTPError as e: return fHTTP 错误{str(e)} except requests.exceptions.Timeout: return 请求超时请检查网络连接 except requests.exceptions.RequestException as e: return f获取网页失败{str(e)}以下给出有道翻译的开发代码详情请在有道翻译官网查看相应文档(目前注册可白嫖100元体验金)import requests import hashlib import random import os import time from dotenv import load_dotenv load_dotenv() app_key os.getenv(TRANSLATE_APP_KEY) api_secret os.getenv(TRANSLATE_API_SECRET) url https://openapi.youdao.com/proxy/http/llm-trans text 你好 salt random.randint(32768, 65536) curtime str(int(time.time())) def get_input_for_sign(text): if len(text) 20: return text else: return text[:10] str(len(text)) text[-10:] input_for_sign get_input_for_sign(text) sign_str f{app_key}{input_for_sign}{salt}{curtime}{api_secret} sign hashlib.sha256(sign_str.encode(utf-8)).hexdigest() params { appKey: app_key, salt: salt, signType: v3, sign: sign, curtime: curtime, i: text, from: auto, to: en, } print(f请求参数{params}) response requests.post(url, dataparams) print(f状态码{response.status_code}) print(f响应头 Content-Type: {response.headers.get(Content-Type)}) print(f原始响应内容{repr(response.text)}) # 使用 repr 查看原始内容工具代码mcp.tool def translate(text: str, from_lang: str auto, to_lang: str en) - str: 翻译文本内容。支持多种语言互译。 appKey os.getenv(TRANSLATE_APP_KEY) api_secret os.getenv(TRANSLATE_API_SECRET) if not appKey or not api_secret: return 翻译 API 配置缺失请检查 .env 文件 url https://openapi.youdao.com/api def get_input_for_sign(text): if len(text) 20: return text else: return text[:10] str(len(text)) text[-10:] salt random.randint(32768, 65536) curtime str(int(time.time())) input_for_sign get_input_for_sign(text) sign_str f{appKey}{input_for_sign}{salt}{curtime}{api_secret} sign hashlib.sha256(sign_str.encode(utf-8)).hexdigest() params { appKey: appKey, salt: salt, signType: v3, sign: sign, curtime: curtime, q: text, from: from_lang, to: to_lang, streamType: full # 指定完整响应 } try: response requests.post(url, dataparams, timeout10) response.raise_for_status() # 解析 SSE 格式响应 result_json None for line in response.iter_lines(): if line: decoded line.decode(utf-8) if decoded.startswith(data:): # 提取 data: 后面的 JSON 内容 json_str decoded[5:].strip() result_json json.loads(json_str) break if not result_json: return f无法解析响应{response.text} # 检查错误 if result_json.get(code) and result_json.get(code) ! 0: return f翻译错误{result_json.get(message, 未知错误)} (错误码{result_json.get(code)}) # 提取翻译结果 trans_result result_json.get(trans_result, []) if trans_result: return .join([item.get(dst, ) for item in trans_result]) translation result_json.get(translation, []) if translation: return translation[0] return 翻译结果为空 except requests.exceptions.RequestException as e: return f翻译请求失败{str(e)} except json.JSONDecodeError as e: return f响应解析失败{str(e)} except Exception as e: return f翻译出错{str(e)}为了使得界面可视化更加便于演示该工具我使用stream设计了一个简单的前端页面该页面通过MCP协议获取工具列表并它使用json格式返回MCP_URL http://127.0.0.1:8000/mcp MCP_HEADERS {Content-Type: application/json, Accept: application/json} def load_tools(): 从 MCP 服务器获取工具列表 try: with httpx.Client(timeout10.0) as c: # MCP 协议tools/list 方法 r c.post(MCP_URL, headersMCP_HEADERS, json{ jsonrpc: 2.0, id: 1, method: tools/list, params: {} }) if r.status_code 200: return r.json().get(result, {}).get(tools, []) except: pass return [] # 页面加载时获取工具 if not st.session_state.tools: st.session_state.tools load_tools()返回格式{ name: translate, description: 翻译文本内容。支持多种语言互译。, inputSchema: { type: object, properties: { text: {type: string}, from_lang: {type: string}, to_lang: {type: string} }, required: [text] } }然后转换为OpenAI Tool格式并传给LLMdef get_tool_defs(): 将 MCP 工具转换为 OpenAI Tool 格式 return [{ type: function, function: { name: t[name], # 工具名称 description: t.get(description, ), # 工具描述 parameters: { type: object, properties: t.get(inputSchema, {}).get(properties, {}), required: t.get(inputSchema, {}).get(required, []) } } } for t in st.session_state.tools] # 调用 LLM 时传入工具定义 response client.chat.completions.create( modeldeepseek-chat, messagesmessages, toolsget_tool_defs(), # ← 关键传入工具列表 tool_choiceauto # 让 LLM 自动决定是否调用工具 )整个工具的调用流程如下用户提问 → LLM 决定调用工具 → Web 应用执行工具 → LLM 生成最终回答# 第 1 步用户输入 prompt 请帮我翻译你好翻译成英文 # 第 2 步LLM 分析并决定调用工具 response client.chat.completions.create( modeldeepseek-chat, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: prompt} ], toolsget_tool_defs(), tool_choiceauto ) # LLM 返回工具调用请求 message.tool_calls [ { id: call_123, function: { name: translate, arguments: {text: 你好, from_lang: zh, to_lang: en} } } ] # 第 3 步Web 应用调用 MCP 工具 def call_mcp_tool(tool_name, arguments): with httpx.Client(timeout30.0) as c: r c.post(MCP_URL, headersMCP_HEADERS, json{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: tool_name, arguments: arguments } }) return r.json()[result][content][0][text] tool_result call_mcp_tool(translate, {text: 你好, from_lang: zh, to_lang: en}) # 返回Hello # 第 4 步将工具结果返回给 LLM生成最终回答 messages.append({ role: assistant, content: , tool_calls: [...] }) messages.append({ role: tool, tool_call_id: call_123, content: Hello }) response2 client.chat.completions.create( modeldeepseek-chat, messagesmessages ) # 最终回答翻译结果是Hello示例界面完整代码1、server.pyFastMCP服务器通过mcp.tool装饰器定义工具工具自动暴露为HTTP服务端口8000from fastmcp import FastMCP from fastmcp import Context import requests import hashlib import random import os import time import json from dotenv import load_dotenv # 加载环境变量 load_dotenv() mcp FastMCP(My MCP Server) mcp.tool def greet(name: str) - str: 向指定用户发送问候 return fHello, {name}! mcp.tool def add(a: int, b: int) - int: 计算两个整数的和 return a b mcp.tool async def summarize(url: str, ctx: Context): 获取任意 HTTP/HTTPS 网页的内容并返回前 3000 字符摘要。 支持所有公开可访问的网页 URL如 https://example.com 或 http://example.com/page await ctx.info(fFetching content from {url}) headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } try: response requests.get(url, timeout10, headersheaders) response.raise_for_status() content response.text[:3000] return f网页内容摘要前 3000 字符:\n{content} except requests.exceptions.HTTPError as e: return fHTTP 错误{str(e)} except requests.exceptions.Timeout: return 请求超时请检查网络连接 except requests.exceptions.RequestException as e: return f获取网页失败{str(e)} mcp.tool def translate(text: str, from_lang: str auto, to_lang: str en) - str: 翻译文本内容。支持多种语言互译。 appKey os.getenv(TRANSLATE_APP_KEY) api_secret os.getenv(TRANSLATE_API_SECRET) if not appKey or not api_secret: return 翻译 API 配置缺失请检查 .env 文件 url https://openapi.youdao.com/api def get_input_for_sign(text): if len(text) 20: return text else: return text[:10] str(len(text)) text[-10:] salt random.randint(32768, 65536) curtime str(int(time.time())) input_for_sign get_input_for_sign(text) sign_str f{appKey}{input_for_sign}{salt}{curtime}{api_secret} sign hashlib.sha256(sign_str.encode(utf-8)).hexdigest() params { appKey: appKey, salt: salt, signType: v3, sign: sign, curtime: curtime, q: text, from: from_lang, to: to_lang, streamType: full # 指定完整响应 } try: response requests.post(url, dataparams, timeout10) response.raise_for_status() # 解析 SSE 格式响应 result_json None for line in response.iter_lines(): if line: decoded line.decode(utf-8) if decoded.startswith(data:): # 提取 data: 后面的 JSON 内容 json_str decoded[5:].strip() result_json json.loads(json_str) break if not result_json: return f无法解析响应{response.text} # 检查错误 if result_json.get(code) and result_json.get(code) ! 0: return f翻译错误{result_json.get(message, 未知错误)} (错误码{result_json.get(code)}) # 提取翻译结果 trans_result result_json.get(trans_result, []) if trans_result: return .join([item.get(dst, ) for item in trans_result]) translation result_json.get(translation, []) if translation: return translation[0] return 翻译结果为空 except requests.exceptions.RequestException as e: return f翻译请求失败{str(e)} except json.JSONDecodeError as e: return f响应解析失败{str(e)} except Exception as e: return f翻译出错{str(e)} if __name__ __main__: mcp.run(transporthttp, port8000)2、app.py客户端文件通过MCP协议tools/list获取工具列表并将MCP工具格式转换为OpenAI tool格式然后调用LLM时传入tool参数接着LLM返回tool_calls客户端执行工具最后将工具结果返回LLM生成最终回答import streamlit as st import httpx import json import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() # 页面配置 st.set_page_config(page_titleMCP 智能助手, page_icon) # 初始化 OpenAI 客户端 client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL), ) # 标题 st.title( MCP 智能助手) # MCP 配置 MCP_URL http://127.0.0.1:8000/mcp MCP_HEADERS {Content-Type: application/json, Accept: application/json} # 系统提示 SYSTEM_PROMPT 你是一个智能助手可以使用以下 MCP 工具 - greet(name): 向用户发送问候 - add(a, b): 计算两个数的和 - summarize(url): 获取网页内容并生成摘要 - translate(text, from_lang, to_lang): 翻译文本支持多语言互译 当用户请求翻译、计算、网页摘要时请调用相应工具。 语言代码zh中文en英文ja日文ko韩文fr法文de德文auto自动检测 # 初始化会话状态 if messages not in st.session_state: st.session_state.messages [] if tools not in st.session_state: st.session_state.tools [] # 侧边栏 - 极简版只有清空对话 with st.sidebar: st.header(⚙️ 操作) # 清空对话按钮 if st.button(️ 清空对话): st.session_state.messages [] st.rerun() # 工具调用函数 def call_mcp_tool(tool_name, arguments): try: with httpx.Client(timeout30.0) as c: r c.post(MCP_URL, headersMCP_HEADERS, json{jsonrpc:2.0,id:1,method:tools/call, params:{name:tool_name,arguments:arguments}}) result r.json() if result in result: return result[result][content][0][text] return f错误{result} except Exception as e: return f调用失败{e} # 获取工具列表页面加载时调用一次 def load_tools(): try: with httpx.Client(timeout10.0) as c: r c.post(MCP_URL, headersMCP_HEADERS, json{jsonrpc:2.0,id:1,method:tools/list,params:{}}) if r.status_code 200: return r.json().get(result, {}).get(tools, []) except: pass return [] # 初始化工具列表 if not st.session_state.tools: st.session_state.tools load_tools() # 获取工具定义 def get_tool_defs(): return [{ type: function, function: { name: t[name], description: t.get(description, ), parameters: { type: object, properties: t.get(inputSchema, {}).get(properties, {}), required: t.get(inputSchema, {}).get(required, []) } } } for t in st.session_state.tools] # 显示历史消息 for msg in st.session_state.messages: with st.chat_message(msg[role]): st.markdown(msg[content]) # 聊天输入 if prompt : st.chat_input(请输入您的问题...): # 用户消息 st.session_state.messages.append({role: user, content: prompt}) with st.chat_message(user): st.markdown(prompt) # AI 回复 with st.chat_message(assistant): with st.spinner(思考中...): try: # 准备消息 messages [{role: system, content: SYSTEM_PROMPT}] messages [{role: m[role], content: m[content]} for m in st.session_state.messages] # 调用 LLM response client.chat.completions.create( modeldeepseek-chat, messagesmessages, toolsget_tool_defs() if st.session_state.tools else None, tool_choiceauto ) message response.choices[0].message # 有工具调用 if message.tool_calls: tool_results [] for tc in message.tool_calls: tool_name tc.function.name tool_args json.loads(tc.function.arguments) # 显示工具调用 with st.status(f 调用 {tool_name}..., expandedTrue) as status: st.json(tool_args) tool_result call_mcp_tool(tool_name, tool_args) st.code(tool_result, languagetext) status.update(labelf✅ {tool_name} 完成, statecomplete) tool_results.append({ name: tool_name, args: tool_args, result: tool_result, id: tc.id }) # 构建工具消息 messages.append({ role: assistant, content: message.content or , tool_calls: [{ id: tr[id], function: { name: tr[name], arguments: json.dumps(tr[args]) }, type: function } for tr in tool_results] }) for tr in tool_results: messages.append({ role: tool, tool_call_id: tr[id], content: tr[result] }) # 第二次调用 response2 client.chat.completions.create( modeldeepseek-chat, messagesmessages ) final_answer response2.choices[0].message.content else: final_answer message.content or # 显示并保存 st.markdown(final_answer) st.session_state.messages.append({role: assistant, content: final_answer}) except Exception as e: st.error(f❌ 错误{e}) st.session_state.messages.append({role: assistant, content: f出错了{e}})3、.env配置文件将api密钥等私密数据进行配置有助于信息安全一旦api密钥泄露会造成不可估量的后果OPENAI_API_KEYAPIKEY OPENAI_BASE_URLhttps://api.deepseek.com本示例用的是deepseek # 翻译 API 配置有道翻译示例 TRANSLATE_APP_KEYAPPID TRANSLATE_API_SECRETAPP密钥启动方式在配置好完整的env文件后首先运行MCP服务器python serve.py然后启动客户端文件streamlit run app.py将会自动调起浏览器如果没有自动调起请手动访问localhost:8501