🔌 MCP 详解

┌─────────────────┐         ┌──────────────────┐
│                 │         │                  │
│   AI Model      │ ◄─────► │   MCP Server     │
│   (Client)      │  MCP    │   (Tools/Data)   │
│                 │ 协议     │                  │
└─────────────────┘         └──────────────────┘
        │                           │
        │ 工具调用                   │ 提供能力
        ▼                           ▼
┌─────────────────┐         ┌──────────────────┐
│   Skills/Agents │         │  Database / API  │
│                 │         │  / File System   │
└─────────────────┘         └──────────────────┘

# 改造前：有状态MCP Server（❌ 不兼容）
class MyMCPServer:
    def __init__(self):
        self.session_state = {}  # 维护状态
    
    def handle_request(self, request):
        user_id = request.user_id
        # 依赖之前保存的状态
        prev_context = self.session_state.get(user_id, {})
        result = process_with_context(request, prev_context)
        self.session_state[user_id] = result.context  # 更新状态
        return result

# 改造后：无状态MCP Server（✅ 兼容）
class MyStatelessMCPServer:
    def handle_request(self, request):
        # 所有上下文都在请求中
        context = request.context  # 由Client传递
        result = process_with_context(request.data, context)
        # 返回新的上下文给Client
        return {
            "result": result.value,
            "next_context": result.next_context  # Client保存
        }

# openclaw/config.yaml 中的MCP配置

mcp_servers:
  # RSS聚合器
  rss_aggregator:
    type: "mcp"
    endpoint: "http://localhost:3100/mcp/rss"
    timeout: 30s
    stateless: true  # 无状态化
    capabilities:
      - fetch_blog
      - fetch_news
      - fetch_trending
  
  # 内容生成器  
  content_generator:
    type: "mcp"
    endpoint: "http://localhost:3101/mcp/content"
    timeout: 60s
    stateless: true
    capabilities:
      - generate_daily_report
      - generate_glossary
  
  # 通知服务
  notifier:
    type: "mcp"
    endpoint: "http://localhost:3102/mcp/notify"
    timeout: 10s
    stateless: true
    capabilities:
      - send_feishu_message
      - send_discord_message

# Skill通过MCP调用
skills:
  rss_aggregation:
    mcp_server: "rss_aggregator"
    tools: ["fetch_blog", "fetch_news"]
  
  report_generation:
    mcp_server: "content_generator"
    tools: ["generate_daily_report"]

"""
妙趣AI - 自定义MCP Server示例
实现了无状态的RSS聚合工具
"""

from datetime import datetime
from typing import Dict, Any
import json, httpx

class MCPServer:
    """MCP Server基础类"""
    
    def __init__(self, name: str, version: str = "1.0.0"):
        self.name = name
        self.version = version
        self.tools = {}  # 注册的工具
    
    def register_tool(self, name: str, handler, description: str, params: list):
        """注册工具"""
        self.tools[name] = {
            "handler": handler,
            "description": description,
            "parameters": params
        }
    
    def discover(self) -> Dict:
        """能力发现接口（Client查询可用工具）"""
        return {
            "server": self.name,
            "version": self.version,
            "tools": {
                name: {
                    "description": info["description"],
                    "parameters": info["parameters"]
                }
                for name, info in self.tools.items()
            }
        }
    
    def handle_request(self, request: Dict) -> Dict:
        """处理MCP请求（无状态）"""
        tool_name = request.get("tool")
        params = request.get("params", {})
        context = request.get("context", {})  # Client传递的上下文
        
        if tool_name not in self.tools:
            return {"error": f"Unknown tool: {tool_name}"}
        
        try:
            result, next_context = self.tools[tool_name]["handler"](params, context)
            return {
                "result": result,
                "next_context": next_context  # 返回给Client
            }
        except Exception as e:
            return {"error": str(e)}


# 创建RSS聚合MCP Server
rss_server = MCPServer("rss-aggregator", "1.0.0")

def fetch_rss(params, context):
    """获取RSS内容（无状态）"""
    sources = params.get("sources", [
        "https://openclaw.ai/blog/feed.xml",
        "https://hnrss.org/frontpage"
    ])
    
    results = []
    for source in sources:
        # 模拟API调用
        results.append({
            "source": source,
            "count": 10,
            "articles": []
        })
    
    # 无状态：返回下次需要的上下文
    next_context = {
        "last_fetch": datetime.now().isoformat(),
        "sources_fetched": sources
    }
    
    return results, next_context

rss_server.register_tool(
    "fetch_rss",
    fetch_rss,
    "聚合多个RSS源的内容",
    [
        {"name": "sources", "type": "array", "required": False},
        {"name": "max_count", "type": "number", "required": False}
    ]
)


# 处理请求示例
def process_mcp_request():
    """完整的MCP请求处理流程"""
    
    # Step 1: 能力发现
    print("🔍 发现能力...")
    capabilities = rss_server.discover()
    print(f"   可用工具: {list(capabilities['tools'].keys())}")
    
    # Step 2: 构造请求（包含上下文）
    print("📤 发送请求...")
    request = {
        "tool": "fetch_rss",
        "params": {
            "sources": ["https://openclaw.ai/blog"],
            "max_count": 5
        },
        "context": {}  # 首次请求，上下文为空
    }
    
    # Step 3: 处理请求
    print("⚙️ 处理中...")
    response = rss_server.handle_request(request)
    
    # Step 4: 处理结果
    print(f"✅ 结果: {response['result']}")
    print(f"📦 下次上下文: {response['next_context']}")
    
    # Step 5: 第二次请求（带上上下文）
    request2 = {
        "tool": "fetch_rss",
        "params": {"sources": ["https://hnrss.org/frontpage"]},
        "context": response['next_context']  # 携带上次的上下文
    }
    
    response2 = rss_server.handle_request(request2)
    print(f"✅ 第二次请求完成")

# 执行
if __name__ == "__main__":
    process_mcp_request()