MCP Server Resource 开发学习文档

MCP Server Resource 开发学习文档

目录

1. 简介
2. 主要依赖与环境准备
3. 服务器核心结构与原理
   • 资源定义与管理
   • 资源列表接口实现
   • 资源读取接口实现
   • 传输方式（stdio 与 sse）
4. 源码详细解析
5. SSE 方式本地部署方法
6. 客户端访问 SSE 服务示例

1. 简介

MCP（Modular Communication Protocol）是一种用于模块化服务通信的协议。本文以 simple-resource 服务器为例，讲解如何开发一个简单的 MCP 资源服务器，将本地文本资源以标准接口暴露给客户端访问。

2. 主要依赖与环境准备

• Python 3.8+
• anyio
• click
• pydantic
• starlette
• uvicorn
• mcp（自定义/三方库）

安装依赖（假设 requirements.txt 已配置）：

pip install -r requirements.txt

3. 服务器核心结构与原理

3.1 资源定义与管理

资源以字典形式存储，key 为资源名，value 为资源内容：

SAMPLE_RESOURCES = {
    "greeting": "Hello! This is a sample text resource.",
    "help": "This server provides a few sample text resources for testing.",
    "about": "This is the simple-resource MCP server implementation.",
}

3.2 资源列表接口实现

通过 @app.list_resources() 装饰器注册资源列表接口，返回所有可用资源的元数据：

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri=AnyUrl(f"file:///{name}.txt"),
            name=name,
            description=f"A sample text resource named {name}",
            mimeType="text/plain",
        )
        for name in SAMPLE_RESOURCES.keys()
    ]

• uri：资源唯一标识，采用 file 协议
• name：资源名
• description：资源描述
• mimeType：资源类型

3.3 资源读取接口实现

通过 @app.read_resource() 装饰器注册资源读取接口，按 uri 读取资源内容：

@app.read_resource()
async def read_resource(uri: AnyUrl) -> str | bytes:
    if uri.path is None:
        raise ValueError(f"Invalid resource path: {uri}")
    name = uri.path.replace(".txt", "").lstrip("/")
    if name not in SAMPLE_RESOURCES:
        raise ValueError(f"Unknown resource: {uri}")
    return SAMPLE_RESOURCES[name]

• 校验 uri 合法性
• 解析资源名
• 返回对应资源内容

3.4 传输方式（stdio 与 sse）

• stdio：标准输入输出，适合本地进程间通信
• sse：Server-Sent Events，基于 HTTP，适合网络通信

通过命令行参数 --transport 选择，默认 stdio。

4. 源码详细解析

4.1 主函数入口

@click.command()
@click.option("--port", default=8000, help="Port to listen on for SSE")
@click.option(
    "--transport",
    type=click.Choice(["stdio", "sse"]),
    default="stdio",
    help="Transport type",
)
def main(port: int, transport: str) -> int:
    app = Server("mcp-simple-resource")
    # ...注册接口...
    if transport == "sse":
        # SSE 相关逻辑
    else:
        # stdio 相关逻辑
    return 0

• 使用 click 实现命令行参数解析
• 根据 transport 参数选择不同的服务启动方式

4.2 SSE 传输实现

if transport == "sse":
    from mcp.server.sse import SseServerTransport
    from starlette.applications import Starlette
    from starlette.responses import Response
    from starlette.routing import Mount, Route

    sse = SseServerTransport("/messages/")

    async def handle_sse(request):
        async with sse.connect_sse(
            request.scope, request.receive, request._send
        ) as streams:
            await app.run(
                streams[0], streams[1], app.create_initialization_options()
            )
        return Response()

    starlette_app = Starlette(
        debug=True,
        routes=[
            Route("/sse", endpoint=handle_sse, methods=["GET"]),
            Mount("/messages/", app=sse.handle_post_message),
        ],
    )

    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=port)

• 使用 Starlette 框架搭建 HTTP 服务
• /sse 路由用于 SSE 连接
• /messages/ 路由用于消息传递
• 通过 uvicorn 启动服务

4.3 stdio 传输实现

else:
    from mcp.server.stdio import stdio_server

    async def arun():
        async with stdio_server() as streams:
            await app.run(
                streams[0], streams[1], app.create_initialization_options()
            )

    anyio.run(arun)

• 适合本地开发和测试

5. SSE 方式本地部署方法

5.1 启动 SSE 服务

在 simple-resource 目录下运行：

uv run mcp-simple-resource --transport sse --port 8000

或直接用 Python 启动：

python mcp_simple_resource/server.py --transport sse --port 8000

• 服务将监听本地 8000 端口，SSE 路径为 /sse

6. 客户端访问 SSE 服务示例

6.1 Python 客户端示例

假设有 mcp 客户端库，示例代码如下：

import asyncio
from mcp.types import AnyUrl
from mcp.client.session import ClientSession
from mcp.client.sse import SseServerParameters, sse_client

async def main():
    async with sse_client(
        SseServerParameters(url="http://localhost:8000/sse")
    ) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 列出所有资源
            resources = await session.list_resources()
            print(resources)

            # 读取特定资源
            resource = await session.read_resource(AnyUrl("file:///greeting.txt"))
            print(resource)

asyncio.run(main())

• SseServerParameters 指定服务端地址
• list_resources() 获取资源列表
• read_resource() 读取具体资源内容

总结

本文详细介绍了 MCP simple-resource 服务器的开发原理、源码结构、接口实现、SSE 部署与客户端访问方法。通过本例，开发者可以快速理解如何基于 MCP 协议开发自定义资源服务，并通过多种传输方式对接客户端。