从Function Calling到MCP：AI工具调用的范式级变化

# MCP异步通信示例
import asyncio
import aiohttp

class AsyncMCPClient:
    """异步MCP Client实现"""
    
    def __init__(self, server_url):
        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 execute_tool(self, tool_name, tool_version, parameters):
        """异步执行工具"""
        request = {
            "id": f"req_{asyncio.get_event_loop().time()}",
            "version": "2.0",
            "timestamp": str(asyncio.get_event_loop().time()),
            "type": "execute",
            "tool": {
                "name": tool_name,
                "version": tool_version,
                "parameters": parameters
            }
        }
        
        async with self.session.post(f"{self.server_url}/execute", json=request) as response:
            return await response.json()
    
    async def execute_multiple_tools(self, tool_calls):
        """并行执行多个工具"""
        tasks = []
        for tool_call in tool_calls:
            task = self.execute_tool(
                tool_call["name"],
                tool_call["version"],
                tool_call["parameters"]
            )
            tasks.append(task)
        
        return await asyncio.gather(*tasks)

# 示例用法
async def main():
    async with AsyncMCPClient("http://localhost:8000") as client:
        # 并行执行多个工具
        results = await client.execute_multiple_tools([
            {
                "name": "file_reader",
                "version": "1.0",
                "parameters": {"path": "/data/file1.txt"}
            },
            {
                "name": "file_reader",
                "version": "1.0",
                "parameters": {"path": "/data/file2.txt"}
            },
            {
                "name": "file_reader",
                "version": "1.0",
                "parameters": {"path": "/data/file3.txt"}
            }
        ])
        
        print(f"并行执行结果: {results}")

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

// MCP安全配置示例
{
  "security": {
    "authentication": {
      "type": "JWT",
      "secret": "your-secret-key",
      "expires_in": 3600
    },
    "authorization": {
      "roles": {
        "user": {
          "permissions": ["file:read", "data:query"]
        },
        "admin": {
          "permissions": ["*:*"]
        }
      }
    },
    "rate_limiting": {
      "global": {
        "requests_per_minute": 1000
      },
      "per_user": {
        "requests_per_minute": 60
      }
    },
    "audit": {
      "enabled": true,
      "log_level": "info",
      "retention_days": 30
    },
    "sandbox": {
      "enabled": true,
      "allowed_paths": ["/data", "/tmp"],
      "allowed_commands": ["ls", "cat", "grep"]
    }
  }
}

from fastapi import FastAPI, HTTPException, Depends, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
import uvicorn
import json
import asyncio
from typing import Dict, List, Any, Optional
import jwt
from datetime import datetime, timedelta

# 安全配置
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 工具注册表
tools_registry: Dict[str, Any] = {}

# 初始化FastAPI应用
app = FastAPI(title="MCP Server", version="2.0")

# JWT认证
bearer_scheme = HTTPBearer()

# 工具定义模型
class ToolDefinition(BaseModel):
    name: str = Field(..., description="工具名称")
    version: str = Field(..., description="工具版本")
    description: str = Field(..., description="工具描述")
    parameters: Dict[str, Any] = Field(..., description="工具参数Schema")
    returns: Dict[str, Any] = Field(..., description="工具返回值Schema")
    security: Dict[str, Any] = Field(default={}, description="安全配置")
    async_supported: bool = Field(default=True, description="是否支持异步调用")

# 工具执行请求模型
class ToolExecuteRequest(BaseModel):
    id: str = Field(..., description="请求ID")
    version: str = Field(..., description="MCP版本")
    timestamp: str = Field(..., description="请求时间戳")
    type: str = Field(..., description="请求类型")
    tool: Dict[str, Any] = Field(..., description="工具信息")
    context: Dict[str, Any] = Field(default={}, description="上下文信息")

# 能力查询请求模型
class CapabilityQueryRequest(BaseModel):
    id: str = Field(..., description="请求ID")
    version: str = Field(..., description="MCP版本")
    timestamp: str = Field(..., description="请求时间戳")
    type: str = Field(..., description="请求类型")
    capabilities: List[str] = Field(default=[], description="查询的能力列表")
    context: Dict[str, Any] = Field(default={}, description="上下文信息")

# 创建访问令牌
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# 验证访问令牌
def verify_token(credentials: HTTPAuthorizationCredentials = Security(bearer_scheme)):
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid authentication credentials")

# 根路由
@app.get("/")
async def root():
    return {"message": "MCP Server is running", "version": "2.0"}

# 健康检查
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# 获取服务器指标
@app.get("/metrics")
async def get_metrics():
    return {
        "tools_count": len(tools_registry),
        "server_version": "2.0"
    }

# 注册工具
@app.post("/tools/register", dependencies=[Depends(verify_token)])
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.get("/tools/{tool_name}")
async def get_tool(tool_name: str, version: Optional[str] = None):
    if version:
        tool_key = f"{tool_name}:{version}"
        if tool_key in tools_registry:
            return tools_registry[tool_key]
        else:
            raise HTTPException(status_code=404, detail=f"Tool {tool_key} not found")
    else:
        # 返回所有版本的工具
        tools = []
        for key, tool in tools_registry.items():
            if key.startswith(f"{tool_name}:"):
                tools.append(tool)
        if not tools:
            raise HTTPException(status_code=404, detail=f"Tool {tool_name} not found")
        return {"tools": tools}

# 能力协商
@app.post("/negotiate")
async def negotiate_capabilities(request: CapabilityQueryRequest):
    # 验证MCP版本
    if request.version != "2.0":
        raise HTTPException(status_code=400, detail="Unsupported MCP version")
    
    # 处理能力查询
    available_capabilities = []
    for tool in tools_registry.values():
        # 检查工具是否支持请求的能力
        if not request.capabilities or any(cap in tool.get("capabilities", []) for cap in request.capabilities):
            available_capabilities.append({
                "name": tool["name"],
                "version": tool["version"],
                "capabilities": tool.get("capabilities", [])
            })
    
    return {
        "id": request.id,
        "version": "2.0",
        "timestamp": str(asyncio.get_event_loop().time()),
        "type": "response",
        "status": "success",
        "capabilities": available_capabilities,
        "context": request.context
    }

# 执行工具
@app.post("/execute")
async def execute_tool(request: ToolExecuteRequest, payload: Dict[str, Any] = Depends(verify_token)):
    # 验证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")
    
    tool_def = tools_registry[tool_key]
    
    # 检查用户是否有执行该工具的权限
    user_roles = payload.get("roles", ["user"])
    tool_permissions = tool_def["security"].get("permissions", [])
    
    # 简化的权限检查，实际应用中应该更复杂
    has_permission = True
    if tool_permissions:
        has_permission = any(role == "admin" or f"{tool_name}:*" in tool_permissions for role in user_roles)
    
    if not has_permission:
        raise HTTPException(status_code=403, detail="Permission denied")
    
    # 这里简化了工具执行逻辑，实际实现需要根据工具定义执行具体的工具
    # 在真实场景中，应该将请求路由到对应的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": str(asyncio.get_event_loop().time()),
                "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": str(asyncio.get_event_loop().time()),
                "type": "response",
                "status": "error",
                "error": {
                    "code": "execution_failed",
                    "message": str(e)
                },
                "context": request.context
            }
    
    # 模拟其他工具执行
    return {
        "id": request.id,
        "version": "2.0",
        "timestamp": str(asyncio.get_event_loop().time()),
        "type": "response",
        "status": "success",
        "result": {
            "message": f"Tool {tool_key} executed successfully",
            "parameters": request.tool["parameters"]
        },
        "context": request.context
    }

# 生成访问令牌
@app.post("/token")
async def generate_token(username: str, password: str):
    # 简化的身份验证，实际应用中应该更复杂
    if username == "admin" and password == "password":
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(
            data={"sub": username, "roles": ["admin"]}, expires_delta=access_token_expires
        )
        return {"access_token": access_token, "token_type": "bearer"}
    elif username == "user" and password == "password":
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(
            data={"sub": username, "roles": ["user"]}, expires_delta=access_token_expires
        )
        return {"access_token": access_token, "token_type": "bearer"}
    else:
        raise HTTPException(status_code=401, detail="Incorrect username or password")

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

import asyncio
import aiohttp
import json
from openai import AsyncOpenAI

class LLM_MCPClient:
    """LLM与MCP Client集成"""
    
    def __init__(self, llm_api_key: str, mcp_server_url: str):
        self.llm_client = AsyncOpenAI(api_key=llm_api_key)
        self.mcp_server_url = mcp_server_url
        self.mcp_session = None
    
    async def __aenter__(self):
        self.mcp_session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.mcp_session.close()
    
    async def list_mcp_tools(self) -> dict:
        """获取MCP工具列表"""
        url = f"{self.mcp_server_url}/tools/list"
        async with self.mcp_session.get(url) as response:
            return await response.json()
    
    async def execute_mcp_tool(self, tool_name: str, tool_version: str, 
                              parameters: dict, user_id: str, session_id: str) -> dict:
        """执行MCP工具"""
        request = {
            "id": f"req_{asyncio.get_event_loop().time()}",
            "version": "2.0",
            "timestamp": str(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
            }
        }
        
        url = f"{self.mcp_server_url}/execute"
        async with self.mcp_session.post(url, json=request) as response:
            return await response.json()
    
    async def generate_tool_call(self, user_query: str, tools: list) -> dict:
        """使用LLM生成工具调用"""
        # 构建提示词
        prompt = f"""
        你是一个智能助手，需要根据用户查询和可用工具，生成合适的工具调用请求。
        
        可用工具：
        {json.dumps(tools, indent=2, ensure_ascii=False)}
        
        用户查询：{user_query}
        
        请生成一个JSON格式的工具调用请求，包含工具名称、版本和参数。
        只返回JSON，不要添加任何其他内容。
        """
        
        # 调用LLM生成工具调用
        response = await self.llm_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个智能工具调用生成器。"},
                {"role": "user", "content": prompt}
            ],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)
    
    async def process_user_query(self, user_query: str, user_id: str, session_id: str) -> str:
        """处理用户查询，智能调用工具"""
        # 获取可用工具列表
        tools_response = await self.list_mcp_tools()
        tools = tools_response.get("tools", [])
        
        if not tools:
            return "没有可用的工具。"
        
        # 使用LLM生成工具调用
        tool_call = await self.generate_tool_call(user_query, tools)
        
        # 执行工具调用
        tool_name = tool_call["name"]
        tool_version = tool_call.get("version", "1.0")
        parameters = tool_call["parameters"]
        
        result = await self.execute_mcp_tool(tool_name, tool_version, parameters, user_id, session_id)
        
        # 使用LLM将工具结果转化为自然语言回答
        prompt = f"""
        你是一个智能助手，需要将工具执行结果转化为自然语言回答，响应用户的原始查询。
        
        用户原始查询：{user_query}
        
        工具调用详情：
        {json.dumps(tool_call, indent=2, ensure_ascii=False)}
        
        工具执行结果：
        {json.dumps(result, indent=2, ensure_ascii=False)}
        
        请将结果转化为自然、友好的自然语言回答，不要添加任何其他内容。
        """
        
        response = await self.llm_client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个智能结果解释器。"},
                {"role": "user", "content": prompt}
            ]
        )
        
        return response.choices[0].message.content

# 示例用法
async def main():
    async with LLM_MCPClient(
        llm_api_key="your-openai-api-key",
        mcp_server_url="http://localhost:8000"
    ) as client:
        # 处理用户查询
        user_query = "请读取/data/sample.txt文件的内容"
        result = await client.process_user_query(user_query, "user_001", "session_123")
        print(f"用户查询：{user_query}")
        print(f"回答：{result}")

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