MCP解决了哪些不可控的AI问题

# MCP智能上下文管理实现
class MCPContextManager:
    def __init__(self, max_context_length=10, max_tokens=2048):
        self.max_context_length = max_context_length
        self.max_tokens = max_tokens
    
    def trim_context(self, context):
        """智能裁剪上下文，减少幻觉生成"""
        # 1. 按长度裁剪
        trimmed = context[-self.max_context_length:]
        
        # 2. 按token数裁剪
        total_tokens = 0
        final_context = []
        for item in reversed(trimmed):
            item_tokens = self.count_tokens(str(item))
            if total_tokens + item_tokens <= self.max_tokens:
                final_context.insert(0, item)
                total_tokens += item_tokens
            else:
                break
        
        return final_context
    
    def count_tokens(self, text):
        """估算token数"""
        return len(text) // 4
    
    def extract_relevant_context(self, query, context):
        """提取与查询相关的上下文"""
        # 简化的相关性提取，实际应用中可使用更复杂的算法
        relevant = []
        query_lower = query.lower()
        
        for item in context:
            content_lower = str(item.get("content", "")).lower()
            if any(keyword in content_lower for keyword in query_lower.split()):
                relevant.append(item)
        
        # 补充最近的上下文
        if len(relevant) < 3:
            recent = context[-3:]
            for item in recent:
                if item not in relevant:
                    relevant.append(item)
        
        return self.trim_context(relevant)

# 示例用法
context_manager = MCPContextManager()
context = [
    {"role": "user", "content": "你好，我想了解MCP协议。"},
    {"role": "assistant", "content": "MCP是Model Communication Protocol的缩写，用于大语言模型与外部工具的安全交互。"},
    {"role": "user", "content": "MCP v2.0有哪些新特性？"},
    {"role": "assistant", "content": "MCP v2.0引入了动态能力协商、分布式架构和生态集成等新特性。"},
    {"role": "user", "content": "如何实现MCP Server？"},
    {"role": "assistant", "content": "可以使用FastAPI等框架实现MCP Server，支持RESTful API和WebSocket。"}
]
query = "MCP如何解决幻觉生成问题？"
relevant_context = context_manager.extract_relevant_context(query, context)
print(f"原始上下文长度：{len(context)}")
print(f"相关上下文长度：{len(relevant_context)}")
print("相关上下文：")
for item in relevant_context:
    print(f"  {item['role']}: {item['content'][:50]}...")

# MCP权限管理实现
class MCPAuthorization:
    def __init__(self):
        # 权限配置：角色 -> 工具 -> 权限
        self.permissions = {
            "user": {
                "file_reader": ["read"],
                "weather_api": ["get"]
            },
            "admin": {
                "file_reader": ["read", "write", "delete"],
                "weather_api": ["get"],
                "system_tools": ["*"]
            }
        }
    
    def has_permission(self, user_role, tool_name, action):
        """检查用户是否有执行工具的权限"""
        # 1. 检查角色是否存在
        if user_role not in self.permissions:
            return False
        
        # 2. 检查工具是否存在
        if tool_name not in self.permissions[user_role]:
            return False
        
        # 3. 检查权限
        role_permissions = self.permissions[user_role][tool_name]
        return action in role_permissions or "*" in role_permissions
    
    def get_allowed_tools(self, user_role):
        """获取用户可以使用的工具列表"""
        if user_role not in self.permissions:
            return []
        return list(self.permissions[user_role].keys())

# 示例用法
auth = MCPAuthorization()

# 检查权限
print(f"user是否可以read file_reader: {auth.has_permission('user', 'file_reader', 'read')}")
print(f"user是否可以write file_reader: {auth.has_permission('user', 'file_reader', 'write')}")
print(f"admin是否可以write file_reader: {auth.has_permission('admin', 'file_reader', 'write')}")
print(f"admin是否可以使用system_tools: {auth.has_permission('admin', 'system_tools', 'execute')}")

# 获取允许的工具列表
print(f"user允许的工具: {auth.get_allowed_tools('user')}")
print(f"admin允许的工具: {auth.get_allowed_tools('admin')}")

// MCP工具定义示例，包含严格的输入验证
{
  "name": "file_reader",
  "version": "1.0",
  "description": "读取文件内容的工具",
  "parameters": {
    "type": "object",
    "properties": {
      "path": {
        "type": "string",
        "description": "文件路径",
        "pattern": "^/[a-zA-Z0-9_/]+$", // 防止路径遍历攻击
        "maxLength": 256 // 限制路径长度
      },
      "encoding": {
        "type": "string",
        "description": "文件编码",
        "default": "utf-8",
        "enum": ["utf-8", "gbk", "ascii"] // 限制允许的编码
      }
    },
    "required": ["path"],
    "additionalProperties": false // 禁止额外属性
  },
  "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
    },
    "sandbox": {
      "enabled": true,
      "allowed_paths": ["/data", "/tmp"], // 限制允许访问的目录
      "allowed_commands": [] // 禁止执行系统命令
    }
  }
}

# MCP认证授权实现
from fastapi import FastAPI, HTTPException, Depends, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from datetime import datetime, timedelta
from typing import Dict, Any, List

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

# 权限配置
permissions = {
    "user": {
        "file_reader": ["read"],
        "weather_api": ["get"]
    },
    "admin": {
        "*": ["*"]  # 管理员拥有所有权限
    }
}

# 创建FastAPI应用
app = FastAPI(title="MCP Security Example")

# JWT认证方案
bearer_scheme = HTTPBearer()

# 创建访问令牌
def create_access_token(data: Dict[str, Any], expires_delta: 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)) -> Dict[str, Any]:
    """验证JWT令牌"""
    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token has expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

# 检查权限
def check_permission(payload: Dict[str, Any], tool_name: str, action: str) -> bool:
    """检查用户是否有执行工具的权限"""
    user_role = payload.get("role", "user")
    
    # 检查角色权限
    if user_role not in permissions:
        return False
    
    role_perms = permissions[user_role]
    
    # 检查是否有所有权限
    if "*" in role_perms:
        return True
    
    # 检查工具权限
    if tool_name not in role_perms:
        return False
    
    tool_perms = role_perms[tool_name]
    return action in tool_perms or "*" in tool_perms

# 示例：需要认证和授权的API端点
@app.post("/execute/{tool_name}")
async def execute_tool(tool_name: str, action: str, 
                     payload: Dict[str, Any] = Depends(verify_token)):
    """执行工具，需要认证和授权"""
    # 检查权限
    if not check_permission(payload, tool_name, action):
        raise HTTPException(status_code=403, detail="Permission denied")
    
    # 执行工具（这里简化为返回成功）
    return {
        "status": "success",
        "message": f"Executed tool {tool_name} with action {action}",
        "user": payload.get("sub"),
        "role": payload.get("role")
    }

# 示例：生成访问令牌的端点
@app.post("/token")
async def generate_token(username: str, password: str, role: str = "user"):
    """生成访问令牌"""
    # 简化的身份验证，实际应用中应连接数据库
    if username == "test" and password == "password":
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(
            data={"sub": username, "role": role}, 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__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# MCP输入验证实现
import jsonschema
from jsonschema import validate, ValidationError
import json

# 工具定义，包含JSON Schema
file_reader_tool = {
    "name": "file_reader",
    "version": "1.0",
    "parameters": {
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "pattern": "^/[a-zA-Z0-9_/]+$",
                "maxLength": 256
            },
            "encoding": {
                "type": "string",
                "enum": ["utf-8", "gbk", "ascii"]
            }
        },
        "required": ["path"],
        "additionalProperties": False
    }
}

# 输入验证函数
def validate_tool_input(tool: dict, input_data: dict) -> bool:
    """验证工具输入是否符合JSON Schema"""
    try:
        validate(instance=input_data, schema=tool["parameters"])
        return True
    except ValidationError as e:
        print(f"输入验证失败: {e.message}")
        print(f"失败路径: {list(e.absolute_path)}")
        return False

# 示例用法

# 有效输入
valid_input = {
    "path": "/data/sample.txt",
    "encoding": "utf-8"
}

# 无效输入1：路径遍历
invalid_input1 = {
    "path": "/data/../etc/passwd",
    "encoding": "utf-8"
}

# 无效输入2：额外属性
invalid_input2 = {
    "path": "/data/sample.txt",
    "encoding": "utf-8",
    "malicious_param": "rm -rf /"
}

# 无效输入3：无效编码
invalid_input3 = {
    "path": "/data/sample.txt",
    "encoding": "invalid_encoding"
}

# 验证测试
print(f"有效输入验证结果: {validate_tool_input(file_reader_tool, valid_input)}")
print(f"无效输入1验证结果: {validate_tool_input(file_reader_tool, invalid_input1)}")
print(f"无效输入2验证结果: {validate_tool_input(file_reader_tool, invalid_input2)}")
print(f"无效输入3验证结果: {validate_tool_input(file_reader_tool, invalid_input3)}")

# MCP性能优化实现
import asyncio
import aiohttp
import json
from typing import Dict, Any, List
from cachetools import TTLCache

class MCPClient:
    """优化的MCP Client，支持异步和缓存"""
    
    def __init__(self, server_url: str, cache_size: int = 100, cache_ttl: int = 300):
        self.server_url = server_url
        self.session = None
        # 工具调用结果缓存，TTL为5分钟
        self.cache = TTLCache(maxsize=cache_size, ttl=cache_ttl)
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.close()
    
    def _get_cache_key(self, tool_name: str, tool_version: str, parameters: Dict[str, Any]) -> str:
        """生成缓存键"""
        return json.dumps({
            "tool_name": tool_name,
            "tool_version": tool_version,
            "parameters": parameters
        }, sort_keys=True)
    
    async def execute_tool(self, tool_name: str, tool_version: str, 
                         parameters: Dict[str, Any], user_id: str, 
                         session_id: str, cache: bool = True) -> Dict[str, Any]:
        """执行工具，支持缓存"""
        # 检查缓存
        cache_key = self._get_cache_key(tool_name, tool_version, parameters)
        if cache and cache_key in self.cache:
            return {
                **self.cache[cache_key],
                "from_cache": True
            }
        
        # 构建请求
        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.server_url}/execute"
        async with self.session.post(url, json=request) as response:
            result = await response.json()
        
        # 缓存结果
        if cache and result.get("status") == "success":
            self.cache[cache_key] = result
        
        return result
    
    async def execute_multiple_tools(self, tool_calls: List[Dict[str, Any]], 
                                   user_id: str, session_id: str) -> List[Dict[str, Any]]:
        """并行执行多个工具，提高性能"""
        tasks = []
        for call in tool_calls:
            task = self.execute_tool(
                call["tool_name"],
                call["tool_version"],
                call["parameters"],
                user_id,
                session_id
            )
            tasks.append(task)
        
        # 并行执行所有任务
        return await asyncio.gather(*tasks)

# 示例用法
async def main():
    async with MCPClient("http://localhost:8000") as client:
        # 并行执行多个工具调用
        tool_calls = [
            {
                "tool_name": "file_reader",
                "tool_version": "1.0",
                "parameters": {"path": "/data/file1.txt"}
            },
            {
                "tool_name": "file_reader",
                "tool_version": "1.0",
                "parameters": {"path": "/data/file2.txt"}
            },
            {
                "tool_name": "weather_api",
                "tool_version": "1.0",
                "parameters": {"location": "北京", "unit": "celsius"}
            }
        ]
        
        # 第一次调用，没有缓存
        results1 = await client.execute_multiple_tools(tool_calls, "user_001", "session_123")
        print(f"第一次调用结果（无缓存）: {results1}")
        
        # 第二次调用，使用缓存
        results2 = await client.execute_multiple_tools(tool_calls, "user_001", "session_123")
        print(f"第二次调用结果（有缓存）: {results2}")
        
        # 检查缓存命中率
        print(f"缓存命中率: {len(client.cache)}/3")

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