在LLM与工具的交互方面，MCP是一个重要的步骤，主要使用FastMCP工具在Python上进行开发。当然，这些工具会不断更新和发展，但截至目前，它仍然被认为是使用最广泛的库。
MCP服务器安装
首先，为了在架构中安装MCP服务器，需要使用以下命令安装fastmcp模块。
pip install fastmcp通过此命令，系统中的 Python 将能够使用 fastmcp 模块。
MCP 服务器配置
在这一步之后，我们来准备一个简单的服务器应用程序。从测试的角度来看，应用程序的最终版本可以想象成如下所示。在这里的示例中，我们将看到一个简单的 mcp 函数工具和一个调用其他 REST API 的不同函数工具。
from fastmcp import FastMCP
import requests
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from mcp.server.sse import SseServerTransport
mcp = FastMCP("测试 MCP 服务器")
@mcp.tool()
def hello(name: str) -> str:
return f"你好 {name}!"
@mcp.tool()
def exTool1(user: str) -> str:
return f"用户 {user} !"
transport = SseServerTransport("/messages/")
async def handle_sse(request):
async with transport.connect_sse(
request.scope,
request.receive,
request._send
) as (in_stream, out_stream):
await mcp._mcp_server.run(
in_stream,
out_stream,
mcp._mcp_server.create_initialization_options()
)
这段代码定义了一个异步函数 `handle_sse`，该函数接受一个请求参数。它使用 `async with` 语句来连接服务器发送事件（SSE），并在连接成功后，使用输入流和输出流来运行 MCP 服务器的相关功能。
@mcp.tool()
def getirHavaDurumu(sehir: str) -> dict:
url = f"https://api.open-meteo.com/v1/forecast?latitude=41.01&longitude=28.97¤t_weather=true"
response = requests.get(url)
data = response.json()
return {"sehir": sehir, "sonuc": data["current_weather"]}
#为两个MCP端点构建一个小型Starlette应用程序
sse_app = Starlette(
routes=[
Route("/sse", handle_sse, methods=["GET"]),
# 注意尾部斜杠以避免307重定向
Mount("/messages/", app=transport.handle_post_message)
]
)
if __name__ == "__main__":
mcp.run(transport="http")我们可以看到每个函数都接收参数。例如，在hello中是name，在exTool1中是user，而在getirHavaDurumu中则是sehir。这些参数在函数内部的使用是非常清晰的。从一个例子来看，上述应用中的sehir参数通过requests模块向open-meteo API发送请求，以获取温度和风速值。
目前我不打算深入其他部分。
在这一步之后，无论您以什么名称保存文件，只需以这种方式调用即可。
python3 server.py 预计该命令的输出如下所示。
╭────────────────────────────────────────────────────────────────────────────╮
│ │
│ _ __ ___ _____ __ __ _____________ ____ ____ │
│ _ __ ___ .'____/___ ______/ /_/ |/ / ____/ __ \ |___ \ / __ \ │
│ _ __ ___ / /_ / __ `/ ___/ __/ /|_/ / / / /_/ / ___/ / / / / / │
│ _ __ ___ / __/ / /_/ (__ ) /_/ / / / /___/ ____/ / __/_/ /_/ / │
│ _ __ ___ /_/ \____/____/\__/_/ /_/\____/_/ /_____(*)____/ │
│
│ FastMCP 2.0 │
│ 🖥️ 服务器名称: 测试 MCP 服务器 │
│ 📦 传输: 可流式传输-HTTP │
│ 🔗 服务器 URL: http://127.0.0.1:8000/mcp │
│ 🏎️ FastMCP 版本: 2.12.0 │
│ 🤝 MCP SDK 版本: 1.13.1 │
│ 📚 文档: https://gofastmcp.com │
│ 🚀 部署: https://fastmcp.cloud │
╰────────────────────────────────────────────────────────────────────────────╯
[09/02/25 23:45:24] 信息 启动 MCP 服务器 '测试 MCP 服务器' server.py:1571
使用传输协议 'http' 在
http://127.0.0.1:8000/mcp
INFO: 启动服务器进程 [281559]
INFO: 正在等待应用程序启动。
INFO: 应用程序启动完成。
INFO: Uvicorn 正在 http://127.0.0.1:8000 上运行（按 CTRL+C 退出）通过这段输出，我们可以确认 MCP 服务器已经在 8000 端口上启动。
现在是 MCP 客户端的配置
为了配置客户端，我们将使用以下应用程序。
首先，我们可以使用以下命令进行安装：
pip install "mcp[cli]"
之后，可以构建一个示例客户端应用程序来使用服务器，如下所示。
import ast
import asyncio
import pprint
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
# --- 配置 ---
SERVER_URL = "http://localhost:8000/mcp"
pp = pprint.PrettyPrinter(indent=2, width=100)
def unwrap_tool_result(resp):
"""
安全地解包来自 FastMCP 工具调用结果对象的内容。
"""
if hasattr(resp, "content") and resp.content:
# 内容是一个包含单个内容对象的列表
content_object = resp.content[0]
# 这可能是 JSON 或纯文本
if hasattr(content_object, "json"):
return content_object.json
if hasattr(content_object, "text"):
try:
# 使用 ast.literal_eval 安全地评估包含 Python 字面量的字符串
return ast.literal_eval(content_object.text)
except (ValueError, SyntaxError):
# 如果不是字面量，返回原始文本
return content_object.text
return resp
async def main():
transport = StreamableHttpTransport(url=SERVER_URL)
client = Client(transport)
print("\n连接到 FastMCP 服务器：", SERVER_URL)
async with client:
# 1. Ping 测试连接性
print("\n正在测试服务器连接...")
await client.ping()
print("✅ 服务器可达!\n")
# 2. 发现服务器能力
print("🛠️ 可用工具:")
pp.pprint(await client.list_tools())
print("\n📚 可用资源:")
pp.pprint(await client.list_resources())
print("\n💬 可用的提示：")
pp.pprint(await client.list_prompts())
# 3. 测试工具
print("\n\n🔍 测试工具：havadurumu")
raw_search = await client.call_tool(
"getirHavaDurumu",
{"sehir": "zonguldak",},
)
search_results = unwrap_tool_result(raw_search)
pp.pprint(search_results)
if __name__ == "__main__":
asyncio.run(main())
在这个应用中，最重要的部分是配置部分的 SERVER_URL，必须填写服务器的地址和端口。
接下来，我们看到将 FastMCP 工具调用转换为可读格式的 unwrap_tool_result 函数。
但真正重要的部分是在 main 函数中的内容。
第一步是检查是否能够成功访问服务器，接下来显示服务器上可用的工具、资源和提示值。目前由于我们只使用了工具，其他的将会是空的。
在工具中，我们查询了天气，并在示例中显示了对“宗古尔达克”省的天气查询。
因此，可以通过以下命令来运行客户端应用程序。
python3 client.py 预计该命令的输出将如下所示。
连接到 FastMCP 服务器： http://localhost:8000/mcp
测试服务器连接...
✅ 服务器可达！
🛠️ 可用工具：
[ Tool(name='hello', title=None, description=None, inputSchema={'properties': {'name': {'title': '姓名', 'type': 'string'}}, 'required': ['name'], 'type': 'object'}, outputSchema={'properties': {'result': {'title': '结果', 'type': 'string'}}, 'required': ['result'], 'title': '_WrappedResult', 'type': 'object', 'x-fastmcp-wrap-result': True}, annotations=None, meta={'_fastmcp': {'tags': []}}),
Tool(name='exTool1', title=None, description=None, inputSchema={'properties': {'user': {'title': '用户', 'type': 'string'}}, 'required': ['user'], 'type': 'object'}, outputSchema={'properties': {'result': {'title': '结果', 'type': 'string'}}, 'required': ['result'], 'title': '_WrappedResult', 'type': 'object', 'x-fastmcp-wrap-result': True}, annotations=None, meta={'_fastmcp': {'tags': []}}),
工具(name=’getirHavaDurumu’, title=None, description=None, inputSchema={‘properties’: {‘sehir’: {‘title’: ‘Sehir’, ‘type’: ‘string’}}, ‘required’: [‘sehir’], ‘type’: ‘object’}, outputSchema={‘additionalProperties’: True, ‘type’: ‘object’}, annotations=None, meta={‘_fastmcp’: {‘tags’: []}})]
🔍 测试工具: havadurumu
<bound method BaseModel.json of TextContent(type=’text’, text='{“sehir”:”zonguldak”,”sonuc”:{“time”:”2025-09-02T20:45″,”interval”:900,”temperature”:22.8,”windspeed”:3.0,”winddirection”:14,”is_day”:0,”weathercode”:1}}’, annotations=None, meta=None)>
通过所有这些步骤，我们实际上是在搭建并使用一个简单的MCP服务器-客户端应用程序。
当然，在这篇文章中，我想简单明了地介绍MCP应用的使用，而不深入探讨其细节、与LLM的关系或其他复杂话题。希望这对初学者来说是一篇有用的文章。
