OpenClaw REST API开发和第三方集成教程

前言

OpenClaw Gateway 不仅为聊天频道提供服务，还暴露了一套完整的 REST API。通过这些 API，你可以将 AI 能力集成到任何应用中——无论是自建的 Web 前端、CRM 系统还是客服工单平台。本文将详细介绍 API 的使用方法和集成实践。

一、API 概览

1.1 基础信息

1.2 核心端点列表

二、认证配置

2.1 生成 API Token

# 生成 API Token
openclaw token create --name "my-app" --scope "chat,send,read"

# 输出：
# Token: oc_tok_a1b2c3d4e5f6g7h8i9j0...
# Scope: chat, send, read
# Created: 2026-04-09

# 查看所有 Token
openclaw token list

# 撤销 Token
openclaw token revoke oc_tok_a1b2c3d4e5f6g7h8i9j0

2.2 在配置文件中设置

// ~/.config/openclaw/openclaw.json5
{
  "api": {
    "enabled": true,
    "port": 18789,
    "auth": {
      "enabled": true,
      "tokens": [
        {
          "name": "my-web-app",
          "token": "oc_tok_a1b2c3d4e5f6g7h8i9j0",
          "scope": ["chat", "send", "read", "admin"],
          "rateLimit": 120  // RPM
        },
        {
          "name": "crm-integration",
          "token": "oc_tok_x9y8z7w6v5u4t3s2r1q0",
          "scope": ["chat", "send"],
          "rateLimit": 30
        }
      ]
    }
  }
}

2.3 认证请求示例

所有 API 请求需要在 Header 中携带 Token：

curl -H "Authorization: Bearer oc_tok_a1b2c3d4e5f6g7h8i9j0" \
  http://localhost:18789/api/v1/models

三、核心 API 使用

3.1 发送消息并获取回复

curl 示例：

curl -X POST http://localhost:18789/api/v1/chat \
  -H "Authorization: Bearer $OPENCLAW_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "message": "帮我用Python写一个快速排序算法",
    "conversationId": "conv_123",
    "model": "claude-sonnet-4-20250514",
    "systemPrompt": "你是一个专业的编程助手",
    "maxTokens": 2000,
    "temperature": 0.7
  }'

响应示例：

{
  "id": "msg_abc123",
  "conversationId": "conv_123",
  "role": "assistant",
  "content": "以下是Python实现的快速排序算法：\n\n```python\ndef quicksort(arr):\n    if len(arr) <= 1:\n        return arr\n    pivot = arr[len(arr) // 2]\n    left = [x for x in arr if x < pivot]\n    middle = [x for x in arr if x == pivot]\n    right = [x for x in arr if x > pivot]\n    return quicksort(left) + middle + quicksort(right)\n```",
  "model": "claude-sonnet-4-20250514",
  "usage": {
    "inputTokens": 45,
    "outputTokens": 156
  },
  "latency": 2341
}

3.2 流式输出

curl 示例（SSE）：

curl -X POST http://localhost:18789/api/v1/chat/stream \
  -H "Authorization: Bearer $OPENCLAW_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "message": "写一篇关于人工智能的短文",
    "model": "claude-sonnet-4-20250514"
  }'

# 返回 SSE 事件流：
# data: {"type":"start","conversationId":"conv_456"}
# data: {"type":"delta","content":"人工"}
# data: {"type":"delta","content":"智能"}
# data: {"type":"delta","content":"（AI）"}
# ...
# data: {"type":"done","usage":{"inputTokens":20,"outputTokens":350}}

3.3 向频道发送消息

# 向 Telegram 用户发送消息
curl -X POST http://localhost:18789/api/v1/send \
  -H "Authorization: Bearer $OPENCLAW_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "channel": "telegram",
    "chatId": "123456789",
    "message": "这是一条通过 API 发送的消息"
  }'

# 向 Discord 频道发送消息
curl -X POST http://localhost:18789/api/v1/send \
  -H "Authorization: Bearer $OPENCLAW_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "channel": "discord",
    "chatId": "CHANNEL_ID",
    "message": "系统通知：服务器维护将于今晚 22:00 开始"
  }'

四、Webhook 接收

4.1 配置 Webhook 回调

当 OpenClaw 收到消息或完成处理时，可以向你的应用发送 Webhook 通知：

{
  "api": {
    "webhooks": {
      "outgoing": [
        {
          "url": "https://your-app.com/api/openclaw-callback",
          "secret": "your-webhook-secret",
          "events": [
            "message.received",
            "message.sent",
            "conversation.created",
            "channel.connected",
            "channel.disconnected"
          ],
          "retry": {
            "maxAttempts": 3,
            "backoffMs": 5000
          }
        }
      ]
    }
  }
}

4.2 Webhook 事件格式

{
  "event": "message.received",
  "timestamp": "2026-04-09T10:30:00Z",
  "data": {
    "messageId": "msg_xyz789",
    "conversationId": "conv_123",
    "channel": "telegram",
    "chatId": "123456789",
    "from": {
      "id": "user_456",
      "name": "Alice"
    },
    "content": "用户发送的消息内容",
    "timestamp": "2026-04-09T10:30:00Z"
  },
  "signature": "sha256=abcdef..."
}

4.3 验证 Webhook 签名

import hmac
import hashlib

def verify_webhook(payload, signature, secret):
    expected = 'sha256=' + hmac.new(
        secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

五、Python 集成示例

5.1 基础封装

import requests
from typing import Optional, Generator

class OpenClawClient:
    def __init__(self, base_url: str = "http://localhost:18789",
                 token: str = ""):
        self.base_url = base_url.rstrip("/")
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

    def chat(self, message: str, conversation_id: Optional[str] = None,
             model: str = "claude-sonnet-4-20250514",
             system_prompt: Optional[str] = None) -> dict:
        """发送消息并获取 AI 回复"""
        payload = {
            "message": message,
            "model": model,
        }
        if conversation_id:
            payload["conversationId"] = conversation_id
        if system_prompt:
            payload["systemPrompt"] = system_prompt

        resp = requests.post(
            f"{self.base_url}/api/v1/chat",
            json=payload,
            headers=self.headers,
            timeout=120
        )
        resp.raise_for_status()
        return resp.json()

    def chat_stream(self, message: str,
                    model: str = "claude-sonnet-4-20250514") -> Generator:
        """流式发送消息"""
        payload = {"message": message, "model": model}
        headers = {**self.headers, "Accept": "text/event-stream"}

        resp = requests.post(
            f"{self.base_url}/api/v1/chat/stream",
            json=payload,
            headers=headers,
            stream=True,
            timeout=120
        )
        resp.raise_for_status()

        for line in resp.iter_lines():
            if line:
                line = line.decode("utf-8")
                if line.startswith("data: "):
                    yield line[6:]

    def send(self, channel: str, chat_id: str, message: str) -> dict:
        """向指定频道发送消息"""
        resp = requests.post(
            f"{self.base_url}/api/v1/send",
            json={
                "channel": channel,
                "chatId": chat_id,
                "message": message
            },
            headers=self.headers
        )
        resp.raise_for_status()
        return resp.json()

    def health(self) -> dict:
        """健康检查"""
        resp = requests.get(f"{self.base_url}/health")
        return resp.json()


# 使用示例
client = OpenClawClient(token="oc_tok_a1b2c3d4e5f6g7h8i9j0")

# 普通对话
result = client.chat("今天天气怎么样？")
print(result["content"])

# 流式对话
for chunk in client.chat_stream("讲一个故事"):
    print(chunk, end="", flush=True)

# 发送到 Telegram
client.send("telegram", "123456789", "Hello from API!")

六、Node.js 集成示例

6.1 基础封装

// openclaw-client.js
const axios = require('axios');

class OpenClawClient {
  constructor(baseUrl = 'http://localhost:18789', token = '') {
    this.baseUrl = baseUrl;
    this.headers = {
      'Authorization': `Bearer ${token}`,
      'Content-Type': 'application/json'
    };
  }

  async chat(message, options = {}) {
    const { conversationId, model = 'claude-sonnet-4-20250514', systemPrompt } = options;

    const payload = { message, model };
    if (conversationId) payload.conversationId = conversationId;
    if (systemPrompt) payload.systemPrompt = systemPrompt;

    const resp = await axios.post(
      `${this.baseUrl}/api/v1/chat`,
      payload,
      { headers: this.headers, timeout: 120000 }
    );
    return resp.data;
  }

  async send(channel, chatId, message) {
    const resp = await axios.post(
      `${this.baseUrl}/api/v1/send`,
      { channel, chatId, message },
      { headers: this.headers }
    );
    return resp.data;
  }

  async health() {
    const resp = await axios.get(`${this.baseUrl}/health`);
    return resp.data;
  }

  async getModels() {
    const resp = await axios.get(
      `${this.baseUrl}/api/v1/models`,
      { headers: this.headers }
    );
    return resp.data;
  }
}

module.exports = OpenClawClient;

// 使用示例
async function main() {
  const client = new OpenClawClient(
    'http://localhost:18789',
    'oc_tok_a1b2c3d4e5f6g7h8i9j0'
  );

  // 普通对话
  const result = await client.chat('用JavaScript实现斐波那契数列', {
    systemPrompt: '你是一个编程导师，解释要清晰易懂'
  });
  console.log(result.content);

  // 发送到 Discord
  await client.send('discord', 'CHANNEL_ID', '来自 API 的消息');

  // 获取可用模型
  const models = await client.getModels();
  console.log('可用模型:', models);
}

main().catch(console.error);

七、与第三方系统集成

7.1 CRM 系统集成

以下示例将 OpenClaw 集成到 CRM，自动分析客户反馈：

# crm_integration.py
from openclaw_client import OpenClawClient

client = OpenClawClient(token="oc_tok_xxx")

def analyze_customer_feedback(feedback_text, customer_id):
    """分析客户反馈并分类"""
    result = client.chat(
        f"请分析以下客户反馈，并返回 JSON 格式结果：\n\n{feedback_text}",
        options={
            "systemPrompt": """你是客户反馈分析专家。请将反馈分类并返回JSON：
{
  "sentiment": "positive/neutral/negative",
  "category": "产品质量/客户服务/价格/物流/其他",
  "urgency": "high/medium/low",
  "summary": "一句话摘要",
  "suggestedAction": "建议处理方式"
}"""
        }
    )
    return result["content"]

# 使用
analysis = analyze_customer_feedback(
    "你们的产品质量太差了，买了两天就坏了，而且客服态度还很差！",
    "CUST_001"
)
print(analysis)

7.2 工单系统集成

# helpdesk_integration.py
def auto_reply_ticket(ticket):
    """自动生成工单初步回复"""
    result = client.chat(
        f"工单标题：{ticket['title']}\n"
        f"工单描述：{ticket['description']}\n"
        f"优先级：{ticket['priority']}\n\n"
        f"请生成一份专业的初步回复。",
        options={
            "systemPrompt": "你是技术支持专家。基于工单内容提供初步解答和处理建议。回复要专业、友善。"
        }
    )

    return {
        "reply": result["content"],
        "auto_generated": True,
        "model": result.get("model"),
        "needs_review": ticket["priority"] == "high"
    }

7.3 自建 Web 前端

<!-- 一个简单的 Web 聊天前端 -->
<!DOCTYPE html>
<html>
<head>
  <title>AI 助手</title>
  <style>
    #chat { max-width: 600px; margin: 0 auto; padding: 20px; }
    .message { margin: 10px 0; padding: 10px; border-radius: 8px; }
    .user { background: #e3f2fd; text-align: right; }
    .assistant { background: #f5f5f5; }
    #input-area { display: flex; gap: 10px; margin-top: 20px; }
    #message-input { flex: 1; padding: 10px; border: 1px solid #ccc; border-radius: 4px; }
    button { padding: 10px 20px; background: #1976d2; color: white; border: none; border-radius: 4px; cursor: pointer; }
  </style>
</head>
<body>
  <div id="chat">
    <h2>AI 助手</h2>
    <div id="messages"></div>
    <div id="input-area">
      <input id="message-input" placeholder="输入消息..." />
      <button onclick="sendMessage()">发送</button>
    </div>
  </div>

  <script>
    const API_URL = 'http://localhost:18789/api/v1/chat';
    const TOKEN = 'oc_tok_a1b2c3d4e5f6g7h8i9j0';
    let conversationId = null;

    async function sendMessage() {
      const input = document.getElementById('message-input');
      const message = input.value.trim();
      if (!message) return;

      appendMessage('user', message);
      input.value = '';

      try {
        const resp = await fetch(API_URL, {
          method: 'POST',
          headers: {
            'Authorization': `Bearer ${TOKEN}`,
            'Content-Type': 'application/json'
          },
          body: JSON.stringify({
            message,
            conversationId,
            model: 'claude-sonnet-4-20250514'
          })
        });

        const data = await resp.json();
        conversationId = data.conversationId;
        appendMessage('assistant', data.content);
      } catch (error) {
        appendMessage('assistant', '发生错误：' + error.message);
      }
    }

    function appendMessage(role, content) {
      const div = document.createElement('div');
      div.className = `message ${role}`;
      div.textContent = content;
      document.getElementById('messages').appendChild(div);
      div.scrollIntoView({ behavior: 'smooth' });
    }

    document.getElementById('message-input')
      .addEventListener('keydown', e => { if (e.key === 'Enter') sendMessage(); });
  </script>
</body>
</html>

八、API 速率限制

8.1 速率限制配置

{
  "api": {
    "rateLimit": {
      // 全局限制
      "global": {
        "windowMs": 60000,    // 1 分钟窗口
        "maxRequests": 120    // 最多 120 次
      },
      // 按 Token 限制
      "perToken": {
        "windowMs": 60000,
        "maxRequests": 60
      }
    }
  }
}

8.2 处理速率限制错误

当触发速率限制时，API 返回 429 Too Many Requests：

{
  "error": "rate_limit_exceeded",
  "message": "请求过于频繁，请稍后再试",
  "retryAfter": 30
}

在客户端实现重试逻辑：

import time

def chat_with_retry(client, message, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.chat(message)
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                retry_after = int(e.response.headers.get('Retry-After', 30))
                print(f"速率限制，{retry_after}秒后重试...")
                time.sleep(retry_after)
            else:
                raise
    raise Exception("超过最大重试次数")

九、安全最佳实践

{
  "api": {
    "security": {
      // IP 白名单
      "allowedIPs": ["10.0.0.0/8", "192.168.1.0/24"],
      // CORS 配置
      "cors": {
        "enabled": true,
        "origins": ["https://your-app.com"],
        "methods": ["GET", "POST", "DELETE"]
      },
      // 请求体大小限制
      "maxBodySize": "1MB"
    }
  }
}

十、调试与排错

# 查看 API 请求日志
openclaw logs | grep -i "api\|request\|response"

# 使用 verbose curl 调试
curl -v -X POST http://localhost:18789/api/v1/chat \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"message":"test"}'

# 检查 API 服务状态
curl -s http://localhost:18789/health | jq .

通过以上 API 集成方案，你可以将 OpenClaw 的 AI 能力无缝融入任何现有系统，构建出强大的智能化应用。