Python简单的MCP服务器-客户端示例

首页 › 论坛 › 置顶 › Python简单的MCP服务器-客户端示例

正在查看 1 个帖子：1-1 (共 1 个帖子)
  • 作者
    帖子
  • 2025-09-05 10:22 #25935
    Q QPY课程团队
    管理员

    在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&current_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")
    在这个应用中，最重要的部分是以 @mcp.tool() 开头的部分。这些部分可以被视为在客户端上可用的工具。例如，在上面的应用中，有三个工具：hello、exTool1 和 getirHavaDurumu。

    我们可以看到每个函数都接收参数。例如，在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("nn🔍 测试工具：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的关系或其他复杂话题。希望这对初学者来说是一篇有用的文章。

  • 作者
    帖子
正在查看 1 个帖子：1-1 (共 1 个帖子)
  • 哎呀，回复话题必需登录。