使用 Python SDK 开发 MCP 服务器

MCP 服务器开发 - 基于 Python SDK

介绍与设置

本指南将解释如何使用 Python SDK 构建 MCP（Model Context Protocol，模型上下文协议）服务器。MCP 服务器是核心组件，负责向各种 MCP 客户端公开资源（Resource，数据）、提示（Prompt，模型接口）和工具（Tool，可执行函数）。

MCP 的 Python SDK 支持快速服务器开发，它提供了简洁的基于装饰器（decorator）的 API，并利用 Python 的 asyncio 库高效处理并发客户端连接。

前提条件：

• Python（建议使用 3.8 或更高版本）。
• pip（Python 的包安装器）。

安装：

从 PyPI（Python 包索引）安装 MCP Python SDK：

pip install modelcontext-mcp

初始化服务器： 首先，你需要创建一个 MCPServer 类的实例。然后，通常使用装饰器定义资源、提示和工具的处理函数。最后，启动服务器以监听传入连接。

import asyncio
from modelcontext_mcp import MCPServer, ResourceContext, PromptContext, ToolContext, MCPErrorData

# 初始化服务器实例，可在全局或主异步函数中进行
server = MCPServer(
    server_id='my-first-python-mcp-server',
    server_description='A demonstration MCP server built with the Python SDK.',
    # 定义服务器能力，供客户端发现
    capabilities={
        'resources': [
            {'id': 'ping', 'description': 'A simple ping resource to check server health.'}
            # 可以在此处列出更多资源能力
        ],
        'prompts': [
            {'id': 'reverse_string', 'description': 'Reverses the input string.'}
            # 更多提示能力
        ],
        'tools': [
            {'id': 'simple_math', 'description': 'Performs basic arithmetic operations.'}
            # 更多工具能力
        ]
    }
)

async def main_server_loop():
    # 处理函数将在后续章节中使用装饰器定义和注册。
    # 目前，假设它们已在别处定义或将在下方添加。

    port = 8080
    host = '0.0.0.0'  # 监听所有可用的网络接口

    try:
        print(f'Starting Python MCP Server on ws://{host}:{port}{server.path}...') Keating on ws://{host}:{port}{server.path}...')
        await server.start(host=host, port=port)
        print('Python MCP Server is running. Press Ctrl+C to stop.')
        print(f'Registered capabilities: {server.get_capabilities()}')
        # 保持服务器无限期运行 (server.start() 负责循环)
        # 如果 server.start() 不是阻塞的，你可能需要像这样：
        # while True:
        #     await asyncio.sleep(3600) # 保持活跃
    except KeyboardInterrupt:
        print('\nKeyboardInterrupt received. Shutting down...')
    except Exception as e:
        print(f'Failed to start or run MCP server: {e}')
    finally:
        if server.is_running():
            print('Stopping server...') Keating server...')
            await server.stop()
            print('Server stopped.')

# 运行服务器 (处理函数将在下方使用装饰器定义)：
# if __name__ == "__main__":
#     # 在运行之前在此处定义处理函数
#     asyncio.run(main_server_loop())

公开资源

资源向客户端提供数据。在 Python SDK 中，你定义一个资源处理函数，并使用 @server.resource(resource_id) 装饰它。该处理函数接收一个 ResourceContext 对象，并应返回一个表示数据载荷（payload）的字典。对于流式数据，处理函数可以是异步生成器（async generator）（使用 async def 和 yield）。

# 在调用 asyncio.run(main_server_loop()) 之前定义这些处理函数
# 'server' 是之前创建的 MCPServer 实例。

# 示例 1：一个简单的 ping 资源（单次响应）
@server.resource(resource_id='ping')
async def ping_handler(context: ResourceContext) -> dict:
    client_ip = context.client_address[0] if context.client_address else 'unknown'
    print(f"Resource 'ping' requested by client: {context.client_id} from IP: {client_ip}")
    return {'message': 'pong', 'timestamp': asyncio.get_event_loop().time()}

# 示例 2：一个流式传输数字的资源（用于流式传输的异步生成器）
@server.resource(resource_id='number_stream')
async def number_stream_handler(context: ResourceContext):
    # 参数可以指定开始、结束、步长等。
    count = context.params.get('count', 5)
    delay = context.params.get('delay', 0.5)
    print(f"Resource 'number_stream' for {count} numbers requested by: {context.client_id}")
    for i in range(1, count + 1):
        await asyncio.sleep(delay)  # 模拟工作或数据获取
        yield {'number': i, 'is_even': i % 2 == 0}
    print(f"完成了为客户端 {context.client_id} 的 'number_stream' 流式传输")

公开提示

提示（Prompt）通常用于涉及某种形式的生成或转换的交互，常与 AI 模型配合使用。使用 @server.prompt(prompt_id) 定义一个提示处理函数。它接收 PromptContext 对象，并返回一个数据载荷字典，或者如果是流式提示，则通过 yield 返回多个数据载荷。

# 在调用 asyncio.run(main_server_loop()) 之前定义这些处理函数

# 示例 1：一个反转字符串的提示（单次响应）
@server.prompt(prompt_id='reverse_string')
async def reverse_string_handler(context: PromptContext) -> dict:
    input_text = context.params.get('text')
    print(f"Prompt 'reverse_string' by {context.client_id} with text: '{input_text}'")
    if not isinstance(input_text, str):
        raise MCPErrorData(code=400, message='"text" parameter must be a string.')
    return {'reversed_text': input_text[::-1], 'original_length': len(input_text)}

# 示例 2：一个模拟 LLM 流式传输提示（异步生成器）
@server.prompt(prompt_id='mock_llm_stream')
async def mock_llm_stream_handler(context: PromptContext):
    prompt_text = context.params.get('prompt', 'Tell me a story.')
    words_to_stream = f"Okay, here's a story about '{prompt_text[:20]}...': Once upon a time...".split()
    print(f"Streaming prompt 'mock_llm_stream' for client {context.client_id}")
    for word in words_to_stream:
        await asyncio.sleep(0.2)  # 模拟 LLM 令牌生成延迟
        yield {'token': word, 'type': 'word'}
    yield {'token': '<EOS>', 'type': 'control'}
    print(f"完成了为客户端 {context.client_id} 的 'mock_llm_stream' 流式传输")

公开工具

工具是客户端可以直接执行的服务器端函数。使用 @server.tool(tool_id) 定义一个工具处理函数。它接收 ToolContext 对象，并返回一个包含工具执行结果的数据载荷字典。

# 在调用 asyncio.run(main_server_loop()) 之前定义这些处理函数

@server.tool(tool_id='simple_math')
async def simple_math_handler(context: ToolContext) -> dict:
    operation = context.params.get('operation')
    a = context.params.get('a')
    b = context.params.get('b')
    print(f"Tool 'simple_math' ({operation}) by {context.client_id} with a={a}, b={b}")

    if not all(isinstance(x, (int, float)) for x in [a, b]):
        raise MCPErrorData(code=400, message='"a" and "b" parameters must be numbers.')

    result: float
    if operation == 'add':
        result = a + b
    elif operation == 'subtract':
        result = a - b
    elif operation == 'multiply':
        result = a * b
    elif operation == 'divide':
        if b == 0:
            raise MCPErrorData(code=400, message='Division by zero is not allowed.')
        result = a / b
    else:
        raise MCPErrorData(code=400, message=f'Unsupported operation: {operation}. Supported: add, subtract, multiply, divide.')

    return {'result': result, 'inputs': {'a': a, 'b': b, 'operation': operation}}

完整服务器示例

这是一个完整的示例，包含了服务器初始化和上面定义的所有处理函数。将此保存为 Python 文件（例如 mcp_py_server.py）并运行它。

import asyncio
from modelcontext_mcp import MCPServer, ResourceContext, PromptContext, ToolContext, MCPErrorData

# 1. 初始化服务器实例
server = MCPServer(
    server_id='python-mcp-server-full',
    server_description='Full Python MCP Server Example with various capabilities.',
    capabilities={
        'resources': [
            {'id': 'ping', 'description': 'Simple health check.'},
            {'id': 'number_stream', 'description': 'Streams a sequence of numbers.'}
        ],
        'prompts': [
            {'id': 'reverse_string', 'description': 'Reverses an input string.'},
            {'id': 'mock_llm_stream', 'description': 'Simulates a streaming LLM response.'}
        ],
        'tools': [
            {'id': 'simple_math', 'description': 'Performs add, subtract, multiply, divide.'}
        ]
    }
)

# 2. 定义资源处理函数
@server.resource(resource_id='ping')
async def ping_handler(context: ResourceContext) -> dict:
    print(f"PING from {context.client_id}")
    return {'message': 'pong from Python', 'server_time': asyncio.get_event_loop().time()}

@server.resource(resource_id='number_stream')
async def number_stream_handler(context: ResourceContext):
    count = context.params.get('count', 3); delay = context.params.get('delay', 0.3)
    print(f"NUMSTREAM for {context.client_id}: count={count}")
    for i in range(1, count + 1):
        await asyncio.sleep(delay)
        yield {'current_number': i, 'is_last': i == count}

# 3. 定义提示处理函数
@server.prompt(prompt_id='reverse_string')
async def reverse_string_handler(context: PromptContext) -> dict:
    text = context.params.get('text', '')
    print(f"REVERSE for {context.client_id}: '{text}'")
    if not isinstance(text, str): raise MCPErrorData(code=400, message='Need string for text.')
    return {'reversed': text[::-1]}

@server.prompt(prompt_id='mock_llm_stream')
async def mock_llm_stream_handler(context: PromptContext):
    text_parts = ['This ', 'is ', 'a ', 'mocked ', 'LLM ', 'stream.']
    print(f"LLMSTREAM for {context.client_id}")
    for part in text_parts:
        await asyncio.sleep(0.1)
        yield {'token': part}
    yield {'token': '<END>'}

# 4. 定义工具处理函数
@server.tool(tool_id='simple_math')
async def simple_math_handler(context: ToolContext) -> dict:
    op, a, b = context.params.get('operation'), context.params.get('a'), context.params.get('b')
    print(f"MATH for {context.client_id}: {a} {op} {b}")
    if not all(isinstance(x, (int, float)) for x in [a,b]): raise MCPErrorData(400, 'a,b num err')
    if op == 'add': result = a + b
    elif op == 'subtract': result = a - b
    elif op == 'multiply': result = a * b
    elif op == 'divide':
        if b == 0: raise MCPErrorData(400, 'div by zero')
        result = a / b
    else: raise MCPErrorData(400, f'bad op: {op}')
    return {'outcome': result}

# 5. 主服务器循环和事件处理函数（可选）
@server.on_connect
async def handle_connect(client_id: str, client_address: tuple):
    print(f"EVENT: Client {client_id} connected from {client_address}")

@server.on_disconnect
async def handle_disconnect(client_id: str):
    print(f"EVENT: Client {client_id} disconnected")

@server.on_error
async def handle_server_error(error: Exception, client_id: str | None = None):
    print(f"EVENT: Server error occurred (Client: {client_id if client_id else 'N/A'}): {error}")

async def run_python_mcp_server():
    port = 8080
    host = '0.0.0.0'
    try:
        print(f'Starting Full Python MCP Server on ws://{host}:{port}{server.path}...')
        await server.start(host=host, port=port)
        print('Server is running. Press Ctrl+C to stop.')
        # server.start() 是阻塞的，因此如果它永久运行，这部分可能不会被执行。
        # 对于非阻塞启动，你可能需要 await server.serve() 并自行管理循环。
        while True: # 如果 server.start 不是完全阻塞的，保持主协程活跃
            await asyncio.sleep(3600)
    except KeyboardInterrupt:
        print('\nCtrl+C pressed. Shutting down the server gracefully...') except Exception as e:
        print(f'An unexpected error occurred while running the server: {e}')
    finally:
        if server.is_running():
            await server.stop()
        print('Python MCP Server has been shut down.')

if __name__ == "__main__":
    try:
        asyncio.run(run_python_mcp_server())
    except KeyboardInterrupt:
        print("应用程序已由用户终止。")

Python MCP 服务器的最佳实践

• 基于装饰器的处理函数： 利用 @server.resource、@server.prompt 和 @server.tool 装饰器，以清晰和声明式的方式定义服务器的能力。
• 异步操作： 确保所有执行 I/O 操作（文件访问、网络调用、数据库查询、大语言模型 LLM 交互）的处理函数都定义为 async def，并使用 await 进行非阻塞操作。
• 类型提示： 在处理函数签名中使用 Python 类型提示（例如 context: ResourceContext，-> dict），以提高代码可读性并辅助静态分析工具，尽管默认情况下它们在运行时不被严格执行。
• 输入验证： 在每个处理函数中严格验证来自 context.params 的参数。对于客户端错误（例如，缺少参数、无效类型），抛出 MCPErrorData(code=..., message=...)。
• 使用异步生成器进行流式传输： 对于需要流式传输多个数据载荷的资源或提示，请将处理函数实现为 async def 函数，并通过 yield 返回每个数据载荷字典。
• 上下文信息： 利用 context 对象（例如 context.client_id、context.params、context.client_address）进行日志记录、个性化响应或授权逻辑。
• 明确的能力声明： 在 MCPServer 实例化期间定义全面的服务器 capabilities（能力），以便客户端能够准确发现你的服务器提供了哪些功能。
• 优雅停机： 处理 KeyboardInterrupt 和其他信号，调用 await server.stop() 进行干净的停机，确保连接正确终止。
• 事件处理函数： 使用 @server.on_connect、@server.on_disconnect 和 @server.on_error 装饰器来实现这些服务器生命周期事件的自定义逻辑，例如日志记录或资源清理。