MCP Client 与 LLM 的协作流程

import asyncio
import httpx
from typing import Dict, Any, Optional

class LLMConnector:
    def __init__(self, api_key: str, base_url: str, model: str):
        self.api_key = api_key
        self.base_url = base_url
        self.model = model
        self.client = httpx.AsyncClient(timeout=30.0)
    
    async def generate(self, prompt: str, tools: Optional[list] = None, **kwargs) -> Dict[str, Any]:
        """调用 LLM API 生成响应"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": self.model,
            "messages": [{"role": "user", "content": prompt}],
            **kwargs
        }
        
        # 如果提供了工具定义，添加到 payload
        if tools:
            payload["tools"] = tools
            payload["tool_choice"] = "auto"
        
        try:
            response = await self.client.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload
            )
            response.raise_for_status()
            return response.json()
        except httpx.HTTPError as e:
            print(f"LLM API 调用失败: {e}")
            raise
        except Exception as e:
            print(f"LLM 生成失败: {e}")
            raise
    
    async def close(self):
        """关闭 HTTP 客户端"""
        await self.client.aclose()

from typing import Dict, List, Any

class PromptGenerator:
    def __init__(self, system_prompt: Optional[str] = None):
        self.system_prompt = system_prompt or "你是一个助手，可以使用工具来帮助用户解决问题。"
    
    def generate_prompt(self, user_query: str, tools: List[Dict[str, Any]], context: Optional[List[Dict[str, Any]]] = None) -> str:
        """根据 MCP Tool Schema 生成优化的 Prompt"""
        # 构建工具描述部分
        tools_desc = "可用工具：\n\n"
        for tool in tools:
            tools_desc += f"## {tool['name']}\n"
            tools_desc += f"**描述：** {tool['description']}\n"
            tools_desc += f"**参数：**\n"
            
            if isinstance(tool['parameters'], dict):
                properties = tool['parameters'].get('properties', {})
                required = tool['parameters'].get('required', [])
                
                for param_name, param_schema in properties.items():
                    required_mark = "*" if param_name in required else ""
                    param_type = param_schema.get('type', 'string')
                    param_desc = param_schema.get('description', '')
                    tools_desc += f"  - {param_name}{required_mark}: {param_type} - {param_desc}\n"
        
        # 构建上下文部分
        context_desc = """
对话历史：
"""
        if context:
            for msg in context:
                role = msg.get('role', 'user')
                content = msg.get('content', '')
                context_desc += f"{role}: {content}\n"
        
        # 构建最终 Prompt
        prompt = f"""{self.system_prompt}

{tools_desc}

{context_desc}

请根据用户请求和可用工具，生成适当的工具调用或直接回答用户问题。

用户请求：
{user_query}

工具调用格式（JSON）：
```json
{{
  "tool_call": {{
    "name": "工具名称",
    "arguments": {{
      "参数名": "参数值"
    }}
  }}
}}

{{
  "content": "直接回答内容"
}}

    return prompt

**代码解析**：
- 根据 MCP Tool Schema 自动生成结构化的工具描述
- 支持对话上下文的集成
- 提供清晰的工具调用格式说明
- 支持自定义系统 Prompt

#### 3.3.3 Response Parser 实现

Response Parser 负责解析 LLM 响应，提取工具调用信息。

**代码示例 3：智能 LLM 响应解析器实现**

```python
import json
from typing import Dict, Any, Optional, Tuple

class ResponseParser:
    def parse_response(self, response: Dict[str, Any]) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
        """解析 LLM 响应，提取工具调用或直接回答"""
        try:
            # 检查是否是 OpenAI 格式的响应
            if "choices" in response:
                choice = response["choices"][0]
                
                # 检查是否有工具调用
                if hasattr(choice, "message") and hasattr(choice.message, "tool_calls") and choice.message.tool_calls:
                    tool_call = choice.message.tool_calls[0]
                    return {
                        "name": tool_call.function.name,
                        "arguments": json.loads(tool_call.function.arguments)
                    }, None
                
                # 直接回答
                elif hasattr(choice, "message") and hasattr(choice.message, "content"):
                    return None, choice.message.content
            
            # 检查是否是 Claude 格式的响应
            elif "content" in response:
                content = response["content"]
                
                # 尝试解析 JSON 格式的工具调用
                if content.strip().startswith("{"):
                    parsed_content = json.loads(content)
                    if "tool_call" in parsed_content:
                        return parsed_content["tool_call"], None
                    elif "content" in parsed_content:
                        return None, parsed_content["content"]
                
                # 直接文本回答
                return None, content
            
            # 未知格式
            return None, str(response)
            
        except Exception as e:
            print(f"解析 LLM 响应失败: {e}")
            # 尝试提取原始文本作为后备
            raw_content = str(response)
            return None, raw_content

from typing import List, Dict, Any
import hashlib

class ContextManager:
    def __init__(self, max_history_length: int = 10, max_context_tokens: int = 8192):
        self.max_history_length = max_history_length
        self.max_context_tokens = max_context_tokens
        self.context: List[Dict[str, Any]] = []
    
    def add_message(self, role: str, content: str) -> None:
        """添加消息到上下文"""
        self.context.append({
            "role": role,
            "content": content,
            "timestamp": self._get_timestamp()
        })
        
        # 裁剪上下文，保持在最大长度以内
        self._trim_context()
    
    def get_context(self, include_tool_calls: bool = True) -> List[Dict[str, Any]]:
        """获取当前上下文"""
        if include_tool_calls:
            return self.context.copy()
        else:
            # 过滤掉工具调用相关消息
            return [msg for msg in self.context if "tool_call" not in msg]
    
    def clear_context(self) -> None:
        """清空上下文"""
        self.context = []
    
    def _trim_context(self) -> None:
        """裁剪上下文，保持在最大长度和令牌数以内"""
        # 1. 按长度裁剪
        if len(self.context) > self.max_history_length:
            self.context = self.context[-self.max_history_length:]
        
        # 2. 按令牌数裁剪（简化实现，实际应使用更精确的令牌计数）
        total_tokens = 0
        trimmed_context = []
        
        for msg in reversed(self.context):
            msg_tokens = self._estimate_tokens(msg["content"])
            if total_tokens + msg_tokens <= self.max_context_tokens:
                trimmed_context.insert(0, msg)
                total_tokens += msg_tokens
            else:
                break
        
        self.context = trimmed_context
    
    def _estimate_tokens(self, text: str) -> int:
        """估算文本的令牌数（简化实现）"""
        return len(text) // 4  # 粗略估算：1 令牌 ≈ 4 字符
    
    def _get_timestamp(self) -> str:
        """获取当前时间戳"""
        from datetime import datetime
        return datetime.now().isoformat()
    
    def get_context_hash(self) -> str:
        """获取上下文的哈希值，用于缓存和比较"""
        context_str = json.dumps(self.context, sort_keys=True)
        return hashlib.sha256(context_str.encode()).hexdigest()

# MCP Client 配置
client:
  name: "my-mcp-client"
  version: "2.0.0"
  timeout: 30.0
  retry_count: 3
  retry_delay: 1.0

# LLM 配置
llm:
  primary:
    type: "openai"
    api_key: "${OPENAI_API_KEY}"
    base_url: "https://api.openai.com/v1"
    model: "gpt-4o"
    temperature: 0.1
  secondary:
    type: "anthropic"
    api_key: "${ANTHROPIC_API_KEY}"
    base_url: "https://api.anthropic.com/v1"
    model: "claude-3-opus-20240229"
    temperature: 0.1
  local:
    type: "ollama"
    base_url: "http://localhost:11434/v1"
    model: "llama3:70b"
    temperature: 0.7

# 上下文管理配置
context:
  max_history_length: 20
  max_context_tokens: 8192
  include_tool_calls: true
  trim_strategy: "both"  # both, length, tokens

# 工具配置
tools:
  - name: "web_search"
    description: "搜索互联网获取最新信息"
    parameters:
      type: "object"
      properties:
        query:
          type: "string"
          description: "搜索查询词"
        num_results:
          type: "integer"
          description: "返回结果数量"
          default: 5
      required: ["query"]
  - name: "calculator"
    description: "执行数学计算"
    parameters:
      type: "object"
      properties:
        expression:
          type: "string"
          description: "数学表达式"
      required: ["expression"]
  - name: "code_interpreter"
    description: "执行代码并返回结果"
    parameters:
      type: "object"
      properties:
        language:
          type: "string"
          description: "代码语言"
          enum: ["python", "javascript", "java", "c++"]
        code:
          type: "string"
          description: "要执行的代码"
      required: ["language", "code"]

# 日志配置
logging:
  level: "INFO"
  format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
  file: "mcp-client.log"

# 监控配置
monitoring:
  enabled: true
  metrics_port: 9090
  tracing_enabled: false

import asyncio
import json
from typing import Dict, Any, List, Optional
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import uvicorn

# 导入之前实现的组件
from mcp_client.llm_connector import LLMConnector
from mcp_client.prompt_generator import PromptGenerator
from mcp_client.response_parser import ResponseParser
from mcp_client.context_manager import ContextManager
from mcp_client.mcp_executor import MCPExecutor

# 初始化 FastAPI 应用
app = FastAPI(title="MCP Client LLM Collaboration Service", version="2.0.0")

# 配置管理
class Config(BaseModel):
    openai_api_key: str
    anthropic_api_key: str
    ollama_base_url: str
    mcp_server_url: str

# 依赖注入：配置
async def get_config():
    return Config(
        openai_api_key="${OPENAI_API_KEY}",
        anthropic_api_key="${ANTHROPIC_API_KEY}",
        ollama_base_url="http://localhost:11434/v1",
        mcp_server_url="http://localhost:8000/mcp"
    )

# 依赖注入：MCP Client 组件
async def get_mcp_client_components(config: Config = Depends(get_config)):
    # 初始化各组件
    llm_connector = LLMConnector(
        api_key=config.openai_api_key,
        base_url="https://api.openai.com/v1",
        model="gpt-4o"
    )
    
    prompt_generator = PromptGenerator()
    response_parser = ResponseParser()
    context_manager = ContextManager(max_history_length=20, max_context_tokens=8192)
    mcp_executor = MCPExecutor(server_url=config.mcp_server_url)
    
    return {
        "llm_connector": llm_connector,
        "prompt_generator": prompt_generator,
        "response_parser": response_parser,
        "context_manager": context_manager,
        "mcp_executor": mcp_executor
    }

# 请求模型
class CollaborationRequest(BaseModel):
    user_query: str
    context: Optional[List[Dict[str, Any]]] = None
    tools: Optional[List[Dict[str, Any]]] = None
    llm_model: Optional[str] = "gpt-4o"

# 响应模型
class CollaborationResponse(BaseModel):
    result: str
    context: List[Dict[str, Any]]
    tool_calls: List[Dict[str, Any]] = []
    status: str

# 主协作接口
@app.post("/collaborate", response_model=CollaborationResponse)
async def collaborate(
    request: CollaborationRequest,
    components: Dict[str, Any] = Depends(get_mcp_client_components)
):
    """MCP Client 与 LLM 协作的主接口"""
    try:
        llm_connector = components["llm_connector"]
        prompt_generator = components["prompt_generator"]
        response_parser = components["response_parser"]
        context_manager = components["context_manager"]
        mcp_executor = components["mcp_executor"]
        
        # 1. 初始化上下文
        if request.context:
            for msg in request.context:
                context_manager.add_message(msg["role"], msg["content"])
        
        # 2. 获取可用工具
        available_tools = request.tools or await mcp_executor.list_tools()
        
        # 3. 生成优化的 Prompt
        prompt = prompt_generator.generate_prompt(
            user_query=request.user_query,
            tools=available_tools,
            context=context_manager.get_context()
        )
        
        # 4. 调用 LLM
        llm_response = await llm_connector.generate(prompt=prompt, tools=available_tools)
        
        # 5. 解析 LLM 响应
        tool_call, direct_response = response_parser.parse_response(llm_response)
        
        tool_calls = []
        final_response = ""
        
        if tool_call:
            # 6. 执行工具调用
            tool_calls.append(tool_call)
            context_manager.add_message("assistant", f"调用工具：{tool_call['name']}")
            
            # 执行工具
            tool_result = await mcp_executor.execute_tool(
                tool_name=tool_call["name"],
                arguments=tool_call["arguments"]
            )
            
            # 7. 将工具结果返回给 LLM，生成最终响应
            context_manager.add_message("tool", f"工具 {tool_call['name']} 执行结果：{json.dumps(tool_result)}")
            
            # 生成第二轮 Prompt
            followup_prompt = prompt_generator.generate_prompt(
                user_query=f"基于工具执行结果，回答用户原始问题：{request.user_query}",
                tools=available_tools,
                context=context_manager.get_context()
            )
            
            # 调用 LLM 生成最终响应
            followup_llm_response = await llm_connector.generate(prompt=followup_prompt)
            _, final_response = response_parser.parse_response(followup_llm_response)
        else:
            # 直接使用 LLM 响应
            final_response = direct_response
        
        # 8. 更新上下文
        context_manager.add_message("user", request.user_query)
        context_manager.add_message("assistant", final_response)
        
        # 9. 返回结果
        return CollaborationResponse(
            result=final_response,
            context=context_manager.get_context(),
            tool_calls=tool_calls,
            status="success"
        )
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"协作失败：{str(e)}")

# 健康检查接口
@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "MCP Client LLM Collaboration Service"}

# 启动服务
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)