当前位置：首页 > news >正文

Model Control Protocol 使用MCP进行各种任务适配，调用工具和资源进行客户端开发

news 2025/7/23 13:41:05

开发环境准备

MCP 开发目前支持5种主流编程语言:Python、TypeScript、Java、Kotlin 和 C#。为了使示例更具代表性且易于理解,使用 Python 在 Windows 系统计算机上。可参考官方文档 https://mcp-docs.cn/quickstart/client，其中包含完整的配置指南和最佳实践。MCP 开发可以借助 uv 进行虚拟环境创建和依赖管理。uv 是新一代 Python 包管理工具，旨在替代传统的 pip、venv 和 pip-tools 工具链。得益于其采用 Rust 语言开发，相比传统工具，uv 具有显著的性能优势。uv 不仅能够更快速地安装和管理 Python 包，还提供了完整的虚拟环境管理功能。uv采用并行下载和智能缓存机制，大幅提升了依赖安装的速度。同时，uv 还提供了更精确的依赖解析和版本控制能力，能够有效避免依赖冲突问题。

安装uv

# 使用 pip 安装
pip install uv# 或使用 Powershell 安装
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

虚拟环境管理

# 创建虚拟环境
uv venv myenv# 激活虚拟环境
env\Scripts\activate# 安装依赖
uv pip install mcp
# 或从 requirements.txt 安装
uv pip install -r requirements.txt

uv也支持直接运行 Python 项目。当项目包含 pyproject.toml 配置文件时，只需一个命令就能完成依赖安装和脚本执行，如下所示:
```
uv run python script.py
```
这个命令实际上整合了传统方式中的两个步骤:先安装依赖，再运行脚本。MCP 项目之所以推荐使用 uv 进行环境管理，主要基于以下两方面的考虑。
- (1)MCP项目通常依赖多个 Python 模块，uv 通过 pyproject.toml 提供了更现代化的依赖管理方案，能够更好地处理复杂的依赖关系。
- (2) uv 优秀的依赖解析机制可以有效避免传统 pip 可能遇到的依赖冲突问题。最重要的是，uv 显著提升的包管理速度对于 MCP 这类需要频繁管理依赖的项目来说，能够明显改善开发体验。
MCP（Model Context Protocol）是Anthropic开发的一个开放协议，用于标准化应用程序如何向大语言模型（LLM）提供上下文。你可以将MCP想象成AI应用的”USB-C接口”，它提供了一种标准化的方式来连接AI模型与不同的数据源和工具。
MCP使用客户端-服务器架构：MCP Host，如Claude Desktop等应用程序，需要通过MCP访问数据；MCP Client，维护与服务器的1:1连接的协议客户端；MCP Server，通过标准化的MCP协议暴露特定功能的轻量级程序

MCP 服务器开发基础

第一个 MCP 服务器：Hello World

使用Python来构建一个简单的MCP Server，它将提供一个工具，可以返回特定名称问好。创建一个简单的MCP服务器。创建一个名为demo_server.py的文件：

from typing import Optional
from mcp.server.fastmcp import FastMCP
# 初始化FastMCP服务器
mcp = FastMCP("first demo")
@mcp.tool()
async def hello(name: Optional[str] = None) -> str:if name:return f"Hello, {name}!"else:return "Hello, World!"
if __name__ == "__main__":# 初始化并运行服务器mcp.run(transport='stdio')

FastMCP类使用Python类型提示和docstrings自动生成工具定义，使MCP工具的创建和维护变得简单。

在 Claude Desktop 中配置 MCP 服务器

为了让Claude Desktop使用我们创建的MCP服务器，我们需要修改Claude Desktop的配置文件，打开Claude Desktop的配置文件：Windows: %APPDATA%\Claude\claude_desktop_config.json
- ```
{"mcpServers": {"hello-world": {"command": "uv","args": ["--directory","/绝对路径/到你的项目目录","run","demo_server.py"]}}
}
```
- 保存文件并重启Claude Desktop，重启后，你应该能在Claude Desktop的输入框下方看到锤子图标，点击它会显示可用工具，其中包括我们创建的”hello”工具。
MCP服务器由以下部分组成：服务器标识；工具定义；工具实现。在Claude Desktop中请求使用工具时，发生了以下步骤：
- Claude分析你的请求并决定使用哪个工具
- Claude请求用户权限调用该工具
- 请求发送到MCP服务器
- 服务器执行工具功能并返回结果
- Claude接收结果并形成自然语言响应

计算器工具

再添加一个工具，可以添加更多工具到我们的服务器，例如一个简单的计算器工具：

@mcp.tool()
async def calculate(expression: str) -> str:"""简单的计算器工具，支持基本数学运算参数:expression: 数学表达式，如"2+3*4""""try:# 注意：实际生产环境中使用 eval 存在安全风险result = eval(expression)return f"计算结果: {result}"except Exception as e:return f"计算错误: {str(e)}"

MCP不仅支持工具，还支持资源。我们可以添加一个简单的资源：

import mcp.types as types
@mcp.list_resources()
async def list_resources() -> list[types.Resource]:"""列出可用资源"""return [types.Resource(uri="hello://example",name="示例资源",description="一个简单的示例资源")]
@mcp.read_resource()
async def read_resource(uri: str) -> str:"""读取资源内容"""if uri == "hello://example":return "这是一个示例资源的内容"raise ValueError(f"未知资源: {uri}")

MCP是一个强大的协议，允许AI模型安全地访问外部数据和功能。通过创建自定义MCP服务器，你可以扩展AI助手的能力，使其能够与你的应用程序、数据和工具进行交互。
还可以将通过 HTTP 请求来查询天气，因此需要安装几个核心依赖包。其中，依赖包 httpx 用于异步发起 HTTP 请求;依赖包 mcp 是使用 MCP 的必要前提。
- ```
uv add httpx mcp
```

现在我们创建一个 weather.py 文件，实现向 OpenWeather 请求天气的功能。具体步骤如下。导入依赖包。导入这个项目所需的依赖包，代码如下:

import json
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP
# 初始化服务器
mcp = FastMCP("WeatherServer")
# API 配置（请替换为自己的 API Key）
API_KEY = "YOUR_API_KEY"
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
USER_AGENT = "weather-app/1.0"
async def fetch_weather(city: str) -> dict[str, Any] | None:"""从 OpenWeather API 获取天气数据"""params = {"q": city,"appid": API_KEY,"units": "metric","lang": "zh_cn"}headers = {"User-Agent": USER_AGENT}async with httpx.AsyncClient() as client:try:response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers,timeout=30.0)response.raise_for_status()return response.json()except httpx.HTTPStatusError as e:return {"error": f"HTTP错误: {e.response.status_code}"}except Exception as e:return {"error": f"请求失败: {str(e)}"}
def format_weather(data: dict[str, Any] | str) -> str:"""格式化天气数据为易读文本"""if isinstance(data, str):try:data = json.loads(data)except Exception as e:return f"无法解析天气数据: {e}"if "error" in data:return f"{data['error']}"city = data.get("name", "未知")country = data.get("sys", {}).get("country", "未知")temp = data.get("main", {}).get("temp", "N/A")humidity = data.get("main", {}).get("humidity", "N/A")wind_speed = data.get("wind", {}).get("speed", "N/A")weather_list = data.get("weather", [{}])description = weather_list[0].get("description", "未知")return (f"{city}, {country}\n"f"温度: {temp}°C\n"f"湿度: {humidity}%\n"f"风速: {wind_speed}m/s\n"f"天气: {description}\n")
@mcp.tool()
async def query_weather(city: str) -> str:"""查询指定城市的天气信息参数:city: 城市名称"""data = await fetch_weather(city)return format_weather(data)
if __name__ == "__main__":mcp.run(transport='stdio')

API配置。进行 OpenWeather 天气査询网站的 API配置并明确通信地址，代码如下:

API KEY="YOUR API KEY"#替换为自己的
Openweather_API_Base= "https://api.openweathermap.org/data/2.5/weather"
USER_AGENT ="weather-app/1.0"

获取天气数据。定义一个异步函数，用于向 OpenWeather 网站请求城市的天气信息，并对可能出现的状态异常进行处理，代码如下:

# 定义一个查询天气的异步函数
async def fetch_weather(city:str)->dict[str,Any]| None:# hTTP请求参数设置params ={"q": city,"appid": API_KEY,"units":"metric","lang":"zh_cn"}# HTTP 请求头设置headers ={"User-Agent":USER AGENT}# 创建HTTP 客户端async with httpx.AsyncClient()as client:try:# 发送 GET 请求查询天气response=await client,get(OPENWEATHER_API_BASE, params=params, headers=headers，timeout=30.0)# 检查响应状态码，不是2xx则抛出异常response.raise_for_status()# 将响应的 JSON 数据解析为字典并返回return response.json()#处理HTTP状态错误(如404等)except httpx.HTTPStatusError as e:return{"error":f"HTTP错误:{e.response.status_code}"}#处理其他可能的问题except Exception ase:return{"error":f"请求失败:{str(e)}

数据格式化。将网站返回的复杂结构数据转换为用户易读的文本输出，代码如下:

# 定义一个名为 format weather的函数，用于处理天气数据
def format_weather(data:dict[str,Any] | str)-> str:#如果传入的是字符串，那么先将其转换为字典if isinstance(data,str):try:data =json.loads(data)except Exception as e:return f"无法解析天气数据:{e}"# 如果数据中有错误信息，那么直接返回错误提示if "error" in data:return f"{data['error']}"# 提取数据时做容错处理，确保缺少数据也能正常工作双city = data.get("name"，"未知")country = data.get("sys"，{}).get("country"，"未知")temp = data.get("main",{}).get("temp","N/A" )humidity = data.get("main",{}).get("humidity","N/A" )wind_speed = data.get("wind",{}).get("speed","N/A")# weather 可能为空列表，因此先提供默认字典赴weather_list=data.get("weather",[{}])description =weather list[0].get("description"，"未知")#使用 f-string 格式化字符串return (f"{city},{country}\n"f"温度:{temp}°C\n"f"湿度:{humidity}%\n"f"风速:{wind_speed}m/s\n"f"天气:{description}\n")

使用 MCP 工具函数。使用 MCP 装饰器封装一个工具，实现天气的查询以及结果的格式化，代码如下:

# 使用 MCP的装饰器将其标记为一个工具
@mcp.tool()
#定义一个异步函数并声明其用途
async def query_weather(city:str)-> str:# 调用查询天气函数data = await fetch_weather(city)# 调用数据格式化函数return format_weather(data)

主程序入口。定义程序入口函数，启动 MCP服务器，并指定使用 STDIO 作为通信方式，代码如下:

if __name__ == "__main__":#以 STDIO 方式运行 MCP 服务器mcp.run(transport='stdio')

上文成功构建了一个能够査询天气的 MCP Server，接下来将该 MCP Server 配置到IDE 中以便实际应用，此处在 Trae 上演示。在Trae 中配置 MCP Server。在 Trae 中将 MCP 的手动配置部分添加 MCP Server,然后就可以启动MCP Server了。这相当于执行命令
- ```
uv --directory .\MCP_demo\example run weather.py
```
- 注意，需将示例中的文件夹路径替换为实际路径。测试 MCP Server。现在测试一下 MCP Server 的效果，在对话界面中，输入“今天武汉天气怎么样”，随后可以看到 LLM 调用了 MCP Server，成功执行并返回了结果,

网页内容提取工具

import httpx
from bs4 import BeautifulSoup
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("WebScraperServer")
@mcp.tool()
async def extract_web_content(url: str, max_length: int = 500) -> str:"""提取网页的主要文本内容参数:url: 网页URL地址max_length: 最大返回长度，默认500字符"""try:async with httpx.AsyncClient() as client:response = await client.get(url, timeout=10.0)response.raise_for_status()soup = BeautifulSoup(response.text, 'html.parser')# 移除脚本和样式for script in soup(["script", "style"]):script.decompose()# 获取文本并清理text = soup.get_text()lines = (line.strip() for line in text.splitlines())chunks = (phrase.strip() for line in lines for phrase in line.split("  "))text = '\n'.join(chunk for chunk in chunks if chunk)# 截断过长的文本if len(text) > max_length:text = text[:max_length] + "..."return f"网页内容提取:\n{text}"except Exception as e:return f"提取网页内容失败: {str(e)}"
if __name__ == "__main__":mcp.run(transport='stdio')

MCP 资源开发

MCP 中的资源是可以被 LLM 访问的数据或内容，通过 list_resources 和 read_resource 方法暴露给客户端。本地文件资源示例：创建一个能够访问本地文本文件的资源服务器：

import os
import mcp.types as types
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("FileResourceServer")
# 定义可访问的文件目录（实际应用中应限制目录访问范围）
ALLOWED_DIRECTORY = "./documents"
@mcp.list_resources()
async def list_resources() -> list[types.Resource]:"""列出可用的文件资源"""resources = []if not os.path.exists(ALLOWED_DIRECTORY):os.makedirs(ALLOWED_DIRECTORY)for filename in os.listdir(ALLOWED_DIRECTORY):path = os.path.join(ALLOWED_DIRECTORY, filename)if os.path.isfile(path) and filename.endswith('.txt'):resources.append(types.Resource(uri=f"file://{filename}",name=filename,description=f"本地文本文件: {filename}"))return resources
@mcp.read_resource()
async def read_resource(uri: str) -> str:"""读取资源内容"""# 解析URI获取文件名if uri.startswith("file://"):filename = uri[7:]file_path = os.path.join(ALLOWED_DIRECTORY, filename)# 安全检查：确保不会访问允许目录外的文件if not os.path.abspath(file_path).startswith(os.path.abspath(ALLOWED_DIRECTORY)):raise ValueError("不允许访问该文件")if os.path.exists(file_path) and os.path.isfile(file_path):with open(file_path, 'r', encoding='utf-8') as f:return f.read()raise ValueError(f"未知资源: {uri}")
if __name__ == "__main__":mcp.run(transport='stdio')

数据库查询资源

创建一个可以查询 SQLite 数据库的资源：

import sqlite3
import mcp.types as types
from mcp.server.fastmcp import FastMCP
from contextlib import asynccontextmanager
mcp = FastMCP("DatabaseResourceServer")
DB_PATH = "sample.db"
# 初始化数据库示例表
def init_db():conn = sqlite3.connect(DB_PATH)cursor = conn.cursor()cursor.execute('''CREATE TABLE IF NOT EXISTS products (id INTEGER PRIMARY KEY,name TEXT,price REAL,description TEXT)''')# 插入示例数据cursor.execute("INSERT OR IGNORE INTO products VALUES (1, '笔记本电脑', 5999.99, '高性能笔记本')")cursor.execute("INSERT OR IGNORE INTO products VALUES (2, '智能手机', 3999.99, '最新款智能手机')")conn.commit()conn.close()
@asynccontextmanager
async def get_db_connection():conn = sqlite3.connect(DB_PATH)conn.row_factory = sqlite3.Row  # 启用行工厂，方便获取字段名try:yield connfinally:conn.close()
@mcp.list_resources()
async def list_resources() -> list[types.Resource]:"""列出数据库资源"""return [types.Resource(uri="db://products",name="产品列表",description="数据库中的产品信息表")]
@mcp.read_resource()
async def read_resource(uri: str) -> str:"""读取数据库资源内容"""if uri == "db://products":async with get_db_connection() as conn:cursor = conn.cursor()cursor.execute("SELECT * FROM products")rows = cursor.fetchall()# 格式化查询结果result = "产品列表:\n"for row in rows:result += f"ID: {row['id']}, 名称: {row['name']}, 价格: {row['price']}, 描述: {row['description']}\n"return resultraise ValueError(f"未知资源: {uri}")
if __name__ == "__main__":init_db()  # 初始化数据库mcp.run(transport='stdio')

MCP 客户端开发

MCP 客户端负责协调 LLM 与 MCP 服务器之间的通信，可以开发一个接入 DeepSeek 模型的 MCP Client。为此,需要安装几个核心依赖包，
- ```
uv add openai python-dotenv
```
- 其中，依赖包 openai 用于调用 OpenAI风格 LLM 的 API;依赖包 python-dotenv 用于从环境变量中读取 API_KEY 信息，这样可以更安全并方便地管理 AP密钥。创建环境变量配置（.env）
- ```
api_key=你的DeepSeek API密钥
base_url=https://api.deepseek.com
model=deepseek-chat
```
- 在项目的 API 访问配置中，采用环境变量进行管理是一种既安全又灵活的最佳实践。为此，我们需要在项目根目录下创建一个名为“.env”的文件，专门用于存储 API的关键配置信息。打开.env 文件，设置3个关键的环境变量。首先是 api_key, 这里需要填入从 DeepSeek平台获取的 API密钥;其次，设置 base_url为 https://api.deepseek.com，这是 DeepSeek 的API服务器地址;最后，将model设置为 deepseek-chat，指定所要使用的模型版本。

现在创建一个 client.py 文件，实现与 LLM 通信以及和 MCP Server 通信的功能。导入依赖包。导入这个项目所需的依赖包，代码如下:

import asyncio
import os
import sys
import json
from typing import Optional
from contextlib import AsyncExitStack
from openai import OpenAI
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
load_dotenv()
class MCPClient:def __init__(self):"""初始化 MCP 客户端"""self.exit_stack = AsyncExitStack()self.openai_api_key = os.getenv("api_key")self.base_url = os.getenv("base_url")self.model = os.getenv("model")if not self.openai_api_key:raise ValueError("未找到 API Key")# 创建 OpenAI 客户端（兼容 DeepSeek API）self.client = OpenAI(api_key=self.openai_api_key,base_url=self.base_url)self.session: Optional[ClientSession] = Noneself.stdio = Noneself.write = Noneasync def connect_to_server(self, server_script_path: str):"""连接到 MCP 服务器"""is_python = server_script_path.endswith('.py')is_js = server_script_path.endswith('.js')if not (is_python or is_js):raise ValueError("服务器脚本必须是 .py 或 .js 文件")command = "python" if is_python else "node"server_params = StdioServerParameters(command=command,args=[server_script_path],env=None)# 启动 MCP Server 并建立通信stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))self.stdio, self.write = stdio_transportself.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))# 初始化会话并列出工具await self.session.initialize()response = await self.session.list_tools()print("\n已连接到服务器，支持以下工具:", [tool.name for tool in response.tools])async def process_query(self, query: str) -> str:"""处理用户查询并与 LLM 交互"""messages = [{"role": "user", "content": query}]# 获取可用工具response = await self.session.list_tools()available_tools = [{"type": "function","function": {"name": tool.name,"description": tool.description,"parameters": tool.input_schema  # 注意：原代码中的 inputSchema 可能应为 input_schema}} for tool in response.tools]# 发送 API 请求response = self.client.chat.completions.create(model=self.model,messages=messages,tools=available_tools,tool_choice="auto")content = response.choices[0]# 判断是否需要调用工具if content.finish_reason == "tool_calls":tool_call = content.message.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)# 调用 MCP 服务器工具result = await self.session.call_tool(tool_name, tool_args)print(f"\n[调用工具 {tool_name}，参数 {tool_args}]\n")# 更新消息列表messages.append(content.message.model_dump())messages.append({"role": "tool","content": result.content[0].text,"tool_call_id": tool_call.id,})# 获取最终响应response = self.client.chat.completions.create(model=self.model,messages=messages,)return response.choices[0].message.content# 不需要调用工具，直接返回结果return content.message.contentasync def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP 客户端已启动! 输入 'quit' 退出")while True:try:query = input("\n你: ").strip()if query.lower() == 'quit':breakresponse = await self.process_query(query)print(f"\nAI: {response}")except Exception as e:print(f"\n发生错误: {str(e)}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose()async def main():if len(sys.argv) < 2:print("用法: python client.py <服务器脚本路径>")sys.exit(1)client = MCPClient()try:await client.connect_to_server(sys.argv[1])await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":asyncio.run(main())

创建 MCP Client。读取.env 中的 api_key，以创建一个与 LLM 对话的 MCP Client 实例，代码如下:

load dotenv()
class MCPClient:def __init__(self ):"""初始化 MCP 客户端"""#用于管理异步上下文self.exit_stack =AsyncExitStack()self.openai_api_key=os.getenv("api_key")# 读取 api keyself.base url=os.getenv("base_url")#读取base urlself.model= os.getenv("model")#读取model# 检査 api keyif not self.openai_api_key:raise ValueError("未找到OpenAI API Key" )#创建openAI Clientself.client =OpenAI(api_key=self.openai_api_key, base_url = self.base_url)#初始化对话self.session:Optional[ClientSession]=None

判断 MCP Server 的文件类型。判断MCP Server 的文件类型是.py 还是.js，并运行相应的命令，代码如下:

#定义一个异步函数
async def connect_to_server(self,server_script_path:str):"""连接到MCP Server 并列出可用工具"""is_python = server_script_path.endswith('.py')#判断MCP Server的文件类型is_js = server_script_path.endswith('.js')if not(is_python or is_js):raise ValueError("服务器脚本必须是·py或·js 文件")command ="python" if is python else "node" #根据ever的类型运行相应的命令server_params=StdioServerParameters(command=comand,args=[server_script_path],env=None)

列出连接的 MCP Server。与 MCP Server 进行通信,并列出 MCP Server 上的工具代码如下:

# 启动 MCP Server 并建立通信
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
# 将 stdio transport 对象解包为两个部分
self.stdio, self.write = stdio_transport
self,session = await self,exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
#初始化会话
await self.session.initialize()
#列出MCP server上的工具
response = await self.session.list_tools()
tools = response.tools
print("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])

向 API 发送请求。通过 API 与 DeepSeek 进行通信，告诉它使用的模型、用户的问题以及可以使用的工具，代码如下:

async def process_query(self,query:str)-> str:#构建一个消息列表messages = [{"role":"user","content":query}]response = await self.session.list_tools() #获取消息列表available tools=[{ #格式化工具列表"type": "function","function":{"name":tool.name,"description":tool.description,"input schema":tool.inputSchema}} for tool in response.tools]# 发送API请求response=self.client.chat.completions.create(model=self.model,messages-messages,tools=available_tools)

处理响应。判断是否需要调用 MCP Server，并将结果返回终端，代码如下:

# 处理返回的内容# 获取模型的响应内容
content=response.choices[0]
if content.finish_reason == "tool_calls": # 判断是否需要调用MCP Server#如何需要使用工具，就解析工具tool_call=content.message.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)#执行MCP Serverresult = await self.session,call_tool(tool_name, tool_args)#输出工具的调用信息用于调试print(f"\n\n[Calling tool {tool name} with args {tool args}]\n\n" )#将模型返回的调用了何种工具的数据以及工具执行完成后的数据都存入 messages 中messages.append(content.message.model_dump())messages.append({"role":"tool","content":result.content[0].text，"tool_call_id":tool_call.id,})# 将上面的结果再返回给大语言模型，用于生成最终的结果response=self.client.chat.completions.create(model=self.model,messages = messages,)#返回最终结果给终端return response.choices[0l.message.content
#如果不需要调用 MCP Server，则直接返回 Deepseek的结果给终端办
return content.message.content

交互式聊天循环。创建一个可以连续对话的MCPClient，代码如下

async def chat_loop(self):"""运行交互式聊天循环"""print("\n MCP客户端已启动!输入'quit'退出")#创建无限循环聊天while True:try:query = input("\n你:").strip()if query.lower()=='quit':#用户输入 quit 就退出循环break#执行对话response =await self.process_query(query)print(f"\n0penAI:{response}")#发生错误后的异常处理except Exception as e:print(f"\n 发生错误:{str(e)}")
# 退出时清理资源
async def cleanup(self):"""清理资源"""await self.exit stack.aclose()

定义主程序入口。定义一个主函数，以此作为程序的入口，代码如下:

async def main():# 检查是否启动了MCPerverif len(sys.argv)< 2:print("Usage: python client.py <path to server script>" )sys.exit(1)# 创建MCPClient 类的实例client = MCPClient()try:# 连接服务器await client.connect_to_server(sys.argv[1])#开始聊天循环await client.chat_loop()finally:#确保执行清理工作await client.cleanup()
#程序入口
if __name__ == "__main__ ":asyncio.run(main())

如何让本地的 MCP Client 与 MCP Server 进行通信。具体步骤如下。启动 MCP Client和 MCP Server。通过命令提示符窗口进入项目文件夹的虚拟环境，同时启动 MCP Client和 MCP Server，命令如下:
- ```
uv run client.py weather.py
```
- 对话测试。待 MCP Client 启动后，输入“今天上海天气怎么样”并按 Enter 键,可以看到MCP Client会通过自动调用MCP Server来查询天气,并返回结果,说明MCP Client与 MCP Server 实现了通信。

多工具协同使用

创建一个能够协同使用多个工具的 MCP 服务器：

from typing import Optional
import httpx
import json
from datetime import datetime
from mcp.server.fastmcp import FastMCPmcp = FastMCP("MultiToolServer")# 1. 时间工具
@mcp.tool()
async def get_current_time(timezone: Optional[str] = "UTC") -> str:"""获取当前时间参数:timezone: 时区，默认为 UTC"""try:# 简单实现，实际应用可使用 pytz 等库处理时区now = datetime.now()return f"当前时间 ({timezone}): {now.strftime('%Y-%m-%d %H:%M:%S')}"except Exception as e:return f"获取时间失败: {str(e)}"# 2. 翻译工具（使用公开API示例）
@mcp.tool()
async def translate_text(text: str, target_lang: str = "en") -> str:"""翻译文本参数:text: 要翻译的文本target_lang: 目标语言，如 'en' 英语, 'zh' 中文, 'fr' 法语"""try:url = "https://translate.argosopentech.com/translate"params = {"q": text,"source": "auto","target": target_lang}async with httpx.AsyncClient() as client:response = await client.get(url, params=params)result = response.json()return f"翻译结果: {result.get('translatedText', '翻译失败')}"except Exception as e:return f"翻译失败: {str(e)}"# 3. 综合信息查询工具（调用其他工具）
@mcp.tool()
async def get_comprehensive_info(city: str) -> str:"""获取城市的综合信息（时间、天气等）参数:city: 城市名称"""# 这里仅作示例，实际中应通过工具调用机制而非直接函数调用time_info = await get_current_time()# 模拟天气查询weather_info = f"模拟天气信息: {city} 晴朗，25°C"return f"{city} 综合信息:\n{time_info}\n{weather_info}"if __name__ == "__main__":mcp.run(transport='stdio')

知识库问答系统

结合资源和工具，创建一个能够基于文档回答问题的系统：

import os
import mcp.types as types
from mcp.server.fastmcp import FastMCP
from sentence_transformers import SentenceTransformer, util
import torch
mcp = FastMCP("KnowledgeBaseServer")
# 文档存储和嵌入模型
DOCUMENTS_DIRECTORY = "./knowledge_docs"
model = SentenceTransformer('all-MiniLM-L6-v2')
document_embeddings = {}
document_texts = {}
def load_documents():"""加载文档并创建嵌入"""if not os.path.exists(DOCUMENTS_DIRECTORY):os.makedirs(DOCUMENTS_DIRECTORY)for filename in os.listdir(DOCUMENTS_DIRECTORY):if filename.endswith('.txt'):path = os.path.join(DOCUMENTS_DIRECTORY, filename)with open(path, 'r', encoding='utf-8') as f:text = f.read()document_texts[filename] = textdocument_embeddings[filename] = model.encode(text, convert_to_tensor=True)@mcp.list_resources()
async def list_resources() -> list[types.Resource]:"""列出知识库中的文档"""load_documents()  # 实际应用中应按需更新而非每次调用都加载return [types.Resource(uri=f"kb://{filename}",name=filename,description="知识库文档") for filename in document_texts.keys()]@mcp.read_resource()
async def read_resource(uri: str) -> str:"""读取知识库文档内容"""if uri.startswith("kb://"):filename = uri[5:]if filename in document_texts:return document_texts[filename][:1000]  # 返回前1000字符raise ValueError(f"未知资源: {uri}")@mcp.tool()
async def answer_from_kb(question: str) -> str:"""从知识库中查找问题答案参数:question: 要回答的问题"""load_documents()if not document_embeddings:return "知识库为空，请先添加文档"# 生成问题嵌入question_embedding = model.encode(question, convert_to_tensor=True)# 查找最相似的文档similarities = {}for filename, embedding in document_embeddings.items():similarity = util.cos_sim(question_embedding, embedding).item()similarities[filename] = similarity# 获取最相似的文档most_similar = max(similarities.items(), key=lambda x: x[1])filename, score = most_similarif score < 0.3:  # 相似度阈值return "知识库中没有找到相关信息"# 返回最相关的文档片段return (f"根据知识库中的 '{filename}' 文档:\n"f"相关度: {score:.2f}\n"f"信息片段: {document_texts[filename][:500]}...")if __name__ == "__main__":load_documents()mcp.run(transport='stdio')

在实际应用中，可以配置多个 MCP 服务器提供不同功能：

{"mcpServers": {"weather-server": {"command": "uv","args": ["--directory", "/path/to/weather", "run", "weather.py"]},"calculator-server": {"command": "uv","args": ["--directory", "/path/to/calculator", "run", "calculator.py"]},"knowledge-base": {"command": "uv","args": ["--directory", "/path/to/kb", "run", "kb_server.py"]}}
}