MCP 在 Agent 架构中的位置

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

class MCPAgentClient:
    def __init__(self, server_url: str = "http://localhost:8000/mcp"):
        self.server_url = server_url
        self.client = httpx.AsyncClient(timeout=30.0)
        self.session_id = None
        
    async def initialize(self):
        """初始化 MCP Client，建立会话"""
        response = await self.client.post(
            f"{self.server_url}/v2/sessions",
            json={"client_info": {"name": "MCP Agent Client", "version": "2.0"}}
        )
        response.raise_for_status()
        result = response.json()
        self.session_id = result["session_id"]
        return result
    
    async def call_tool(self, tool_name: str, arguments: Dict[str, Any], context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """调用指定工具"""
        if not self.session_id:
            await self.initialize()
        
        payload = {
            "tool_name": tool_name,
            "arguments": arguments,
            "context": context or {},
            "session_id": self.session_id
        }
        
        try:
            response = await self.client.post(
                f"{self.server_url}/v2/tool/call",
                json=payload
            )
            response.raise_for_status()
            return response.json()
        except httpx.HTTPError as e:
            print(f"工具调用失败: {e}")
            raise
        except Exception as e:
            print(f"MCP Client 调用失败: {e}")
            raise
    
    async def get_tool_list(self) -> List[Dict[str, Any]]:
        """获取可用工具列表"""
        response = await self.client.get(f"{self.server_url}/v2/tools")
        response.raise_for_status()
        return response.json().get("tools", [])
    
    async def close(self):
        """关闭 MCP Client，清理资源"""
        if self.session_id:
            await self.client.delete(f"{self.server_url}/v2/sessions/{self.session_id}")
        await self.client.aclose()

# 使用示例
async def main():
    # 创建 MCP Client
    mcp_client = MCPAgentClient()
    
    try:
        # 初始化客户端
        await mcp_client.initialize()
        
        # 获取可用工具列表
        tools = await mcp_client.get_tool_list()
        print(f"可用工具: {[tool['name'] for tool in tools]}")
        
        # 调用示例工具
        result = await mcp_client.call_tool(
            tool_name="weather_api",
            arguments={"city": "北京", "date": "2026-01-01"}
        )
        print(f"工具调用结果: {json.dumps(result, indent=2, ensure_ascii=False)}")
    finally:
        # 关闭客户端
        await mcp_client.close()

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

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

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

# 工具注册表
tools_registry = {}

# 会话管理
sessions = {}

# 请求模型
class ToolCallRequest(BaseModel):
    tool_name: str
    arguments: Dict[str, Any]
    context: Optional[Dict[str, Any]] = None
    session_id: str

class ToolRegisterRequest(BaseModel):
    name: str
    description: str
    parameters: Dict[str, Any]
    implementation: Dict[str, Any]

class SessionCreateRequest(BaseModel):
    client_info: Dict[str, Any]

# 注册示例工具
def register_example_tools():
    """注册示例工具"""
    # 天气 API 工具
    tools_registry["weather_api"] = {
        "name": "weather_api",
        "description": "获取指定城市和日期的天气信息",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称"
                },
                "date": {
                    "type": "string",
                    "description": "日期，格式：YYYY-MM-DD"
                }
            },
            "required": ["city", "date"]
        },
        "implementation": {
            "type": "local",
            "function": lambda city, date: {
                "temperature": 22,
                "humidity": 45,
                "condition": "晴",
                "city": city,
                "date": date
            }
        }
    }
    
    # 计算器工具
    tools_registry["calculator"] = {
        "name": "calculator",
        "description": "执行简单的数学计算",
        "parameters": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "数学表达式，如：1+2*3"
                }
            },
            "required": ["expression"]
        },
        "implementation": {
            "type": "local",
            "function": lambda expression: {
                "result": eval(expression),
                "expression": expression
            }
        }
    }

# 会话管理@app.post("/v2/sessions")async def create_session(request: SessionCreateRequest):
    """创建新会话"""
    session_id = f"session_{len(sessions) + 1}"
    sessions[session_id] = {
        "id": session_id,
        "client_info": request.client_info,
        "created_at": "2026-01-01T00:00:00Z",
        "last_used": "2026-01-01T00:00:00Z"
    }
    return {"session_id": session_id}

@app.delete("/v2/sessions/{session_id}")async def delete_session(session_id: str):
    """删除会话"""
    if session_id not in sessions:
        raise HTTPException(status_code=404, detail="Session not found")
    del sessions[session_id]
    return {"message": "Session deleted successfully"}

# 工具管理@app.get("/v2/tools")async def list_tools():
    """列出所有可用工具"""
    return {"tools": list(tools_registry.values())}

@app.post("/v2/tools")async def register_tool(request: ToolRegisterRequest):
    """注册新工具"""
    tools_registry[request.name] = {
        "name": request.name,
        "description": request.description,
        "parameters": request.parameters,
        "implementation": request.implementation
    }
    return {"message": "Tool registered successfully", "tool": tools_registry[request.name]}

# 工具调用@app.post("/v2/tool/call")async def call_tool(request: ToolCallRequest):
    """调用指定工具"""
    # 验证会话
    if request.session_id not in sessions:
        raise HTTPException(status_code=401, detail="Invalid session")
    
    # 更新会话最后使用时间
    sessions[request.session_id]["last_used"] = "2026-01-01T00:00:00Z"
    
    # 验证工具
    if request.tool_name not in tools_registry:
        raise HTTPException(status_code=404, detail="Tool not found")
    
    tool = tools_registry[request.tool_name]
    
    try:
        # 执行工具调用
        if tool["implementation"]["type"] == "local":
            # 本地工具实现
            result = tool["implementation"]["function"](**request.arguments)
        else:
            # 远程工具实现（简化处理）
            result = {"error": "Remote tool implementation not supported in this example"}
        
        # 返回结果
        return {
            "success": True,
            "result": result,
            "tool_name": request.tool_name,
            "session_id": request.session_id
        }
    except Exception as e:
        return {
            "success": False,
            "error": str(e),
            "tool_name": request.tool_name,
            "session_id": request.session_id
        }

# 启动服务器
if __name__ == "__main__":
    # 注册示例工具
    register_example_tools()
    
    # 启动服务器
    uvicorn.run(app, host="0.0.0.0", port=8000)

from typing import Dict, Any, Optional, List
import asyncio
from mcp_client import MCPAgentClient
from llm_client import LLMClient  # 假设的 LLM 客户端
from memory_manager import MemoryManager  # 假设的记忆管理器

class MCPAgentController:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.mcp_client = MCPAgentClient(config.get("mcp_server_url", "http://localhost:8000/mcp"))
        self.llm_client = LLMClient(config.get("llm_config", {}))
        self.memory_manager = MemoryManager(config.get("memory_config", {}))
        self.state = "idle"  # idle, thinking, planning, executing
    
    async def initialize(self):
        """初始化 Agent 控制器"""
        await self.mcp_client.initialize()
        await self.llm_client.initialize()
        await self.memory_manager.initialize()
        self.state = "idle"
    
    async def handle_request(self, user_request: str, context: Optional[Dict[str, Any]] = None) -> str:
        """处理用户请求"""
        self.state = "thinking"
        
        try:
            # 1. 检索相关记忆
            relevant_memories = await self.memory_manager.retrieve_memories(user_request)
            
            # 2. 构建完整上下文
            full_context = {
                "user_request": user_request,
                "relevant_memories": relevant_memories,
                "system_prompt": self.config.get("system_prompt", "你是一个智能助手，可以使用工具帮助用户解决问题。")
            }
            
            # 3. 模型推理，生成工具调用请求
            self.state = "planning"
            model_response = await self.llm_client.generate(full_context)
            
            # 4. 解析模型响应，提取工具调用请求
            tool_calls = self._parse_tool_calls(model_response)
            
            if tool_calls:
                # 5. 执行工具调用
                self.state = "executing"
                tool_results = []
                
                for tool_call in tool_calls:
                    # 调用 MCP Client 执行工具
                    result = await self.mcp_client.call_tool(
                        tool_name=tool_call["name"],
                        arguments=tool_call["arguments"],
                        context=context
                    )
                    tool_results.append(result)
                
                # 6. 结合工具结果，生成最终响应
                self.state = "thinking"
                final_context = {
                    **full_context,
                    "tool_results": tool_results
                }
                final_response = await self.llm_client.generate(final_context)
            else:
                # 7. 直接生成最终响应
                final_response = model_response
            
            # 8. 存储对话记忆
            await self.memory_manager.store_conversation(
                user_request=user_request,
                assistant_response=final_response,
                tool_calls=tool_calls,
                tool_results=tool_results if tool_calls else []
            )
            
            self.state = "idle"
            return final_response
        except Exception as e:
            self.state = "idle"
            print(f"Agent 处理请求失败: {e}")
            raise
    
    def _parse_tool_calls(self, model_response: str) -> List[Dict[str, Any]]:
        """解析模型响应，提取工具调用请求"""
        # 这里简化实现，实际应根据模型输出格式进行解析
        # 示例：假设模型输出 JSON 格式的工具调用请求
        import json
        try:
            response_data = json.loads(model_response)
            if "tool_calls" in response_data:
                return response_data["tool_calls"]
            return []
        except:
            return []
    
    async def get_agent_status(self) -> Dict[str, Any]:
        """获取 Agent 状态"""
        return {
            "state": self.state,
            "config": self.config,
            "mcp_status": "connected" if self.mcp_client.session_id else "disconnected",
            "llm_status": "initialized" if self.llm_client.initialized else "not_initialized",
            "memory_status": "initialized" if self.memory_manager.initialized else "not_initialized"
        }
    
    async def close(self):
        """关闭 Agent 控制器，清理资源"""
        await self.mcp_client.close()
        await self.llm_client.close()
        await self.memory_manager.close()
        self.state = "idle"

# 使用示例
async def main():
    # Agent 配置
    config = {
        "mcp_server_url": "http://localhost:8000/mcp",
        "llm_config": {
            "model_name": "gpt-4o",
            "temperature": 0.1
        },
        "memory_config": {
            "vector_db_url": "http://localhost:3000",
            "similarity_threshold": 0.8
        },
        "system_prompt": "你是一个智能助手，可以使用工具帮助用户解决问题。请根据用户请求，生成适当的工具调用或直接回答用户问题。"
    }
    
    # 创建 Agent 控制器
    agent_controller = MCPAgentController(config)
    
    try:
        # 初始化 Agent
        await agent_controller.initialize()
        
        # 获取 Agent 状态
        status = await agent_controller.get_agent_status()
        print(f"Agent 状态: {status}")
        
        # 处理用户请求
        user_request = "北京今天的天气怎么样？"
        response = await agent_controller.handle_request(user_request)
        print(f"用户请求: {user_request}")
        print(f"Agent 响应: {response}")
        
        # 处理另一个请求
        user_request = "1+2*3等于多少？"
        response = await agent_controller.handle_request(user_request)
        print(f"用户请求: {user_request}")
        print(f"Agent 响应: {response}")
    finally:
        # 关闭 Agent
        await agent_controller.close()

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

# MCP Agent 架构配置
agent_config:
  # 基本配置
  name: "MCP Agent"
  version: "2.0"
  description: "基于 MCP v2.0 的智能 Agent 系统"
  
  # 系统提示词
  system_prompt: "你是一个智能助手，可以使用工具帮助用户解决问题。请根据用户请求，生成适当的工具调用或直接回答用户问题。"
  
  # MCP 配置
  mcp:
    server_url: "http://localhost:8000/mcp"
    client_config:
      timeout: 30.0
      retry_count: 3
      retry_delay: 1.0
      connection_pool_size: 10
  
  # LLM 配置
  llm:
    type: "openai"
    model_name: "gpt-4o"
    api_key: "${OPENAI_API_KEY}"
    temperature: 0.1
    max_tokens: 2048
  
  # 记忆管理配置
  memory:
    type: "vector_db"
    url: "http://localhost:3000"
    similarity_threshold: 0.8
    max_memory_count: 1000
    memory_ttl: 3600  # 秒
  
  # API 网关配置
  api_gateway:
    host: "0.0.0.0"
    port: 8080
    cors: true
    auth: false  # 开发环境关闭认证
  
  # 安全配置
  security:
    enable_rate_limiting: true
    rate_limit: "100 requests per minute"
    enable_ip_filtering: false
    allowed_ips: []
  
  # 日志配置
  logging:
    level: "INFO"
    format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    file: "/var/log/mcp_agent.log"
  
  # 监控配置
  monitoring:
    enable_prometheus: true
    prometheus_port: 9090
    enable_tracing: true
    tracing_exporter: "jaeger"
    tracing_url: "http://localhost:14268/api/traces"
  
  # 部署配置
  deployment:
    mode: "single"  # single, distributed
    replicas: 1
    resource_limits:
      cpu: "1"
      memory: "2Gi"