OpenClaw 自定义MCP服务器开发指南

什么是MCP？为什么需要自定义？

世界上有一种协议叫MCP（Model Context Protocol），它就像AI世界的"USB接口"——任何支持MCP的工具，都能插进OpenClaw。

但问题来了：你的公司内部API、私有数据库、自研工具，它们可不支持MCP。这时候，你需要自己写一个MCP Server，把它们"翻译"成OpenClaw能懂的语言。

🎯 自定义MCP Server的五大场景：

私有数据库：让Agent直接查询公司内部MySQL/PostgreSQL
内部API：对接公司ERP、CRM、OA系统
自研工具：把你们团队的工具变成Agent可调用的能力
硬件设备：让Agent控制IoT设备、机器人、传感器
特殊数据源：调用付费API、内网服务、加密数据源

MCP Server 核心概念

MCP Server需要实现三个核心方法：

① tools/list - 告诉Agent"我有哪些工具可用"

② tools/call - 执行Agent指定的工具调用

③ resources/list - （可选）提供可读取的数据资源

实战：构建一个"公司HR系统"MCP Server

假设你的公司有一个HR系统API，你想让Agent能查询员工信息、请假记录等。

第一步：设计工具清单

// hr-mcp-server/tools-definition.json
{
  "tools": [
    {
      "name": "get_employee_info",
      "description": "查询员工基本信息（姓名、部门、职级）",
      "inputSchema": {
        "type": "object",
        "properties": {
          "employee_id": {
            "type": "string",
            "description": "员工ID，格式：EMP + 6位数字"
          }
        },
        "required": ["employee_id"]
      }
    },
    {
      "name": "list_leave_requests",
      "description": "查询员工的请假记录",
      "inputSchema": {
        "type": "object",
        "properties": {
          "employee_id": {"type": "string"},
          "start_date": {"type": "string", "format": "date"},
          "end_date": {"type": "string", "format": "date"}
        },
        "required": ["employee_id"]
      }
    },
    {
      "name": "submit_leave_request",
      "description": "提交新的请假申请",
      "inputSchema": {
        "type": "object",
        "properties": {
          "employee_id": {"type": "string"},
          "leave_type": {"type": "string", "enum": ["年假", "病假", "事假", "婚假"]},
          "start_date": {"type": "string"},
          "end_date": {"type": "string"},
          "reason": {"type": "string"}
        },
        "required": ["employee_id", "leave_type", "start_date", "end_date"]
      }
    }
  ]
}

第二步：实现MCP Server（Python版）

# hr-mcp-server/server.py
import json
import asyncio
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# 模拟HR系统API（实际项目中替换为真实API调用）
HR_SYSTEM_API = "https://hr.internal.company.com/api"

server = Server("company-hr-mcp-server")

@server.list_tools()
async def list_tools() -> list[Tool]:
    """告诉Agent有哪些工具可用"""
    return [
        Tool(
            name="get_employee_info",
            description="查询员工基本信息（姓名、部门、职级）",
            inputSchema={
                "type": "object",
                "properties": {
                    "employee_id": {"type": "string", "description": "员工ID"}
                },
                "required": ["employee_id"]
            }
        ),
        Tool(
            name="list_leave_requests",
            description="查询员工的请假记录",
            inputSchema={
                "type": "object",
                "properties": {
                    "employee_id": {"type": "string"},
                    "start_date": {"type": "string", "format": "date"},
                    "end_date": {"type": "string", "format": "date"}
                },
                "required": ["employee_id"]
            }
        ),
        Tool(
            name="submit_leave_request",
            description="提交新的请假申请",
            inputSchema={
                "type": "object",
                "properties": {
                    "employee_id": {"type": "string"},
                    "leave_type": {"type": "string", "enum": ["年假", "病假", "事假", "婚假"]},
                    "start_date": {"type": "string"},
                    "end_date": {"type": "string"},
                    "reason": {"type": "string"}
                },
                "required": ["employee_id", "leave_type", "start_date", "end_date"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """执行工具调用"""
    if name == "get_employee_info":
        emp_id = arguments.get("employee_id")
        # 实际项目中：调用HR系统API
        # response = requests.get(f"{HR_SYSTEM_API}/employees/{emp_id}", headers=auth_headers)
        mock_data = {
            "employee_id": emp_id,
            "name": "张三",
            "department": "技术部",
            "level": "P7",
            "join_date": "2020-03-15"
        }
        return [TextContent(type="text", text=json.dumps(mock_data, ensure_ascii=False))]

    elif name == "list_leave_requests":
        emp_id = arguments.get("employee_id")
        mock_leaves = [
            {"date": "2026-05-01", "type": "年假", "days": 3, "status": "已批准"},
            {"date": "2026-04-15", "type": "病假", "days": 1, "status": "已批准"}
        ]
        return [TextContent(type="text", text=json.dumps(mock_leaves, ensure_ascii=False))]

    elif name == "submit_leave_request":
        # 实际项目中：POST到HR系统API
        result = {"status": "success", "request_id": "LR-2026-0521-001"}
        return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]

    return [TextContent(type="text", text="未知工具")]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

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

第三步：配置到OpenClaw

# ~/.openclaw/config.yaml 中添加MCP服务器配置
mcpServers:
  company-hr:
    command: "python"
    args:
      - "/path/to/hr-mcp-server/server.py"
    env:
      HR_API_KEY: "${HR_API_KEY}"  # 从环境变量读取
    # 可选：添加描述，帮助Agent理解何时使用
    description: "公司HR系统工具，可以查询员工信息、请假记录、提交请假申请"

# 重启OpenClaw使配置生效
openclaw gateway restart

第四步：在OpenClaw中使用

# Agent现在可以调用你的HR工具了
openclaw chat --message "查询员工EMP123456的基本信息，然后看看他最近有没有请假"

# Agent会自动：
# 1. 调用 get_employee_info(employee_id="EMP123456")
# 2. 调用 list_leave_requests(employee_id="EMP123456")
# 3. 整合结果，生成回答

# 你也可以直接在提示中指定
openclaw chat --message "帮我提交一个请假申请" \
  --tool company-hr/submit_leave_request \
  --args '{"employee_id": "EMP123456", "leave_type": "年假", "start_date": "2026-06-01", "end_date": "2026-06-03", "reason": "家庭旅行"}'

进阶：添加认证和安全

⚠️ 安全提示：

MCP Server直接对接内部系统，必须做好安全防护！

# hr-mcp-server/auth.py - 添加认证中间件
import os
from functools import wraps

def require_auth(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # 从环境变量读取API Key
        expected_key = os.getenv("HR_MCP_API_KEY")
        provided_key = kwargs.get("context", {}).get("api_key")
        
        if not expected_key or provided_key != expected_key:
            return [TextContent(type="text", text="认证失败：无效的API Key")]
        
        return await func(*args, **kwargs)
    return wrapper

# 在call_tool中使用
@server.call_tool()
@require_auth
async def call_tool(name: str, arguments: Any):
    # ... 原有逻辑
    pass

常见MCP Server模板

Server类型	适用场景	核心工具
数据库MCP	查询MySQL/PostgreSQL	query, insert, update, schema_info
API代理MCP	包装第三方API	get, post, put, delete
文件系统MCP	读写服务器文件	read_file, write_file, list_dir
IoT设备MCP	控制智能设备	get_status, set_config, send_command
消息推送MCP	发送通知/邮件	send_email, send_sms, push_wechat

调试技巧

stdio模式测试：直接用 echo '{"jsonrpc":"2.0","method":"tools/list"}' | python server.py 测试
日志输出：在Server中加 print(json.dumps(log_entry), file=sys.stderr)，OpenClaw会捕获stderr
OpenClaw日志：tail -f ~/.openclaw/logs/mcp-company-hr.log 查看调用记录
工具测试：openclaw mcp test company-hr get_employee_info --args '{"employee_id":"EMP001"}'

🎯 妙趣提示：

写MCP Server就像给Agent当翻译——把"人类系统的语言"翻译成"AI能懂的语言"。翻译得越准确，Agent干活越利索。

性能优化

连接池：数据库MCP Server用连接池，避免频繁建连
结果缓存：对查询类工具加缓存，相同参数直接返回
超时设置：每个工具调用加超时（建议5-30秒），避免卡死
批量操作：支持批量查询的工具，比单个查询快N倍