OpenClaw MCP 与扩展 —— Server 配置、工具扩展、结果处理

┌──────────────────────────────────────────────────────────┐
│                    MCP 架构概览                            │
│                                                          │
│   AI Client (OpenClaw)                                   │
│         │                                                │
│         │ MCP Protocol                                   │
│         ▼                                                │
│   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐     │
│   │ MCP Server  │  │ MCP Server  │  │ MCP Server  │     │
│   │ (文件系统)   │  │ (数据库)    │  │ (API)       │     │
│   └─────────────┘  └─────────────┘  └─────────────┘     │
│         │                │                │             │
│         ▼                ▼                ▼             │
│   本地文件系统      PostgreSQL      第三方 API          │
│                                                          │
│   标准化接口：Tools / Resources / Prompts                 │
└──────────────────────────────────────────────────────────┘

Tools（工具）
├── 可被 AI 调用的操作
├── 有输入参数和输出结果
├── 示例：execute_query, send_email, search_docs
│
Resources（资源）
├── 可被 AI 读取的数据源
├── 有 URI 标识
├── 示例：file:///path/to/config.yaml, db://users/123
│
Prompts（提示词）
├── 预定义的提示词模板
├── 可带参数
├── 示例：code_review_template, bug_report_template

┌──────────────────────────────────────────────────────────┐
│              OpenClaw + MCP 的价值                         │
│                                                          │
│  没有 MCP 的 OpenClaw：                                    │
│  ├── 能读写本地文件 ✅                                     │
│  ├── 能执行受限 Shell 命令 ✅                               │
│  ├── 能生成和修改代码 ✅                                    │
│  └── 但无法连接外部服务 ❌                                  │
│                                                          │
│  有 MCP 的 OpenClaw：                                     │
│  ├── 查询数据库 ✅                                         │
│  ├── 调用外部 API ✅                                       │
│  ├── 搜索文档/知识库 ✅                                    │
│  ├── 发送通知（Slack/邮件）✅                               │
│  ├── 操作云资源（AWS/GCP）✅                                │
│  └── 连接任何 MCP Server ✅                                │
└──────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│                  MCP vs 传统插件                          │
├──────────────────┬──────────────────┬────────────────────┤
│ 维度             │ MCP              │ 传统插件           │
├──────────────────┼──────────────────┼────────────────────┤
│ 标准化           │ ✅ 统一协议       │ ❌ 每个工具不同     │
│ 跨平台           │ ✅ 语言无关       │ ❌ 通常绑定语言     │
│ 发现机制         │ ✅ 自动发现       │ ❌ 手动配置         │
│ 安全性           │ ✅ 权限隔离       │ ⚠️ 取决于实现      │
│ 组合性           │ ✅ 多 Server 组合 │ ⚠️ 有限            │
│ 调试             │ ✅ 标准调试工具   │ ❌ 工具各异         │
└──────────────────┴──────────────────┴────────────────────┘

# ~/.config/openclaw/mcp.yaml
# OpenClaw 的 MCP Server 配置

mcp:
  # 是否启用 MCP
  enabled: true

  # 超时设置（秒）
  timeout: 30

  # 最大并发连接数
  max_connections: 5

  # Server 配置列表
  servers:
    # 文件系统 Server（内置）
    filesystem:
      enabled: true
      type: builtin
      config:
        allowed_paths:
          - ~/projects
          - ~/documents
        blocked_paths:
          - ~/.ssh
          - ~/.gnupg

    # 数据库 Server
    database:
      enabled: true
      type: stdio
      command: npx
      args:
        - -y
        - @modelcontextprotocol/server-postgres
        - postgresql://localhost:5432/mydb
      env:
        DATABASE_URL: postgresql://localhost:5432/mydb

    # 自定义 API Server
    weather-api:
      enabled: true
      type: stdio
      command: python
      args:
        - /opt/mcp-servers/weather_server.py
      env:
        WEATHER_API_KEY: ${WEATHER_API_KEY}

    # 远程 Server（SSE 传输）
    docs-search:
      enabled: true
      type: sse
      url: https://mcp.example.com/sse
      headers:
        Authorization: Bearer ${DOCS_API_TOKEN}

# 文件系统 Server（默认启用）
servers:
  filesystem:
    enabled: true
    config:
      # 允许访问的路径
      allowed_paths:
        - ~/projects
        - ~/documents
        - ./workspace

      # 禁止访问的路径（优先级更高）
      blocked_paths:
        - ~/.ssh
        - ~/.gnupg
        - /etc/shadow

      # 最大读取文件大小（MB）
      max_read_size: 10

      # 允许的文件类型
      allowed_extensions:
        - .py
        - .js
        - .ts
        - .md
        - .yaml
        - .json
        - .txt
        - .sql

# PostgreSQL Server
servers:
  postgres:
    enabled: true
    type: stdio
    command: npx
    args:
      - -y
      - @modelcontextprotocol/server-postgres
      - postgresql://user:pass@localhost:5432/dbname
    permissions:
      read_only: true
      allowed_tables:
        - users
        - orders
        - products
      blocked_operations:
        - DROP
        - TRUNCATE
        - ALTER

# SQLite Server（轻量级选择）
servers:
  sqlite:
    enabled: true
    type: stdio
    command: python
    args:
      - -m
      - mcp_sqlite_server
      - /path/to/database.db
    permissions:
      read_only: false  # SQLite 通常允许写入

# SSE 传输的远程 Server
servers:
  remote-docs:
    enabled: true
    type: sse
    url: https://docs-mcp.example.com/sse
    timeout: 60
    headers:
      Authorization: Bearer ${DOCS_TOKEN}
      X-Project-ID: my-project

  remote-api:
    enabled: true
    type: sse
    url: http://localhost:8080/mcp
    # 本地开发不需要认证
    timeout: 30

# 检查 MCP Server 状态
openclaw mcp status

# 输出示例：
# MCP Server 状态：
# ✅ filesystem - 运行中
# ✅ postgres   - 运行中
# ❌ weather    - 连接失败（API Key 未设置）
# ✅ docs-search - 运行中

# 启动/停止 Server
openclaw mcp start weather
openclaw mcp stop weather

# 重启所有 Server
openclaw mcp restart

# 查看 Server 日志
openclaw mcp logs postgres

# 诊断 Server 连接
openclaw mcp doctor

# weather_server.py：一个简单的 MCP Server 示例
"""天气查询 MCP Server"""

import asyncio
import httpx
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("weather-server")

@app.list_tools()
async def list_tools():
    """列出可用的工具"""
    return [
        Tool(
            name="get_weather",
            description="获取指定城市的天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 Beijing, Shanghai"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "default": "celsius",
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="get_forecast",
            description="获取指定城市的天气预报（未来 7 天）",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    },
                    "days": {
                        "type": "integer",
                        "minimum": 1,
                        "maximum": 7,
                        "default": 3,
                        "description": "预报天数"
                    }
                },
                "required": ["city"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    """执行工具调用"""
    api_key = os.environ.get("WEATHER_API_KEY")

    if name == "get_weather":
        city = arguments["city"]
        unit = arguments.get("unit", "celsius")

        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"https://api.weather.com/v1/{city}",
                params={"key": api_key, "units": unit}
            )
            data = response.json()

            return [TextContent(
                type="text",
                text=f"{city} 天气：\n"
                     f"温度：{data['temperature']}°{unit[0].upper()}\n"
                     f"湿度：{data['humidity']}%\n"
                     f"天气：{data['condition']}\n"
                     f"风速：{data['wind_speed']} km/h"
            )]

    elif name == "get_forecast":
        # 类似实现...
        pass

if __name__ == "__main__":
    import mcp.server.stdio
    asyncio.run(mcp.server.stdio.run_server(app))

// code-analyzer-server.ts：代码分析 MCP Server
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "code-analyzer",
  version: "1.0.0",
});

// 注册工具
server.tool(
  "analyze_complexity",
  "分析代码的圈复杂度",
  {
    code: z.string().describe("要分析的代码"),
    language: z.enum(["python", "javascript", "typescript", "go", "java"])
      .describe("代码语言"),
  },
  async ({ code, language }) => {
    const result = analyzeComplexity(code, language);

    return {
      content: [
        {
          type: "text",
          text: `代码复杂度分析报告：\n\n` +
                `圈复杂度: ${result.complexity}\n` +
                `函数数: ${result.functionCount}\n` +
                `最高复杂度函数: ${result.maxFunction}\n` +
                `建议: ${result.recommendations}`,
        },
      ],
    };
  }
);

server.tool(
  "find_duplicates",
  "查找代码中的重复片段",
  {
    files: z.array(z.string()).describe("文件路径列表"),
    threshold: z.number().min(0).max(1).default(0.8)
      .describe("相似度阈值"),
  },
  async ({ files, threshold }) => {
    const duplicates = findDuplicates(files, threshold);

    return {
      content: [
        {
          type: "text",
          text: `发现 ${duplicates.length} 处重复代码：\n` +
                duplicates.map(d =>
                  `- ${d.file1}:${d.line1} ↔ ${d.file2}:${d.line2}\n` +
                  `  相似度: ${(d.similarity * 100).toFixed(1)}%`
                ).join("\n"),
        },
      ],
    };
  }
);

// 启动 Server
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

main().catch(console.error);

# 在 OpenClaw 配置中注册自定义工具
# ~/.config/openclaw/config.yaml

tools:
  custom:
    # 天气查询
    weather:
      enabled: true
      type: mcp
      server: weather-api
      tool: get_weather

    # 代码分析
    code_analyzer:
      enabled: true
      type: mcp
      server: code-analyzer
      tool: analyze_complexity

    # 通知发送
    notifier:
      enabled: true
      type: mcp
      server: slack-notifier
      tool: send_message

# 测试 MCP 工具
openclaw mcp test weather get_weather '{"city": "Beijing"}'

# 输出示例：
# 测试工具: weather.get_weather
# 参数: {"city": "Beijing"}
# 结果:
# 北京 天气：
# 温度：25°C
# 湿度：65%
# 天气：晴
# 风速：12 km/h
# ✅ 工具调用成功

# 测试所有已配置的工具
openclaw mcp test-all

# PostgreSQL 集成
servers:
  postgres:
    enabled: true
    command: npx
    args: ["-y", "@modelcontextprotocol/server-postgres", "postgresql://localhost/mydb"]
    permissions:
      read_only: true
      max_query_time: 10

# 使用示例
openclaw "查询 users 表中最近 7 天注册的用户数量，按天分组"

# 带 MCP 工具的复杂查询
openclaw "使用数据库工具执行以下操作：
1. 查询 orders 表，找出 top 10 消费用户
2. 查询这些用户的 profile 信息
3. 输出格式：用户名 | 消费总额 | 注册时间 | 最后登录"

# 文档搜索 Server
servers:
  docs-search:
    enabled: true
    type: sse
    url: https://docs-mcp.internal/sse
    config:
      indices:
        - api-docs
        - internal-wiki
        - codebase-index
      max_results: 10

# 使用示例
openclaw "搜索我们的 API 文档，找出所有与用户认证相关的端点"

openclaw "在我们的内部 wiki 中搜索关于部署流程的文档"

# Prometheus 监控 Server
servers:
  prometheus:
    enabled: true
    command: python
    args: ["/opt/mcp-servers/prometheus_server.py"]
    env:
      PROMETHEUS_URL: http://prometheus:9090

# 告警 Server
servers:
  alerting:
    enabled: true
    command: python
    args: ["/opt/mcp-servers/alerting_server.py"]
    env:
      SLACK_WEBHOOK: ${SLACK_WEBHOOK_URL}
      PAGERDUTY_KEY: ${PAGERDUTY_KEY}

# AWS Server
servers:
  aws:
    enabled: true
    command: npx
    args: ["-y", "@modelcontextprotocol/server-aws"]
    env:
      AWS_REGION: us-east-1
    permissions:
      allowed_services:
        - s3
        - ec2
        - lambda
      read_only: true

# 使用示例
openclaw "列出我们 S3 中所有超过 1GB 的文件"

openclaw "检查 EC2 实例的运行状态，找出未使用的实例"

# 标准输出管道
openclaw "查询数据库中的用户统计" | grep -E "^[0-9]"

# JSON 输出管道
openclaw --json "获取系统状态" | jq '.services[] | select(.status == "error")'

# 链式处理
openclaw "分析日志文件中的错误" \
  | grep "ERROR" \
  | sort \
  | uniq -c \
  | sort -rn \
  | head -20

#!/bin/bash
# process-result.sh：AI 输出的后处理

# 获取 AI 输出
result=$(openclaw --quiet "分析以下日志，提取错误信息：
$(cat /var/log/app.log)")

# 提取关键信息
errors=$(echo "$result" | grep -oP "ERROR: \K.*")

# 生成报告
{
  echo "# 日志分析报告"
  echo ""
  echo "## 错误摘要"
  echo ""
  echo "$errors" | while read -r error; do
    echo "- $error"
  done
  echo ""
  echo "## 建议"
  echo ""
  openclaw --quiet "基于以下错误，给出修复建议：
$errors"
} > report.md

# CSV 转换
openclaw --json "查询用户数据" | \
  jq -r '.users[] | [.name, .email, .created_at] | @csv' \
  > users.csv

# Markdown 表格
openclaw --json "获取项目列表" | \
  jq -r '.projects[] | "| \(.name) | \(.status) | \(.owner) |"' \
  > projects_table.md

# HTML 报告
openclaw --json "生成测试报告数据" | \
  python3 -c "
import json, sys
data = json.load(sys.stdin)
print('<html><body>')
print('<h1>测试报告</h1>')
print(f'<p>通过率: {data[\"pass_rate\"]}%</p>')
print('<table>')
for test in data['tests']:
    color = 'green' if test['passed'] else 'red'
    print(f'<tr style=\"color:{color}\">')
    print(f'<td>{test[\"name\"]}</td>')
    print(f'<td>{test[\"duration\"]}s</td>')
    print('</tr>')
print('</table></body></html>')
" > report.html

#!/bin/bash
# route-result.sh：根据 AI 输出路由到不同处理流程

result=$(openclaw --quiet "分析以下系统指标并分类：
CPU: $(uptime | awk -F'load average:' '{print $2}')
内存: $(free -m | grep Mem)
磁盘: $(df -h / | tail -1 | awk '{print $5}')

输出格式：
STATUS: NORMAL/WARNING/CRITICAL
DETAIL: 详细说明")

status=$(echo "$result" | grep "^STATUS:" | cut -d: -f2 | tr -d ' ')

case $status in
  NORMAL)
    echo "✅ 系统状态正常"
    ;;
  WARNING)
    echo "⚠️ 系统状态警告，发送 Slack 通知"
    curl -X POST "$SLACK_WEBHOOK" -d "{
      \"text\": \"系统警告：$(echo "$result" | grep '^DETAIL:')\"
    }"
    ;;
  CRITICAL)
    echo "🚨 系统状态严重，触发告警"
    # 发送 PagerDuty 告警
    curl -X POST "https://events.pagerduty.com/v2/enqueue" \
      -d "{
        \"routing_key\": \"$PAGERDUTY_KEY\",
        \"event_action\": \"trigger\",
        \"payload\": {
          \"summary\": \"系统严重警告\",
          \"severity\": \"critical\",
          \"source\": \"openclaw-monitor\",
          \"details\": \"$(echo "$result" | grep '^DETAIL:')\"
        }
      }"
    ;;
esac

#!/usr/bin/env python3
# structured_result.py：处理 OpenClaw 的结构化输出

import json
import subprocess
import sys

def run_openclaw(prompt):
    """运行 OpenClaw 并获取 JSON 输出"""
    result = subprocess.run(
        ["openclaw", "--json", prompt],
        capture_output=True,
        text=True
    )
    return json.loads(result.stdout)

def process_users():
    """处理用户数据"""
    data = run_openclaw(
        "查询数据库 users 表，返回 JSON 格式：\n"
        "{users: [{id, name, email, status, created_at}]}"
    )

    # 过滤活跃用户
    active_users = [
        u for u in data["users"]
        if u["status"] == "active"
    ]

    # 统计
    stats = {
        "total": len(data["users"]),
        "active": len(active_users),
        "inactive": len(data["users"]) - len(active_users),
    }

    # 输出报告
    print("用户统计报告")
    print(f"总用户数: {stats['total']}")
    print(f"活跃用户: {stats['active']}")
    print(f"非活跃用户: {stats['inactive']}")

    # 生成邮件列表
    emails = [u["email"] for u in active_users]
    return emails

if __name__ == "__main__":
    emails = process_users()
    # 可以用于后续的邮件发送等

# github_server.py：GitHub API MCP Server
import os
import httpx
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("github-server")
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")
HEADERS = {
    "Authorization": f"Bearer {GITHUB_TOKEN}",
    "Accept": "application/vnd.github.v3+json"
}

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="list_issues",
            description="列出仓库的 Issues",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "state": {"type": "string", "enum": ["open", "closed", "all"]},
                    "labels": {"type": "string", "description": "逗号分隔的标签"}
                },
                "required": ["owner", "repo"]
            }
        ),
        Tool(
            name="create_issue",
            description="创建新的 Issue",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "title": {"type": "string"},
                    "body": {"type": "string"},
                    "labels": {"type": "array", "items": {"type": "string"}}
                },
                "required": ["owner", "repo", "title"]
            }
        ),
        Tool(
            name="list_prs",
            description="列出 Pull Requests",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "state": {"type": "string", "enum": ["open", "closed", "merged", "all"]}
                },
                "required": ["owner", "repo"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "list_issues":
        owner = arguments["owner"]
        repo = arguments["repo"]
        state = arguments.get("state", "open")

        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"https://api.github.com/repos/{owner}/{repo}/issues",
                headers=HEADERS,
                params={"state": state}
            )
            issues = response.json()

            text = f"仓库 {owner}/{repo} 的 Issues（{state}）：\n\n"
            for issue in issues[:20]:
                text += f"#{issue['number']} {issue['title']} - {issue['state']}\n"
                if issue.get("labels"):
                    labels = ", ".join([l["name"] for l in issue["labels"]])
                    text += f"  标签: {labels}\n"

            return [TextContent(type="text", text=text)]

if __name__ == "__main__":
    import asyncio
    import mcp.server.stdio
    asyncio.run(mcp.server.stdio.run_server(app))

# email_server.py：邮件通知 MCP Server
import os
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("email-server")

SMTP_CONFIG = {
    "server": os.environ.get("SMTP_SERVER", "smtp.gmail.com"),
    "port": int(os.environ.get("SMTP_PORT", "587")),
    "username": os.environ.get("SMTP_USERNAME"),
    "password": os.environ.get("SMTP_PASSWORD"),
}

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="send_email",
            description="发送邮件通知",
            inputSchema={
                "type": "object",
                "properties": {
                    "to": {"type": "string", "description": "收件人邮箱"},
                    "subject": {"type": "string", "description": "邮件主题"},
                    "body": {"type": "string", "description": "邮件正文"},
                    "body_type": {
                        "type": "string",
                        "enum": ["plain", "html"],
                        "default": "plain"
                    }
                },
                "required": ["to", "subject", "body"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "send_email":
        msg = MIMEMultipart()
        msg["From"] = SMTP_CONFIG["username"]
        msg["To"] = arguments["to"]
        msg["Subject"] = arguments["subject"]

        body_type = arguments.get("body_type", "plain")
        msg.attach(MIMEText(arguments["body"], body_type))

        with smtplib.SMTP(SMTP_CONFIG["server"], SMTP_CONFIG["port"]) as server:
            server.starttls()
            server.login(SMTP_CONFIG["username"], SMTP_CONFIG["password"])
            server.send_message(msg)

        return [TextContent(
            type="text",
            text=f"✅ 邮件已发送至 {arguments['to']}"
        )]

# slack_server.py：Slack 通知 MCP Server
import os
import httpx
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("slack-server")
SLACK_WEBHOOK = os.environ.get("SLACK_WEBHOOK_URL")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="send_slack_message",
            description="发送 Slack 消息",
            inputSchema={
                "type": "object",
                "properties": {
                    "channel": {"type": "string", "description": "频道名，如 #general"},
                    "text": {"type": "string", "description": "消息内容"},
                    "blocks": {
                        "type": "array",
                        "description": "Slack Block Kit 块（可选）"
                    }
                },
                "required": ["text"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "send_slack_message":
        payload = {"text": arguments["text"]}
        if arguments.get("channel"):
            payload["channel"] = arguments["channel"]
        if arguments.get("blocks"):
            payload["blocks"] = arguments["blocks"]

        async with httpx.AsyncClient() as client:
            response = await client.post(SLACK_WEBHOOK, json=payload)

        if response.status_code == 200:
            return [TextContent(type="text", text="✅ 消息已发送")]
        else:
            return [TextContent(
                type="text",
                text=f"❌ 发送失败: {response.status_code} {response.text}"
            )]

# 安装新的 MCP Server
# 方式一：使用官方包
npx -y @modelcontextprotocol/server-postgres postgresql://localhost/db

# 方式二：使用 pip
pip install mcp-sqlite-server

# 方式三：从源码安装
git clone https://github.com/example/custom-mcp-server.git
cd custom-mcp-server
pip install -e .

# 验证安装
openclaw mcp verify custom-server

# 配置 Server
openclaw mcp configure custom-server
# 交互式配置向导
# > 输入 Server 名称: my-custom-server
# > 输入命令: python
# > 输入参数: -m my_server
# > 是否设置环境变量？(y/n): y
# > API_KEY: sk-xxx...

# 检查更新
openclaw mcp check-updates

# 更新所有 Server
openclaw mcp update-all

# 更新指定 Server
openclaw mcp update postgres

# 回滚到之前的版本
openclaw mcp rollback postgres --to 1.2.0

# 诊断所有 Server
openclaw mcp doctor

# 输出示例：
# MCP 诊断报告
# =============
#
# Server: filesystem
#   状态: ✅ 运行中
#   延迟: 12ms
#   工具数: 5
#   最后错误: 无
#
# Server: postgres
#   状态: ✅ 运行中
#   延迟: 45ms
#   工具数: 3
#   最后错误: 无
#
# Server: weather-api
#   状态: ❌ 错误
#   延迟: -
#   工具数: 0
#   最后错误: Connection refused (ECONNREFUSED)
#
# Server: docs-search
#   状态: ⚠️ 慢
#   延迟: 3500ms
#   工具数: 2
#   最后错误: 无（但响应时间超过阈值）

# 详细调试模式
openclaw mcp debug postgres --verbose

# 查看 Server 日志
openclaw mcp logs postgres --follow
openclaw mcp logs postgres --tail 100

# 监控 MCP Server 性能
openclaw mcp metrics

# 输出示例：
# MCP Server 性能指标
# ===================
#
# Server: filesystem
#   总调用: 1,234
#   平均延迟: 15ms
#   P95 延迟: 45ms
#   错误率: 0.1%
#
# Server: postgres
#   总调用: 567
#   平均延迟: 89ms
#   P95 延迟: 250ms
#   错误率: 0.5%
#
# Server: weather-api
#   总调用: 89
#   平均延迟: 1200ms
#   P95 延迟: 3000ms
#   错误率: 2.3%

# ~/.config/openclaw/mcp-templates.yaml
# 常用 MCP Server 配置模板

templates:
  postgres:
    type: stdio
    command: npx
    args:
      - -y
      - @modelcontextprotocol/server-postgres
      - "${DATABASE_URL}"
    env:
      DATABASE_URL: postgresql://localhost:5432/mydb
    permissions:
      read_only: true

  sqlite:
    type: stdio
    command: python
    args:
      - -m
      - mcp_sqlite_server
      - "${DB_PATH}"
    env:
      DB_PATH: ./data/app.db

  github:
    type: stdio
    command: npx
    args:
      - -y
      - @modelcontextprotocol/server-github
    env:
      GITHUB_PERSONAL_ACCESS_TOKEN: "${GITHUB_TOKEN}"

┌──────────────────────────────────────────────────────────┐
│              MCP 权限模型                                  │
│                                                          │
│  Server 级权限                                            │
│  ├── enabled: true/false（是否启用）                       │
│  ├── read_only: true/false（只读模式）                     │
│  └── timeout: 30（超时限制）                               │
│                                                          │
│  工具级权限                                               │
│  ├── allowed_tools: [tool1, tool2]（白名单）               │
│  ├── blocked_tools: [tool3]（黑名单）                      │
│  └── require_approval: true（需要人工确认）                 │
│                                                          │
│  数据级权限                                               │
│  ├── allowed_paths: [/path1, /path2]（路径白名单）          │
│  ├── blocked_paths: [/etc]（路径黑名单）                    │
│  ├── allowed_tables: [users, orders]（表白名单）            │
│  └── max_rows: 1000（最大返回行数）                         │
└──────────────────────────────────────────────────────────┘

# 细粒度权限配置
mcp:
  servers:
    postgres:
      enabled: true
      permissions:
        # 工具级
        allowed_tools:
          - execute_query
          - list_tables
        blocked_tools:
          - execute_ddl
          - drop_table

        # 数据级
        allowed_tables:
          - users
          - orders
          - products
        max_rows: 1000
        max_query_time: 30

        # 需要人工确认的操作
        require_approval:
          - DELETE 操作
          - UPDATE 操作（影响 > 100 行）

┌──────────────────────────────────────────────────────────┐
│              MCP 安全最佳实践                              │
│                                                          │
│  🔒 认证与授权                                            │
│  ├── 使用环境变量存储密钥，不要硬编码                        │
│  ├── 定期轮换 API Key 和 Token                             │
│  └── 使用最小权限原则配置访问权限                           │
│                                                          │
│  🔒 输入验证                                              │
│  ├── 所有外部输入都需要验证                                 │
│  ├── 使用参数化查询防止 SQL 注入                           │
│  └── 限制查询复杂度和执行时间                               │
│                                                          │
│  🔒 输出控制                                              │
│  ├── 限制返回数据量（行数、大小）                           │
│  ├── 敏感字段自动脱敏                                      │
│  └── 审计日志记录所有操作                                   │
│                                                          │
│  🔒 网络隔离                                              │
│  ├── MCP Server 运行在受限网络环境中                        │
│  ├── 外部 API 调用使用代理                                 │
│  └── 禁止 Server 访问内网敏感服务                           │
└──────────────────────────────────────────────────────────┘

# 启用审计日志
mcp:
  audit:
    enabled: true
    log_file: ~/.config/openclaw/mcp-audit.log
    log_level: info

    # 记录的内容
    log:
      tool_calls: true      # 工具调用
      query_execution: true # 查询执行
      errors: true          # 错误
      permissions: true     # 权限检查

┌──────────────────────────────────────────────────────────┐
│              MCP 与扩展最佳实践                            │
│                                                          │
│  1. 渐进式扩展                                            │
│     ├── 先使用内置 Server（文件系统）                       │
│     ├── 再添加常用的外部 Server（数据库）                    │
│     └── 最后开发自定义 Server                              │
│                                                          │
│  2. 权限最小化                                            │
│     ├── 只启用需要的 Server                                │
│     ├── 只开放必要的权限                                    │
│     └── 敏感操作需要人工确认                                │
│                                                          │
│  3. 错误处理                                              │
│     ├── Server 失败不影响 OpenClaw 核心功能                  │
│     ├── 清晰的错误信息和降级策略                            │
│     └── 自动重试和超时机制                                 │
│                                                          │
│  4. 性能优化                                              │
│     ├── 按需加载 Server（不使用时不启动）                    │
│     ├── 合理设置超时和并发限制                              │
│     └── 缓存频繁使用的数据                                  │
│                                                          │
│  5. 版本管理                                              │
│     ├── 固定 Server 版本，避免意外升级                       │
│     ├── 定期更新安全补丁                                    │
│     └── 更新前在测试环境验证                                │
└──────────────────────────────────────────────────────────┘

# Q1: MCP Server 连接失败
openclaw mcp doctor
# 检查：命令是否正确？环境变量是否设置？网络是否可达？

# Q2: 工具调用超时
# 解决：增加 timeout 设置
# mcp:
#   servers:
#     my-server:
#       timeout: 60  # 从默认 30s 增加到 60s

# Q3: 权限不足
# 解决：检查 allowed_tools 和 blocked_tools 配置

# Q4: Server 启动慢
# 解决：使用按需加载，设置 lazy: true
# mcp:
#   servers:
#     heavy-server:
#       lazy: true  # 首次调用时才启动

# Q5: 结果格式不对
# 解决：检查 Server 的输出格式是否符合 MCP 协议规范
# 使用 openclaw mcp debug <server> --verbose 查看详细通信

┌──────────────────────────────────────────────────────────┐
│              推荐扩展路线图                                │
│                                                          │
│  阶段 1：基础                                            │
│  ├── ✅ 文件系统 Server（内置）                             │
│  └── ✅ Shell 命令执行（内置）                              │
│                                                          │
│  阶段 2：数据层                                          │
│  ├── 📦 SQLite/PostgreSQL Server                           │
│  └── 📦 Redis Server                                      │
│                                                          │
│  阶段 3：协作层                                          │
│  ├── 📦 GitHub/GitLab Server                              │
│  ├── 📦 Slack/钉钉通知 Server                              │
│  └── 📦 邮件通知 Server                                    │
│                                                          │
│  阶段 4：运维层                                          │
│  ├── 📦 Prometheus/监控 Server                             │
│  ├── 📦 AWS/GCP 云服务 Server                              │
│  └── 📦 Docker/K8s Server                                 │
│                                                          │
│  阶段 5：自定义                                          │
│  ├── 📦 业务特定 API Server                                │
│  ├── 📦 内部工具 Server                                    │
│  └── 📦 团队知识库 Server                                  │
└──────────────────────────────────────────────────────────┘