model-context

MCP（Model Context Protocol）——AI 原生服务开发

概述

Model Context Protocol（MCP）是一种开放标准，用于将 AI 助手连接到外部数据源和工具。构建可提供 工具（大模型可调用的函数）、资源（大模型可读取的数据）和 提示词（大模型可使用的模板）的服务。

关键概念：

• 工具：大模型可执行的函数（读取文件、查询 API、运行命令）
• 资源：大模型可访问的数据源（文件、数据库、API）
• 提示词：包含常见任务所需参数的可复用模板
• 客户端-服务端：MCP 服务提供功能，客户端（如 Claude Desktop）使用这些功能
• 传输方式：STDIO（本地）、SSE（Server-Sent Events）、HTTP（网络）

官方 SDK：

• TypeScript：@modelcontextprotocol/sdk
• Python：mcp

安装：

# TypeScript server
npx @modelcontextprotocol/create-server@latest my-server
cd my-server && npm install

# Python server
pip install mcp
# Or use uv (recommended)
uv pip install mcp

快速入门——TypeScript 服务

1. 使用 CLI 创建服务

# Interactive setup
npx @modelcontextprotocol/create-server@latest my-filesystem-server

# Options prompt:
# - Server name: my-filesystem-server
# - Language: TypeScript
# - Include example tools: Yes

2. 定义工具

// src/index.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/promises";
import * as path from "path";

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

// List available tools
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "read_file",
        description: "Read contents of a file",
        inputSchema: {
          type: "object",
          properties: {
            path: {
              type: "string",
              description: "Path to the file to read",
            },
          },
          required: ["path"],
        },
      },
      {
        name: "list_directory",
        description: "List contents of a directory",
        inputSchema: {
          type: "object",
          properties: {
            path: {
              type: "string",
              description: "Directory path to list",
            },
          },
          required: ["path"],
        },
      },
    ],
  };
});

// Handle tool calls
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "read_file": {
      const filePath = args.path as string;
      const content = await fs.readFile(filePath, "utf-8");
      return {
        content: [{ type: "text", text: content }],
      };
    }

    case "list_directory": {
      const dirPath = args.path as string;
      const entries = await fs.readdir(dirPath, { withFileTypes: true });
      const listing = entries
        .map((entry) => `${entry.isDirectory() ? "📁" : "📄"} ${entry.name}`)
        .join("\n");
      return {
        content: [{ type: "text", text: listing }],
      };
    }

    default:
      throw new Error(`Unknown tool: ${name}`);
  }
});

// Start server
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Filesystem MCP server running on stdio");
}

main();

3. 配置 Claude Desktop

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// %APPDATA%/Claude/claude_desktop_config.json (Windows)
{
  "mcpServers": {
    "filesystem": {
      "command": "node",
      "args": ["/absolute/path/to/my-filesystem-server/build/index.js"]
    }
  }
}

4. 构建和测试

# Build TypeScript
npm run build

# Restart Claude Desktop (Cmd+Q and reopen)
# Server appears in 🔌 menu

快速入门——Python 服务

1. 创建服务

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

# Create server instance
app = Server("filesystem-server")

# Define tools
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="read_file",
            description="Read contents of a file",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path to read"}
                },
                "required": ["path"],
            },
        ),
        Tool(
            name="list_directory",
            description="List directory contents",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Directory path"}
                },
                "required": ["path"],
            },
        ),
    ]

# Handle tool calls
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "read_file":
        file_path = arguments["path"]
        with open(file_path, "r") as f:
            content = f.read()
        return [TextContent(type="text", text=content)]

    elif name == "list_directory":
        dir_path = arguments["path"]
        entries = os.listdir(dir_path)
        listing = "\n".join(
            f"{'📁' if os.path.isdir(os.path.join(dir_path, e)) else '📄'} {e}"
            for e in entries
        )
        return [TextContent(type="text", text=listing)]

    else:
        raise ValueError(f"Unknown tool: {name}")

# Start server
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

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

2. 配置 Claude Desktop

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

3. 测试

# Test server standalone
python server.py

# Restart Claude Desktop
# Tools appear in 🔌 menu

资源——向大模型提供数据

资源提供对数据源的只读访问。

TypeScript 资源示例

import {
  ListResourcesRequestSchema,
  ReadResourceRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

// List available resources
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  return {
    resources: [
      {
        uri: "file:///docs/readme.md",
        name: "README",
        description: "Project README documentation",
        mimeType: "text/markdown",
      },
      {
        uri: "file:///config/settings.json",
        name: "Settings",
        description: "Application settings",
        mimeType: "application/json",
      },
    ],
  };
});

// Handle resource reads
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const uri = request.params.uri;

  if (uri.startsWith("file://")) {
    const filePath = uri.replace("file://", "");
    const content = await fs.readFile(filePath, "utf-8");
    return {
      contents: [
        {
          uri,
          mimeType: "text/plain",
          text: content,
        },
      ],
    };
  }

  throw new Error(`Unknown resource: ${uri}`);
});

Python 资源示例

from mcp.types import Resource, ResourceContent

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="file:///docs/readme.md",
            name="README",
            description="Project README",
            mimeType="text/markdown",
        ),
        Resource(
            uri="file:///config/settings.json",
            name="Settings",
            description="App settings",
            mimeType="application/json",
        ),
    ]

@app.read_resource()
async def read_resource(uri: str) -> ResourceContent:
    if uri.startswith("file://"):
        file_path = uri.replace("file://", "")
        with open(file_path, "r") as f:
            content = f.read()
        return ResourceContent(uri=uri, mimeType="text/plain", text=content)

    raise ValueError(f"Unknown resource: {uri}")

提示词——可复用模板

提示词是大模型可通过参数使用的模板。

TypeScript 提示词示例

import {
  ListPromptsRequestSchema,
  GetPromptRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

// List prompts
server.setRequestHandler(ListPromptsRequestSchema, async () => {
  return {
    prompts: [
      {
        name: "code_review",
        description: "Review code for best practices",
        arguments: [
          {
            name: "language",
            description: "Programming language",
            required: true,
          },
          {
            name: "code",
            description: "Code to review",
            required: true,
          },
        ],
      },
    ],
  };
});

// Handle prompt requests
server.setRequestHandler(GetPromptRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "code_review") {
    const language = args?.language || "unknown";
    const code = args?.code || "";

    return {
      messages: [
        {
          role: "user",
          content: {
            type: "text",
            text: `Review this ${language} code for best practices, security issues, and improvements:\n\n\`\`\`${language}\n${code}\n\`\`\``,
          },
        },
      ],
    };
  }

  throw new Error(`Unknown prompt: ${name}`);
});

Python 提示词示例

from mcp.types import Prompt, PromptMessage, PromptArgument

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="code_review",
            description="Review code for best practices",
            arguments=[
                PromptArgument(
                    name="language", description="Programming language", required=True
                ),
                PromptArgument(name="code", description="Code to review", required=True),
            ],
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
    if name == "code_review":
        language = arguments.get("language", "unknown")
        code = arguments.get("code", "")

        return [
            PromptMessage(
                role="user",
                content={
                    "type": "text",
                    "text": f"Review this {language} code:\n\n```{language}\n{code}\n```",
                },
            )
        ]

    raise ValueError(f"Unknown prompt: {name}")

高级模式

1. 错误处理

// TypeScript
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  try {
    const { name, arguments: args } = request.params;

    if (name === "risky_operation") {
      // Validate inputs
      if (!args.required_param) {
        throw new Error("Missing required parameter: required_param");
      }

      // Perform operation with proper error handling
      const result = await performRiskyOperation(args.required_param);

      return {
        content: [{ type: "text", text: JSON.stringify(result) }],
      };
    }
  } catch (error) {
    // Return error to LLM with helpful message
    return {
      content: [
        {
          type: "text",
          text: `Error: ${error instanceof Error ? error.message : String(error)}`,
        },
      ],
      isError: true,
    };
  }
});

# Python
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    try:
        if name == "risky_operation":
            if "required_param" not in arguments:
                raise ValueError("Missing required parameter: required_param")

            result = await perform_risky_operation(arguments["required_param"])
            return [TextContent(type="text", text=json.dumps(result))]

    except Exception as e:
        return [TextContent(type="text", text=f"Error: {str(e)}", isError=True)]

2. 异步操作

// TypeScript - Async API calls
import axios from "axios";

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "fetch_api_data") {
    const url = args.url as string;
    const response = await axios.get(url);

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(response.data, null, 2),
        },
      ],
    };
  }
});

# Python - Async database queries
import aiosqlite

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "query_database":
        query = arguments["query"]

        async with aiosqlite.connect("database.db") as db:
            async with db.execute(query) as cursor:
                rows = await cursor.fetchall()
                result = json.dumps(rows)

        return [TextContent(type="text", text=result)]

3. 环境变量和配置

// TypeScript - Load config from environment
import dotenv from "dotenv";
dotenv.config();

const API_KEY = process.env.API_KEY;
const BASE_URL = process.env.BASE_URL || "https://api.example.com";

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "api_call") {
    const response = await fetch(`${BASE_URL}/endpoint`, {
      headers: { Authorization: `Bearer ${API_KEY}` },
    });
    // ...
  }
});

# Python - Load config from environment
import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("API_KEY")
BASE_URL = os.getenv("BASE_URL", "https://api.example.com")

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "api_call":
        headers = {"Authorization": f"Bearer {API_KEY}"}
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{BASE_URL}/endpoint", headers=headers) as resp:
                data = await resp.json()
                return [TextContent(type="text", text=json.dumps(data))]

4. 流式响应（大型数据）

// TypeScript - Stream large file contents
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "read_large_file") {
    const filePath = request.params.arguments.path as string;
    const stream = fs.createReadStream(filePath, { encoding: "utf-8" });

    let content = "";
    for await (const chunk of stream) {
      content += chunk;
      // Could yield chunks incrementally in future MCP versions
    }

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

5. 动态注册工具

// TypeScript - Register tools from config
interface ToolConfig {
  name: string;
  description: string;
  schema: object;
  handler: (args: any) => Promise<any>;
}

const toolRegistry = new Map<string, ToolConfig>();

function registerTool(config: ToolConfig) {
  toolRegistry.set(config.name, config);
}

// Register custom tools
registerTool({
  name: "custom_tool",
  description: "Dynamically registered tool",
  schema: {
    type: "object",
    properties: {
      input: { type: "string" },
    },
  },
  handler: async (args) => {
    return { result: `Processed: ${args.input}` };
  },
});

// List tools dynamically
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: Array.from(toolRegistry.values()).map((tool) => ({
      name: tool.name,
      description: tool.description,
      inputSchema: tool.schema,
    })),
  };
});

// Call tools dynamically
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const tool = toolRegistry.get(request.params.name);
  if (!tool) {
    throw new Error(`Unknown tool: ${request.params.name}`);
  }

  const result = await tool.handler(request.params.arguments);
  return {
    content: [{ type: "text", text: JSON.stringify(result) }],
  };
});

传输类型

1. STDIO（本地执行）

Claude Desktop 集成的默认方式。服务以子进程形式运行。

// Claude Desktop config
{
  "mcpServers": {
    "my-server": {
      "command": "node",
      "args": ["/path/to/server/build/index.js"],
      "env": {
        "API_KEY": "your-api-key"
      }
    }
  }
}

2. SSE（Server-Sent Events）

适用于通过 HTTP 传输的长时间运行服务。

// TypeScript SSE server
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import express from "express";

const app = express();
const server = new Server(/* ... */);

app.get("/sse", async (req, res) => {
  const transport = new SSEServerTransport("/messages", res);
  await server.connect(transport);
});

app.post("/messages", async (req, res) => {
  // Handle incoming messages
});

app.listen(3000, () => {
  console.log("MCP server listening on http://localhost:3000");
});

// Claude Desktop config for SSE
{
  "mcpServers": {
    "remote-server": {
      "url": "http://localhost:3000/sse"
    }
  }
}

3. HTTP（可流式传输的 HTTP）

适用于 REST 风格的 MCP 服务。

# Python HTTP server with FastAPI
from fastapi import FastAPI
from mcp.server.fastapi import create_fastapi_app

app = FastAPI()
mcp_app = Server("http-server")

# Define tools/resources/prompts as usual
# ...

# Mount MCP routes
app.mount("/mcp", create_fastapi_app(mcp_app))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

调试 MCP 服务

1. 服务日志

// TypeScript - Console.error for logs (STDOUT is for MCP protocol)
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  console.error(`Tool called: ${request.params.name}`);
  console.error(`Arguments: ${JSON.stringify(request.params.arguments)}`);
  // ...
});

# Python - Use logging module
import logging

logging.basicConfig(level=logging.DEBUG, filename="mcp-server.log")
logger = logging.getLogger(__name__)

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    logger.debug(f"Tool called: {name}")
    logger.debug(f"Arguments: {arguments}")
    # ...

2. MCP Inspector

# Install MCP Inspector (browser-based debugging tool)
npm install -g @modelcontextprotocol/inspector

# Run inspector with your server
mcp-inspector node /path/to/server/build/index.js

# Opens browser at http://localhost:6274
# - Test tools manually
# - See request/response payloads
# - Debug JSON schema validation

3. Claude Desktop 日志

# macOS
tail -f ~/Library/Logs/Claude/mcp*.log

# Windows
# Check %APPDATA%/Claude/logs/

4. 独立测试服务

// TypeScript - Add test harness
if (process.argv.includes("--test")) {
  // Simulate tool call
  const testRequest = {
    params: {
      name: "read_file",
      arguments: { path: "./test.txt" },
    },
  };

  server
    .setRequestHandler(CallToolRequestSchema, async (request) => {
      // ... your handler
    })
    .then((handler) => handler(testRequest))
    .then((result) => console.log(JSON.stringify(result, null, 2)))
    .catch((error) => console.error(error));
} else {
  // Normal server startup
  main();
}

# Test without Claude Desktop
node build/index.js --test

实际服务示例

1. GitHub 集成服务

// github-server/src/index.ts
import { Octokit } from "@octokit/rest";

const octokit = new Octokit({ auth: process.env.GITHUB_TOKEN });

server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "list_repos",
        description: "List repositories for a user/org",
        inputSchema: {
          type: "object",
          properties: {
            owner: { type: "string", description: "Username or org name" },
          },
          required: ["owner"],
        },
      },
      {
        name: "create_issue",
        description: "Create a GitHub issue",
        inputSchema: {
          type: "object",
          properties: {
            owner: { type: "string" },
            repo: { type: "string" },
            title: { type: "string" },
            body: { type: "string" },
          },
          required: ["owner", "repo", "title"],
        },
      },
    ],
  };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "list_repos") {
    const { data } = await octokit.repos.listForUser({
      username: args.owner as string,
    });
    const repos = data.map((r) => ({
      name: r.name,
      description: r.description,
      stars: r.stargazers_count,
    }));
    return {
      content: [{ type: "text", text: JSON.stringify(repos, null, 2) }],
    };
  }

  if (name === "create_issue") {
    const { data } = await octokit.issues.create({
      owner: args.owner as string,
      repo: args.repo as string,
      title: args.title as string,
      body: args.body as string,
    });
    return {
      content: [
        { type: "text", text: `Issue created: ${data.html_url}` },
      ],
    };
  }
});

2. 数据库查询服务

# database-server/server.py
import asyncpg
import json
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("database-server")

async def get_db_pool():
    return await asyncpg.create_pool(
        host="localhost",
        database="mydb",
        user="user",
        password="password"
    )

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="query",
            description="Execute SQL query (SELECT only)",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SQL query"},
                },
                "required": ["sql"],
            },
        ),
        Tool(
            name="list_tables",
            description="List all tables in database",
            inputSchema={"type": "object", "properties": {}},
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    pool = await get_db_pool()

    if name == "query":
        sql = arguments["sql"]
        # Security: Only allow SELECT
        if not sql.strip().upper().startswith("SELECT"):
            raise ValueError("Only SELECT queries allowed")

        async with pool.acquire() as conn:
            rows = await conn.fetch(sql)
            result = [dict(row) for row in rows]

        return [TextContent(type="text", text=json.dumps(result, indent=2))]

    elif name == "list_tables":
        async with pool.acquire() as conn:
            tables = await conn.fetch("""
                SELECT table_name FROM information_schema.tables
                WHERE table_schema = 'public'
            """)
            result = [row["table_name"] for row in tables]

        return [TextContent(type="text", text=json.dumps(result, indent=2))]

3. 网页抓取服务

// scraper-server/src/index.ts
import * as cheerio from "cheerio";
import axios from "axios";

server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "scrape_url",
        description: "Scrape content from a URL",
        inputSchema: {
          type: "object",
          properties: {
            url: { type: "string", description: "URL to scrape" },
            selector: {
              type: "string",
              description: "CSS selector for content",
            },
          },
          required: ["url"],
        },
      },
    ],
  };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "scrape_url") {
    const url = request.params.arguments.url as string;
    const selector = request.params.arguments.selector as string;

    const response = await axios.get(url);
    const $ = cheerio.load(response.data);

    let content: string;
    if (selector) {
      content = $(selector).text();
    } else {
      content = $("body").text();
    }

    return {
      content: [
        {
          type: "text",
          text: content.trim(),
        },
      ],
    };
  }
});

最佳实践

1. 工具设计

✅ 应该：

• 使用清晰且描述性强的工具名称（用 read_file，而不是 rf）
• 提供详细说明，帮助大模型理解
• 定义完整且带有说明的 JSON Schema
• 尽可能返回结构化数据（JSON）
• 妥善处理错误，并提供有帮助的信息

❌ 不应该：

• 创建范围过广的工具（将复杂操作拆分开）
• 返回超大载荷（对大型数据集进行分页）
• 使用含糊的参数名称
• 假设大模型了解特定领域的术语

2. 安全性

关键规则：

• 验证所有输入（类型、范围、格式）
• 清理文件路径（防止目录遍历）
• 对命令/操作使用允许列表
• 绝不要在响应中暴露敏感凭据
• 对开销较大的操作实施速率限制
• 默认使用只读访问

// Example: Path validation
function validatePath(inputPath: string): string {
  const normalized = path.normalize(inputPath);
  const allowed = path.resolve("/safe/directory");

  if (!normalized.startsWith(allowed)) {
    throw new Error("Path outside allowed directory");
  }

  return normalized;
}

3. 性能

• 缓存开销较大的操作
• 尽可能对大型响应进行流式传输
• 对列表操作使用分页
• 设置合理的超时时间
• 为受速率限制的 API 实施请求排队

4. 测试

# Use MCP Inspector for manual testing
mcp-inspector node build/index.js

# Unit test tool handlers
npm test

# Integration test with Claude Desktop
# 1. Add to config
# 2. Restart Claude
# 3. Test in conversation

5. 文档

• 在代码注释中记录所有工具
• 在说明中提供用法示例
• 在文档中列出错误场景
• 维护记录服务更新的 CHANGELOG

常见问题

❌ 向 STDOUT 写入内容（会破坏 MCP 协议）：

// WRONG
console.log("Debug message");  // STDOUT is for MCP protocol

// CORRECT
console.error("Debug message");  // STDERR for logs

❌ 未处理错误：

// WRONG - unhandled promise rejection
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const data = await riskyOperation();  // Can throw
  return { content: [{ type: "text", text: data }] };
});

// CORRECT
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  try {
    const data = await riskyOperation();
    return { content: [{ type: "text", text: data }] };
  } catch (error) {
    return {
      content: [{ type: "text", text: `Error: ${error.message}` }],
      isError: true,
    };
  }
});

❌ 阻塞操作：

# WRONG - synchronous file read blocks event loop
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    with open("large_file.txt", "r") as f:  # Blocks!
        content = f.read()
    return [TextContent(type="text", text=content)]

# CORRECT - async file I/O
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    async with aiofiles.open("large_file.txt", "r") as f:
        content = await f.read()
    return [TextContent(type="text", text=content)]

❌ Schema 中缺少必填字段：

// WRONG - missing "required" field
{
  name: "search",
  inputSchema: {
    type: "object",
    properties: {
      query: { type: "string" }
    }
    // Missing: required: ["query"]
  }
}

资源

• 官方文档：https://modelcontextprotocol.io
• TypeScript SDK：https://github.com/modelcontextprotocol/typescript-sdk
• Python SDK：https://github.com/modelcontextprotocol/python-sdk
• 服务示例：https://github.com/modelcontextprotocol/servers
• MCP Inspector：https://github.com/modelcontextprotocol/inspector
• 规范：https://spec.modelcontextprotocol.io

相关技能

构建 MCP 服务时，可以考虑以下互补技能：

• typescript-core：TypeScript 类型安全、tsconfig 优化和高级模式
• asyncio：使用 async/await 为 MCP 服务实现 Python 异步模式
• openrouter：用于多模型支持的替代 LLM API 集成

快速 TypeScript MCP 模式（内联，可独立使用）

// Type-safe MCP server with TypeScript
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 { z } from 'zod';

// Define tool schemas with runtime validation
const SearchSchema = z.object({
  query: z.string().min(1).max(500),
  limit: z.number().int().min(1).max(100).default(10),
});

type SearchArgs = z.infer<typeof SearchSchema>;

class TypeSafeMCPServer {
  private server: Server;

  constructor() {
    this.server = new Server(
      {
        name: 'typed-search-server',
        version: '1.0.0',
      },
      {
        capabilities: {
          tools: {},
        },
      }
    );

    this.setupToolHandlers();
  }

  private setupToolHandlers() {
    // List tools with full type inference
    this.server.setRequestHandler(ListToolsRequestSchema, async () => ({
      tools: [
        {
          name: 'search',
          description: 'Search with type-safe parameters',
          inputSchema: {
            type: 'object',
            properties: {
              query: { type: 'string', minLength: 1, maxLength: 500 },
              limit: { type: 'number', minimum: 1, maximum: 100, default: 10 },
            },
            required: ['query'],
          },
        },
      ],
    }));

    // Type-safe tool execution
    this.server.setRequestHandler(CallToolRequestSchema, async (request) => {
      if (request.params.name === 'search') {
        // Runtime validation with Zod
        const args = SearchSchema.parse(request.params.arguments);

        // Type-safe implementation
        const results = await this.performSearch(args);

        return {
          content: [{ type: 'text', text: JSON.stringify(results) }],
        };
      }

      throw new Error(`Unknown tool: ${request.params.name}`);
    });
  }

  private async performSearch(args: SearchArgs): Promise<Array<{ title: string; url: string }>> {
    // Implementation with full type safety
    // args.query is string, args.limit is number
    return [];
  }

  async run() {
    const transport = new StdioServerTransport();
    await this.server.connect(transport);
  }
}

// Start server
const server = new TypeSafeMCPServer();
server.run();

快速 Python Async MCP 模式（内联，可独立使用）

# Async MCP server with Python
import asyncio
import logging
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field

# Type-safe argument models with Pydantic
class SearchArgs(BaseModel):
    query: str = Field(..., min_length=1, max_length=500)
    limit: int = Field(10, ge=1, le=100)

class AsyncMCPServer:
    def __init__(self):
        self.server = Server("async-search-server")
        self.setup_handlers()

    def setup_handlers(self):
        @self.server.list_tools()
        async def list_tools() -> list[Tool]:
            return [
                Tool(
                    name="search",
                    description="Search with async processing",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "query": {"type": "string", "minLength": 1, "maxLength": 500},
                            "limit": {"type": "number", "minimum": 1, "maximum": 100, "default": 10},
                        },
                        "required": ["query"],
                    },
                )
            ]

        @self.server.call_tool()
        async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
            if name == "search":
                # Validate with Pydantic
                args = SearchArgs(**arguments)

                # Async implementation
                results = await self.perform_search(args)

                return [
                    TextContent(
                        type="text",
                        text=str(results)
                    )
                ]

            raise ValueError(f"Unknown tool: {name}")

    async def perform_search(self, args: SearchArgs) -> list[dict[str, str]]:
        """Async search implementation"""
        # Simulate async I/O
        await asyncio.sleep(0.1)

        # Use validated args (args.query is str, args.limit is int)
        return [
            {"title": f"Result for {args.query}", "url": "https://example.com"}
        ]

async def main():
    """Run async MCP server"""
    logging.basicConfig(level=logging.INFO)
    server = AsyncMCPServer()

    # Use stdio transport for Claude Desktop
    async with stdio_server() as (read_stream, write_stream):
        await server.server.run(
            read_stream,
            write_stream,
            server.server.create_initialization_options()
        )

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

快速多语言 MCP 模式（内联，可独立使用）

TypeScript 与 Python 的权衡：

特性	TypeScript	Python
类型安全	编译时 + 运行时（Zod）	仅运行时（Pydantic）
性能	启动更快，但有 Node.js 开销	启动较慢，更适合 CPU 密集型任务
异步支持	原生 async/await、事件循环	asyncio，特别适合 I/O
生态系统	npm 包、前端工具	数据科学、ML 库
最适合	Web API、实时工具	数据处理、ML 集成

两者共有的常见模式：

1. 输入验证

   • TypeScript：Zod 模式
   • Python：Pydantic 模型

2. 错误处理

   • 两者：使用特定错误类型进行 Try/catch
   • 在 MCP 响应中返回错误内容

3. 资源管理

   • TypeScript：使用 try/finally 的 async/await
   • Python：异步上下文管理器

4. 测试

   • TypeScript：使用 mock transport 的 Vitest/Jest
   • Python：使用 pytest-asyncio 的 pytest

选择实现语言：

// TypeScript - Best for:
// - File system operations
// - Web scraping/HTTP requests
// - JSON/API manipulation
// - Real-time data streams

// Python - Best for:
// - Data analysis (pandas, numpy)
// - Machine learning (scikit-learn, torch)
// - Database ETL operations
// - Scientific computing

[如果一同部署，完整的 TypeScript、Python async 和 OpenRouter 模式可在相应技能中查看]

本地 MCP 集成模式（你的代码库）

.mcp.json 布局（项目范围）

{
  "mcpServers": {
    "kuzu-memory": {
      "type": "stdio",
      "command": "kuzu-memory",
      "args": ["mcp"]
    },
    "mcp-vector-search": {
      "type": "stdio",
      "command": "uv",
      "args": ["run", "mcp-vector-search", "mcp"],
      "env": {
        "MCP_ENABLE_FILE_WATCHING": "true"
      }
    }
  }
}

CLI 约定

• 为 stdio 服务提供 mcp 子命令。
• 使用 setup 完成端到端初始化和集成（mcp-vector-search、kuzu-memory）。
• 使用 install / uninstall 指定目标客户端（mcp-ticketer）。
• 提供 doctor 命令以验证依赖项。

环境约定

• 文件监视：MCP_ENABLE_FILE_WATCHING=true
• Kuzu 内存：KUZU_MEMORY_PROJECT_ROOT、KUZU_MEMORY_DB
• Ticketer 适配器：MCP_TICKETER_ADAPTER、GITHUB_TOKEN、GITHUB_OWNER、GITHUB_REPO

常见服务名称

kuzu-memory、mcp-vector-search、mcp-ticketer、mcp-skillset、mcp-browser

总结

• MCP 支持以 AI 为核心的服务开发，提供工具、资源和提示词
• 工具 是大模型可以调用的函数（读取文件、查询 API、执行操作）
• 资源 提供只读数据访问（文件、数据库、文档）
• 提示词 是包含参数、可用于常见任务的可复用模板
• SDK 提供 TypeScript 和 Python 版本，并全面支持异步操作
• 通过 JSON 配置集成 Claude Desktop（STDIO 传输）
• 使用 MCP Inspector、服务日志和独立测试进行 调试
• 安全 至关重要：验证输入、清理路径、使用允许列表
• 最佳实践：使用清晰的命名、完善的模式、错误处理和性能优化