目 录CONTENT

文章目录
MCP

【LLM】构建和使用MCP Server:理解python-sdk与fastmcp

EulerBlind
2025-07-02 / 0 评论 / 0 点赞 / 0 阅读 / 0 字

Model Context Protocol(MCP)是一种开放标准,旨在标准化大型语言模型(LLMs)与外部上下文和工具的交互方式。在本文中,我们将深入探讨如何使用Python构建MCP服务器和客户端,并比较官方的python-sdk与fastmcp这两个项目的区别与联系。

MCP的核心概念

在开始实际编码之前,让我们先了解MCP的三个核心概念:

  1. 工具(Tools):允许LLM执行操作,例如计算、API调用或数据处理。
  2. 资源(Resources):为LLM提供只读数据,如文件内容、数据库记录等。
  3. 提示(Prompts):帮助LLM更有效地与服务器交互的模板。

这三个概念共同构成了MCP的基础功能,使AI模型能够安全地访问外部工具和数据。

python-sdk vs fastmcp

在实现MCP服务器之前,我们需要了解两个主要的Python实现:

python-sdk

python-sdk是Model Context Protocol的官方Python实现。作为官方实现,它提供了完整而灵活的API,支持MCP协议的所有功能。

特点:

  • 完整实现MCP规范
  • 提供低级和高级API
  • 支持多种传输方式(SSE和STDIO)
  • 灵活性高,适合需要精细控制的场景
  • 持续维护和更新

fastmcp

fastmcp是由Jerad Lowin创建的更高级别的封装库,目标是提供更Pythonic、更简洁的MCP服务器开发体验。

特点:

  • 简洁的装饰器语法
  • 自动类型处理
  • 集成的CLI工具
  • 快速开发体验

值得注意的是,fastmcp已经被整合到官方python-sdk中,原仓库不再单独维护。所以现在推荐直接使用官方python-sdk。

代码对比

让我们通过代码示例来比较这两种方式:

使用python-sdk低级API:

from mcp.server.lowlevel import Server
from mcp.types import Tool, TextContent

# 创建MCP服务器
mcp_server = Server("MCPDemoServer")

# 定义工具
ADD_TOOL = Tool(
    name="add",
    description="Add two numbers",
    inputSchema={
        "type": "object",
        "properties": {
            "a": {"type": "integer", "description": "First number"},
            "b": {"type": "integer", "description": "Second number"}
        },
        "required": ["a", "b"]
    }
)

@mcp_server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "add":
        a = arguments.get("a", 0)
        b = arguments.get("b", 0)
        result = a + b
        return [TextContent(type="text", text=str(result))]
    else:
        raise ValueError(f"Unknown tool: {name}")

使用fastmcp高级API:

from fastmcp import FastMCP

mcp = FastMCP("Demo 🚀")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

可以看到,fastmcp大大简化了代码,使开发者能够专注于业务逻辑而非协议细节。

实现MCP服务器

现在,让我们看看如何实现一个完整的MCP服务器,支持SSE和STDIO两种传输方式。

服务器核心组件

我们的服务器需要以下组件:

  1. 工具定义和处理
  2. 资源定义和处理
  3. SSE传输支持
  4. STDIO传输支持

1. 工具定义和处理

# 定义工具
ADD_TOOL = Tool(
    name="add",
    description="Add two numbers",
    inputSchema={
        "type": "object",
        "properties": {
            "a": {"type": "integer", "description": "First number"},
            "b": {"type": "integer", "description": "Second number"}
        },
        "required": ["a", "b"]
    }
)

AVAILABLE_TOOLS = [ADD_TOOL]

@mcp_server.call_tool()
async def call_tool(name: str, arguments: dict):
    logger.info(f"[mcp_server.call_tool] name={name}, arguments={arguments}")
    if name == "add":
        a = arguments.get("a", 0)
        b = arguments.get("b", 0)
        result = a + b
        return [TextContent(type="text", text=str(result))]
    else:
        raise ValueError(f"Unknown tool: {name}")

@mcp_server.list_tools()
async def list_tools():
    logger.info("[mcp_server.list_tools] called")
    return AVAILABLE_TOOLS

2. 资源定义和处理

# 定义资源
GREETING_RESOURCE = {
    "uri_pattern": "greeting://{name}",
    "name": "greeting",
    "description": "Get a personalized greeting"
}

AVAILABLE_RESOURCES = [GREETING_RESOURCE]

@mcp_server.list_resources()
async def list_resources():
    logger.info("[mcp_server.list_resources] called")
    return AVAILABLE_RESOURCES

@mcp_server.read_resource()
async def read_resource(uri: str):
    logger.info(f"[mcp_server.read_resource] uri={uri}")
    if uri.startswith("greeting://"):
        name = uri.replace("greeting://", "")
        return f"Hello, {name}!"
    else:
        raise ValueError(f"Unknown resource URI: {uri}")

3. SSE传输支持

SSE(Server-Sent Events)允许服务器向客户端推送事件,适合Web应用的实时通信。

# SSE端点
async def sse_endpoint(request: Request):
    logger.info("[SSE] => sse_endpoint called")

    # 创建SSE传输
    sse_transport = SseServerTransport("/sse/messages")
    scope = request.scope
    receive = request.receive
    send = request._send

    async with sse_transport.connect_sse(scope, receive, send) as (read_st, write_st):
        logger.info("[SSE] run mcp_server with SSE transport => calling 'mcp_server.run'")
        init_opts = mcp_server.create_initialization_options()
        # 运行服务器
        await mcp_server.run(read_st, write_st, init_opts)

    logger.info("[SSE] => SSE session ended")
    return

我们使用Starlette框架提供HTTP端点,并使用 SseServerTransport处理SSE连接。

4. STDIO传输支持

STDIO传输使用标准输入/输出流进行通信,适合命令行应用和本地集成。

# 运行Stdio服务器
def run_stdio_server():
    """运行标准输入输出服务器"""
    import sys
    import asyncio
  
    logger.info("Starting MCP server with stdio transport")
  
    # 创建初始化选项
    init_opts = mcp_server.create_initialization_options()
  
    # 使用低级API运行服务器
    async def run():
        try:
            # 创建传输
            from mcp.server.stdio import StdioServerTransport
            transport = StdioServerTransport(sys.stdin.buffer, sys.stdout.buffer)
          
            # 使用连接上下文管理器
            async with transport.connect() as (read_stream, write_stream):
                await mcp_server.run(read_stream, write_stream, init_opts)
        except Exception as e:
            logger.error(f"STDIO server error: {e}")
            import traceback
            logger.error(traceback.format_exc())
            sys.exit(1)
  
    # 运行异步函数
    asyncio.run(run())

完整服务器

将这些组件组合起来,我们得到一个完整的MCP服务器,可以根据命令行参数选择使用SSE或STDIO传输:

def main():
    """主函数,根据参数运行不同类型的服务器"""
    import argparse
  
    parser = argparse.ArgumentParser(description="MCP Demo Server")
    parser.add_argument("--transport", choices=["sse", "stdio"], default="sse",
                        help="Transport type (sse or stdio)")
    args = parser.parse_args()
  
    if args.transport == "stdio":
        run_stdio_server()
    else:  # 默认使用SSE
        uvicorn.run(app, host="0.0.0.0", port=8000)

if __name__ == "__main__":
    main()

实现MCP客户端

在服务器实现之后,我们需要一个客户端来与服务器交互。客户端也需要支持SSE和STDIO两种传输方式。

客户端核心组件

  1. 传输方式选择
  2. 客户端会话管理
  3. 工具调用和资源访问

1. 传输方式选择

async def run_client(transport_type: str, server_command: Optional[str] = None, server_url: Optional[str] = None):
    """运行MCP客户端"""
    logger.info(f"Starting MCP client with {transport_type} transport")
  
    if transport_type == "stdio":
        if not server_command:
            server_path = os.path.join(SCRIPT_DIR, "server.py")
            server_command = f"{sys.executable} {server_path} --transport stdio"
      
        # 创建服务器参数
        cmd_parts = server_command.split()
        server_params = StdioServerParameters(
            command=cmd_parts[0],
            args=cmd_parts[1:] if len(cmd_parts) > 1 else []
        )
      
        # 连接到服务器
        async with stdio_client(server_params) as (read, write):
            # 创建客户端会话
            async with ClientSession(read, write) as session:
                # 初始化连接
                await session.initialize()
              
                # 调用工具和读取资源
                await demo_client_operations(session)
              
    elif transport_type == "sse":
        if not server_url:
            server_url = "http://localhost:8000/sse"
      
        # 连接到服务器
        async with sse_client(server_url) as (read, write):
            # 创建客户端会话
            async with ClientSession(read, write) as session:
                # 初始化连接
                await session.initialize()
              
                # 调用工具和读取资源
                await demo_client_operations(session)

2. 工具调用和资源访问

async def demo_client_operations(session: ClientSession):
    """执行演示操作"""
    try:
        # 列出可用工具
        logger.info("Listing available tools...")
        tools = await session.list_tools()
        logger.info(f"Available tools: {tools}")
      
        # 调用add工具
        logger.info("Calling 'add' tool...")
        result = await session.call_tool("add", arguments={"a": 5, "b": 3})
        logger.info(f"Result of add(5, 3): {result}")
      
        # 列出可用资源
        logger.info("Listing available resources...")
        resources = await session.list_resources()
        logger.info(f"Available resources: {resources}")
      
        # 读取greeting资源
        logger.info("Reading 'greeting' resource...")
        try:
            uri = "greeting://World"
            content, mime_type = await session.read_resource(uri)
            logger.info(f"Greeting content: {content}, mime type: {mime_type}")
        except Exception as e:
            logger.error(f"Error reading resource: {e}")
    except Exception as e:
        logger.error(f"Error during client operations: {e}")

测试MCP实现

为了确保我们的实现正常工作,我们需要编写测试代码来验证服务器和客户端的功能。

def test_stdio():
    """测试标准输入/输出连接"""
    logger.info("=== 测试 STDIO 连接 ===")
  
    # 运行客户端(客户端会自动启动服务器)
    return_code = run_client(transport="stdio")
  
    if return_code == 0:
        logger.info("STDIO 测试成功!")
    else:
        logger.error(f"STDIO 测试失败,返回码: {return_code}")
  
    return return_code

def test_sse():
    """测试 SSE 连接"""
    logger.info("=== 测试 SSE 连接 ===")
  
    # 启动服务器
    server_proc = run_server(transport="sse")
  
    try:
        # 等待服务器启动
        logger.info("等待服务器启动...")
        time.sleep(2)
      
        # 运行客户端
        return_code = run_client(transport="sse", server_url="http://localhost:8000/sse")
      
        if return_code == 0:
            logger.info("SSE 测试成功!")
        else:
            logger.error(f"SSE 测试失败,返回码: {return_code}")
      
        return return_code
  
    finally:
        # 停止服务器
        logger.info("停止服务器...")
        server_proc.terminate()
        server_proc.wait()

实际应用与挑战

在实现MCP服务器和客户端的过程中,我们遇到了一些挑战:

  1. 类型兼容性问题:MCP API需要特定的类型(如AnyUrl),但在不同版本中可能有差异。
  2. 错误处理:需要处理各种异常情况,如连接失败、协议错误等。
  3. 调试复杂性:由于涉及到网络通信和异步编程,调试可能变得复杂。

选择哪个框架?

  • 如果你是MCP新手:建议使用官方python-sdk中的高级API(原fastmcp)。
  • 如果你需要最大灵活性:使用python-sdk的低级API。
  • 如果你正在构建生产系统:使用官方python-sdk,因为它有持续的维护和更新。

总结

Model Context Protocol为大型语言模型提供了一种标准化的方式来访问外部工具和数据。通过官方python-sdk或其高级封装,我们可以轻松构建MCP服务器和客户端。

fastmcp作为一个高层封装,极大地简化了MCP服务器的开发,但现在已经被整合到官方python-sdk中,所以推荐直接使用官方SDK。

无论选择哪种方式,MCP都为AI应用开发提供了强大的工具,使LLM能够安全、标准化地与外部世界交互。

参考资源

0

评论区