3.MCP开发实践

# 3.1 环境准备

# Python 环境

# 安装 uv（推荐的 Python 包管理器）
# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

# 创建项目
uv init my-mcp-server
cd my-mcp-server

# 添加 MCP SDK
uv add mcp

1
2
3
4
5
6
7
8
9
10
11
12
13

# TypeScript 环境

# 创建项目
mkdir my-mcp-server
cd my-mcp-server
npm init -y

# 安装 MCP SDK
npm install @modelcontextprotocol/sdk

# 安装 TypeScript（如果需要）
npm install -D typescript @types/node
npx tsc --init

1
2
3
4
5
6
7
8
9
10
11

# 3.2 Claude Desktop 配置

Claude Desktop 是最常用的 MCP Host，配置文件位置：

macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json

# 基础配置示例

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/username/Documents"
      ]
    },
    "sqlite": {
      "command": "uvx",
      "args": [
        "mcp-server-sqlite",
        "--db-path",
        "/path/to/database.db"
      ]
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 配置字段说明

字段	说明	示例
`command`	启动 Server 的命令	`"uvx"`, `"npx"`, `"python"`
`args`	命令参数数组	`["mcp-server-xxx", "--option"]`
`env`	环境变量	`{"API_KEY": "xxx"}`
`disabled`	是否禁用	`true` / `false`
`autoApprove`	自动批准的工具列表	`["read_file"]`

# 使用环境变量

{
  "mcpServers": {
    "github": {
      "command": "uvx",
      "args": ["mcp-server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11

# 3.3 创建 MCP Server

# Python 版本

创建一个简单的 MCP Server，提供文件搜索功能：

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import os
import glob

# 创建 Server 实例
server = Server("file-search-server")

# 定义工具列表
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="search_files",
            description="在指定目录中搜索匹配的文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "directory": {
                        "type": "string",
                        "description": "搜索的目录路径"
                    },
                    "pattern": {
                        "type": "string",
                        "description": "文件匹配模式，如 *.py"
                    }
                },
                "required": ["directory", "pattern"]
            }
        )
    ]

# 实现工具调用
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "search_files":
        directory = arguments["directory"]
        pattern = arguments["pattern"]
        
        # 执行文件搜索
        search_path = os.path.join(directory, "**", pattern)
        files = glob.glob(search_path, recursive=True)
        
        result = f"找到 {len(files)} 个文件:\n"
        for f in files[:20]:  # 限制返回数量
            result += f"- {f}\n"
        
        return [TextContent(type="text", text=result)]
    
    raise ValueError(f"未知工具: {name}")

# 启动 Server
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())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

# TypeScript 版本

// server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import * as fs from "fs";
import * as path from "path";
import { glob } from "glob";

// 创建 Server 实例
const server = new Server(
  {
    name: "file-search-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

// 定义工具列表
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "search_files",
        description: "在指定目录中搜索匹配的文件",
        inputSchema: {
          type: "object",
          properties: {
            directory: {
              type: "string",
              description: "搜索的目录路径",
            },
            pattern: {
              type: "string",
              description: "文件匹配模式，如 *.py",
            },
          },
          required: ["directory", "pattern"],
        },
      },
    ],
  };
});

// 实现工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "search_files") {
    const directory = args?.directory as string;
    const pattern = args?.pattern as string;

    const searchPath = path.join(directory, "**", pattern);
    const files = await glob(searchPath);

    let result = `找到 ${files.length} 个文件:\n`;
    files.slice(0, 20).forEach((f) => {
      result += `- ${f}\n`;
    });

    return {
      content: [{ type: "text", text: result }],
    };
  }

  throw new Error(`未知工具: ${name}`);
});

// 启动 Server
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("File Search MCP Server running on stdio");
}

main().catch(console.error);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

# 3.4 定义 Tools

# 工具设计原则

单一职责：每个工具只做一件事
清晰描述：description 要准确描述功能
参数验证：使用 JSON Schema 定义参数
错误处理：返回有意义的错误信息

# 复杂工具示例

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="execute_sql",
            description="执行 SQL 查询并返回结果",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "SQL 查询语句"
                    },
                    "database": {
                        "type": "string",
                        "description": "数据库名称",
                        "enum": ["users", "products", "orders"]
                    },
                    "limit": {
                        "type": "integer",
                        "description": "返回结果数量限制",
                        "default": 100,
                        "minimum": 1,
                        "maximum": 1000
                    }
                },
                "required": ["query", "database"]
            }
        ),
        Tool(
            name="create_file",
            description="创建新文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "文件路径"
                    },
                    "content": {
                        "type": "string",
                        "description": "文件内容"
                    },
                    "overwrite": {
                        "type": "boolean",
                        "description": "是否覆盖已存在的文件",
                        "default": False
                    }
                },
                "required": ["path", "content"]
            }
        )
    ]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

# 3.5 暴露 Resources

Resources 用于向 AI 提供数据内容：

from mcp.types import Resource, TextResourceContents

# 定义资源列表
@server.list_resources()
async def list_resources():
    return [
        Resource(
            uri="config://app/settings",
            name="应用配置",
            description="当前应用的配置信息",
            mimeType="application/json"
        ),
        Resource(
            uri="file://logs/latest",
            name="最新日志",
            description="应用的最新日志内容",
            mimeType="text/plain"
        )
    ]

# 读取资源内容
@server.read_resource()
async def read_resource(uri: str):
    if uri == "config://app/settings":
        config = {
            "version": "1.0.0",
            "debug": True,
            "max_connections": 100
        }
        return TextResourceContents(
            uri=uri,
            mimeType="application/json",
            text=json.dumps(config, indent=2)
        )
    
    if uri == "file://logs/latest":
        # 读取最新日志
        with open("/var/log/app/latest.log", "r") as f:
            content = f.read()
        return TextResourceContents(
            uri=uri,
            mimeType="text/plain",
            text=content
        )
    
    raise ValueError(f"未知资源: {uri}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# 3.6 定义 Prompts

Prompts 提供预定义的提示词模板：

from mcp.types import Prompt, PromptArgument, PromptMessage, TextContent

@server.list_prompts()
async def list_prompts():
    return [
        Prompt(
            name="code_review",
            description="代码审查提示词",
            arguments=[
                PromptArgument(
                    name="code",
                    description="需要审查的代码",
                    required=True
                ),
                PromptArgument(
                    name="language",
                    description="编程语言",
                    required=False
                )
            ]
        )
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
    if name == "code_review":
        code = arguments.get("code", "")
        language = arguments.get("language", "unknown")
        
        return {
            "messages": [
                PromptMessage(
                    role="user",
                    content=TextContent(
                        type="text",
                        text=f"""请审查以下 {language} 代码，关注：
1. 代码质量和可读性
2. 潜在的 bug 和安全问题
3. 性能优化建议
4. 最佳实践建议

代码：
```{language}
{code}
```"""
                    )
                )
            ]
        }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

# 3.7 调试与测试

# 使用 MCP Inspector

MCP Inspector 是官方提供的调试工具：

# 安装并运行 Inspector
npx @modelcontextprotocol/inspector python server.py

1
2

Inspector 提供 Web 界面，可以：

查看 Server 提供的 Tools、Resources、Prompts
手动调用工具并查看结果
查看通信日志

# 日志调试

在 Server 中添加日志：

import logging

# 配置日志（输出到 stderr，不影响 stdio 通信）
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    logger.debug(f"调用工具: {name}, 参数: {arguments}")
    # ... 工具实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 单元测试

import pytest
from server import search_files

@pytest.mark.asyncio
async def test_search_files():
    result = await search_files("/tmp", "*.txt")
    assert "找到" in result

1
2
3
4
5
6
7

# 3.8 部署配置

# 本地部署（Claude Desktop）

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

1
2
3
4
5
6
7
8

# 使用 uvx 部署

将 Server 发布到 PyPI 后：

{
  "mcpServers": {
    "my-server": {
      "command": "uvx",
      "args": ["my-mcp-server"]
    }
  }
}

1
2
3
4
5
6
7
8

# 3.9 小结

本节介绍了 MCP Server 开发的完整流程：

环境准备：安装 Python/TypeScript SDK
配置 Host：设置 Claude Desktop 配置文件
开发 Server：实现 Tools、Resources、Prompts
调试测试：使用 Inspector 和日志调试
部署上线：配置本地或远程部署

下一节我们将介绍 MCP 生态系统中的常用 Server 和学习资源。

编辑

#MCP #开发 #Python #TypeScript

上次更新: 2025/12/19, 15:17:48

← 2.MCP核心架构 4.MCP生态与资源→