精通 MCP Server和Client 01

❝

喜欢阅读，写作和摄影的「coder」

❞

精通MCP Server|Client

为什么用MCP？

MCP是什么我就不解释了，最近MCP的相关博客，推文已经烂大街了，我这里只讲硬核的东西，你一般见不到的内容。

在我看来，MCP跟传统的tool call 最大的区别就是动态可插拔的特性：

某个tool 组件有哪些tool可以调用这个是动态的，当我们在某个tool group新增了tool或者删改了tool，所有能连接到当前「tool group server」的client都能感知到。这是非常有用的。

当然了，除此之外它的通用性，层次架构更加清晰，拆分了resource，tools，root等概念。

另外，一个技术值不值得学，只要一个标准就是「生态」和「大厂背书」，这个就毋庸置疑了，除了OpenAI公开兼容，目前阿里，腾讯已经开始全面支持，这是一个tool call的大趋势。

tool call 从原始的「刀耕火种」时代迈入了「青铜器」时代。

SSE Transport

官方的示例是基于stdio模式的「https://modelcontextprotocol.io/quickstart/server」，这是最初的一种模式。

在这种模式下，你的server和client要在同一台机器。客户端启动「https://modelcontextprotocol.io/quickstart/client」需要这样：

uv run client.py path/to/server.py # python server

client的代码是这样的：

server_params = StdioServerParameters(
  command="python",
  args=[server_script_path],
  env=None
)

stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))

网上99%的教程也都是这种，但你是不是有一种「担心」：如果我们需要的server 很多，这要全部在一台机器上，要改，所有的机器也都要重启，这不是回到原来的尴尬境地了吗？

你想到的，MCP肯定也想到啦。因此，除了stdio（「Standard Input/Output」）模式，MCP还支持sse（「Server-Sent Events」）模式，这是一种基于网络 远程的调用方式。

def run(self, transport: Literal["stdio", "sse"] = "stdio")
....

官方这样解释的：

❝

SSE 传输通过 HTTP POST 请求实现服务器到客户端的流式传输，从而实现客户端到服务器的通信。

在以下情况下使用 SSE：

• 仅需要服务器到客户端的流式传输
• 使用受限网络
• 实现简单更新

❞

因为是基于网络的工作方式，官方对于SSE还加了额外「注意事项」说明：

❝

安全警告：DNS重新绑定攻击

如果没有妥善保护，SSE 传输可能容易受到 DNS 重新绑定攻击。为了防止这种情况发生：

1. 始终验证传入 SSE 连接上的 Origin 标头，以确保它们来自预期的来源
2. 在本地运行时，避免将服务器绑定到所有网络接口（0.0.0.0） - 仅绑定到本地主机（127.0.0.1）
3. 为所有 SSE 连接实施适当的身份验证

如果没有这些保护措施，攻击者可以使用 DNS 重新绑定从远程网站与本地 MCP 服务器进行交互。

❞

好的，基本前置已经准备完毕，下面开始进入期待的demo环节。

SSE Server

对于构建Server官方SDK构建了「mcp.server.fastmcp.FastMCP」用于快速构建。当然，你也可以使用「mcp.server.Server」构建，原理是一样的，FastMCP封装了一层，更加便捷。

因为是示例demo，我们这里先只以最简单的方式构建：构建两个经典的tool：天气查询和货币汇率转换。

1 ⚙初始化FastMCP server

from mcp.server.fastmcp import FastMCP

# 默认的host是0.0.0.0，port是8000
mcp = FastMCP("sse_weather", host="localhost", port=9990)

更多可配置项参考：「mcp.server.fastmcp.server.Settings」 类。

2 ?编写tool，这个看起来很熟悉，类似「langchain」的方式。

@mcp.tool()
async def get_weather(city: str) -> str:
    """Get weather information for a city.

    Args:
        city: Name of the city
    """

    return f"{city} is sunny, enjoy it!"

@mcp.tool()
def convert(amount: float, currency_from: str, currency_to: str) -> float:
    """use latest exchange rate to convert currency

    Args:
      amount: the amount of currency to convert
      currency_from: the currency to convert from
      currency_to: the currency to convert to
    """
    return amount * 0.8

3 ?运行 server

if __name__ == "__main__":
    mcp.run(transport='sse')

一切??，程咬金的三板?有没有。

我是使用的uv「https://docs.astral.sh/uv」作为Python的包管理器。

这个工具使用起来还是很方便的，有非常浓重的?Rust包管理的味道（因为它是Rust语言开发的）。最新版本的pychaorm也支持了uv作为包管理器，创建项目也非常方便。

将上面的代码保存为sse_weather.py

于是乎我使用如下命令启动：

 uv run sse_weather.py

如果看到下面显示，则成功了

INFO:     Started server process [24973]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:9990 (Press CTRL+C to quit)

SSE Client

1 创建Client对象

import asyncio
from contextlib import AsyncExitStack
from typing import Optional

from mcp import ClientSession, ListToolsResult
from mcp.client.sse import sse_client
from mcp.types import TextContent, ImageContent, EmbeddedResource, Tool, CallToolResult


class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

2 ?增加连接server的方法

    async def connect_to_server(self, server_url: str):
        """Connect to an MCP server

        Args:
            server_url: URL of the server
        """

        sc = sse_client(url=server_url)
        sse_transport = await self.exit_stack.enter_async_context(sc)
        sse, write = sse_transport
        cs = ClientSession(sse, write)
        self.session = await self.exit_stack.enter_async_context(cs)

        await self.session.initialize()

3 ?增加tool_list 方法和tool_call方法

    async def list_tools(self) -> list[Tool]:
        """List available tools"""
        response: ListToolsResult = await self.session.list_tools()
        print(response.model_dump())
        return response.tools


    async def tool_call(self, tool_name: str, args: dict) -> list[TextContent | ImageContent | EmbeddedResource]:
        """Call a tool by name with arguments"""
        response: CallToolResult = await self.session.call_tool(tool_name, args)
        return response.content

4 ♻️资源清理方法

    async def cleanup(self):
        """Clean up resources"""
        await self.exit_stack.aclose()

然后就可以测试了：

async def main():
    client = MCPClient()
    try:
        await client.connect_to_server("http://127.0.0.1:9990/sse")
        await client.list_tools()
        r = await client.tool_call("get_weather", {"city": "Beijing"})
        print(r)

    finally:
        await client.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

同样的，使用uv 可以启动

uv run sse_client.py

看一下输出

首先是看一下「ListToolsResult」对象：

{
  "meta": "None",
"nextCursor": "None",
"tools": [
    {
      "name": "get_weather",
      "description": "Get weather information for a city.\n\n    Args:\n        city: Name of the city\n    ",
      "inputSchema": {
        "properties": {
          "city": {
            "title": "City",
            "type": "string"
          }
        },
        "required": [
          "city"
        ],
        "title": "get_weatherArguments",
        "type": "object"
      }
    },
    {
      "name": "convert",
      "description": "use latest exchange rate to convert currency\n\n    Args:\n      amount: the amount of currency to convert\n      currency_from: the currency to convert from\n      currency_to: the currency to convert to\n    ",
      "inputSchema": {
        "properties": {
          "amount": {
            "title": "Amount",
            "type": "number"
          },
          "currency_from": {
            "title": "Currency From",
            "type": "string"
          },
          "currency_to": {
            "title": "Currency To",
            "type": "string"
          }
        },
        "required": [
          "amount",
          "currency_from",
          "currency_to"
        ],
        "title": "convertArguments",
        "type": "object"
      }
    }
  ]
}

这个格式应该很熟悉了吧。

然后是调用天气的工具之后，server端的输出：

[TextContent(type='text', text='Beijing is sunny, enjoy it!', annotations=None)]

到这里，基本的东西就介绍完毕了，后面我们会讲如何跟我们熟悉的LLM结合起来实现tool call。

关注不迷路。