MCP Server开发

Server的职责

• 资源提供：暴露可读取的数据源
• 工具实现：提供可执行的功能
• 提示模板：定义可重用的提示词模板
• 能力声明：告知客户端支持的功能
• 权限控制：确保安全访问

Python环境设置

# 安装MCP Python SDK
pip install mcp

# 或使用uv
uv add mcp

# 项目结构
my-mcp-server/
├── src/
│   └── my_server/
│       ├── __init__.py
│       └── server.py
├── pyproject.toml
└── README.md

TypeScript环境设置

# 安装MCP TypeScript SDK
npm install @modelcontextprotocol/sdk

# 项目结构
my-mcp-server/
├── src/
│   └── index.ts
├── package.json
├── tsconfig.json
└── README.md

Python基础服务器

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server

# 创建服务器实例
server = Server("my-server")

@server.list_resources()
async def list_resources():
    return []

@server.read_resource()
async def read_resource(uri: str):
    return ""

@server.list_tools()
async def list_tools():
    return []

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    return []

# 启动服务器
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

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

TypeScript基础服务器

// index.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server(
  { name: "my-server", version: "1.0.0" },
  { capabilities: { resources: {}, tools: {} } }
);

server.setRequestHandler(ListResourcesRequestSchema, async () => {
  return { resources: [] };
});

server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  return { contents: [] };
});

server.setRequestHandler(ListToolsRequestSchema, async () => {
  return { tools: [] };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  return { content: [] };
});

// 启动传输
const transport = new StdioServerTransport();
await server.connect(transport);

资源实现示例

# Python实现
from mcp.types import Resource, TextContent

@server.list_resources()
async def list_resources() -> list[Resource]:
    """返回可用资源列表"""
    return [
        Resource(
            uri="file:///data/users.json",
            name="用户数据",
            description="系统用户列表",
            mimeType="application/json"
        ),
        Resource(
            uri="file:///data/config.yaml",
            name="配置文件",
            description="系统配置",
            mimeType="text/yaml"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    """读取指定资源内容"""
    if uri == "file:///data/users.json":
        with open("data/users.json") as f:
            return f.read()
    elif uri == "file:///data/config.yaml":
        with open("data/config.yaml") as f:
            return f.read()
    else:
        raise ValueError(f"未知资源: {uri}")

工具实现示例

# Python实现
from mcp.types import Tool, TextContent
import subprocess

@server.list_tools()
async def list_tools() -> list[Tool]:
    """返回可用工具列表"""
    return [
        Tool(
            name="run_command",
            description="执行shell命令",
            inputSchema={
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "要执行的命令"
                    }
                },
                "required": ["command"]
            }
        ),
        Tool(
            name="read_file",
            description="读取文件内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "文件路径"
                    }
                },
                "required": ["path"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    """执行工具调用"""
    if name == "run_command":
        result = subprocess.run(
            arguments["command"],
            shell=True,
            capture_output=True,
            text=True
        )
        return [
            TextContent(
                type="text",
                text=result.stdout or result.stderr
            )
        ]
    elif name == "read_file":
        with open(arguments["path"]) as f:
            return [
                TextContent(
                    type="text",
                    text=f.read()
                )
            ]
    else:
        raise ValueError(f"未知工具: {name}")

提示模板实现

# Python实现
from mcp.types import Prompt, PromptArgument, GetPromptResult, PromptMessage

@server.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="code_review",
            description="代码审查模板",
            arguments=[
                PromptArgument(
                    name="language",
                    description="编程语言",
                    required=True
                ),
                PromptArgument(
                    name="focus",
                    description="审查重点",
                    required=False
                )
            ]
        )
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> GetPromptResult:
    if name == "code_review":
        language = arguments.get("language", "代码")
        focus = arguments.get("focus", "代码质量")
        
        prompt_text = f"""请审查以下{language}代码，重点关注{focus}方面：

1. 代码结构和可读性
2. 潜在的bug和错误
3. 性能优化建议
4. 最佳实践建议

请提供详细的审查意见和改进建议。"""
        
        return GetPromptResult(
            description=f"{language}代码审查模板",
            messages=[
                PromptMessage(
                    role="user",
                    content=TextContent(type="text", text=prompt_text)
                )
            ]
        )

错误处理最佳实践

# 自定义错误
class ResourceNotFoundError(Exception):
    """资源未找到"""
    pass

class ToolExecutionError(Exception):
    """工具执行失败"""
    pass

# 使用错误
@server.read_resource()
async def read_resource(uri: str):
    try:
        # 尝试读取资源
        content = await fetch_resource(uri)
        return content
    except FileNotFoundError:
        raise ResourceNotFoundError(f"资源不存在: {uri}")
    except PermissionError:
        raise PermissionError(f"无权限访问: {uri}")

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    try:
        result = await execute_tool(name, arguments)
        return [TextContent(type="text", text=result)]
    except Exception as e:
        # 返回错误信息而非抛出异常
        return [
            TextContent(
                type="text",
                text=f"执行失败: {str(e)}"
            )
        ]

使用MCP Inspector测试

# 安装Inspector
npx @modelcontextprotocol/inspector

# 测试本地服务器
npx @modelcontextprotocol/inspector python server.py

# Inspector功能
- 查看服务器能力
- 测试资源列表和读取
- 测试工具调用
- 查看通信日志

单元测试

# test_server.py
import pytest
from my_server import server

@pytest.mark.asyncio
async def test_list_resources():
    resources = await server.list_resources()
    assert len(resources) > 0
    assert all(r.uri for r in resources)

@pytest.mark.asyncio
async def test_call_tool():
    result = await server.call_tool(
        "read_file",
        {"path": "test.txt"}
    )
    assert result is not None

Claude Desktop配置

// ~/Library/Application Support/Claude/claude_desktop_config.json
{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}