一、前言
MCP 已经火了一段时间了,目前也有许多现有的 MCP 服务器。今天我们要做的是自己开发一个 MCP 服务器,并编写 MCP 客户端,调用 MCP 服务器。并使用本地大语言模型与之联动。
除了自己编写 MCP 客户端,我们写好的 MCP 服务器也可以提供给 Cline、CherryStudio 等平台调用。
二、FastMCP
FastMCP 是一个用于快速构建 MCP 服务器、客户端的 Python 模块。可以通过 pip 安装:
pip install fastmcp
2.1 创建 MCP 服务器
首先我们来创建一个 MCP 服务器。创建 MCP 服务器的步骤如下:
- 创建 FastMCP 实例创建工具运行 MCP 服务器
在 FastMCP 中提供了 FastMCP 类用于构建服务器,另外提供了@mcp.tool注解用来添加工具,上述流程完整代码如下:
from fastmcp import FastMCP# 1、创建 FastMCP 实例mcp = FastMCP()# 2、创建工具@mcp.tool()def get_weather(city: str, date: str): """ 获取 city 的天气情况 :param city: 城市 :param date: 日期 :return: 城市天气情况的描述 """ return f"{city} {date} 天晴,18度。"# 3、运行MCP服务器if __name__ == '__main__': mcp.run()
2.2 创建 MCP 客户端
MCP 客户端是用于从服务器获取上下文的程序。MCP 需要做的事情有:
- 连接 MCP 服务器获取工具、资源、提示词调用工具
FastMCP 提供了 Client 类用于创建 MCP 客户端,代码如下:
import asynciofrom fastmcp import Clientasync def run(): # 连接 MCP 服务器 client = Client('server.py') async with client: # 列出工具 tools = await client.list_tools() tool = tools[0] # 调用工具 tool_result = await client.call_tool(tool.name, {"city": "南昌"}) print(tool_result)if __name__ == '__main__': asyncio.run(run())
三、大语言模型调用MCP服务器
大语言模型调用 MCP 的原理就是,使用 MCP 客户端从 MCP 服务器拿到所有工具,在调用大语言模型时,以 function calling 的方式把工具传入,当大语言模型返回 tool_call 消息时,利用 MCP 客户端调用工具,并将调用结果传递给大语言模型。
3.1 OpenAI模块调用ollama中模型
首先看看如何调用 ollama 模型:
from openai import OpenAI# 创建OpenAI客户端openai_client = OpenAI( base_url='http://127.0.0.1:11434/v1', api_key="None")# 调用大语言模型response = openai_client.chat.completions.create( model='qwen3:0.6b', messages=[ { 'role': 'user', 'content': '今天北京天气怎么样?' } ])
这里我们做了两件事,分别是创建 OpenAI 客户端和调用大语言模型。在调用大语言模型时,我们可以传入工具列表。
3.2 大语言模型调用 MCP 服务器
大语言模型需要依赖 MCP 客户端调用服务器,因此还需要创建 MCP 客户端。具体代码如下:
import asyncioimport jsonfrom fastmcp import Clientfrom openai import OpenAI# 创建MCP客户端mcp_client = Client('server.py')# 创建OpenAI客户端openai_client = OpenAI( base_url='http://127.0.0.1:11434/v1', api_key="None")async def main(): async with mcp_client: # 获取工具列表,并转换格式 tools = await mcp_client.list_tools() tools = [ { "type": "function", "function": { "name": tool.name, "description": tool.description, "input_schema": tool.inputSchema } } for tool in tools ] # 调用语言模型时,传入工具 response = openai_client.chat.completions.create( model='qwen3:0.6b', messages=[ { 'role': 'user', 'content': '今天北京天气怎么样?' } ], tools=tools ) # 如果返回tool_calls,则调用工具,否则直接回答 if response.choices[0].finish_reason != 'tool_calls': print(response) else: for tool_call in response.choices[0].message.tool_calls: # 调用工具,参数由大语言模型给出 tool_result = await mcp_client.call_tool(tool_call.function.name, json.loads(tool_call.function.arguments)) print(tool_result)if __name__ == '__main__': asyncio.run(main())
这里有几点需要注意。
- 工具格式,MCP 客户端返回的工具格式不能直接使用,而是要转换成如下形式:
{ "type": "function", "function": { "name": "", "description": "", "input_schema": {} }}
- 大语言模型的 tool_call 中会返回 arguments,表示调用工具时携带的参数,但是这个参数是字符串形式,而 call_tool 需要 json,因此需要转换成 json
这样,我们整个内容就完整了。
四、总结
FastMCP 是一个非常方便的 MCP 开发模块,我们上面开发的 MCP 服务器不止可以用客户端代码调用,也可以用 Cline 等客户端调用。
另外 FastMCP 还提供了 prompts、resources 等功能,感兴趣的读者可以查看官方文档。