架构师AI杜

Appearance

第7天：MCP协议发布（2024年底） ​

学习目标 ​

  • 理解MCP（Model Context Protocol）的诞生背景
  • 掌握MCP协议的核心思想和架构
  • 了解MCP Server和Client的实现
  • 理解MCP与Claude Desktop的集成
  • 掌握MCP的应用场景和最佳实践

课程内容 ​

1. MCP的诞生背景 ​

1.1 AI Agent的集成挑战 ​

1.1.1 集成复杂度高

问题：

  • 每个工具都需要单独集成
  • API接口不统一
  • 数据格式不一致
  • 错误处理机制各异

示例：

python
# 集成文件系统
def read_file(path):
    # 实现文件读取
    pass

# 集成数据库
def query_database(sql):
    # 实现数据库查询
    pass

# 集成API
def call_api(url, params):
    # 实现API调用
    pass

# 每个工具都需要单独实现

1.1.2 可扩展性差

问题：

  • 添加新工具需要修改核心代码
  • 工具之间难以协作
  • 难以动态加载工具
  • 维护成本高

1.1.3 标准缺失

问题：

  • 没有统一的协议
  • 各家实现不兼容
  • 难以共享工具
  • 生态碎片化

1.2 MCP的诞生 ​

2024年底：Anthropic发布MCP（Model Context Protocol）

核心思想：

  • 标准化AI Agent与外部系统的交互
  • 提供统一的接口规范
  • 降低集成复杂度
  • 促进工具生态发展

目标：

  • 简化工具集成
  • 提高互操作性
  • 促进生态繁荣
  • 降低开发成本

2. MCP协议详解 ​

2.1 MCP的核心概念 ​

2.1.1 定义

MCP（Model Context Protocol）：一个开放协议，用于标准化AI模型与外部系统（如工具、数据库、API等）之间的交互。

2.1.2 核心组件

MCP Server：

  • 提供工具和资源的端点
  • 实现MCP协议
  • 暴露能力给AI模型

MCP Client：

  • 连接到MCP Server
  • 调用Server提供的工具
  • 访问Server提供的资源

MCP Host：

  • 托管MCP Client的环境
  • 如Claude Desktop、VS Code等

2.1.3 核心能力

工具（Tools）：

  • 可执行的操作
  • 带参数和返回值
  • 类似函数调用

资源（Resources）：

  • 可访问的数据
  • 支持读取和更新
  • 类似文件系统

提示（Prompts）：

  • 预定义的提示模板
  • 可参数化
  • 简化常用任务

2.2 MCP协议架构 ​

2.2.1 整体架构

┌─────────────────────────────────────────────────┐
│                 MCP Host                    │
│         (Claude Desktop, VS Code)          │
└────────────────┬────────────────────────────┘
                 │
                 │ MCP Protocol
                 │
┌────────────────▼────────────────────────────┐
│              MCP Client                   │
│         (连接到MCP Server)               │
└────────────────┬────────────────────────────┘
                 │
                 │ MCP Protocol
                 │
┌────────────────▼────────────────────────────┐
│             MCP Server                    │
│         (提供工具和资源)                 │
│  ┌──────────────────────────────────┐    │
│  │  Tools (工具)                 │    │
│  │  - file_read                  │    │
│  │  - file_write                 │    │
│  │  - database_query            │    │
│  └──────────────────────────────────┘    │
│  ┌──────────────────────────────────┐    │
│  │  Resources (资源)             │    │
│  │  - files/                    │    │
│  │  - database/                 │    │
│  │  - api/                      │    │
│  └──────────────────────────────────┘    │
│  ┌──────────────────────────────────┐    │
│  │  Prompts (提示)               │    │
│  │  - summarize_file            │    │
│  │  - analyze_data             │    │
│  └──────────────────────────────────┘    │
└──────────────────────────────────────────┘

2.2.2 通信流程

1. 初始化：

Client → Server: initialize
Server → Client: server_info

2. 工具调用：

Client → Server: tools/list
Server → Client: tools_list

Client → Server: tools/call
Server → Client: tool_result

3. 资源访问：

Client → Server: resources/list
Server → Client: resources_list

Client → Server: resources/read
Server → Client: resource_content

Client → Server: resources/write
Server → Client: write_result

4. 提示使用：

Client → Server: prompts/list
Server → Client: prompts_list

Client → Server: prompts/get
Server → Client: prompt_content

2.3 MCP协议规范 ​

2.3.1 JSON-RPC 2.0

MCP基于JSON-RPC 2.0协议。

请求格式：

json
{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "file_read",
    "arguments": {
      "path": "/path/to/file.txt"
    }
  },
  "id": 1
}

响应格式：

json
{
  "jsonrpc": "2.0",
  "result": {
    "content": "file content here"
  },
  "id": 1
}

错误格式：

json
{
  "jsonrpc": "2.0",
  "error": {
    "code": -32602,
    "message": "Invalid params"
  },
  "id": 1
}

2.3.2 工具定义

工具列表：

json
{
  "tools": [
    {
      "name": "file_read",
      "description": "Read a file from the filesystem",
      "inputSchema": {
        "type": "object",
        "properties": {
          "path": {
            "type": "string",
            "description": "Path to the file"
          }
        },
        "required": ["path"]
      }
    },
    {
      "name": "file_write",
      "description": "Write content to a file",
      "inputSchema": {
        "type": "object",
        "properties": {
          "path": {
            "type": "string",
            "description": "Path to the file"
          },
          "content": {
            "type": "string",
            "description": "Content to write"
          }
        },
        "required": ["path", "content"]
      }
    }
  ]
}

工具调用：

json
{
  "name": "file_read",
  "arguments": {
    "path": "/path/to/file.txt"
  }
}

工具结果：

json
{
  "content": [
    {
      "type": "text",
      "text": "file content here"
    }
  ]
}

2.3.3 资源定义

资源列表：

json
{
  "resources": [
    {
      "uri": "file:///path/to/file.txt",
      "name": "file.txt",
      "description": "A text file",
      "mimeType": "text/plain"
    },
    {
      "uri": "database://users",
      "name": "users",
      "description": "User database",
      "mimeType": "application/json"
    }
  ]
}

资源读取：

json
{
  "uri": "file:///path/to/file.txt"
}

资源内容：

json
{
  "contents": [
    {
      "uri": "file:///path/to/file.txt",
      "mimeType": "text/plain",
      "text": "file content here"
    }
  ]
}

2.3.4 提示定义

提示列表：

json
{
  "prompts": [
    {
      "name": "summarize_file",
      "description": "Summarize a file",
      "arguments": [
        {
          "name": "path",
          "description": "Path to the file",
          "required": true
        }
      ]
    },
    {
      "name": "analyze_data",
      "description": "Analyze data",
      "arguments": [
        {
          "name": "data",
          "description": "Data to analyze",
          "required": true
        }
      ]
    }
  ]
}

提示获取：

json
{
  "name": "summarize_file",
  "arguments": {
    "path": "/path/to/file.txt"
  }
}

提示内容：

json
{
  "messages": [
    {
      "role": "user",
      "content": {
        "type": "resource",
        "resource": {
          "uri": "file:///path/to/file.txt"
        }
      }
    },
    {
      "role": "user",
      "content": {
        "type": "text",
        "text": "Please summarize this file."
      }
    }
  ]
}

3. MCP Server开发 ​

3.1 Server基础 ​

3.1.1 Server初始化

python
from mcp.server import Server
from mcp.types import Tool, Resource

# 创建Server
server = Server(
    name="my-mcp-server",
    version="1.0.0"
)

# 定义工具
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="file_read",
            description="Read a file from the filesystem",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path to the file"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="file_write",
            description="Write content to a file",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path to the file"
                    },
                    "content": {
                        "type": "string",
                        "description": "Content to write"
                    }
                },
                "required": ["path", "content"]
            }
        )
    ]

# 定义资源
@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="file:///path/to/file.txt",
            name="file.txt",
            description="A text file",
            mimeType="text/plain"
        )
    ]

# 启动Server
if __name__ == "__main__":
    import asyncio
    asyncio.run(server.run())

3.1.2 工具实现

python
import os

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    if name == "file_read":
        path = arguments.get("path")
        try:
            with open(path, 'r') as f:
                content = f.read()
            return content
        except Exception as e:
            return f"Error reading file: {str(e)}"
    
    elif name == "file_write":
        path = arguments.get("path")
        content = arguments.get("content")
        try:
            os.makedirs(os.path.dirname(path), exist_ok=True)
            with open(path, 'w') as f:
                f.write(content)
            return f"Successfully wrote to {path}"
        except Exception as e:
            return f"Error writing file: {str(e)}"
    
    else:
        return f"Unknown tool: {name}"

3.1.3 资源实现

python
@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri.startswith("file://"):
        path = uri[7:]  # Remove "file://" prefix
        try:
            with open(path, 'r') as f:
                content = f.read()
            return content
        except Exception as e:
            return f"Error reading resource: {str(e)}"
    else:
        return f"Unsupported URI scheme: {uri}"

@server.write_resource()
async def write_resource(uri: str, content: str) -> str:
    if uri.startswith("file://"):
        path = uri[7:]  # Remove "file://" prefix
        try:
            os.makedirs(os.path.dirname(path), exist_ok=True)
            with open(path, 'w') as f:
                f.write(content)
            return f"Successfully wrote to {path}"
        except Exception as e:
            return f"Error writing resource: {str(e)}"
    else:
        return f"Unsupported URI scheme: {uri}"

3.2 高级功能 ​

3.2.1 数据库集成

python
import sqlite3

# 数据库工具
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    if name == "database_query":
        sql = arguments.get("sql")
        try:
            conn = sqlite3.connect('database.db')
            cursor = conn.cursor()
            cursor.execute(sql)
            results = cursor.fetchall()
            conn.close()
            return str(results)
        except Exception as e:
            return f"Error executing query: {str(e)}"
    
    # ... 其他工具

# 数据库资源
@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri.startswith("database://"):
        table = uri[11:]  # Remove "database://" prefix
        try:
            conn = sqlite3.connect('database.db')
            cursor = conn.cursor()
            cursor.execute(f"SELECT * FROM {table}")
            results = cursor.fetchall()
            conn.close()
            return str(results)
        except Exception as e:
            return f"Error reading database: {str(e)}"
    
    # ... 其他资源

3.2.2 API集成

python
import requests

# API工具
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    if name == "api_call":
        url = arguments.get("url")
        method = arguments.get("method", "GET")
        params = arguments.get("params", {})
        headers = arguments.get("headers", {})
        data = arguments.get("data", {})
        
        try:
            if method == "GET":
                response = requests.get(url, params=params, headers=headers)
            elif method == "POST":
                response = requests.post(url, json=data, headers=headers)
            else:
                return f"Unsupported method: {method}"
            
            return response.text
        except Exception as e:
            return f"Error calling API: {str(e)}"
    
    # ... 其他工具

3.2.3 提示模板

python
@server.list_prompts()
async def list_prompts() -> list[dict]:
    return [
        {
            "name": "summarize_file",
            "description": "Summarize a file",
            "arguments": [
                {
                    "name": "path",
                    "description": "Path to the file",
                    "required": True
                }
            ]
        },
        {
            "name": "analyze_data",
            "description": "Analyze data",
            "arguments": [
                {
                    "name": "data",
                    "description": "Data to analyze",
                    "required": True
                }
            ]
        }
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[dict]:
    if name == "summarize_file":
        path = arguments.get("path")
        return [
            {
                "role": "user",
                "content": {
                    "type": "resource",
                    "resource": {
                        "uri": f"file://{path}"
                    }
                }
            },
            {
                "role": "user",
                "content": {
                    "type": "text",
                    "text": "Please summarize this file."
                }
            }
        ]
    
    # ... 其他提示

4. MCP Client开发 ​

4.1 Client基础 ​

4.1.1 连接Server

python
from mcp.client import Client

# 创建Client
client = Client(
    name="my-mcp-client",
    version="1.0.0"
)

# 连接到Server
async def connect_to_server():
    await client.connect("stdio")  # 或 "tcp://localhost:8080"
    
    # 获取Server信息
    server_info = await client.get_server_info()
    print(f"Connected to server: {server_info['name']}")

# 运行Client
if __name__ == "__main__":
    import asyncio
    asyncio.run(connect_to_server())

4.1.2 调用工具

python
async def call_tools():
    # 列出可用工具
    tools = await client.list_tools()
    print("Available tools:")
    for tool in tools:
        print(f"  - {tool['name']}: {tool['description']}")
    
    # 调用工具
    result = await client.call_tool(
        name="file_read",
        arguments={
            "path": "/path/to/file.txt"
        }
    )
    print(f"Tool result: {result}")
    
    # 调用另一个工具
    result = await client.call_tool(
        name="file_write",
        arguments={
            "path": "/path/to/output.txt",
            "content": "Hello, MCP!"
        }
    )
    print(f"Tool result: {result}")

4.1.3 访问资源

python
async def access_resources():
    # 列出可用资源
    resources = await client.list_resources()
    print("Available resources:")
    for resource in resources:
        print(f"  - {resource['uri']}: {resource['name']}")
    
    # 读取资源
    content = await client.read_resource(
        uri="file:///path/to/file.txt"
    )
    print(f"Resource content: {content}")
    
    # 写入资源
    result = await client.write_resource(
        uri="file:///path/to/output.txt",
        content="Hello, MCP!"
    )
    print(f"Write result: {result}")

4.2 高级功能 ​

4.2.1 批量操作

python
async def batch_operations():
    # 批量调用工具
    results = await asyncio.gather(
        client.call_tool("file_read", {"path": "/file1.txt"}),
        client.call_tool("file_read", {"path": "/file2.txt"}),
        client.call_tool("file_read", {"path": "/file3.txt"}),
    )
    
    for i, result in enumerate(results):
        print(f"File {i+1}: {result}")

4.2.2 错误处理

python
async def safe_tool_call(name: str, arguments: dict):
    try:
        result = await client.call_tool(name, arguments)
        return result
    except Exception as e:
        print(f"Error calling tool {name}: {str(e)}")
        return None

async def safe_resource_read(uri: str):
    try:
        content = await client.read_resource(uri)
        return content
    except Exception as e:
        print(f"Error reading resource {uri}: {str(e)}")
        return None

4.2.3 事件监听

python
@client.on("notification")
async def handle_notification(notification):
    print(f"Received notification: {notification}")

@client.on("error")
async def handle_error(error):
    print(f"Received error: {error}")

5. MCP与Claude Desktop集成 ​

5.1 Claude Desktop配置 ​

5.1.1 配置文件位置

macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

Windows: %APPDATA%\Claude\claude_desktop_config.json

Linux: ~/.config/Claude/claude_desktop_config.json

5.1.2 配置文件格式

json
{
  "mcpServers": {
    "filesystem": {
      "command": "python",
      "args": ["/path/to/filesystem_server.py"]
    },
    "database": {
      "command": "python",
      "args": ["/path/to/database_server.py"]
    },
    "api": {
      "command": "python",
      "args": ["/path/to/api_server.py"]
    }
  }
}

5.1.3 配置示例

json
{
  "mcpServers": {
    "my-custom-server": {
      "command": "python",
      "args": ["/Users/username/mcp-servers/my_server.py"],
      "env": {
        "API_KEY": "your-api-key",
        "DATABASE_URL": "postgresql://user:password@localhost/db"
      }
    }
  }
}

5.2 Claude Desktop使用 ​

5.2.1 使用工具

在Claude Desktop中，你可以直接调用MCP Server提供的工具。

示例对话：

用户：请读取 /path/to/file.txt 文件的内容
Claude：[调用 file_read 工具]
文件内容如下：
[文件内容]

用户：请将这段内容写入 /path/to/output.txt
Claude：[调用 file_write 工具]
已成功写入文件。

5.2.2 访问资源

Claude Desktop可以自动访问MCP Server提供的资源。

示例对话：

用户：请分析 database://users 资源
Claude：[读取 database://users 资源]
数据库内容如下：
[数据库内容]

分析结果：
[分析结果]

5.2.3 使用提示

Claude Desktop可以使用MCP Server提供的预定义提示。

示例对话：

用户：请使用 summarize_file 提示来总结 /path/to/file.txt
Claude：[调用 summarize_file 提示]
文件总结如下：
[总结内容]

6. MCP的应用场景 ​

6.1 文件系统操作 ​

场景：AI Agent需要读写文件。

工具：

  • file_read: 读取文件
  • file_write: 写入文件
  • file_list: 列出文件
  • file_delete: 删除文件

示例：

python
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    if name == "file_read":
        path = arguments.get("path")
        with open(path, 'r') as f:
            return f.read()
    
    elif name == "file_write":
        path = arguments.get("path")
        content = arguments.get("content")
        with open(path, 'w') as f:
            f.write(content)
        return f"Successfully wrote to {path}"
    
    elif name == "file_list":
        path = arguments.get("path", ".")
        files = os.listdir(path)
        return "\n".join(files)
    
    elif name == "file_delete":
        path = arguments.get("path")
        os.remove(path)
        return f"Successfully deleted {path}"

6.2 数据库操作 ​

场景：AI Agent需要查询和更新数据库。

工具：

  • database_query: 执行SQL查询
  • database_execute: 执行SQL语句
  • database_schema: 获取数据库schema

示例：

python
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    if name == "database_query":
        sql = arguments.get("sql")
        conn = sqlite3.connect('database.db')
        cursor = conn.cursor()
        cursor.execute(sql)
        results = cursor.fetchall()
        conn.close()
        return str(results)
    
    elif name == "database_execute":
        sql = arguments.get("sql")
        conn = sqlite3.connect('database.db')
        cursor = conn.cursor()
        cursor.execute(sql)
        conn.commit()
        conn.close()
        return "Successfully executed SQL"
    
    elif name == "database_schema":
        table = arguments.get("table")
        conn = sqlite3.connect('database.db')
        cursor = conn.cursor()
        cursor.execute(f"PRAGMA table_info({table})")
        schema = cursor.fetchall()
        conn.close()
        return str(schema)

6.3 API调用 ​

场景：AI Agent需要调用外部API。

工具：

  • api_get: GET请求
  • api_post: POST请求
  • api_put: PUT请求
  • api_delete: DELETE请求

示例：

python
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    if name == "api_get":
        url = arguments.get("url")
        params = arguments.get("params", {})
        headers = arguments.get("headers", {})
        response = requests.get(url, params=params, headers=headers)
        return response.text
    
    elif name == "api_post":
        url = arguments.get("url")
        data = arguments.get("data", {})
        headers = arguments.get("headers", {})
        response = requests.post(url, json=data, headers=headers)
        return response.text
    
    # ... 其他HTTP方法

6.4 云服务集成 ​

场景：AI Agent需要与云服务交互。

工具：

  • aws_s3_upload: 上传文件到S3
  • aws_s3_download: 从S3下载文件
  • gcp_storage_upload: 上传文件到GCS
  • azure_blob_upload: 上传文件到Azure Blob

示例：

python
import boto3

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    if name == "aws_s3_upload":
        bucket = arguments.get("bucket")
        key = arguments.get("key")
        file_path = arguments.get("file_path")
        
        s3 = boto3.client('s3')
        s3.upload_file(file_path, bucket, key)
        return f"Successfully uploaded {file_path} to s3://{bucket}/{key}"
    
    elif name == "aws_s3_download":
        bucket = arguments.get("bucket")
        key = arguments.get("key")
        file_path = arguments.get("file_path")
        
        s3 = boto3.client('s3')
        s3.download_file(bucket, key, file_path)
        return f"Successfully downloaded s3://{bucket}/{key} to {file_path}"

7. MCP的最佳实践 ​

7.1 Server开发最佳实践 ​

7.1.1 错误处理

python
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    try:
        # 验证参数
        if "path" not in arguments:
            return "Error: Missing required parameter 'path'"
        
        # 执行操作
        result = execute_tool(name, arguments)
        
        return result
    except FileNotFoundError as e:
        return f"Error: File not found - {str(e)}"
    except PermissionError as e:
        return f"Error: Permission denied - {str(e)}"
    except Exception as e:
        return f"Error: {str(e)}"

7.1.2 参数验证

python
from pydantic import BaseModel, ValidationError

class FileReadArgs(BaseModel):
    path: str

class FileWriteArgs(BaseModel):
    path: str
    content: str

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    try:
        if name == "file_read":
            args = FileReadArgs(**arguments)
            return read_file(args.path)
        
        elif name == "file_write":
            args = FileWriteArgs(**arguments)
            return write_file(args.path, args.content)
    
    except ValidationError as e:
        return f"Error: Invalid arguments - {str(e)}"

7.1.3 日志记录

python
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    logger.info(f"Calling tool {name} with arguments {arguments}")
    
    try:
        result = execute_tool(name, arguments)
        logger.info(f"Tool {name} executed successfully")
        return result
    except Exception as e:
        logger.error(f"Error executing tool {name}: {str(e)}")
        return f"Error: {str(e)}"

7.2 Client开发最佳实践 ​

7.2.1 连接管理

python
class MCPClientManager:
    def __init__(self):
        self.clients = {}
    
    async def connect(self, name: str, config: dict):
        if name in self.clients:
            return self.clients[name]
        
        client = Client(name=name, version="1.0.0")
        await client.connect(config.get("transport", "stdio"))
        self.clients[name] = client
        return client
    
    async def disconnect(self, name: str):
        if name in self.clients:
            await self.clients[name].disconnect()
            del self.clients[name]
    
    async def disconnect_all(self):
        for name in list(self.clients.keys()):
            await self.disconnect(name)

7.2.2 重试机制

python
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def call_tool_with_retry(client: Client, name: str, arguments: dict) -> str:
    return await client.call_tool(name, arguments)

7.2.3 缓存

python
from functools import lru_cache

class CachedMCPClient:
    def __init__(self, client: Client):
        self.client = client
        self.cache = {}
    
    @lru_cache(maxsize=100)
    async def call_tool_cached(self, name: str, arguments: tuple) -> str:
        return await self.client.call_tool(name, dict(arguments))
    
    async def call_tool(self, name: str, arguments: dict) -> str:
        # 将arguments转换为可哈希的tuple
        args_tuple = tuple(sorted(arguments.items()))
        return await self.call_tool_cached(name, args_tuple)

实践任务 ​

任务1：实现文件系统MCP Server ​

目标：实现一个文件系统MCP Server。

要求：

  1. 实现文件读写工具
  2. 实现文件列表工具
  3. 实现文件删除工具
  4. 实现文件资源
  5. 测试Server功能

代码框架：

python
from mcp.server import Server
from mcp.types import Tool, Resource
import os

# 创建Server
server = Server(
    name="filesystem-server",
    version="1.0.0"
)

# 定义工具
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="file_read",
            description="Read a file from the filesystem",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path to the file"
                    }
                },
                "required": ["path"]
            }
        ),
        # ... 其他工具
    ]

# 实现工具
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    if name == "file_read":
        path = arguments.get("path")
        try:
            with open(path, 'r') as f:
                return f.read()
        except Exception as e:
            return f"Error: {str(e)}"
    
    # ... 其他工具

# 定义资源
@server.list_resources()
async def list_resources() -> list[Resource]:
    resources = []
    for root, dirs, files in os.walk("."):
        for file in files:
            path = os.path.join(root, file)
            resources.append(
                Resource(
                    uri=f"file://{path}",
                    name=file,
                    description=f"File: {path}",
                    mimeType="text/plain"
                )
            )
    return resources

# 启动Server
if __name__ == "__main__":
    import asyncio
    asyncio.run(server.run())

任务2：实现数据库MCP Server ​

目标：实现一个数据库MCP Server。

要求：

  1. 实现数据库查询工具
  2. 实现数据库执行工具
  3. 实现数据库schema工具
  4. 实现数据库资源
  5. 测试Server功能

代码框架：

python
from mcp.server import Server
from mcp.types import Tool, Resource
import sqlite3

# 创建Server
server = Server(
    name="database-server",
    version="1.0.0"
)

# 定义工具
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="database_query",
            description="Execute a SQL query",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SQL query to execute"
                    }
                },
                "required": ["sql"]
            }
        ),
        # ... 其他工具
    ]

# 实现工具
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> str:
    if name == "database_query":
        sql = arguments.get("sql")
        try:
            conn = sqlite3.connect('database.db')
            cursor = conn.cursor()
            cursor.execute(sql)
            results = cursor.fetchall()
            conn.close()
            return str(results)
        except Exception as e:
            return f"Error: {str(e)}"
    
    # ... 其他工具

# 启动Server
if __name__ == "__main__":
    import asyncio
    asyncio.run(server.run())

任务3：集成到Claude Desktop ​

目标：将MCP Server集成到Claude Desktop。

要求：

  1. 实现一个MCP Server
  2. 配置Claude Desktop
  3. 测试工具调用
  4. 测试资源访问
  5. 测试提示使用

配置文件：

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

课后作业 ​

作业1：MCP Server开发 ​

题目：开发一个特定领域的MCP Server。

要求：

  1. 选择一个领域（如天气、新闻、股票等）
  2. 设计工具和资源
  3. 实现MCP Server
  4. 测试Server功能
  5. 集成到Claude Desktop

作业2：MCP Client开发 ​

题目：开发一个MCP Client应用。

要求：

  1. 实现MCP Client
  2. 连接到多个Server
  3. 实现工具调用
  4. 实现资源访问
  5. 测试Client功能

作业3：MCP生态调研 ​

题目：调研MCP生态。

要求：

  1. 调研现有的MCP Server
  2. 分析Server的功能和特点
  3. 比较不同Server的优劣
  4. 撰写2000字调研报告

参考资料 ​

必读文献 ​

  1. Anthropic (2024). "Model Context Protocol (MCP) Specification".

    • MCP协议规范

  2. Anthropic (2024). "MCP Server Development Guide".

    • MCP Server开发指南

  3. Anthropic (2024). "MCP Client Development Guide".

    • MCP Client开发指南

推荐阅读 ​

  1. JSON-RPC 2.0 Specification: https://www.jsonrpc.org/specification

    • JSON-RPC 2.0规范

  2. Claude Desktop Documentation: https://docs.anthropic.com/claude/docs/mcp

    • Claude Desktop MCP文档

在线资源 ​

  1. MCP GitHub: https://github.com/modelcontextprotocol

    • MCP GitHub组织

  2. MCP Examples: https://github.com/modelcontextprotocol/servers

    • MCP Server示例

  3. MCP SDK: https://github.com/modelcontextprotocol/python-sdk

    • MCP Python SDK

扩展阅读 ​

MCP前沿 ​

  • Anthropic (2024). "MCP Best Practices".

    • MCP最佳实践

  • Community (2024). "MCP Server Showcase".

    • MCP Server展示

协议设计 ​

  • Fielding, R. T. (2000). "Architectural Styles and the Design of Network-based Software Architectures". PhD Dissertation.

    • REST架构设计

  • W3C (2006). "Web Services Addressing (WS-Addressing)".

    • Web服务协议

下节预告 ​

下一节我们将学习Skills标准发布（2025年），了解Anthropic Skills标准的诞生背景、核心思想、文档驱动开发理念，以及它如何标准化AI能力的描述和发现。

架构师AI杜公众号二维码

扫描二维码关注"架构师AI杜"公众号，获取更多技术内容和最新动态