[MCP][03]使用FastMCP开发MCP应用

前言

之前的示例用的都是MCP的官方SDK（版本 1.14.0），简单使用还是没问题的，但对于Sampling、Elicitation这些相对高级的功能，官方没有提供Demo，而且因为比较新，网上也没搜到能用的案例。以我自己的水平折腾了一天也没捣鼓出来。在翻mcp源码时意外发现了其内置的FastMCP，顺藤摸瓜找到了FastMCP的官网，在官方文档中找到了相关用法。这里我们就用FastMCP来实现之前用mcp官方sdk做的功能，看看它有什么优势。
安装

截至本文日期的fastmcp版本为 2.12.2

1. # uv
2. uv add fastmcp
4. # pip
5. python -m pip install fastmcp

复制代码

MCP Server

MCP Server的写法跟之前使用mcp官方sdk差不多，只是导入FastMCP的地方和运行配置不太一样。

1. from fastmcp import FastMCP
2. from typing import TypeAlias, Union
3. from datetime import datetime
4. import asyncio
5. import asyncssh
7. mcp = FastMCP("custom")
9. Number: TypeAlias = Union[int, float]
10. @mcp.tool()
11. def add(a: Number, b: Number) -> Number:
12.     """Add two numbers"""
13.     return a + b
15. @mcp.tool()
16. def multiply(a: Number, b: Number) -> Number:
17.     """Multiply two numbers"""
18.     return a * b
20. @mcp.tool()
21. def is_greater_than(a: Number, b: Number) -> bool:
22.     """Check if a is greater than b
23.    
24.     Args:
25.         a (Number): The first number
26.         b (Number): The second number
28.     Returns:
29.         bool: True if a is greater than b, False otherwise
30.     """
31.     return a > b
33. @mcp.tool()
34. async def get_weather(city: str) -> str:  
35.     """Get weather for a given city."""
36.     return f"It's always sunny in {city}!"
38. @mcp.tool()
39. async def get_date() -> str:
40.     """Get today's date."""
41.     return datetime.now().strftime("%Y-%m-%d")
43. @mcp.tool()
44. async def execute_ssh_command_remote(hostname: str, command: str) -> str:
45.     """Execute an SSH command on a remote host.
46.    
47.     Args:
48.         hostname (str): The hostname of the remote host.
49.         command (str): The SSH command to execute.
51.     Returns:
52.         str: The output of the SSH command.
53.     """
54.     try:
55.         async with asyncssh.connect(hostname, username="rainux", connect_timeout=10) as conn:
56.             result = await conn.run(command, timeout=10)
57.             stdout = result.stdout
58.             stderr = result.stderr
59.             content = str(stdout if stdout else stderr)
60.             return content
61.     except Exception as e:
62.         return f"Error executing command '{command}' on host '{hostname}': {str(e)}"
63.    
64. @mcp.tool()
65. async def execute_command_local(command: str, timeout: int = 10) -> str:
66.     """Execute a shell command locally.
67.    
68.     Args:
69.         command (str): The shell command to execute.
70.         timeout (int): Timeout in seconds for command execution. default is 10 seconds.
72.     Returns:
73.         str: The output of the shell command.
74.     """
75.     try:
76.         proc = await asyncio.create_subprocess_shell(
77.             command,
78.             stdout=asyncio.subprocess.PIPE,
79.             stderr=asyncio.subprocess.PIPE
80.         )
81.         stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
82.         stdout_str = stdout.decode().strip()
83.         stderr_str = stderr.decode().strip()
84.         # content = stdout.decode() if stdout else stderr.decode()
85.         if stdout_str:
86.             return f"Stdout: {stdout_str}"
87.         elif stderr_str:
88.             return f"Stderr: {stderr_str}"
89.         else:
90.             return "Command executed successfully with no output"
91.     except asyncio.TimeoutError:
92.         if proc and not proc.returncode:
93.             try:
94.                 proc.terminate()
95.                 await proc.wait()
96.             except:
97.                 pass
98.         return f"Error: Command '{command}' timed out after {timeout} seconds"
99.     except Exception as e:
100.         return f"Error executing command '{command}': {str(e)}"
102. if __name__ == "__main__":
103.     mcp.run(transport="http", host="localhost", port=8001, show_banner=False)

复制代码

因为使用http协议，所以运行client前要先运行server，顺带测试下能否正常启动。
MCP Client

FastMCP的client写法与mcp官方sdk用法大致上也差不多，但在一些细节上更加友好。

1. """
2. MCP客户端示例程序
3. 该程序演示了如何使用MCP协议与服务器进行交互，并通过LLM处理用户查询。
4. """
6. import asyncio
7. import json
8. import readline  # For enhanced input editing
9. import traceback
10. from typing import cast
11. from openai.types.chat import ChatCompletionMessageFunctionToolCall
12. from fastmcp import Client
13. from openai import AsyncOpenAI
14. from pkg.config import cfg
15. from pkg.log import logger
18. class MCPHost:
19.     """MCP主机类，用于管理与MCP服务器的连接和交互"""
20.    
21.     def __init__(self, server_uri: str):
22.         """
23.         初始化MCP客户端
24.         
25.         Args:
26.             server_uri (str): MCP服务器的URI地址
27.         """
28.         # 初始化MCP客户端连接
29.         self.mcp_client: Client = Client(server_uri)
30.         # 初始化异步OpenAI客户端用于与LLM交互
31.         self.llm = AsyncOpenAI(
32.             base_url=cfg.llm_base_url,
33.             api_key=cfg.llm_api_key,
34.         )
35.         # 存储对话历史消息
36.         self.messages = []
38.     async def close(self):
39.         """关闭MCP客户端连接"""
40.         if self.mcp_client:
41.             await self.mcp_client.close()
43.     async def process_query(self, query: str) -> str:
44.         """Process a user query by interacting with the MCP server and LLM.
45.         
46.         Args:
47.             query (str): The user query to process.
49.         Returns:
50.             str: The response from the MCP server.
51.         """
52.         # 将用户查询添加到消息历史中
53.         self.messages.append({
54.             "role": "user",
55.             "content": query,
56.         })
58.         # 使用异步上下文管理器确保MCP客户端连接正确建立和关闭
59.         async with self.mcp_client:
60.             # 从MCP服务器获取可用工具列表
61.             tools = await self.mcp_client.list_tools()
62.             # 构造LLM可以理解的工具格式
63.             available_tools = []
65.             # 将MCP工具转换为OpenAI格式
66.             for tool in tools:
67.                 available_tools.append({
68.                     "type": "function",
69.                     "function": {
70.                         "name": tool.name,
71.                         "description": tool.description,
72.                         "parameters": tool.inputSchema,
73.                     }
74.                 })
75.             logger.info(f"Available tools: {[tool['function']['name'] for tool in available_tools]}")
77.             # 调用LLM，传入对话历史和可用工具
78.             resp = await self.llm.chat.completions.create(
79.                 model=cfg.llm_model,
80.                 messages=self.messages,
81.                 tools=available_tools,
82.                 temperature=0.3,
83.             )
85.             # 存储最终响应文本
86.             final_text = []
87.             # 获取LLM的首个响应消息
88.             message = resp.choices[0].message
89.             # 如果响应包含直接内容，则添加到结果中
90.             if hasattr(message, "content") and message.content:
91.                 final_text.append(message.content)
93.             # 循环处理工具调用，直到没有更多工具调用为止
94.             while message.tool_calls:
95.                 # 遍历所有工具调用
96.                 for tool_call in message.tool_calls:
97.                     # 确保工具调用有函数信息
98.                     if not hasattr(tool_call, "function"):
99.                         continue
101.                     # 类型转换以获取函数调用详情
102.                     function_call = cast(ChatCompletionMessageFunctionToolCall, tool_call)
103.                     function = function_call.function
104.                     tool_name = function.name
105.                     # 解析函数参数
106.                     tool_args = json.loads(function.arguments)
108.                     # 检查MCP客户端是否已连接
109.                     if not self.mcp_client.is_connected():
110.                         raise RuntimeError("Session not initialized. Cannot call tool.")
111.                     
112.                     # 调用MCP服务器上的指定工具
113.                     result = await self.mcp_client.call_tool(tool_name, tool_args)
115.                     # 将助手的工具调用添加到消息历史中
116.                     self.messages.append({
117.                         "role": "assistant",
118.                         "tool_calls": [
119.                             {
120.                                 "id": tool_call.id,
121.                                 "type": "function",
122.                                 "function": {
123.                                     "name": function.name,
124.                                     "arguments": function.arguments
125.                                 }
126.                             }
127.                         ]
128.                     })
130.                     # 将工具调用结果添加到消息历史中
131.                     self.messages.append({
132.                         "role": "tool",
133.                         "tool_call_id":tool_call.id,
134.                         "content": str(result.content) if result.content else ""
135.                     })
136.                
137.                 # 基于工具调用结果再次调用LLM
138.                 final_resp = await self.llm.chat.completions.create(
139.                     model=cfg.llm_model,
140.                     messages=self.messages,
141.                     tools=available_tools,
142.                     temperature=0.3,
143.                 )
144.                 # 更新消息为最新的LLM响应
145.                 message = final_resp.choices[0].message
146.                 # 如果响应包含内容，则添加到最终结果中
147.                 if message.content:
148.                     final_text.append(message.content)
149.             
150.             # 返回连接后的完整响应
151.             return "\n".join(final_text)
153.     async def chat_loop(self):
154.         """主聊天循环，处理用户输入并显示响应"""
155.         print("Welcome to the MCP chat! Type 'quit' to exit.")
157.         # 持续处理用户输入直到用户退出
158.         while True:
159.             try:
160.                 # 获取用户输入
161.                 query = input("You: ").strip()
163.                 # 检查退出命令
164.                 if query.lower() == "quit":
165.                     print("Exiting chat. Goodbye!")
166.                     break
168.                 # 跳过空输入
169.                 if not query:
170.                     continue
172.                 # 处理用户查询并获取响应
173.                 resp = await self.process_query(query)
174.                 print(f"Assistant: {resp}")
175.             
176.             # 捕获并记录聊天循环中的任何异常
177.             except Exception as e:
178.                 logger.error(f"Error in chat loop: {str(e)}")
179.                 logger.error(traceback.format_exc())
182. async def main():
183.     """主函数，程序入口点"""
184.     # 创建MCP主机实例
185.     client = MCPHost(server_uri="http://localhost:8001/mcp")
186.     try:
187.         # 启动聊天循环
188.         await client.chat_loop()
189.     except Exception as e:
190.         # 记录主程序中的任何异常
191.         logger.error(f"Error in main: {str(e)}")
192.         logger.error(traceback.format_exc())
193.     finally:
194.         # 确保客户端连接被正确关闭
195.         await client.close()
196.    
198. if __name__ == "__main__":
199.     # 运行主程序
200.     asyncio.run(main())

复制代码

FastMCP的客户端API设计更加直观，特别是在连接管理和工具调用方面，代码更简洁易懂。
client运行输出：

1. Welcome to the MCP chat! Type 'quit' to exit.
2. You: 今天的日期是什么
3. Assistant: 今天的日期是2025年9月13日。
4. You: 检查下 tx 服务器和本地的内存占用情况
5. Assistant: 以下是 tx 服务器和本地的内存占用情况：
7. ### tx 服务器
9.                total        used        free      shared  buff/cache   available
10. Mem:           3.7Gi       2.2Gi       207Mi       142Mi       1.7Gi       1.5Gi
11. Swap:             0B          0B          0B
14. ### 本地
16.                total        used        free      shared  buff/cache   available
17. Mem:           62Gi        14Gi        38Gi       487Mi        10Gi        48Gi
18. Swap:         3.8Gi          0B       3.8Gi
21. 从这些信息中可以看出，tx 服务器的内存使用较高，而本地系统仍有较多可用内存。如果需要进一步分析或采取措施，请告诉我！
22. You: 再查下硬盘
23. Assistant: 已检查 tx 服务器和本地的内存及硬盘占用情况。以下是总结：
25. ### tx 服务器
26. - **内存占用**:
27.   - 总内存: 3.7 Gi
28.   - 已用内存: 2.2 Gi
29.   - 可用内存: 1.5 Gi
30. - **硬盘使用**:
31.   - 根目录 `/`: 总大小 69G，已用 17G，可用 53G，使用率 24%
33. ### 本地
34. - **内存占用**:
35.   - 总内存: 62 Gi
36.   - 已用内存: 14 Gi
37.   - 可用内存: 48 Gi
38. - **硬盘使用**:
39.   - 根目录 `/`: 总大小 234G，已用 30G，使用率 14%
40.   - `/home`: 总大小 676G，已用 197G，使用率 31%
42. 如果需要进一步操作，请告知！
43. You: quit
44. Exiting chat. Goodbye!

复制代码

可以看到，FastMCP的基本运行逻辑是正常的，跟使用MCP官方SDK相差不大，而且还更简洁一点。
小结

使用FastMCP与使用mcp官方sdk相比，整体体验更加友好。FastMCP不仅保持了与官方SDK的兼容性，还在API设计上做了优化，使得代码更加简洁易懂。后续博客中我们会继续使用FastMCP来介绍Sampling、Elicitation等MCP的高级功能。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！