MCP (Model Context Protocol) 全面指南：从协议原理到生产级 Agent 集成

┌─────────────────────────────────────────┐
│              Host Application            │
│    (Claude Desktop / Cursor / IDE)       │
│                                          │
│  ┌──────────────────────────────────┐   │
│  │         MCP Client                │   │
│  │  (协议实现 + 能力发现 + 安全层)    │   │
│  └──────────────────────────────────┘   │
│              ▲    ▲    ▲                 │
│              │    │    │                 │
│  ┌───────────┼────┼────┼───────────┐    │
│  │  MCP Server │ MCP Server │ MCP Server │   │
│  │  (GitHub)   │  (File)    │  (Database)  │   │
│  └───────────┴────┴────┴───────────┘    │
└─────────────────────────────────────────┘

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "search_github",
    "arguments": {"query": "MCP protocol", "limit": 5}
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {"type": "text", "text": "Found 5 repositories..."}
    ]
  }
}

pip install mcp

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, ImageContent
import asyncio
import json
import os
from pathlib import Path

# 创建 MCP Server 实例
app = Server("ai-master-github-tools")

# === 能力发现：告诉 Client 我们提供了什么 ===
@app.list_tools()
async def list_tools():
    """返回所有可用工具列表"""
    return [
        Tool(
            name="search_github",
            description="搜索 GitHub 仓库、用户或代码",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    },
                    "type": {
                        "type": "string",
                        "enum": ["repositories", "users", "code"],
                        "description": "搜索类型",
                        "default": "repositories"
                    },
                    "sort": {
                        "type": "string",
                        "enum": ["stars", "forks", "updated"],
                        "description": "排序方式",
                        "default": "stars"
                    },
                    "per_page": {
                        "type": "integer",
                        "description": "每页结果数",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="read_file",
            description="读取本地文件内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "文件路径"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="list_directory",
            description="列出目录内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "目录路径"
                    }
                },
                "required": ["path"]
            }
        )
    ]

# === 工具调用：实际执行逻辑 ===
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    """处理工具调用请求"""
    
    if name == "search_github":
        import urllib.request
        
        query = arguments.get("query", "")
        search_type = arguments.get("type", "repositories")
        sort = arguments.get("sort", "stars")
        per_page = arguments.get("per_page", 5)
        
        # 构建 GitHub API 请求
        url = f"https://api.github.com/search/{search_type}?q={query}&sort={sort}&per_page={per_page}"
        
        req = urllib.request.Request(url)
        req.add_header("Accept", "application/vnd.github.v3+json")
        
        # 如果有 token 则添加
        token = os.environ.get("GITHUB_TOKEN")
        if token:
            req.add_header("Authorization", f"Bearer {token}")
        
        try:
            with urllib.request.urlopen(req, timeout=10) as response:
                data = json.loads(response.read().decode())
                
            # 格式化结果
            items = data.get("items", [])
            total = data.get("total_count", 0)
            
            result_lines = [f"📊 找到 {total} 个结果（显示前 {len(items)} 个）\n"]
            
            for item in items:
                if search_type == "repositories":
                    result_lines.append(
                        f"⭐ {item['full_name']}\n"
                        f"   {item.get('description', 'N/A')}\n"
                        f"   Stars: {item['stargazers_count']:,} | "
                        f"Forks: {item['forks_count']:,} | "
                        f"Language: {item.get('language', 'N/A')}\n"
                        f"   🔗 {item['html_url']}\n"
                    )
                elif search_type == "users":
                    result_lines.append(
                        f"👤 {item['login']}\n"
                        f"   {item.get('bio', 'N/A')}\n"
                        f"   Repos: {item['public_repos']} | "
                        f"Followers: {item['followers']}\n"
                        f"   🔗 {item['html_url']}\n"
                    )
            
            return [TextContent(type="text", text="\n".join(result_lines))]
            
        except Exception as e:
            return [TextContent(type="text", text=f"❌ 搜索失败: {str(e)}")]
    
    elif name == "read_file":
        path = arguments.get("path", "")
        try:
            # 安全检查：限制在特定目录
            base_dir = Path("/workspace")
            full_path = (base_dir / path).resolve()
            if not str(full_path).startswith(str(base_dir)):
                return [TextContent(type="text", text="❌ 路径超出允许范围")]
            
            content = full_path.read_text(encoding="utf-8")
            return [TextContent(type="text", text=f"📄 {path}:\n\n{content[:5000]}")]
        except Exception as e:
            return [TextContent(type="text", text=f"❌ 读取失败: {str(e)}")]
    
    elif name == "list_directory":
        path = arguments.get("path", ".")
        try:
            base_dir = Path("/workspace")
            full_path = (base_dir / path).resolve()
            if not str(full_path).startswith(str(base_dir)):
                return [TextContent(type="text", text="❌ 路径超出允许范围")]
            
            entries = list(full_path.iterdir())
            lines = [f"📁 {path}/\n"]
            
            for entry in sorted(entries):
                icon = "📂" if entry.is_dir() else "📄"
                size = "" if entry.is_dir() else f" ({entry.stat().st_size:,} bytes)"
                lines.append(f"  {icon} {entry.name}{size}")
            
            return [TextContent(type="text", text="\n".join(lines))]
        except Exception as e:
            return [TextContent(type="text", text=f"❌ 列出失败: {str(e)}")]
    
    else:
        return [TextContent(type="text", text=f"❌ 未知工具: {name}")]

# === 资源暴露：让 Agent 可以读取数据 ===
@app.list_resources()
async def list_resources():
    """返回所有可用资源列表"""
    return []

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

{
  "mcpServers": {
    "ai-master-tools": {
      "command": "python",
      "args": ["/path/to/ai-master-mcp-server.py"],
      "env": {
        "GITHUB_TOKEN": "your_token_here"
      }
    }
  }
}

┌─────────────────────────────────────────────┐
│              用户 / Agent                    │
│              (不可信环境)                     │
├─────────────────────────────────────────────┤
│           MCP Client (安全层)                │
│  • 权限控制：决定 Agent 可以调用哪些工具       │
│  • 输入验证：过滤和验证所有请求参数            │
│  • 速率限制：防止滥用和 DoS                   │
│  • 审计日志：记录所有工具调用                  │
├─────────────────────────────────────────────┤
│           MCP Server (执行层)                │
│  • 业务逻辑执行                               │
│  • 数据访问控制                               │
│  • 敏感信息过滤                               │
├─────────────────────────────────────────────┤
│           外部系统 (数据库/API)              │
│  • 最小权限访问                               │
│  • 数据脱敏                                   │
└─────────────────────────────────────────────┘

# ❌ 危险：允许 Agent 访问整个文件系统
@app.list_resources()
async def list_resources():
    return [Resource(uri="file:///*", name="All Files")]

# ✅ 安全：限制在特定目录
@app.list_resources()
async def list_resources():
    return [
        Resource(
            uri="file:///workspace/docs",
            name="Project Documentation"
        ),
        Resource(
            uri="file:///workspace/src",
            name="Source Code"
        )
    ]

import re

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "execute_command":
        cmd = arguments.get("command", "")
        # 禁止危险命令
        dangerous_patterns = ['rm -rf', 'sudo', 'dd ', '> /dev/', '| bash']
        for pattern in dangerous_patterns:
            if pattern in cmd:
                raise ValueError(f"命令包含禁止模式: {pattern}")
        # 限制命令长度
        if len(cmd) > 1000:
            raise ValueError("命令过长")
        # 只允许白名单命令
        allowed_prefixes = ['ls ', 'cat ', 'grep ', 'head ', 'wc ']
        if not any(cmd.startswith(p) for p in allowed_prefixes):
            raise ValueError("命令不在白名单中")

import re

def sanitize_output(text: str) -> str:
    """从输出中移除敏感信息"""
    # 移除 API Key 模式
    text = re.sub(r'sk-[a-zA-Z0-9]{48}', '[REDACTED_API_KEY]', text)
    # 移除密码
    text = re.sub(r'"password":\s*"[^"]*"', '"password": "[REDACTED]"', text)
    # 移除 IP 地址（可选）
    text = re.sub(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', '[REDACTED_IP]', text)
    return text

import logging
import json
from datetime import datetime

audit_logger = logging.getLogger("mcp-audit")

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    # 记录审计日志
    audit_logger.info(json.dumps({
        "timestamp": datetime.utcnow().isoformat(),
        "tool": name,
        "arguments": {k: v for k, v in arguments.items() 
                      if k not in ["password", "token", "api_key"]},
        "status": "success"
    }))
    
    # 正常处理...

用户 ──→ [个人 AI 助手] ──→ [MCP Client] ──→ [MCP Server] ──→ [Headless Service]
                                              │
                                              ├──→ GitHub API
                                              ├──→ 数据库
                                              ├──→ 文件系统
                                              └──→ 任何 MCP 服务

from mcp.server import Server
from mcp.server.fastmcp import FastMCP
import httpx
import os

# 使用 FastMCP 简化开发
mcp = FastMCP("headless-service")

@mcp.tool()
async def query_database(sql: str, params: dict = None) -> str:
    """查询数据库（只读）"""
    # 安全：只允许 SELECT
    if not sql.strip().upper().startswith("SELECT"):
        return "❌ 只允许 SELECT 查询"
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://database-service:8080/query",
            json={"sql": sql, "params": params},
            headers={"Authorization": f"Bearer {os.environ['DB_TOKEN']}"}
        )
        return response.json()["result"]

@mcp.tool()
async def create_record(table: str, data: dict) -> str:
    """创建数据记录"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"http://api-service:8080/{table}",
            json=data,
            headers={"Authorization": f"Bearer {os.environ['API_TOKEN']}"}
        )
        return f"✅ 创建成功: {response.json()['id']}"

@mcp.resource("config://app")
async def get_app_config() -> str:
    """获取应用配置信息"""
    return """
    {
      "app_name": "AI Master",
      "version": "2.0.0",
      "features": ["knowledge", "tools", "blog", "news"],
      "api_endpoint": "https://ai-master.cc/api"
    }
    """

if __name__ == "__main__":
    mcp.run()