MCP是什么？从Function Calling到AI工具生态中枢的范式转变

// MCP请求格式示例
{
  "id": "req_1234567890",
  "version": "2.0",
  "timestamp": "2025-12-30T12:00:00Z",
  "type": "execute",
  "tool": {
    "name": "file_reader",
    "version": "1.0",
    "parameters": {
      "path": "/data/sample.txt",
      "encoding": "utf-8"
    }
  },
  "context": {
    "user_id": "user_001",
    "session_id": "session_123"
  }
}

// MCP响应格式示例
{
  "id": "req_1234567890",
  "version": "2.0",
  "timestamp": "2025-12-30T12:00:01Z",
  "type": "response",
  "status": "success",
  "result": {
    "content": "Hello, MCP!",
    "metadata": {
      "file_size": 12,
      "read_time": 0.001
    }
  },
  "context": {
    "user_id": "user_001",
    "session_id": "session_123"
  }
}

// MCP工具定义示例
{
  "name": "file_reader",
  "version": "1.0",
  "description": "读取文件内容的工具",
  "parameters": {
    "type": "object",
    "properties": {
      "path": {
        "type": "string",
        "description": "文件路径",
        "pattern": "^/[a-zA-Z0-9_/]+$" // 限制路径格式，防止路径遍历攻击
      },
      "encoding": {
        "type": "string",
        "description": "文件编码",
        "default": "utf-8",
        "enum": ["utf-8", "gbk", "ascii"]
      }
    },
    "required": ["path"]
  },
  "returns": {
    "type": "object",
    "properties": {
      "content": {
        "type": "string",
        "description": "文件内容"
      },
      "metadata": {
        "type": "object",
        "properties": {
          "file_size": {
            "type": "integer",
            "description": "文件大小（字节）"
          },
          "read_time": {
            "type": "number",
            "description": "读取时间（秒）"
          }
        }
      }
    }
  },
  "security": {
    "permissions": ["file:read"],
    "rate_limit": {
      "requests_per_minute": 60
    }
  }
}

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
import json

app = FastAPI(title="MCP Server", version="2.0")

# 工具注册表
tools_registry = {}

# 工具定义模型
class ToolDefinition(BaseModel):
    name: str
    version: str
    description: str
    parameters: dict
    returns: dict
    security: dict = {}

# 工具执行请求模型
class ToolExecuteRequest(BaseModel):
    id: str
    version: str
    timestamp: str
    type: str
    tool: dict
    context: dict = {}

# 注册工具
@app.post("/tools/register")
async def register_tool(tool_def: ToolDefinition):
    tool_key = f"{tool_def.name}:{tool_def.version}"
    tools_registry[tool_key] = tool_def.dict()
    return {"status": "success", "message": f"Tool {tool_key} registered successfully"}

# 获取工具列表
@app.get("/tools/list")
async def list_tools():
    return {"tools": list(tools_registry.values())}

# 执行工具
@app.post("/execute")
async def execute_tool(request: ToolExecuteRequest):
    # 验证MCP版本
    if request.version != "2.0":
        raise HTTPException(status_code=400, detail="Unsupported MCP version")
    
    tool_name = request.tool["name"]
    tool_version = request.tool.get("version", "1.0")
    tool_key = f"{tool_name}:{tool_version}"
    
    # 检查工具是否存在
    if tool_key not in tools_registry:
        raise HTTPException(status_code=404, detail=f"Tool {tool_key} not found")
    
    # 这里简化了工具执行逻辑，实际实现需要根据工具定义执行具体的工具
    # 在真实场景中，应该将请求路由到对应的MCP Host执行
    
    # 模拟文件读取工具执行
    if tool_name == "file_reader":
        path = request.tool["parameters"].get("path")
        encoding = request.tool["parameters"].get("encoding", "utf-8")
        
        try:
            with open(path, "r", encoding=encoding) as f:
                content = f.read()
            
            return {
                "id": request.id,
                "version": "2.0",
                "timestamp": "2025-12-30T12:00:01Z",
                "type": "response",
                "status": "success",
                "result": {
                    "content": content,
                    "metadata": {
                        "file_size": len(content),
                        "read_time": 0.001
                    }
                },
                "context": request.context
            }
        except Exception as e:
            return {
                "id": request.id,
                "version": "2.0",
                "timestamp": "2025-12-30T12:00:01Z",
                "type": "response",
                "status": "error",
                "error": {
                    "code": "execution_failed",
                    "message": str(e)
                },
                "context": request.context
            }
    
    # 模拟其他工具执行
    return {
        "id": request.id,
        "version": "2.0",
        "timestamp": "2025-12-30T12:00:01Z",
        "type": "response",
        "status": "success",
        "result": {
            "message": f"Tool {tool_key} executed successfully"
        },
        "context": request.context
    }

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

import asyncio
import aiohttp
import json
from typing import Dict, Any, Optional

class MCPError(Exception):
    """MCP相关错误"""
    pass

class MCPClient:
    """MCP Client实现"""
    
    def __init__(self, server_url: str):
        self.server_url = server_url
        self.session = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.close()
    
    async def _send_request(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
        """发送MCP请求"""
        url = f"{self.server_url}{endpoint}"
        async with self.session.post(url, json=data) as response:
            if response.status != 200:
                raise MCPError(f"Request failed with status {response.status}: {await response.text()}")
            return await response.json()
    
    async def list_tools(self) -> Dict[str, Any]:
        """获取可用工具列表"""
        url = f"{self.server_url}/tools/list"
        async with self.session.get(url) as response:
            if response.status != 200:
                raise MCPError(f"Failed to list tools: {await response.text()}")
            return await response.json()
    
    async def execute_tool(self, tool_name: str, tool_version: str, parameters: Dict[str, Any], 
                         user_id: str, session_id: str) -> Dict[str, Any]:
        """执行工具"""
        request = {
            "id": f"req_{asyncio.get_event_loop().time()}",
            "version": "2.0",
            "timestamp": json.dumps(asyncio.get_event_loop().time()),
            "type": "execute",
            "tool": {
                "name": tool_name,
                "version": tool_version,
                "parameters": parameters
            },
            "context": {
                "user_id": user_id,
                "session_id": session_id
            }
        }
        
        return await self._send_request("/execute", request)

# 示例用法
async def main():
    async with MCPClient("http://localhost:8000") as client:
        # 获取工具列表
        tools = await client.list_tools()
        print("可用工具:", json.dumps(tools, indent=2, ensure_ascii=False))
        
        # 执行文件读取工具（假设已注册）
        try:
            result = await client.execute_tool(
                tool_name="file_reader",
                tool_version="1.0",
                parameters={"path": "/data/sample.txt"},
                user_id="user_001",
                session_id="session_123"
            )
            print("执行结果:", json.dumps(result, indent=2, ensure_ascii=False))
        except MCPError as e:
            print(f"执行失败: {e}")

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

import asyncio
import aiohttp
import json

async def register_tool():
    """注册文件读取工具到MCP Server"""
    tool_definition = {
        "name": "file_reader",
        "version": "1.0",
        "description": "读取文件内容的工具",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "文件路径",
                    "pattern": "^/[a-zA-Z0-9_/]+$"
                },
                "encoding": {
                    "type": "string",
                    "description": "文件编码",
                    "default": "utf-8",
                    "enum": ["utf-8", "gbk", "ascii"]
                }
            },
            "required": ["path"]
        },
        "returns": {
            "type": "object",
            "properties": {
                "content": {
                    "type": "string",
                    "description": "文件内容"
                },
                "metadata": {
                    "type": "object",
                    "properties": {
                        "file_size": {
                            "type": "integer",
                            "description": "文件大小（字节）"
                        },
                        "read_time": {
                            "type": "number",
                            "description": "读取时间（秒）"
                        }
                    }
                }
            }
        },
        "security": {
            "permissions": ["file:read"],
            "rate_limit": {
                "requests_per_minute": 60
            }
        }
    }
    
    async with aiohttp.ClientSession() as session:
        url = "http://localhost:8000/tools/register"
        async with session.post(url, json=tool_definition) as response:
            if response.status != 200:
                print(f"注册失败: {await response.text()}")
            else:
                result = await response.json()
                print(f"注册成功: {result}")

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