MCP Server 开发实战：从零构建标准化工具服务

MCP Server 开发实战：从零构建标准化工具服务

什么是 MCP Server？

MCP (Model Context Protocol) Server 是实现 MCP 协议的服务端，向客户端（Claude Desktop、Cursor、Continue 等）暴露：

• Tools：可执行的函数（如：读文件、查数据库、调 API、跑命令）
• Resources：可读取的数据源（如：文件内容、数据库记录、网页快照）
• Prompts：可复用的提示词模板（如：代码审查、文案生成、SQL 编写）

客户端通过标准化协议发现并调用这些能力，实现「一次编写，多端复用」。

协议核心概念

JSON-RPC 2.0 基础

// 请求
{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "read_file", "arguments": {"path": "/tmp/test.txt"}}}

// 响应
{"jsonrpc": "2.0", "id": 1, "result": {"content": [{"type": "text", "text": "文件内容..."}]}}

// 错误响应
{"jsonrpc": "2.0", "id": 1, "error": {"code": -32602, "message": "Invalid params", "data": "..."}}

核心方法表

| 方法 | 方向 | 说明 | |------|------|------| | initialize | 客户端→服务端 | 握手，协商协议版本、能力 | | tools/list | 客户端→服务端 | 获取可用工具列表 | | tools/call | 客户端→服务端 | 调用工具 | | resources/list | 客户端→服务端 | 获取可用资源列表 | | resources/read | 客户端→服务端 | 读取资源内容 | | resources/subscribe | 客户端→服务端 | 订阅资源变更 | | prompts/list | 客户端→服务端 | 获取提示词模板列表 | | prompts/get | 客户端→服务端 | 获取渲染后的提示词 | | notifications/* | 服务端→客户端 | 进度、日志、资源变更通知 |

---

1. 环境搭建

# 创建项目
mkdir mcp-server-example && cd mcp-server-example
python -m venv .venv
source .venv/bin/activate

# 安装 MCP SDK (官方 Python 实现)
pip install mcp

# 开发依赖
pip install pydantic pydantic-settings pytest pytest-asyncio httpx

项目结构

mcp-server-example/
├── src/
│   └── mcp_server_example/
│       ├── __init__.py
│       ├── server.py          # 核心 Server 实现
│       ├── tools/             # Tools 实现
│       │   ├── __init__.py
│       │   ├── filesystem.py
│       │   ├── web_search.py
│       │   └── database.py
│       ├── resources/         # Resources 实现
│       │   ├── __init__.py
│       │   └── filesystem.py
│       ├── prompts/           # Prompts 实现
│       │   ├── __init__.py
│       │   └── templates.py
│       └── transport/         # 传输层
│           ├── __init__.py
│           ├── stdio.py
│           └── sse.py
├── tests/
├── pyproject.toml
└── README.md

---

2. 核心 Server 实现

2.1 基础框架 (src/mcp_server_example/server.py)

"""MCP Server 核心实现"""
import asyncio
import json
import sys
from contextlib import asynccontextmanager
from typing import Any, Dict, List, Optional, AsyncIterator
from dataclasses import dataclass

from mcp import types
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.server.sse

from .tools import register_tools
from .resources import register_resources
from .prompts import register_prompts


@dataclass
class ServerContext:
    """服务端上下文，保存共享状态"""
    config: Dict[str, Any]
    db_pool: Optional[Any] = None


class MCPServer:
    def __init__(self, name: str = "example-server", version: str = "1.0.0"):
        self.name = name
        self.version = version
        self.server = Server(name)
        self.context = ServerContext(config={})
        self._setup_handlers()
    
    def _setup_handlers(self):
        """注册协议处理器"""
        # Tools
        @self.server.list_tools()
        async def list_tools() -> List[types.Tool]:
            return register_tools(self.context)
        
        @self.server.call_tool()
        async def call_tool(name: str, arguments: Dict[str, Any]) -> List[types.TextContent | types.ImageContent | types.EmbeddedResource]:
            return await self._call_tool(name, arguments)
        
        # Resources
        @self.server.list_resources()
        async def list_resources() -> List[types.Resource]:
            return register_resources(self.context)
        
        @self.server.read_resource()
        async def read_resource(uri: str) -> str:
            return await self._read_resource(uri)
        
        # Prompts
        @self.server.list_prompts()
        async def list_prompts() -> List[types.Prompt]:
            return register_prompts(self.context)
        
        @self.server.get_prompt()
        async def get_prompt(name: str, arguments: Dict[str, str]) -> types.GetPromptResult:
            return await self._get_prompt(name, arguments)
    
    async def _call_tool(self, name: str, arguments: Dict[str, Any]) -> List[types.TextContent | types.ImageContent | types.EmbeddedResource]:
        # 实际实现中分发到具体工具处理器
        # 这里简化演示
        return [types.TextContent(type="text", text=f"Tool {name} called with {arguments}")]
    
    async def _read_resource(self, uri: str) -> str:
        # 实际实现中分发到具体资源处理器
        return f"Resource content for {uri}"
    
    async def _get_prompt(self, name: str, arguments: Dict[str, str]) -> types.GetPromptResult:
        # 实际实现中分发到具体提示词处理器
        return types.GetPromptResult(
            description=f"Prompt {name}",
            messages=[types.PromptMessage(role="user", content=types.TextContent(type="text", text=f"Prompt {name} with {arguments}"))]
        )
    
    async def run_stdio(self):
        """通过 stdio 运行 (本地进程通信)"""
        async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
            await self.server.run(
                read_stream,
                write_stream,
                InitializationOptions(
                    server_name=self.name,
                    server_version=self.version,
                    capabilities=self.server.get_capabilities(
                        notification_options=NotificationOptions(),
                        experimental_capabilities={}
                    )
                )
            )
    
    async def run_sse(self, host: str = "0.0.0.0", port: int = 8000):
        """通过 SSE 运行 (HTTP 服务器)"""
        from starlette.applications import Starlette
        from starlette.routing import Route, Mount
        from starlette.responses import Response
        
        sse = mcp.server.sse.SseServerTransport("/messages/")
        
        async def handle_sse(request):
            async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream):
                await self.server.run(
                    read_stream,
                    write_stream,
                    InitializationOptions(
                        server_name=self.name,
                        server_version=self.version,
                        capabilities=self.server.get_capabilities(
                            notification_options=NotificationOptions(),
                            experimental_capabilities={}
                        )
                    )
                )
        
        starlette_app = Starlette(
            routes=[
                Route("/sse", handle_sse),
                Mount("/messages/", app=sse.handle_post_message),
            ]
        )
        
        import uvicorn
        config = uvicorn.Config(starlette_app, host=host, port=port, log_level="info")
        server = uvicorn.Server(config)
        await server.serve()


async def main():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--transport", choices=["stdio", "sse"], default="stdio")
    parser.add_argument("--host", default="0.0.0.0")
    parser.add_argument("--port", type=int, default=8000)
    args = parser.parse_args()
    
    server = MCPServer()
    
    if args.transport == "stdio":
        await server.run_stdio()
    else:
        await server.run_sse(args.host, args.port)


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

---

3. Tools 实现

3.1 文件系统工具 (src/mcp_server_example/tools/filesystem.py)

"""文件系统操作工具"""
import os
import json
from pathlib import Path
from typing import Dict, Any, List
from mcp import types
from pydantic import BaseModel, Field


class ReadFileArgs(BaseModel):
    path: str = Field(description="文件绝对路径")
    encoding: str = Field(default="utf-8", description="编码")


class WriteFileArgs(BaseModel):
    path: str = Field(description="文件绝对路径")
    content: str = Field(description="写入内容")
    encoding: str = Field(default="utf-8", description="编码")


class ListDirArgs(BaseModel):
    path: str = Field(description="目录绝对路径")
    recursive: bool = Field(default=False, description="是否递归")


FILESYSTEM_TOOLS = [
    types.Tool(
        name="read_file",
        description="读取文件内容",
        inputSchema=ReadFileArgs.model_json_schema(),
    ),
    types.Tool(
        name="write_file",
        description="写入文件内容 (需启用写权限)",
        inputSchema=WriteFileArgs.model_json_schema(),
    ),
    types.Tool(
        name="list_directory",
        description="列出目录内容",
        inputSchema=ListDirArgs.model_json_schema(),
    ),
]


async def handle_read_file(arguments: Dict[str, Any]) -> List[types.TextContent]:
    args = ReadFileArgs(**arguments)
    try:
        path = Path(args.path).resolve()
        # 安全检查：限制在允许的根目录下
        if not is_path_allowed(path):
            return [types.TextContent(type="text", text=f"Error: Path not allowed: {path}")]
        
        content = path.read_text(encoding=args.encoding)
        return [types.TextContent(type="text", text=content)]
    except Exception as e:
        return [types.TextContent(type="text", text=f"Error reading file: {e}")]


async def handle_write_file(arguments: Dict[str, Any]) -> List[types.TextContent]:
    args = WriteFileArgs(**arguments)
    try:
        path = Path(args.path).resolve()
        if not is_path_allowed(path):
            return [types.TextContent(type="text", text=f"Error: Path not allowed: {path}")]
        
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(args.content, encoding=args.encoding)
        return [types.TextContent(type="text", text=f"File written: {path}")]
    except Exception as e:
        return [types.TextContent(type="text", text=f"Error writing file: {e}")]


async def handle_list_directory(arguments: Dict[str, Any]) -> List[types.TextContent]:
    args = ListDirArgs(**arguments)
    try:
        path = Path(args.path).resolve()
        if not is_path_allowed(path):
            return [types.TextContent(type="text", text=f"Error: Path not allowed: {path}")]
        
        if args.recursive:
            items = []
            for p in path.rglob("*"):
                rel = p.relative_to(path)
                items.append({"path": str(rel), "type": "dir" if p.is_dir() else "file", "size": p.stat().st_size if p.is_file() else None})
            return [types.TextContent(type="text", text=json.dumps(items, indent=2, ensure_ascii=False))]
        else:
            items = []
            for p in path.iterdir():
                items.append({"name": p.name, "type": "dir" if p.is_dir() else "file", "size": p.stat().st_size if p.is_file() else None})
            return [types.TextContent(type="text", text=json.dumps(items, indent=2, ensure_ascii=False))]
    except Exception as e:
        return [types.TextContent(type="text", text=f"Error listing directory: {e}")]


# 安全策略：允许的根目录
ALLOWED_ROOTS = [Path.cwd(), Path.home() / "Documents", Path("/tmp")]

def is_path_allowed(path: Path) -> bool:
    try:
        path.resolve().relative_to(Path.cwd())
        return True
    except ValueError:
        for root in ALLOWED_ROOTS:
            try:
                path.resolve().relative_to(root.resolve())
                return True
            except ValueError:
                continue
    return False


TOOL_HANDLERS = {
    "read_file": handle_read_file,
    "write_file": handle_write_file,
    "list_directory": handle_list_directory,
}

3.2 工具注册 (src/mcp_server_example/tools/__init__.py)

"""Tools 注册入口"""
from typing import Dict, Any, List
from mcp import types
from .filesystem import FILESYSTEM_TOOLS, TOOL_HANDLERS
from .web_search import WEB_SEARCH_TOOLS, WEB_SEARCH_HANDLERS
from .database import DATABASE_TOOLS, DATABASE_HANDLERS


ALL_TOOLS = FILESYSTEM_TOOLS + WEB_SEARCH_TOOLS + DATABASE_TOOLS
ALL_HANDLERS = {**TOOL_HANDLERS, **WEB_SEARCH_HANDLERS, **DATABASE_HANDLERS}


def register_tools(context: Any) -> List[types.Tool]:
    """注册所有工具，可根据 context.config 动态启用/禁用"""
    enabled = context.config.get("enabled_tools", ["filesystem", "web_search", "database"])
    tools = []
    if "filesystem" in enabled:
        tools.extend(FILESYSTEM_TOOLS)
    if "web_search" in enabled:
        tools.extend(WEB_SEARCH_TOOLS)
    if "database" in enabled:
        tools.extend(DATABASE_TOOLS)
    return tools


async def call_tool(name: str, arguments: Dict[str, Any], context: Any) -> List[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """分发工具调用"""
    handler = ALL_HANDLERS.get(name)
    if not handler:
        return [types.TextContent(type="text", text=f"Unknown tool: {name}")]
    return await handler(arguments)

---

4. Resources 实现

4.1 文件系统资源 (src/mcp_server_example/resources/filesystem.py)

"""文件系统资源"""
from pathlib import Path
from typing import List
from mcp import types


FILESYSTEM_RESOURCES = [
    types.Resource(
        uri="file:///workspace/{path}",
        name="Workspace File",
        description="工作区内的文件",
        mimeType="text/plain",
    ),
    types.Resource(
        uri="file:///home/{path}",
        name="Home Directory File",
        description="用户文档目录下的文件",
        mimeType="text/plain",
    ),
]


async def read_resource(uri: str) -> str:
    """读取资源内容"""
    # 解析 URI 模板
    if uri.startswith("file:///workspace/"):
        rel_path = uri[len("file:///workspace/"):]
        abs_path = Path.cwd() / rel_path
    elif uri.startswith("file:///home/"):
        rel_path = uri[len("file:///home/"):]
        abs_path = Path.home() / "Documents" / rel_path
    else:
        raise ValueError(f"Unknown resource URI: {uri}")
    
    # 安全检查
    try:
        abs_path.resolve().relative_to(Path.cwd())
    except ValueError:
        raise PermissionError(f"Access denied: {uri}")
    
    if not abs_path.exists():
        raise FileNotFoundError(f"Resource not found: {uri}")
    
    if abs_path.is_dir():
        # 返回目录列表
        items = [{"name": p.name, "type": "dir" if p.is_dir() else "file"} for p in abs_path.iterdir()]
        return json.dumps(items, ensure_ascii=False, indent=2)
    
    return abs_path.read_text(encoding="utf-8")


def register_resources(context: Any) -> List[types.Resource]:
    enabled = context.config.get("enabled_resources", ["filesystem"])
    resources = []
    if "filesystem" in enabled:
        resources.extend(FILESYSTEM_RESOURCES)
    return resources

---

5. Prompts 实现

5.1 提示词模板 (src/mcp_server_example/prompts/templates.py)

"""提示词模板"""
from typing import Dict, List
from mcp import types


PROMPTS = [
    types.Prompt(
        name="code_review",
        description="代码审查提示词",
        arguments=[
            types.PromptArgument(name="code", description="待审查代码", required=True),
            types.PromptArgument(name="language", description="编程语言", required=False),
            types.PromptArgument(name="focus", description="重点关注方面 (security/performance/style)", required=False),
        ],
    ),
    types.Prompt(
        name="sql_generation",
        description="自然语言生成 SQL",
        arguments=[
            types.PromptArgument(name="question", description="自然语言问题", required=True),
            types.PromptArgument(name="schema", description="数据库 Schema (DDL)", required=True),
            types.PromptArgument(name="dialect", description="SQL 方言 (postgresql/mysql/sqlite)", required=False),
        ],
    ),
    types.Prompt(
        name="ecommerce_copywriting",
        description="电商文案生成",
        arguments=[
            types.PromptArgument(name="product_info", description="商品信息 (JSON)", required=True),
            types.PromptArgument(name="platform", description="目标平台 (tiktok/douyin/xiaohongshu)", required=True),
            types.PromptArgument(name="tone", description="语气风格", required=False),
        ],
    ),
]


PROMPT_TEMPLATES = {
    "code_review": """你是资深代码审查专家。请审查以下 {language} 代码，重点关注 {focus}：

{code}

请从以下维度给出反馈：
1. **正确性**：逻辑错误、边界条件、异常处理
2. **安全性**：注入风险、权限控制、敏感数据泄露
3. **性能**：时间/空间复杂度、数据库查询优化、缓存策略
4. **可维护性**：命名、结构、复用性、测试覆盖
5. **最佳实践**：语言惯用法、设计模式、库使用

输出格式：
- ✅ 优点
- ⚠️ 问题 (严重/中等/轻微)
- 💡 建议
- 📝 重构示例 (如适用)""",
    
    "sql_generation": """你是 SQL 专家。根据以下数据库 Schema，将自然语言问题转换为 {dialect} SQL。

**Schema:**

{schema}

**问题:** {question}

**要求:**
1. 只输出 SQL，不包含解释
2. 使用参数化查询 (占位符 $1, $2 或 ?, ?)
3. 明确列出 SELECT 列，避免 SELECT *
4. 添加必要的索引提示
5. 处理 NULL、去重、分页

**SQL:**""",
    
    "ecommerce_copywriting": """你是资深电商文案专家。根据商品信息，生成 {platform} 平台风格的带货文案。

**商品信息:**

{product_info}

**平台特性:**
- TikTok: 短视频口播脚本、前3秒钩子、节奏感强、行动号召明确
- 抖音: 痛点开头、场景化、情绪共鸣、评论区互动引导
- 小红书: 种草笔记、真实体验、emoji丰富、标签优化

**语气风格:** {tone}

**输出:**
1. 标题 (3-5个备选)
2. 正文/脚本
3. 标签/话题
4. 发布建议 (时间、频次、互动策略)""",
}


async def get_prompt(name: str, arguments: Dict[str, str]) -> types.GetPromptResult:
    template = PROMPT_TEMPLATES.get(name)
    if not template:
        raise ValueError(f"Unknown prompt: {name}")
    
    rendered = template.format(**arguments)
    return types.GetPromptResult(
        description=f"Rendered prompt: {name}",
        messages=[
            types.PromptMessage(
                role="user",
                content=types.TextContent(type="text", text=rendered)
            )
        ],
    )


def register_prompts(context: Any) -> List[types.Prompt]:
    return PROMPTS

---

6. 传输层与部署

6.1 stdio 模式 (本地)

# 直接运行，Claude Desktop 配置:
# {
#   "mcpServers": {
#     "example": {
#       "command": "python",
#       "args": ["-m", "mcp_server_example.server"],
#       "cwd": "/path/to/mcp-server-example"
#     }
#   }
# }
python -m mcp_server_example.server --transport stdio

6.2 SSE 模式 (远程/网络)

# 启动 HTTP 服务器
python -m mcp_server_example.server --transport sse --host 0.0.0.0 --port 8000

# 客户端连接: SSE endpoint = http://host:8000/sse

6.3 Docker 部署

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY pyproject.toml .
COPY src/ ./src/
RUN pip install -e .

# 非 root 用户
RUN useradd -m -u 1000 mcp && chown -R mcp:mcp /app
USER mcp

# stdio 模式入口
ENTRYPOINT ["python", "-m", "mcp_server_example.server", "--transport", "stdio"]

# 构建
docker build -t mcp-server-example .

# 运行 (stdio 模式需特殊处理，建议用 SSE)
docker run -p 8000:8000 mcp-server-example python -m mcp_server_example.server --transport sse

---

7. 测试

7.1 单元测试 (tests/test_tools.py)

import pytest
from mcp_server_example.tools.filesystem import handle_read_file, handle_list_directory
from mcp_server_example.tools.web_search import handle_web_search


@pytest.mark.asyncio
async def test_read_file(tmp_path):
    test_file = tmp_path / "test.txt"
    test_file.write_text("Hello MCP!")
    
    # 临时允许测试目录
    import mcp_server_example.tools.filesystem as fs_mod
    fs_mod.ALLOWED_ROOTS.append(tmp_path)
    
    result = await handle_read_file({"path": str(test_file)})
    assert result[0].text == "Hello MCP!"
    
    fs_mod.ALLOWED_ROOTS.remove(tmp_path)


@pytest.mark.asyncio
async def test_list_directory(tmp_path):
    (tmp_path / "a.txt").write_text("a")
    (tmp_path / "b").mkdir()
    (tmp_path / "b" / "c.txt").write_text("c")
    
    import mcp_server_example.tools.filesystem as fs_mod
    fs_mod.ALLOWED_ROOTS.append(tmp_path)
    
    result = await handle_list_directory({"path": str(tmp_path), "recursive": True})
    items = json.loads(result[0].text)
    assert len(items) == 3
    assert any(i["path"] == "a.txt" for i in items)
    assert any(i["path"] == "b/c.txt" for i in items)
    
    fs_mod.ALLOWED_ROOTS.remove(tmp_path)

7.2 集成测试 (tests/test_integration.py)

import pytest
from mcp import types
from mcp.client.session import ClientSession
from mcp.client.stdio import stdio_client


@pytest.mark.asyncio
async def test_server_stdio():
    """测试 stdio 传输下的完整流程"""
    async with stdio_client(
        command="python",
        args=["-m", "mcp_server_example.server", "--transport", "stdio"],
    ) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # 初始化
            await session.initialize()
            
            # 列出工具
            tools = await session.list_tools()
            assert len(tools.tools) > 0
            tool_names = [t.name for t in tools.tools]
            assert "read_file" in tool_names
            
            # 调用工具
            result = await session.call_tool("read_file", {"path": __file__})
            assert len(result.content) > 0
            assert "test_integration" in result.content[0].text
            
            # 列出资源
            resources = await session.list_resources()
            assert len(resources.resources) > 0
            
            # 列出提示词
            prompts = await session.list_prompts()
            assert len(prompts.prompts) > 0

---

8. 发布与分发

8.1 PyPI 发布

# pyproject.toml
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "mcp-server-example"
version = "1.0.0"
description = "Example MCP Server"
readme = "README.md"
license = {text = "MIT"}
authors = [{name = "Your Name", email = "you@example.com"}]
requires-python = ">=3.10"
dependencies = [
    "mcp>=1.0.0",
    "pydantic>=2.0",
    "pydantic-settings>=2.0",
]
[project.optional-dependencies]
dev = ["pytest", "pytest-asyncio", "httpx"]
sse = ["uvicorn", "starlette"]

[project.scripts]
mcp-server-example = "mcp_server_example.server:main"

[tool.hatch.build.targets.wheel]
packages = ["src/mcp_server_example"]

# 发布
pip install build twine
python -m build
twine upload dist/*

8.2 客户端接入配置

Claude Desktop (~/Library/Application Support/Claude/claude_desktop_config.json):

{
  "mcpServers": {
    "example-stdio": {
      "command": "python",
      "args": ["-m", "mcp_server_example.server", "--transport", "stdio"],
      "cwd": "/path/to/mcp-server-example",
      "env": {
        "ENABLED_TOOLS": "filesystem,web_search,database",
        "ENABLED_RESOURCES": "filesystem"
      }
    },
    "example-sse": {
      "command": "npx",
      "args": ["mcp-remote", "http://localhost:8000/sse"]
    }
  }
}

Cursor (Settings → MCP):

{
  "mcpServers": {
    "example": {
      "url": "http://localhost:8000/sse"
    }
  }
}

Continue.dev (~/.continue/config.json):

{
  "mcpServers": [
    {
      "name": "example",
      "transport": "stdio",
      "command": "python",
      "args": ["-m", "mcp_server_example.server", "--transport", "stdio"]
    }
  ]
}

---

9. 最佳实践清单

安全

• [ ] 最小权限：仅暴露必要工具，默认禁用写操作
• [ ] 路径遍历防护：所有文件操作规范化路径、检查允许根目录
• [ ] SQL 注入防护：数据库工具仅允许参数化查询
• [ ] 认证授权：SSE 模式集成 OAuth 2.1 / API Key
• [ ] 审计日志：记录所有工具调用、参数、结果、耗时

可靠性

• [ ] 超时控制：每个工具设置合理超时 (默认 30s)
• [ ] 进度通知：长任务发送 notifications/progress
• [ ] 取消支持：响应 notifications/cancelled
• [ ] 幂等性：写操作设计幂等键
• [ ] 熔断降级：外部依赖失败时优雅降级

可观测性

• [ ] 结构化日志：JSON 格式、包含 trace_id
• [ ] 指标收集：调用次数、延迟分位数、错误率
• [ ] 分布式追踪：OpenTelemetry 集成
• [ ] 健康检查：/health 端点 (SSE 模式)

开发体验

• [ ] 完整类型提示：Pydantic 模型 + MCP types
• [ ] 文档字符串：每个工具/资源/提示词有清晰描述
• [ ] 示例丰富：README 包含常用调用示例
• [ ] 版本语义化：遵循 SemVer，破坏性变更大版本号

---

进阶主题

1. 动态工具发现：从 OpenAPI/Swagger 自动生成 Tools 2. 资源订阅与变更通知：WebSocket/SSE 推送文件/数据库变更 3. 工具组合与管道：tool_chain 元工具，串联多工具 4. 多租户隔离：同一 Server 服务多用户，数据隔离 5. 插件系统：运行时加载/卸载工具模块 6. 性能优化：连接池、缓存、批量处理、异步并发

---

参考资源

• MCP 官方规范
• MCP Python SDK
• MCP Server 示例集
• Claude Desktop MCP 配置
• [本站 MCP 关键词详解](/keywords/mcp)
• [本站 MCP Builder Skill](/skills/mcp-builder)