教程

MCP Server开发与工具集成完全教程

2026年06月01日 10:18 23 次阅读
教程简介

本教程全面讲解MCP（Model Context Protocol）Server开发与工具集成的核心技术，涵盖MCP协议架构与规范、Python/TypeScript SDK开发、Resources与Tools实现、Prompt模板设计、Sampling采样机制、本地与远程Server部署、数据库/文件系统/API集成、安全权限管理、与Claude Desktop/LangChain集成等核心内容，通过完整的数据库查询和文件操作MCP Server实战案例帮助开发者掌握MCP生态开发。

MCP Server开发与工具集成完全教程

本教程全面讲解MCP（Model Context Protocol）Server开发与工具集成的核心技术，通过丰富的代码示例和实战案例，帮助开发者掌握MCP生态开发。

目录

  1. MCP协议概述
  2. 协议架构与核心概念
  3. Python SDK开发
  4. TypeScript SDK开发
  5. Resources资源实现
  6. Tools工具实现
  7. Prompt模板设计
  8. Sampling采样机制
  9. 本地与远程Server部署
  10. 数据库集成
  11. 文件系统集成
  12. 外部API集成
  13. 安全权限管理
  14. 与Claude Desktop集成
  15. 与LangChain集成
  16. 实战：数据库查询MCP Server
  17. 实战：文件操作MCP Server
  18. 最佳实践
  19. 总结

MCP协议概述

Model Context Protocol（MCP）是由Anthropic于2024年11月发布的一项开放协议，旨在为大语言模型（LLM）应用提供标准化的外部数据源和工具连接方式。MCP可以被理解为AI应用的"USB-C接口"——它提供了一种统一的方式，让AI模型能够安全地访问各种外部资源。

为什么需要MCP？

在MCP出现之前，每个AI应用与外部工具的集成都是"点对点"的定制开发：

传统方式（N×M复杂度）：
┌────────┐     ┌────────┐
│ Claude │────▶│ Tool A │
│        │────▶│ Tool B │
│        │────▶│ Tool C │
└────────┘     └────────┘
┌────────┐     ┌────────┐
│ GPT    │────▶│ Tool A │  ← 每个应用都要为每个工具写适配代码
│        │────▶│ Tool B │
└────────┘     └────────┘

MCP方式（N+M复杂度）：
┌────────┐         ┌─────────┐         ┌────────┐
│ Claude │◀──MCP──▶│  MCP    │◀──MCP──▶│ Server │
│ GPT    │◀──MCP──▶│ Protocol│◀──MCP──▶│ Server │
│ Cursor │◀──MCP──▶│         │◀──MCP──▶│ Server │
└────────┘         └─────────┘         └────────┘
  MCP客户端         标准协议层            MCP服务端

MCP通过标准化协议将N×M的复杂度降低为N+M，使得：

  • 工具开发者只需实现一次MCP Server，所有支持MCP的客户端都能使用
  • 应用开发者只需实现MCP客户端，即可接入所有MCP Server
  • 用户获得了更丰富的工具生态系统

MCP的核心价值

  1. 标准化：统一的协议规范，消除碎片化集成
  2. 安全性：内置权限控制和安全边界
  3. 可组合性：Server可以相互组合，构建复杂工作流
  4. 生态效应：社区共建共享的工具生态

协议架构与核心概念

整体架构

┌─────────────────────────────────────────────────────────┐
│                    MCP Host Application                   │
│              （如Claude Desktop、Cursor IDE）              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐      │
│  │ MCP Client  │  │ MCP Client  │  │ MCP Client  │      │
│  │  (实例1)    │  │  (实例2)    │  │  (实例3)    │      │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘      │
└─────────┼────────────────┼────────────────┼──────────────┘
          │                │                │
    ┌─────▼─────┐   ┌─────▼─────┐   ┌─────▼─────┐
    │ MCP Server│   │ MCP Server│   │ MCP Server│
    │ (数据库)  │   │ (文件系统)│   │ (Web API) │
    └───────────┘   └───────────┘   └───────────┘

核心概念

MCP协议定义了三种核心原语（Primitives）：

原语 控制方 描述 类比
Resources 应用程序控制 上下文数据，由客户端决定何时获取 GET请求
Tools 模型控制 由AI模型决定何时调用的函数 POST请求
Prompts 用户控制 预定义的模板，由用户主动选择 快捷指令

通信协议

MCP基于JSON-RPC 2.0协议，支持两种传输方式：

1. stdio（标准输入/输出）—— 本地通信
   Client ──stdin──▶ Server
   Client ◀──stdout── Server

2. HTTP + SSE（Server-Sent Events）—— 远程通信
   Client ──HTTP POST──▶ Server
   Client ◀──SSE stream── Server

生命周期

Client                           Server
  │                                │
  │──── initialize ─────────────▶│
  │◀─── initialize result ───────│
  │──── initialized ────────────▶│
  │                                │
  │     ◀── 正常请求/响应 ──▶     │
  │                                │
  │──── shutdown ────────────────▶│
  │◀─── exit ────────────────────│

Python SDK开发

MCP官方提供了Python SDK（mcp包），是开发MCP Server最常用的方式。

安装与基础设置

# 安装MCP Python SDK
pip install mcp

# 推荐使用uv进行项目管理
pip install uv

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

最小MCP Server

# server.py —— 最简单的MCP Server
from mcp.server.fastmcp import FastMCP

# 创建MCP Server实例
mcp = FastMCP("my-first-server")

# 定义一个工具
@mcp.tool()
def add(a: int, b: int) -> int:
    """将两个数字相加

    Args:
        a: 第一个数字
        b: 第二个数字
    """
    return a + b

# 定义一个资源
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """获取个性化问候语"""
    return f"你好，{name}！欢迎使用MCP Server。"

# 定义一个prompt模板
@mcp.prompt()
def review_code(code: str) -> str:
    """代码审查prompt"""
    return f"请审查以下代码并提供改进建议：\n\n```python\n{code}\n```"

# 启动Server
if __name__ == "__main__":
    mcp.run()

异步工具实现

import asyncio
import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("async-server")

@mcp.tool()
async def fetch_weather(city: str) -> dict:
    """获取城市天气信息

    Args:
        city: 城市名称（中文或英文）
    """
    async with httpx.AsyncClient() as client:
        # 使用wttr.in获取天气
        response = await client.get(
            f"https://wttr.in/{city}?format=j1",
            timeout=10.0
        )
        if response.status_code == 200:
            data = response.json()
            current = data.get("current_condition", [{}])[0]
            return {
                "city": city,
                "temperature": current.get("temp_C", "N/A"),
                "humidity": current.get("humidity", "N/A"),
                "description": current.get("weatherDesc", [{}])[0].get("value", "N/A"),
                "wind_speed": current.get("windspeedKmph", "N/A")
            }
        return {"error": f"无法获取{city}的天气信息"}

@mcp.tool()
async def search_web(query: str, max_results: int = 5) -> list[dict]:
    """搜索网页内容

    Args:
        query: 搜索关键词
        max_results: 最大返回结果数
    """
    async with httpx.AsyncClient() as client:
        # 这里使用一个简化的搜索API示例
        response = await client.get(
            "https://api.search.example.com/search",
            params={"q": query, "limit": max_results},
            timeout=15.0
        )
        results = response.json().get("results", [])
        return [
            {
                "title": r.get("title"),
                "url": r.get("url"),
                "snippet": r.get("snippet", "")[:200]
            }
            for r in results[:max_results]
        ]

带上下文的工具

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("context-aware-server")

@mcp.tool()
async def long_running_task(data: str, ctx: Context) -> str:
    """执行一个长时间运行的任务，支持进度报告

    Args:
        data: 输入数据
    """
    steps = ["解析数据", "验证格式", "处理中", "生成结果"]

    for i, step in enumerate(steps):
        # 报告进度
        await ctx.report_progress(i + 1, len(steps))
        await ctx.info(f"正在执行: {step}")
        await asyncio.sleep(1)  # 模拟耗时操作

    return f"任务完成！处理了数据: {data[:50]}..."

@mcp.tool()
async def read_resource_content(uri: str, ctx: Context) -> str:
    """读取指定资源的内容

    Args:
        uri: 资源URI
    """
    # 通过上下文读取资源
    try:
        content = await ctx.read_resource(uri)
        return content
    except Exception as e:
        return f"读取资源失败: {e}"

TypeScript SDK开发

对于前端开发者，MCP也提供了TypeScript SDK。

安装与设置

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

# 安装依赖
npm install @modelcontextprotocol/sdk zod

# TypeScript配置
npm install -D typescript @types/node

TypeScript MCP Server实现

// src/server.ts
import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

// 创建MCP Server实例
const server = new McpServer({
  name: "typescript-mcp-server",
  version: "1.0.0",
});

// 定义工具
server.tool(
  "calculate",
  "执行数学计算",
  {
    expression: z.string().describe("数学表达式，如 '2 + 3 * 4'"),
  },
  async ({ expression }) => {
    try {
      // 注意：生产环境中不应使用eval，这里仅为演示
      const result = Function(`"use strict"; return (${expression})`)();
      return {
        content: [
          {
            type: "text",
            text: `${expression} = ${result}`,
          },
        ],
      };
    } catch (error) {
      return {
        content: [
          {
            type: "text",
            text: `计算错误: ${error}`,
          },
        ],
        isError: true,
      };
    }
  }
);

// 定义资源
server.resource(
  "config",
  "config://app",
  async (uri) => ({
    contents: [
      {
        uri: uri.href,
        text: JSON.stringify({
          appName: "My MCP App",
          version: "1.0.0",
          features: ["tools", "resources", "prompts"],
        }, null, 2),
      },
    ],
  })
);

// 带模板的资源
server.resource(
  "user-profile",
  new ResourceTemplate("users://{userId}/profile", { list: undefined }),
  async (uri, { userId }) => ({
    contents: [
      {
        uri: uri.href,
        text: JSON.stringify({
          id: userId,
          name: `User ${userId}`,
          joined: new Date().toISOString(),
        }),
      },
    ],
  })
);

// 定义Prompt模板
server.prompt(
  "summarize",
  "总结文本内容",
  {
    text: z.string().describe("需要总结的文本"),
    style: z.enum(["brief", "detailed", "bullet-points"]).optional(),
  },
  async ({ text, style }) => ({
    messages: [
      {
        role: "user",
        content: {
          type: "text",
          text: `请以${style === "brief" ? "简洁" : style === "bullet-points" ? "要点列表" : "详细"}的方式总结以下内容：\n\n${text}`,
        },
      },
    ],
  })
);

// 启动Server
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP Server已启动 (stdio模式)");
}

main().catch(console.error);

TypeScript配置文件

// tsconfig.json
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./build",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true,
    "declaration": true
  },
  "include": ["src/**/*"]
}

// package.json
{
  "name": "my-mcp-server-ts",
  "version": "1.0.0",
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node build/server.js"
  },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.0.0",
    "zod": "^3.22.0"
  }
}

Resources资源实现

Resources是MCP中由应用程序控制的数据源，类似于REST API中的GET端点。

静态资源

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("resource-server")

# 简单的静态资源
@mcp.resource("config://database")
def get_db_config() -> str:
    """获取数据库配置信息"""
    return json.dumps({
        "host": "localhost",
        "port": 5432,
        "database": "myapp",
        "pool_size": 10
    }, indent=2)

# 文件资源
@mcp.resource("file://readme")
def get_readme() -> str:
    """获取项目README内容"""
    with open("README.md", "r", encoding="utf-8") as f:
        return f.read()

动态资源（带参数模板）

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("dynamic-resources")

# 使用URI模板的动态资源
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> dict:
    """获取用户资料

    Args:
        user_id: 用户ID
    """
    # 模拟从数据库查询
    users_db = {
        "001": {"name": "张三", "email": "zhangsan@example.com", "role": "admin"},
        "002": {"name": "李四", "email": "lisi@example.com", "role": "user"},
    }
    user = users_db.get(user_id, {"error": "用户不存在"})
    return {"user_id": user_id, **user}

# 带查询参数的资源
@mcp.resource("logs://{service}/recent")
def get_recent_logs(service: str, lines: int = 50) -> list[str]:
    """获取服务的最近日志

    Args:
        service: 服务名称
        lines: 返回的日志行数
    """
    import subprocess
    try:
        result = subprocess.run(
            ["tail", f"-n{lines}", f"/var/log/{service}.log"],
            capture_output=True, text=True, timeout=5
        )
        return result.stdout.strip().split("\n")
    except Exception as e:
        return [f"读取日志失败: {e}"]

列举资源

@mcp.resource("files://documents")
def list_documents() -> list[dict]:
    """列出所有可用文档"""
    docs_dir = Path("./documents")
    if not docs_dir.exists():
        return []

    documents = []
    for file_path in docs_dir.glob("*.md"):
        documents.append({
            "name": file_path.name,
            "path": str(file_path),
            "size": file_path.stat().st_size,
            "modified": file_path.stat().st_mtime
        })
    return documents

Tools工具实现

Tools是MCP中最强大的原语，它让AI模型能够执行具体的操作。

基础工具定义

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field

mcp = FastMCP("tools-server")

# 简单工具
@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> dict:
    """计算BMI指数

    Args:
        weight_kg: 体重（千克）
        height_m: 身高（米）
    """
    bmi = weight_kg / (height_m ** 2)

    if bmi < 18.5:
        category = "偏瘦"
    elif bmi < 24:
        category = "正常"
    elif bmi < 28:
        category = "偏胖"
    else:
        category = "肥胖"

    return {
        "bmi": round(bmi, 2),
        "category": category,
        "healthy_range": "18.5 - 24.0"
    }

# 使用Pydantic模型定义复杂参数
class EmailMessage(BaseModel):
    to: str = Field(description="收件人邮箱地址")
    subject: str = Field(description="邮件主题")
    body: str = Field(description="邮件正文")
    cc: list[str] = Field(default_factory=list, description="抄送列表")
    priority: str = Field(default="normal", description="优先级: low/normal/high")

@mcp.tool()
async def send_email(message: EmailMessage) -> dict:
    """发送邮件

    Args:
        message: 邮件消息对象
    """
    # 这里是邮件发送逻辑的示例
    print(f"发送邮件到: {message.to}")
    print(f"主题: {message.subject}")
    print(f"优先级: {message.priority}")

    # 模拟发送
    return {
        "status": "sent",
        "message_id": "msg_123456",
        "to": message.to,
        "subject": message.subject
    }

带错误处理的工具

from mcp.server.fastmcp import FastMCP
from mcp.types import TextContent

mcp = FastMCP("robust-tools")

@mcp.tool()
async def safe_database_query(sql: str, database: str = "default") -> list[dict]:
    """安全执行数据库查询

    Args:
        sql: SQL查询语句（仅支持SELECT）
        database: 数据库名称
    """
    # 安全检查
    sql_upper = sql.strip().upper()
    if not sql_upper.startswith("SELECT"):
        return [{"error": "仅支持SELECT查询，不允许修改数据"}]

    # 危险关键词检查
    dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE"]
    for keyword in dangerous_keywords:
        if keyword in sql_upper:
            return [{"error": f"检测到危险关键词: {keyword}"}]

    try:
        # 执行查询
        # result = await execute_query(database, sql)
        return [{"status": "success", "rows": [], "message": "查询已执行"}]
    except Exception as e:
        return [{"error": f"查询执行失败: {str(e)}"}]

工具的高级特性

from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("advanced-tools")

@mcp.tool()
async def process_file(
    file_path: str,
    operation: str,
    output_format: str = "json",
    ctx: Context = None
) -> dict:
    """处理文件，支持多种操作

    Args:
        file_path: 文件路径
        operation: 操作类型 (read/analyze/convert)
        output_format: 输出格式 (json/text/csv)
    """
    from pathlib import Path

    path = Path(file_path)

    # 安全检查：防止路径遍历
    if ".." in file_path or file_path.startswith("/etc"):
        return {"error": "不允许访问该路径"}

    if not path.exists():
        return {"error": f"文件不存在: {file_path}"}

    if ctx:
        await ctx.info(f"开始处理文件: {file_path}")

    if operation == "read":
        content = path.read_text(encoding="utf-8")
        return {"content": content[:10000], "truncated": len(content) > 10000}

    elif operation == "analyze":
        stat = path.stat()
        return {
            "name": path.name,
            "size_bytes": stat.st_size,
            "extension": path.suffix,
            "modified": stat.st_mtime,
            "is_binary": not path.suffix in {".txt", ".md", ".py", ".json", ".csv"}
        }

    elif operation == "convert":
        # 文件格式转换逻辑
        return {"status": "converted", "output_format": output_format}

    return {"error": f"未知操作: {operation}"}

Prompt模板设计

Prompts是用户控制的模板，可以帮助用户更高效地与AI交互。

基础Prompt模板

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("prompt-server")

@mcp.prompt()
def explain_code(code: str, language: str = "python") -> str:
    """解释代码的功能和逻辑

    Args:
        code: 需要解释的代码
        language: 编程语言
    """
    return f"""请详细解释以下{language}代码的功能、逻辑和关键概念：

```{language}
{code}

请从以下几个方面进行解释：

  1. 代码的整体功能
  2. 关键函数/类的作用
  3. 重要的设计模式或算法
  4. 潜在的改进建议"""

@mcp.prompt() def debug_error(error_message: str, code_context: str = "") → list[dict]: """帮助调试错误

Args:
    error_message: 错误信息
    code_context: 相关代码上下文
"""
messages = [
    {
        "role": "system",
        "content": "你是一个经验丰富的软件工程师，擅长调试和解决问题。"
    },
    {
        "role": "user",
        "content": f"""请帮我分析和解决以下错误：

错误信息：

{error_message}

{"相关代码：" + chr(10) + "" + chr(10) + code_context + chr(10) + "" if code_context else ""}

请提供：

  1. 错误原因分析
  2. 可能的解决方案
  3. 预防此类错误的建议""" } ] return messages

@mcp.prompt() def generate_tests(function_signature: str, description: str = "") → str: """为函数生成测试用例

Args:
    function_signature: 函数签名
    description: 函数描述
"""
return f"""请为以下函数生成全面的测试用例：

函数签名：{function_signature} {f"函数描述：" if description else ""}

要求：

  1. 正常输入测试
  2. 边界条件测试
  3. 异常输入测试
  4. 使用pytest框架
  5. 包含测试docstring"""

### 多轮对话Prompt

```python
@mcp.prompt()
def code_review_session(code: str) -> list[dict]:
    """启动代码审查会话

    Args:
        code: 需要审查的代码
    """
    return [
        {
            "role": "system",
            "content": """你是一个专业的代码审查员。请按照以下标准审查代码：
- 代码质量和可读性
- 性能和效率
- 安全性
- 错误处理
- 测试覆盖

对每个发现的问题，请给出具体的改进建议和代码示例。"""
        },
        {
            "role": "user",
            "content": f"请审查以下代码：\n\n```python\n{code}\n```"
        }
    ]

Sampling采样机制

Sampling是MCP的一项高级功能，允许Server在处理请求时向LLM发起补全请求。这意味着Server可以利用AI的能力来增强自身的处理逻辑。

工作原理

Client (Host)              MCP Server
    │                         │
    │◀── createMessage ──────│  Server请求AI补全
    │     (sampling request)  │
    │                         │
    │── (Host调用LLM) ──▶    │  Host决定是否允许
    │                         │
    │──── sampling result ──▶│  返回AI生成结果
    │                         │

Python实现Sampling

from mcp.server.fastmcp import FastMCP, Context
import mcp.types as types

mcp = FastMCP("sampling-server")

@mcp.tool()
async def intelligent_analyze(data: str, analysis_type: str, ctx: Context) -> dict:
    """使用AI进行智能数据分析

    Args:
        data: 待分析的数据
        analysis_type: 分析类型 (sentiment/summary/extract)
    """
    # 构建sampling请求
    if analysis_type == "sentiment":
        prompt = f"分析以下文本的情感倾向（正面/负面/中性），并给出置信度：\n\n{data}"
    elif analysis_type == "summary":
        prompt = f"请用3句话总结以下内容：\n\n{data}"
    elif analysis_type == "extract":
        prompt = f"从以下文本中提取关键实体（人名、地点、组织、日期）：\n\n{data}"
    else:
        return {"error": f"不支持的分析类型: {analysis_type}"}

    # 通过sampling请求LLM
    result = await ctx.session.create_message(
        messages=[
            types.SamplingMessage(
                role="user",
                content=types.TextContent(type="text", text=prompt)
            )
        ],
        max_tokens=1000,
        temperature=0.3
    )

    # 处理结果
    if result.content.type == "text":
        return {
            "analysis_type": analysis_type,
            "result": result.content.text,
            "model": result.model
        }

    return {"error": "无法获取AI分析结果"}

实际应用：AI辅助数据处理

@mcp.tool()
async def smart_csv_processor(
    csv_content: str,
    task: str,
    ctx: Context
) -> dict:
    """AI辅助处理CSV数据

    Args:
        csv_content: CSV内容
        task: 处理任务描述
    """
    # 第一步：让AI理解数据结构
    schema_result = await ctx.session.create_message(
        messages=[
            types.SamplingMessage(
                role="user",
                content=types.TextContent(
                    type="text",
                    text=f"分析以下CSV数据的结构，列出列名和数据类型：\n\n{csv_content[:2000]}"
                )
            )
        ],
        max_tokens=500
    )

    # 第二步：根据任务生成处理代码
    code_result = await ctx.session.create_message(
        messages=[
            types.SamplingMessage(
                role="user",
                content=types.TextContent(
                    type="text",
                    text=f"""数据结构：{schema_result.content.text}

任务：{task}

请生成Python pandas代码来完成这个任务。只返回代码，不要解释。"""
                )
            )
        ],
        max_tokens=1000
    )

    return {
        "schema_analysis": schema_result.content.text,
        "generated_code": code_result.content.text,
        "status": "ready_to_execute"
    }

本地与远程Server部署

本地stdio模式

本地模式是最简单的部署方式，通过标准输入/输出与客户端通信。

# server_stdio.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("local-server")

@mcp.tool()
def hello(name: str) -> str:
    """打招呼

    Args:
        name: 名字
    """
    return f"你好，{name}！"

if __name__ == "__main__":
    # 默认就是stdio模式
    mcp.run()

    # 或者显式指定
    # mcp.run(transport="stdio")

在Claude Desktop中配置stdio Server：

{
  "mcpServers": {
    "my-local-server": {
      "command": "python",
      "args": ["/path/to/server_stdio.py"],
      "env": {
        "DATABASE_URL": "sqlite:///data.db"
      }
    }
  }
}

远程HTTP+SSE模式

对于需要远程访问的场景，使用HTTP + SSE传输。

# server_sse.py
from mcp.server.fastmcp import FastMCP
import os

mcp = FastMCP("remote-server")

@mcp.tool()
def get_system_info() -> dict:
    """获取系统信息"""
    import platform
    return {
        "system": platform.system(),
        "node": platform.node(),
        "release": platform.release(),
        "python_version": platform.python_version()
    }

if __name__ == "__main__":
    # 以SSE模式运行
    mcp.run(
        transport="sse",
        host="0.0.0.0",
        port=8080
    )

使用Starlette自定义HTTP Server

# server_http.py
from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route, Mount
import uvicorn

mcp = FastMCP("custom-http-server")

@mcp.tool()
def echo(text: str) -> str:
    """回显文本"""
    return text

# 创建Starlette应用
def create_app():
    sse = SseServerTransport("/messages/")

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

    return Starlette(
        routes=[
            Route("/sse", endpoint=handle_sse),
            Mount("/messages/", app=sse.handle_post_message),
        ]
    )

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

Docker部署

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 暴露端口（SSE模式）
EXPOSE 8080

# 启动Server
CMD ["python", "server_sse.py"]

# docker-compose.yml
version: '3.8'

services:
  mcp-server:
    build: .
    ports:
      - "8080:8080"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
      - API_KEY=${API_KEY}
    volumes:
      - ./data:/app/data
    restart: unless-stopped

  db:
    image: postgres:15
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

数据库集成

将数据库集成到MCP Server中，让AI能够直接查询和操作数据。

SQLite集成

import sqlite3
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("sqlite-server")

DB_PATH = Path("data/app.db")

def get_db():
    """获取数据库连接"""
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    return conn

@mcp.tool()
def query_database(sql: str) -> list[dict]:
    """执行SQL查询

    Args:
        sql: SQL查询语句（仅支持SELECT）
    """
    # 安全检查
    sql_clean = sql.strip()
    if not sql_clean.upper().startswith("SELECT"):
        return [{"error": "仅支持SELECT查询"}]

    try:
        conn = get_db()
        cursor = conn.execute(sql_clean)
        columns = [desc[0] for desc in cursor.description] if cursor.description else []
        rows = cursor.fetchall()
        result = [dict(zip(columns, row)) for row in rows]
        conn.close()
        return result
    except Exception as e:
        return [{"error": str(e)}]

@mcp.tool()
def list_tables() -> list[dict]:
    """列出数据库中的所有表"""
    conn = get_db()
    cursor = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
    )
    tables = []
    for row in cursor.fetchall():
        table_name = row[0]
        # 获取表结构
        schema_cursor = conn.execute(f"PRAGMA table_info({table_name})")
        columns = [
            {"name": r[1], "type": r[2], "notnull": bool(r[3]), "pk": bool(r[5])}
            for r in schema_cursor.fetchall()
        ]
        # 获取行数
        count_cursor = conn.execute(f"SELECT COUNT(*) FROM {table_name}")
        row_count = count_cursor.fetchone()[0]

        tables.append({
            "name": table_name,
            "columns": columns,
            "row_count": row_count
        })
    conn.close()
    return tables

@mcp.resource("db://schema")
def get_database_schema() -> str:
    """获取完整数据库结构"""
    tables = list_tables()
    schema_text = "数据库结构：\n\n"
    for table in tables:
        schema_text += f"表: {table['name']} ({table['row_count']} 行)\n"
        for col in table['columns']:
            pk_mark = " [PK]" if col['pk'] else ""
            null_mark = " NOT NULL" if col['notnull'] else ""
            schema_text += f"  - {col['name']}: {col['type']}{pk_mark}{null_mark}\n"
        schema_text += "\n"
    return schema_text

PostgreSQL集成（异步）

import asyncpg
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("postgres-server")

# 连接池
pool = None

async def init_pool():
    global pool
    pool = await asyncpg.create_pool(
        "postgresql://user:password@localhost:5432/mydb",
        min_size=2,
        max_size=10
    )

@mcp.tool()
async def pg_query(sql: str, params: list = None) -> list[dict]:
    """执行PostgreSQL查询

    Args:
        sql: SQL查询语句
        params: 查询参数列表
    """
    if not pool:
        await init_pool()

    # 安全检查
    sql_upper = sql.strip().upper()
    dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE", "CREATE"]
    for kw in dangerous:
        if kw in sql_upper:
            return [{"error": f"不允许执行包含 {kw} 的语句"}]

    try:
        async with pool.acquire() as conn:
            if params:
                rows = await conn.fetch(sql, *params)
            else:
                rows = await conn.fetch(sql)
            return [dict(row) for row in rows]
    except Exception as e:
        return [{"error": str(e)}]

@mcp.tool()
async def pg_table_info(table_name: str) -> dict:
    """获取PostgreSQL表的详细信息

    Args:
        table_name: 表名
    """
    if not pool:
        await init_pool()

    async with pool.acquire() as conn:
        # 获取列信息
        columns = await conn.fetch("""
            SELECT column_name, data_type, is_nullable, column_default
            FROM information_schema.columns
            WHERE table_name = $1
            ORDER BY ordinal_position
        """, table_name)

        # 获取行数
        count = await conn.fetchval(f'SELECT COUNT(*) FROM "{table_name}"')

        # 获取索引
        indexes = await conn.fetch("""
            SELECT indexname, indexdef
            FROM pg_indexes
            WHERE tablename = $1
        """, table_name)

        return {
            "table": table_name,
            "columns": [dict(c) for c in columns],
            "row_count": count,
            "indexes": [dict(i) for i in indexes]
        }

@mcp.prompt()
def sql_assistant(db_type: str = "postgresql") -> str:
    """SQL查询助手"""
    return f"""你是一个{db_type}数据库专家。用户会描述他们想要查询的数据，你需要帮助他们编写SQL语句。

规则：
1. 只生成SELECT查询
2. 使用参数化查询防止SQL注入
3. 优化查询性能
4. 解释查询逻辑"""

文件系统集成

文件系统MCP Server是最常用的Server之一，让AI能够安全地读写文件。

安全文件系统Server

import os
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("filesystem-server")

# 配置允许访问的目录
ALLOWED_DIRS = [
    Path.home() / "documents",
    Path.home() / "projects",
    Path("/tmp/workspace")
]

def is_path_allowed(path: Path) -> bool:
    """检查路径是否在允许范围内"""
    try:
        resolved = path.resolve()
        return any(
            resolved.is_relative_to(allowed.resolve())
            for allowed in ALLOWED_DIRS
        )
    except:
        return False

@mcp.tool()
def read_file(file_path: str) -> str:
    """读取文件内容

    Args:
        file_path: 文件路径
    """
    path = Path(file_path)
    if not is_path_allowed(path):
        return f"错误: 没有权限访问 {file_path}"

    if not path.exists():
        return f"错误: 文件不存在 {file_path}"

    if path.stat().st_size > 10 * 1024 * 1024:  # 10MB限制
        return "错误: 文件过大（超过10MB）"

    try:
        return path.read_text(encoding="utf-8")
    except UnicodeDecodeError:
        return "错误: 文件不是文本格式（可能是二进制文件）"

@mcp.tool()
def write_file(file_path: str, content: str) -> dict:
    """写入文件

    Args:
        file_path: 文件路径
        content: 文件内容
    """
    path = Path(file_path)
    if not is_path_allowed(path):
        return {"error": f"没有权限写入 {file_path}"}

    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding="utf-8")
        return {
            "status": "success",
            "path": str(path),
            "size": path.stat().st_size
        }
    except Exception as e:
        return {"error": f"写入失败: {e}"}

@mcp.tool()
def list_directory(dir_path: str = ".", pattern: str = "*") -> list[dict]:
    """列出目录内容

    Args:
        dir_path: 目录路径
        pattern: 文件名匹配模式
    """
    path = Path(dir_path)
    if not is_path_allowed(path):
        return [{"error": f"没有权限访问 {dir_path}"}]

    if not path.is_dir():
        return [{"error": f"{dir_path} 不是目录"}]

    items = []
    for item in sorted(path.glob(pattern)):
        stat = item.stat()
        items.append({
            "name": item.name,
            "type": "directory" if item.is_dir() else "file",
            "size": stat.st_size if item.is_file() else None,
            "modified": stat.st_mtime,
            "path": str(item)
        })

    return items

@mcp.tool()
def search_files(
    directory: str,
    query: str,
    file_extensions: list[str] = None
) -> list[dict]:
    """在文件中搜索内容

    Args:
        directory: 搜索目录
        query: 搜索关键词
        file_extensions: 限制文件扩展名
    """
    path = Path(directory)
    if not is_path_allowed(path):
        return [{"error": f"没有权限访问 {directory}"}]

    results = []
    extensions = set(file_extensions) if file_extensions else {".txt", ".md", ".py", ".js", ".ts"}

    for file_path in path.rglob("*"):
        if not file_path.is_file():
            continue
        if file_path.suffix not in extensions:
            continue
        if file_path.stat().st_size > 5 * 1024 * 1024:  # 跳过大文件
            continue

        try:
            content = file_path.read_text(encoding="utf-8")
            if query.lower() in content.lower():
                # 找到匹配的行
                lines = content.split("\n")
                matching_lines = [
                    {"line_num": i + 1, "content": line.strip()}
                    for i, line in enumerate(lines)
                    if query.lower() in line.lower()
                ][:5]  # 最多5行

                results.append({
                    "file": str(file_path),
                    "matches": len(matching_lines),
                    "lines": matching_lines
                })
        except:
            continue

    return results[:20]  # 最多20个文件

@mcp.resource("fs://workspace")
def get_workspace_info() -> dict:
    """获取工作空间信息"""
    info = {}
    for allowed_dir in ALLOWED_DIRS:
        if allowed_dir.exists():
            info[str(allowed_dir)] = {
                "exists": True,
                "file_count": sum(1 for _ in allowed_dir.rglob("*") if _.is_file()),
                "total_size": sum(f.stat().st_size for f in allowed_dir.rglob("*") if f.is_file())
            }
    return info

外部API集成

将外部API服务集成到MCP Server中。

多API集成Server

import httpx
from mcp.server.fastmcp import FastMCP
from datetime import datetime

mcp = FastMCP("api-integration-server")

# 通用HTTP客户端
async def api_request(
    url: str,
    method: str = "GET",
    headers: dict = None,
    params: dict = None,
    json_data: dict = None,
    timeout: float = 30.0
) -> dict:
    """通用API请求函数"""
    async with httpx.AsyncClient() as client:
        response = await client.request(
            method=method,
            url=url,
            headers=headers,
            params=params,
            json=json_data,
            timeout=timeout
        )
        return {
            "status_code": response.status_code,
            "data": response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text
        }

# 天气API
@mcp.tool()
async def get_weather(city: str) -> dict:
    """获取城市天气信息

    Args:
        city: 城市名称
    """
    result = await api_request(
        f"https://wttr.in/{city}",
        params={"format": "j1"},
        timeout=10.0
    )

    if result["status_code"] == 200:
        data = result["data"]
        current = data.get("current_condition", [{}])[0]
        return {
            "city": city,
            "temperature_c": current.get("temp_C"),
            "feels_like_c": current.get("FeelsLikeC"),
            "humidity": current.get("humidity"),
            "wind_kmph": current.get("windspeedKmph"),
            "description": current.get("weatherDesc", [{}])[0].get("value"),
            "updated": datetime.now().isoformat()
        }
    return {"error": f"获取天气失败: HTTP {result['status_code']}"}

# 翻译API
@mcp.tool()
async def translate_text(
    text: str,
    target_language: str = "zh",
    source_language: str = "auto"
) -> dict:
    """翻译文本

    Args:
        text: 待翻译的文本
        target_language: 目标语言代码
        source_language: 源语言代码
    """
    # 使用免费翻译API示例
    result = await api_request(
        "https://api.mymemory.translated.net/get",
        params={
            "q": text,
            "langpair": f"{source_language}|{target_language}"
        }
    )

    if result["status_code"] == 200:
        translated = result["data"].get("responseData", {}).get("translatedText", "")
        return {
            "original": text,
            "translated": translated,
            "source_language": source_language,
            "target_language": target_language
        }
    return {"error": "翻译失败"}

# JSONPlaceholder示例（用于学习）
@mcp.tool()
async def get_todos(user_id: int = 1, completed: bool = None) -> list[dict]:
    """获取用户待办事项

    Args:
        user_id: 用户ID
        completed: 过滤完成状态
    """
    result = await api_request(
        "https://jsonplaceholder.typicode.com/todos",
        params={"userId": user_id}
    )

    if result["status_code"] == 200:
        todos = result["data"]
        if completed is not None:
            todos = [t for t in todos if t.get("completed") == completed]
        return todos
    return [{"error": "获取待办事项失败"}]

@mcp.tool()
async def create_post(title: str, body: str, user_id: int = 1) -> dict:
    """创建文章

    Args:
        title: 文章标题
        body: 文章内容
        user_id: 作者用户ID
    """
    result = await api_request(
        "https://jsonplaceholder.typicode.com/posts",
        method="POST",
        json_data={
            "title": title,
            "body": body,
            "userId": user_id
        }
    )

    if result["status_code"] in (200, 201):
        return result["data"]
    return {"error": "创建文章失败"}

安全权限管理

安全是MCP Server开发中最重要的考量之一。

权限控制框架

import hashlib
import hmac
import time
from functools import wraps
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("secure-server")

# 权限定义
PERMISSIONS = {
    "read": "读取数据",
    "write": "写入数据",
    "admin": "管理操作",
    "execute": "执行代码"
}

# 用户权限映射（实际应用中应从数据库或配置文件读取）
USER_PERMISSIONS = {
    "user1": ["read"],
    "user2": ["read", "write"],
    "admin": ["read", "write", "admin", "execute"]
}

class SecurityManager:
    """安全管理器"""

    def __init__(self):
        self.rate_limits = {}  # {user_id: [(timestamp, ...)]}
        self.max_requests_per_minute = 60

    def check_permission(self, user: str, required_permission: str) -> bool:
        """检查用户权限"""
        user_perms = USER_PERMISSIONS.get(user, [])
        return required_permission in user_perms

    def check_rate_limit(self, user: str) -> bool:
        """检查频率限制"""
        now = time.time()
        if user not in self.rate_limits:
            self.rate_limits[user] = []

        # 清理过期记录
        self.rate_limits[user] = [
            t for t in self.rate_limits[user]
            if now - t < 60
        ]

        if len(self.rate_limits[user]) >= self.max_requests_per_minute:
            return False

        self.rate_limits[user].append(now)
        return True

    def sanitize_input(self, text: str) -> str:
        """清理输入，防止注入攻击"""
        # 移除潜在的危险字符
        dangerous_patterns = [
            "'; DROP", "'; DELETE", "'; UPDATE", "'; INSERT",
            "<script>", "javascript:", "onerror=",
            "../", "..\\"
        ]
        for pattern in dangerous_patterns:
            if pattern.lower() in text.lower():
                raise ValueError(f"检测到不安全的输入模式")
        return text

security = SecurityManager()

def require_permission(permission: str):
    """权限检查装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 从上下文获取用户信息
            # 注意：实际实现中需要从MCP上下文中获取认证信息
            user = kwargs.get("_user", "anonymous")

            if not security.check_permission(user, permission):
                return {"error": f"权限不足: 需要 {permission} 权限"}

            if not security.check_rate_limit(user):
                return {"error": "请求过于频繁，请稍后再试"}

            return await func(*args, **kwargs)
        return wrapper
    return decorator

@mcp.tool()
@require_permission("read")
def read_data(query: str, _user: str = "anonymous") -> dict:
    """读取数据（需要read权限）

    Args:
        query: 查询条件
    """
    sanitized = security.sanitize_input(query)
    return {"status": "success", "data": [], "query": sanitized}

@mcp.tool()
@require_permission("write")
def write_data(data: dict, _user: str = "anonymous") -> dict:
    """写入数据（需要write权限）

    Args:
        data: 要写入的数据
    """
    return {"status": "success", "message": "数据已写入"}

@mcp.tool()
@require_permission("admin")
def admin_action(action: str, _user: str = "anonymous") -> dict:
    """管理操作（需要admin权限）

    Args:
        action: 管理操作类型
    """
    return {"status": "success", "action": action}

输入验证与清理

from pydantic import BaseModel, Field, validator
import re

class SafeQueryInput(BaseModel):
    """安全的查询输入模型"""
    query: str = Field(..., max_length=1000, description="查询内容")
    limit: int = Field(default=10, ge=1, le=100, description="返回数量限制")

    @validator("query")
    def validate_query(cls, v):
        # 检查SQL注入
        sql_patterns = [
            r"(\b(SELECT|INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|EXEC)\b)",
            r"(--|;|\/\*|\*\/)",
            r"(\b(UNION|OR|AND)\b.*\b(SELECT|INSERT|UPDATE|DELETE)\b)"
        ]
        for pattern in sql_patterns:
            if re.search(pattern, v, re.IGNORECASE):
                raise ValueError("查询包含不安全的内容")
        return v

@mcp.tool()
def safe_search(input_data: SafeQueryInput) -> dict:
    """安全搜索

    Args:
        input_data: 验证后的输入数据
    """
    return {
        "query": input_data.query,
        "limit": input_data.limit,
        "results": []
    }

与Claude Desktop集成

Claude Desktop是MCP的主要客户端之一，配置简单直观。

配置文件位置

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

配置示例

{
  "mcpServers": {
    "my-tools": {
      "command": "python",
      "args": ["/absolute/path/to/my_server.py"],
      "env": {
        "DATABASE_URL": "sqlite:///data.db",
        "API_KEY": "your-api-key-here"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/username/documents",
        "/Users/username/projects"
      ]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxx"
      }
    }
  }
}

使用uv运行Python Server

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

测试集成

配置完成后，重启Claude Desktop，你可以在对话中看到可用的工具图标。测试方法：

  1. 在Claude Desktop中开启新的对话
  2. 输入需要使用工具的请求，如"帮我查询当前天气"
  3. Claude会自动识别并调用MCP Server提供的工具
  4. 查看工具调用结果

与LangChain集成

将MCP Server集成到LangChain生态中，构建更复杂的AI应用。

MCP适配LangChain

from langchain.tools import BaseTool
from langchain.callbacks.manager import CallbackManagerForToolRun
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from typing import Optional
import asyncio

class MCPToolAdapter(BaseTool):
    """将MCP Tool适配为LangChain Tool"""

    name: str
    description: str
    server_command: str
    server_args: list[str]
    tool_name: str
    _session: Optional[ClientSession] = None

    def _run(
        self,
        *args,
        run_manager: Optional[CallbackManagerForToolRun] = None,
        **kwargs
    ) -> str:
        """同步执行MCP工具"""
        return asyncio.run(self._arun(*args, **kwargs))

    async def _arun(self, *args, **kwargs) -> str:
        """异步执行MCP工具"""
        if not self._session:
            await self._connect()

        result = await self._session.call_tool(
            self.tool_name,
            arguments=kwargs
        )

        # 提取文本内容
        texts = [
            content.text for content in result.content
            if hasattr(content, "text")
        ]
        return "\n".join(texts)

    async def _connect(self):
        """连接到MCP Server"""
        server_params = StdioServerParameters(
            command=self.server_command,
            args=self.server_args
        )

        self._stdio_transport = stdio_client(server_params)
        read, write = await self._stdio_transport.__aenter__()
        self._session = ClientSession(read, write)
        await self._session.__aenter__()
        await self._session.initialize()


# 使用示例
def create_mcp_tools():
    """从MCP Server创建LangChain工具集"""
    weather_tool = MCPToolAdapter(
        name="get_weather",
        description="获取指定城市的天气信息",
        server_command="python",
        server_args=["/path/to/weather_server.py"],
        tool_name="get_weather"
    )

    return [weather_tool]

使用LangChain Agent调用MCP工具

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

async def run_mcp_agent():
    """运行使用MCP工具的LangChain Agent"""

    # 创建MCP工具
    tools = create_mcp_tools()

    # 创建LLM
    llm = ChatOpenAI(model="gpt-4o", temperature=0)

    # 创建Prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个有帮助的助手，可以使用工具来完成任务。"),
        MessagesPlaceholder(variable_name="chat_history", optional=True),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad")
    ])

    # 创建Agent
    agent = create_openai_tools_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    # 执行
    result = await agent_executor.ainvoke({
        "input": "北京今天的天气怎么样？"
    })

    print(result["output"])

实战：数据库查询MCP Server

下面我们构建一个完整的数据库查询MCP Server，集成SQLite数据库，支持表结构查看、数据查询、数据分析等功能。

"""
database_mcp_server.py - 完整的数据库查询MCP Server
"""
import sqlite3
import json
import os
from pathlib import Path
from datetime import datetime
from mcp.server.fastmcp import FastMCP, Context

# 创建MCP Server
mcp = FastMCP(
    "database-query-server",
    version="1.0.0"
)

# 数据库路径
DB_PATH = os.environ.get("DATABASE_PATH", "data/app.db")

def get_connection() -> sqlite3.Connection:
    """获取数据库连接"""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    return conn

def init_sample_data():
    """初始化示例数据"""
    conn = get_connection()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS employees (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            department TEXT NOT NULL,
            position TEXT NOT NULL,
            salary REAL NOT NULL,
            hire_date TEXT NOT NULL,
            email TEXT UNIQUE
        );

        CREATE TABLE IF NOT EXISTS departments (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT UNIQUE NOT NULL,
            manager TEXT,
            budget REAL,
            location TEXT
        );

        CREATE TABLE IF NOT EXISTS projects (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            department_id INTEGER,
            start_date TEXT,
            end_date TEXT,
            status TEXT DEFAULT 'active',
            budget REAL,
            FOREIGN KEY (department_id) REFERENCES departments(id)
        );
    """)

    # 插入示例数据（如果表为空）
    cursor = conn.execute("SELECT COUNT(*) FROM employees")
    if cursor.fetchone()[0] == 0:
        conn.executescript("""
            INSERT INTO departments (name, manager, budget, location) VALUES
            ('技术部', '张三', 500000, 'A栋3楼'),
            ('市场部', '李四', 300000, 'B栋2楼'),
            ('财务部', '王五', 200000, 'A栋5楼'),
            ('人事部', '赵六', 150000, 'B栋1楼');

            INSERT INTO employees (name, department, position, salary, hire_date, email) VALUES
            ('张三', '技术部', '技术总监', 35000, '2020-01-15', 'zhangsan@company.com'),
            ('李四', '市场部', '市场总监', 30000, '2020-03-20', 'lisi@company.com'),
            ('王五', '财务部', '财务总监', 32000, '2019-11-01', 'wangwu@company.com'),
            ('赵六', '人事部', '人事总监', 28000, '2021-02-10', 'zhaoliu@company.com'),
            ('钱七', '技术部', '高级工程师', 25000, '2021-06-15', 'qianqi@company.com'),
            ('孙八', '技术部', '工程师', 20000, '2022-01-10', 'sunba@company.com'),
            ('周九', '市场部', '市场经理', 22000, '2022-03-01', 'zhoujiu@company.com'),
            ('吴十', '技术部', '初级工程师', 15000, '2023-07-01', 'wushi@company.com');

            INSERT INTO projects (name, department_id, start_date, end_date, status, budget) VALUES
            ('AI平台开发', 1, '2024-01-01', '2024-12-31', 'active', 200000),
            ('品牌推广', 2, '2024-03-01', '2024-09-30', 'active', 100000),
            ('财务系统升级', 3, '2024-02-01', '2024-08-31', 'completed', 80000);
        """)

    conn.commit()
    conn.close()

# 工具：查询数据
@mcp.tool()
def query(sql: str) -> dict:
    """执行SQL查询（仅支持SELECT）

    Args:
        sql: SQL查询语句
    """
    sql_clean = sql.strip()

    # 安全检查
    if not sql_clean.upper().startswith("SELECT"):
        return {"error": "仅支持SELECT查询语句"}

    dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE"]
    sql_upper = sql_clean.upper()
    for kw in dangerous:
        if kw in sql_upper:
            return {"error": f"安全限制: 不允许执行包含 {kw} 的语句"}

    try:
        conn = get_connection()
        cursor = conn.execute(sql_clean)
        columns = [desc[0] for desc in cursor.description] if cursor.description else []
        rows = cursor.fetchall()
        result = [dict(zip(columns, row)) for row in rows]
        conn.close()

        return {
            "status": "success",
            "columns": columns,
            "row_count": len(result),
            "data": result
        }
    except Exception as e:
        return {"error": f"查询执行失败: {str(e)}"}

# 工具：列出所有表
@mcp.tool()
def list_tables() -> list[dict]:
    """列出数据库中的所有表及其结构"""
    conn = get_connection()
    cursor = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
    )

    tables = []
    for row in cursor.fetchall():
        table_name = row[0]
        schema = conn.execute(f"PRAGMA table_info({table_name})").fetchall()
        count = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]

        tables.append({
            "name": table_name,
            "columns": [
                {
                    "name": s[1],
                    "type": s[2],
                    "nullable": not s[3],
                    "default": s[4],
                    "primary_key": bool(s[5])
                }
                for s in schema
            ],
            "row_count": count
        })

    conn.close()
    return tables

# 工具：统计分析
@mcp.tool()
def analyze_table(table_name: str, column: str = None) -> dict:
    """对表进行统计分析

    Args:
        table_name: 表名
        column: 要分析的列名（可选）
    """
    # 安全检查：验证表名
    conn = get_connection()
    valid_tables = [r[0] for r in conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table'"
    ).fetchall()]

    if table_name not in valid_tables:
        conn.close()
        return {"error": f"表 {table_name} 不存在"}

    result = {"table": table_name}

    # 基本统计
    count = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
    result["total_rows"] = count

    if column:
        # 列统计
        stats = conn.execute(f"""
            SELECT
                COUNT({column}) as non_null_count,
                COUNT(DISTINCT {column}) as unique_count
            FROM {table_name}
        """).fetchone()

        result["column_stats"] = {
            "column": column,
            "non_null_count": stats[0],
            "unique_count": stats[1]
        }

        # 尝试数值统计
        try:
            numeric_stats = conn.execute(f"""
                SELECT
                    MIN({column}) as min_val,
                    MAX({column}) as max_val,
                    AVG({column}) as avg_val,
                    SUM({column}) as sum_val
                FROM {table_name}
                WHERE {column} IS NOT NULL
            """).fetchone()

            result["numeric_stats"] = {
                "min": numeric_stats[0],
                "max": numeric_stats[1],
                "average": round(numeric_stats[2], 2) if numeric_stats[2] else None,
                "sum": numeric_stats[3]
            }
        except:
            pass  # 非数值列，跳过数值统计

        # 值分布（前10）
        distribution = conn.execute(f"""
            SELECT {column}, COUNT(*) as count
            FROM {table_name}
            WHERE {column} IS NOT NULL
            GROUP BY {column}
            ORDER BY count DESC
            LIMIT 10
        """).fetchall()

        result["top_values"] = [
            {"value": r[0], "count": r[1]} for r in distribution
        ]

    conn.close()
    return result

# 工具：数据导出
@mcp.tool()
def export_table(table_name: str, format: str = "json", limit: int = 1000) -> str:
    """导出表数据

    Args:
        table_name: 表名
        format: 导出格式 (json/csv)
        limit: 最大导出行数
    """
    conn = get_connection()

    try:
        cursor = conn.execute(f"SELECT * FROM {table_name} LIMIT ?", (limit,))
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()

        if format == "json":
            data = [dict(zip(columns, row)) for row in rows]
            return json.dumps(data, ensure_ascii=False, indent=2)
        elif format == "csv":
            lines = [",".join(columns)]
            for row in rows:
                lines.append(",".join(str(v) for v in row))
            return "\n".join(lines)
        else:
            return f"不支持的格式: {format}"
    except Exception as e:
        return f"导出失败: {e}"
    finally:
        conn.close()

# 资源：数据库概览
@mcp.resource("db://overview")
def get_database_overview() -> dict:
    """获取数据库概览信息"""
    conn = get_connection()

    tables = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
    ).fetchall()

    overview = {
        "database_path": DB_PATH,
        "database_size": os.path.getsize(DB_PATH) if os.path.exists(DB_PATH) else 0,
        "table_count": len(tables),
        "tables": {}
    }

    for (table_name,) in tables:
        count = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
        overview["tables"][table_name] = {"row_count": count}

    conn.close()
    return overview

# Prompt：SQL助手
@mcp.prompt()
def sql_assistant() -> str:
    """SQL查询助手，帮助编写和优化SQL查询"""
    return """你是一个SQL数据库专家助手。你可以帮助用户：

1. 编写SQL查询语句
2. 解释查询结果
3. 优化查询性能
4. 设计数据库结构

数据库包含以下表：
- employees: 员工信息（id, name, department, position, salary, hire_date, email）
- departments: 部门信息（id, name, manager, budget, location）
- projects: 项目信息（id, name, department_id, start_date, end_date, status, budget）

请使用query工具执行SQL查询，并用中文解释结果。"""

# 初始化并启动
if __name__ == "__main__":
    # 确保数据目录存在
    Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)

    # 初始化示例数据
    init_sample_data()

    print(f"数据库MCP Server已启动")
    print(f"数据库路径: {DB_PATH}")

    mcp.run()

实战：文件操作MCP Server

"""
filesystem_mcp_server.py - 完整的文件操作MCP Server
"""
import os
import shutil
from pathlib import Path
from datetime import datetime
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("filesystem-server")

# 配置允许的目录
WORKSPACE = Path(os.environ.get("WORKSPACE_DIR", os.path.expanduser("~/workspace")))
ALLOWED_DIRS = [WORKSPACE, Path("/tmp")]

def validate_path(path_str: str) -> Path:
    """验证并规范化路径"""
    path = Path(path_str).expanduser().resolve()

    # 安全检查
    if not any(path.is_relative_to(d.resolve()) for d in ALLOWED_DIRS):
        raise PermissionError(f"没有权限访问: {path_str}")

    return path

@mcp.tool()
def read_file(path: str) -> dict:
    """读取文件内容

    Args:
        path: 文件路径
    """
    try:
        file_path = validate_path(path)

        if not file_path.exists():
            return {"error": f"文件不存在: {path}"}

        if not file_path.is_file():
            return {"error": f"不是文件: {path}"}

        size = file_path.stat().st_size
        if size > 10 * 1024 * 1024:  # 10MB限制
            return {"error": f"文件过大 ({size} bytes)，超过10MB限制"}

        content = file_path.read_text(encoding="utf-8")
        return {
            "status": "success",
            "path": str(file_path),
            "size": size,
            "content": content
        }
    except PermissionError as e:
        return {"error": str(e)}
    except UnicodeDecodeError:
        return {"error": "文件不是UTF-8文本格式"}
    except Exception as e:
        return {"error": f"读取失败: {e}"}

@mcp.tool()
def write_file(path: str, content: str, create_dirs: bool = True) -> dict:
    """写入文件

    Args:
        path: 文件路径
        content: 文件内容
        create_dirs: 是否自动创建父目录
    """
    try:
        file_path = validate_path(path)

        if create_dirs:
            file_path.parent.mkdir(parents=True, exist_ok=True)

        file_path.write_text(content, encoding="utf-8")
        return {
            "status": "success",
            "path": str(file_path),
            "size": len(content.encode("utf-8")),
            "created": datetime.now().isoformat()
        }
    except PermissionError as e:
        return {"error": str(e)}
    except Exception as e:
        return {"error": f"写入失败: {e}"}

@mcp.tool()
def list_directory(path: str = ".", show_hidden: bool = False) -> list[dict]:
    """列出目录内容

    Args:
        path: 目录路径
        show_hidden: 是否显示隐藏文件
    """
    try:
        dir_path = validate_path(path)

        if not dir_path.is_dir():
            return [{"error": f"不是目录: {path}"}]

        items = []
        for item in sorted(dir_path.iterdir()):
            if not show_hidden and item.name.startswith("."):
                continue

            stat = item.stat()
            items.append({
                "name": item.name,
                "type": "directory" if item.is_dir() else "file",
                "size": stat.st_size if item.is_file() else None,
                "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
                "permissions": oct(stat.st_mode)[-3:]
            })

        return items
    except PermissionError as e:
        return [{"error": str(e)}]
    except Exception as e:
        return [{"error": f"列出目录失败: {e}"}]

@mcp.tool()
def copy_file(source: str, destination: str) -> dict:
    """复制文件

    Args:
        source: 源文件路径
        destination: 目标路径
    """
    try:
        src_path = validate_path(source)
        dst_path = validate_path(destination)

        if not src_path.exists():
            return {"error": f"源文件不存在: {source}"}

        dst_path.parent.mkdir(parents=True, exist_ok=True)

        if src_path.is_dir():
            shutil.copytree(str(src_path), str(dst_path))
        else:
            shutil.copy2(str(src_path), str(dst_path))

        return {
            "status": "success",
            "source": str(src_path),
            "destination": str(dst_path)
        }
    except PermissionError as e:
        return {"error": str(e)}
    except Exception as e:
        return {"error": f"复制失败: {e}"}

@mcp.tool()
def delete_file(path: str, confirm: bool = False) -> dict:
    """删除文件或目录（需要确认）

    Args:
        path: 文件路径
        confirm: 确认删除（必须为true才会执行）
    """
    if not confirm:
        return {
            "warning": "请设置confirm=true确认删除操作",
            "path": path
        }

    try:
        target_path = validate_path(path)

        if not target_path.exists():
            return {"error": f"路径不存在: {path}"}

        if target_path.is_dir():
            shutil.rmtree(str(target_path))
        else:
            target_path.unlink()

        return {
            "status": "deleted",
            "path": str(target_path)
        }
    except PermissionError as e:
        return {"error": str(e)}
    except Exception as e:
        return {"error": f"删除失败: {e}"}

@mcp.tool()
def search_files(
    directory: str,
    pattern: str = "*",
    content_search: str = None,
    max_results: int = 50
) -> list[dict]:
    """搜索文件

    Args:
        directory: 搜索目录
        pattern: 文件名匹配模式
        content_search: 文件内容搜索关键词
        max_results: 最大结果数
    """
    try:
        dir_path = validate_path(directory)
        results = []

        for file_path in dir_path.rglob(pattern):
            if len(results) >= max_results:
                break

            if not file_path.is_file():
                continue

            item = {
                "name": file_path.name,
                "path": str(file_path),
                "size": file_path.stat().st_size,
                "modified": datetime.fromtimestamp(file_path.stat().st_mtime).isoformat()
            }

            if content_search:
                try:
                    content = file_path.read_text(encoding="utf-8")
                    if content_search.lower() in content.lower():
                        lines = content.split("\n")
                        matching = [
                            {"line": i + 1, "text": l.strip()}
                            for i, l in enumerate(lines)
                            if content_search.lower() in l.lower()
                        ][:3]
                        item["matches"] = matching
                    else:
                        continue  # 内容不匹配，跳过
                except:
                    continue  # 无法读取，跳过

            results.append(item)

        return results
    except PermissionError as e:
        return [{"error": str(e)}]
    except Exception as e:
        return [{"error": f"搜索失败: {e}"}]

@mcp.tool()
def get_file_info(path: str) -> dict:
    """获取文件详细信息

    Args:
        path: 文件路径
    """
    try:
        file_path = validate_path(path)

        if not file_path.exists():
            return {"error": f"路径不存在: {path}"}

        stat = file_path.stat()
        return {
            "name": file_path.name,
            "path": str(file_path),
            "type": "directory" if file_path.is_dir() else "file",
            "size": stat.st_size,
            "created": datetime.fromtimestamp(stat.st_ctime).isoformat(),
            "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
            "accessed": datetime.fromtimestamp(stat.st_atime).isoformat(),
            "permissions": oct(stat.st_mode),
            "is_symlink": file_path.is_symlink()
        }
    except PermissionError as e:
        return {"error": str(e)}
    except Exception as e:
        return {"error": f"获取信息失败: {e}"}

# 资源：工作空间概览
@mcp.resource("workspace://info")
def get_workspace_info() -> dict:
    """获取工作空间信息"""
    if not WORKSPACE.exists():
        return {"error": "工作空间不存在"}

    total_files = 0
    total_size = 0
    file_types = {}

    for f in WORKSPACE.rglob("*"):
        if f.is_file():
            total_files += 1
            total_size += f.stat().st_size
            ext = f.suffix or "无扩展名"
            file_types[ext] = file_types.get(ext, 0) + 1

    return {
        "workspace": str(WORKSPACE),
        "total_files": total_files,
        "total_size_mb": round(total_size / (1024 * 1024), 2),
        "file_types": dict(sorted(file_types.items(), key=lambda x: -x[1])[:10])
    }

# Prompt：文件操作助手
@mcp.prompt()
def file_assistant() -> str:
    """文件操作助手"""
    return f"""你是文件系统操作助手。工作空间位于: {WORKSPACE}

你可以帮助用户：
1. 读取和查看文件内容
2. 创建和写入文件
3. 搜索文件和内容
4. 复制、移动、删除文件
5. 获取文件信息

安全限制：
- 只能访问工作空间目录和/tmp目录
- 删除操作需要用户确认
- 单个文件读取限制10MB

请安全、谨慎地执行文件操作。"""

if __name__ == "__main__":
    WORKSPACE.mkdir(parents=True, exist_ok=True)
    print(f"文件系统MCP Server已启动")
    print(f"工作空间: {WORKSPACE}")
    mcp.run()

最佳实践

1. 工具设计原则

# ✓ 好的工具设计
@mcp.tool()
def get_user(user_id: str) -> dict:
    """根据ID获取用户信息

    Args:
        user_id: 用户唯一标识符（UUID格式）
    """
    # 清晰的参数描述
    # 明确的返回格式
    # 完整的错误处理
    pass

# ✗ 不好的工具设计
@mcp.tool()
def do_stuff(x, y):
    """处理数据"""
    # 参数含义不明确
    # 没有类型标注
    # 描述过于模糊
    pass

2. 安全检查清单

在开发MCP Server时，确保遵循以下安全最佳实践：

  • 输入验证：所有输入参数都经过验证和清理
  • 路径遍历防护：防止 .. 等路径遍历攻击
  • SQL注入防护：使用参数化查询，禁止危险SQL关键词
  • 频率限制：实施请求频率限制
  • 权限检查：验证操作权限
  • 大小限制：限制文件大小、返回数据量
  • 错误信息：错误消息不应暴露内部实现细节
  • 日志记录：记录关键操作的审计日志

3. 性能优化

# 使用连接池
import asyncpg

pool = None

async def get_pool():
    global pool
    if pool is None:
        pool = await asyncpg.create_pool(
            "postgresql://...",
            min_size=2,
            max_size=10
        )
    return pool

# 缓存频繁访问的资源
from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_schema(table_name: str) -> dict:
    """缓存表结构信息"""
    # ... 查询数据库
    pass

# 批量操作
@mcp.tool()
async def batch_insert(records: list[dict]) -> dict:
    """批量插入记录"""
    pool = await get_pool()
    async with pool.acquire() as conn:
        await conn.executemany(
            "INSERT INTO table (col1, col2) VALUES ($1, $2)",
            [(r["col1"], r["col2"]) for r in records]
        )
    return {"inserted": len(records)}

4. 错误处理模式

from enum import Enum

class ErrorCode(Enum):
    INVALID_INPUT = "INVALID_INPUT"
    NOT_FOUND = "NOT_FOUND"
    PERMISSION_DENIED = "PERMISSION_DENIED"
    INTERNAL_ERROR = "INTERNAL_ERROR"
    RATE_LIMITED = "RATE_LIMITED"

def create_error(code: ErrorCode, message: str, details: dict = None) -> dict:
    """创建标准化的错误响应"""
    return {
        "error": {
            "code": code.value,
            "message": message,
            "details": details or {}
        }
    }

@mcp.tool()
def robust_tool(input_data: str) -> dict:
    """健壮的工具实现示例"""
    # 输入验证
    if not input_data:
        return create_error(ErrorCode.INVALID_INPUT, "输入数据不能为空")

    try:
        # 业务逻辑
        result = process(input_data)
        return {"status": "success", "data": result}
    except PermissionError:
        return create_error(ErrorCode.PERMISSION_DENIED, "没有权限执行此操作")
    except FileNotFoundError:
        return create_error(ErrorCode.NOT_FOUND, "资源不存在")
    except Exception as e:
        # 记录日志但不暴露内部细节
        logger.error(f"内部错误: {e}")
        return create_error(ErrorCode.INTERNAL_ERROR, "服务器内部错误")

5. 文档与测试

@mcp.tool()
def welldocumented_tool(
    query: str,
    max_results: int = 10,
    include_metadata: bool = False
) -> dict:
    """搜索数据库中的记录

    这个工具支持全文搜索，返回匹配的记录列表。
    支持模糊匹配和精确匹配两种模式。

    Args:
        query: 搜索关键词，支持中英文。示例: "张三"、"project-alpha"
        max_results: 最大返回结果数，范围1-100，默认10
        include_metadata: 是否包含元数据（创建时间、修改时间等），默认False

    Returns:
        包含以下字段的字典:
        - results: 匹配记录列表
        - total: 总匹配数
        - query: 原始查询
        - execution_time_ms: 执行时间（毫秒）

    Raises:
        当查询为空或max_results超出范围时返回错误。
    """
    pass

总结

MCP协议为AI应用与外部工具的集成提供了一个优雅、标准化的解决方案。通过本教程的学习，你应该掌握了以下核心技能：

基础技能：

  1. MCP协议的架构和核心概念（Resources、Tools、Prompts）
  2. Python和TypeScript两种SDK的Server开发方法
  3. 三种核心原语的设计和实现

进阶技能： 4. Sampling采样机制，让Server能够调用LLM增强自身能力 5. 本地stdio和远程HTTP+SSE两种部署方式 6. 数据库、文件系统、外部API的集成方法

实战技能： 7. 完整的数据库查询MCP Server开发 8. 完整的文件操作MCP Server开发 9. 安全权限管理和输入验证 10. 与Claude Desktop和LangChain的集成

最佳实践： 11. 工具设计原则和文档规范 12. 性能优化和错误处理模式 13. 安全检查清单

MCP生态正在快速发展，越来越多的工具和平台开始支持MCP协议。建议开发者：

  • 从简单的工具开始，逐步构建复杂的Server
  • 充分利用社区已有的MCP Server（如官方的filesystem、github等）
  • 关注MCP协议的更新，及时适配新特性
  • 参与MCP社区，分享自己的Server实现

随着AI应用的普及，MCP将成为连接AI与现实世界的重要桥梁。掌握MCP开发技能，将为你的AI开发生涯打开新的大门。

本教程持续更新中，欢迎反馈和建议。

内容声明

本文内容为AI技术学习教程，仅供学习参考。如涉及技术问题，欢迎通过 xurj005@163.com 与我们交流。

相关教程

Python入门完全教程

29,193 次阅读

Java基础编程教程

116 次阅读

Go语言实战教程

110 次阅读

Python编程入门完全教程

93 次阅读
目录
快速导航
返回首页 全部教程 AI资讯