OpenClaw MCP 工具开发指南

什么是 MCP？

MCP (Model Context Protocol) 是由 Anthropic 提出的开放协议，旨在标准化 AI 模型与外部数据源、工具之间的连接方式。

在 OpenClaw 中，MCP 是 Agent Skills 的核心通信机制，让 AI 能够：

• 🔌 统一接口访问各种数据源
• 🔒 安全地执行外部工具
• 🔄 动态发现和调用能力
• 📊 获取实时上下文信息

MCP 架构原理

┌─────────────────┐      MCP Protocol      ┌─────────────────┐
│   AI Agent      │ ◄────────────────────► │  MCP Server     │
│  (OpenClaw)     │   JSON-RPC over SSE    │  (Your Skill)   │
└─────────────────┘                        └─────────────────┘
                                                    │
                       ┌────────────────────────────┼────────────────────────────┐
                       ▼                            ▼                            ▼
                ┌──────────────┐           ┌──────────────┐           ┌──────────────┐
                │   Database   │           │    APIs      │           │   Files      │
                └──────────────┘           └──────────────┘           └──────────────┘
          

核心概念

环境搭建

1. 安装 MCP SDK

# Node.js
npm install @modelcontextprotocol/sdk

# Python
pip install mcp

2. OpenClaw MCP 配置

# ~/.openclaw/mcp.json
{
  "servers": {
    "my-database": {
      "command": "node",
      "args": ["/path/to/my-mcp-server.js"],
      "env": {
        "DB_URL": "postgresql://localhost/mydb"
      }
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "your-token"
      }
    }
  }
}

实现 MCP Server

基础 Server (Node.js)

import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from '@modelcontextprotocol/sdk/types.js';

// 创建 Server
const server = new Server(
  {
    name: 'my-data-server',
    version: '1.0.0',
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

// 声明可用工具
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: 'query_data',
        description: '查询数据库数据',
        inputSchema: {
          type: 'object',
          properties: {
            table: { type: 'string', description: '表名' },
            filters: { type: 'object', description: '过滤条件' },
          },
          required: ['table'],
        },
      },
    ],
  };
});

// 处理工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === 'query_data') {
    const { table, filters } = request.params.arguments;
    
    // 执行查询
    const results = await queryDatabase(table, filters);
    
    return {
      content: [
        {
          type: 'text',
          text: JSON.stringify(results, null, 2),
        },
      ],
    };
  }
  throw new Error(`未知工具: ${request.params.name}`);
});

// 启动 Server
const transport = new StdioServerTransport();
await server.connect(transport);

Python 版本

from mcp.server import Server
from mcp.types import TextContent
import asyncio

app = Server("my-data-server")

@app.list_tools()
async def list_tools():
    return [
        {
            "name": "analyze_csv",
            "description": "分析CSV文件",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string"},
                    "operations": {"type": "array"}
                },
                "required": ["file_path"]
            }
        }
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "analyze_csv":
        result = await analyze_csv_file(
            arguments["file_path"],
            arguments.get("operations", [])
        )
        return [TextContent(type="text", text=result)]
    raise ValueError(f"未知工具: {name}")

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

Resources 资源暴露

让 AI 能够读取结构化数据：

// 声明资源
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  return {
    resources: [
      {
        uri: 'database://users/schema',
        name: '用户表结构',
        mimeType: 'application/json',
      },
      {
        uri: 'database://orders/recent',
        name: '最近订单',
        mimeType: 'application/json',
      },
    ],
  };
});

// 读取资源
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const uri = request.params.uri;
  
  if (uri === 'database://users/schema') {
    const schema = await getTableSchema('users');
    return {
      contents: [
        {
          uri,
          mimeType: 'application/json',
          text: JSON.stringify(schema, null, 2),
        },
      ],
    };
  }
  
  throw new Error(`未知资源: ${uri}`);
});

高级功能

进度通知

// 向客户端发送进度更新
await server.notification({
  method: 'notifications/progress',
  params: {
    progressToken: request.params.meta?.progressToken,
    progress: 50,
    total: 100,
    message: '正在处理数据...',
  },
});

Prompts 模板

server.setRequestHandler(ListPromptsRequestSchema, async () => {
  return {
    prompts: [
      {
        name: 'code_review',
        description: '代码审查模板',
        arguments: [
          {
            name: 'code',
            description: '要审查的代码',
            required: true,
          },
        ],
      },
    ],
  };
});

server.setRequestHandler(GetPromptRequestSchema, async (request) => {
  if (request.params.name === 'code_review') {
    const code = request.params.arguments?.code;
    return {
      description: '代码审查',
      messages: [
        {
          role: 'user',
          content: {
            type: 'text',
            text: `请审查以下代码:\n\n\`\`\`\n${code}\n\`\`\``,\n          },
        },
      ],
    };
  }
});

安全最佳实践

// 限制可访问的表
const ALLOWED_TABLES = ['users', 'orders', 'products'];

async function queryDatabase(table: string) {
  if (!ALLOWED_TABLES.includes(table)) {
    throw new Error(`无权访问表: ${table}`);
  }
  // ...
}

测试 MCP Server

# 使用 MCP Inspector 测试
npx @modelcontextprotocol/inspector node ./my-server.js

# 或使用 OpenClaw CLI
openclaw mcp test my-server

# 运行集成测试
openclaw mcp test my-server --scenario full

OpenClaw 集成

// 在 Skill 中使用 MCP
import { MCPClient } from '@openclaw/mcp';

const client = new MCPClient();

export async function analyzeWithMCP(data: string) {
  // 连接到 MCP Server
  await client.connect('my-data-server');
  
  // 调用工具
  const result = await client.callTool('analyze_data', {
    input: data,
    format: 'json'
  });
  
  return result;
}

常见问题排查

相关资源

• OpenClaw Agent Skills 开发教程
• MCP 官方文档
• OpenClaw 子Agent编排
• ClawHub 技能市场