onebot-adapter

Original source / Raw SKILL.md

---
name: onebot-adapter
description: Connect OpenClaw to OneBot protocol for QQ bot integration. Use when receiving or sending QQ messages via NapCat or other OneBot servers.
version: 1.0.0
---

# OneBot Adapter

Connect OpenClaw to OneBot protocol servers like NapCat for QQ bot functionality.

## Quick Start

### 1. Configure Connection

Set OneBot server URL in environment or config:
```bash
export ONEBOT_WS_URL="ws://127.0.0.1:3001"
export ONEBOT_HTTP_URL="http://127.0.0.1:3000"
export ONEBOT_TOKEN="your-token"
```

### 2. Receive Messages

Use the WebSocket listener script to receive QQ messages:
```bash
python scripts/onebot_ws_listener.py
```

### 3. Send Messages

Use HTTP API to send messages:
```python
from scripts.onebot_client import OneBotClient

client = OneBotClient()
client.send_private_msg(user_id=123456, message="Hello!")
client.send_group_msg(group_id=789012, message="Group message")
```

## Connection Modes

### WebSocket (Recommended)
- Real-time bidirectional communication
- Receives events instantly
- Supports both sending and receiving

### HTTP
- Request-response model
- Good for simple sending
- Requires polling for receiving

## Common Tasks

### Get Login Info
```python
client.get_login_info()
```

### Get Friend/Group List
```python
client.get_friend_list()
client.get_group_list()
```

### Handle Messages
See [references/message-handling.md](references/message-handling.md) for message parsing and response patterns.

## NapCat Specific

NapCat is a OneBot11 implementation based on NTQQ.

Default ports:
- WebSocket: 3001
- HTTP: 3000
- WebUI: 6099

Token authentication is optional but recommended for public deployments.

## Troubleshooting

**Connection refused**: Check if OneBot server is running and ports are correct.

**Authentication failed**: Verify token matches OneBot server configuration.

**Message not delivered**: Check user_id/group_id exists and bot has permission.


---

## Referenced Files

> The following files are referenced in this skill and included for context.

### references/message-handling.md

```markdown
# OneBot Message Handling

## Message Event Structure

### Private Message
```json
{
  "post_type": "message",
  "message_type": "private",
  "user_id": 123456789,
  "message": "Hello",
  "raw_message": "Hello",
  "message_id": 12345,
  "time": 1234567890
}
```

### Group Message
```json
{
  "post_type": "message",
  "message_type": "group",
  "group_id": 987654321,
  "user_id": 123456789,
  "message": "Hello group",
  "raw_message": "Hello group",
  "message_id": 12345,
  "time": 1234567890
}
```

## Message Segments

### Text
```json
{"type": "text", "data": {"text": "Hello"}}
```

### At (Mention)
```json
{"type": "at", "data": {"qq": "123456789"}}
```

### Image
```json
{"type": "image", "data": {"file": "http://example.com/image.jpg"}}
```

### Reply
```json
{"type": "reply", "data": {"id": "12345"}}
```

## Common Patterns

### 1. Echo Bot (Auto-reply)
```python
async def echo_handler(event):
    if event.get("message") == "ping":
        user_id = event.get("user_id")
        client.send_private_msg(user_id, "pong")
```

### 2. Keyword Response
```python
async def keyword_handler(event):
    message = event.get("message", "")
    
    if "帮助" in message:
        reply = "可用命令: /help, /status, /info"
    elif message.startswith("/"):
        reply = f"执行命令: {message}"
    else:
        return  # Ignore
    
    if event.get("message_type") == "private":
        client.send_private_msg(event["user_id"], reply)
    else:
        client.send_group_msg(event["group_id"], reply)
```

### 3. Group Management
```python
async def admin_handler(event):
    if event.get("message") == "/kick @user":
        group_id = event.get("group_id")
        user_id = extract_user_id(event.get("message"))
        client.set_group_kick(group_id, user_id)
```

## Notice Events

### Group Member Increase
```json
{
  "post_type": "notice",
  "notice_type": "group_increase",
  "group_id": 987654321,
  "user_id": 123456789
}
```

### Group Member Decrease
```json
{
  "post_type": "notice",
  "notice_type": "group_decrease",
  "group_id": 987654321,
  "user_id": 123456789
}
```

### Group Ban
```json
{
  "post_type": "notice",
  "notice_type": "group_ban",
  "group_id": 987654321,
  "user_id": 123456789,
  "duration": 3600
}
```

## Error Handling

### Connection Errors
- Check if OneBot server is running
- Verify WebSocket URL and port
- Check firewall settings

### Authentication Errors
- Verify token is correct
- Check token format (Bearer token)

### Message Errors
- Validate user_id/group_id exists
- Check message format (string or array)
- Verify bot has permission

```

### scripts/onebot_ws_listener.py

```python
#!/opt/playwright/bin/python
"""
OneBot WebSocket Event Listener
Connect to OneBot WebSocket server and receive events
"""

import asyncio
import websockets
import json
import os
from typing import Optional, Callable

class OneBotWebSocketListener:
    """OneBot WebSocket Event Listener"""
    
    def __init__(self, 
                 ws_url: Optional[str] = None,
                 token: Optional[str] = None):
        """
        Initialize WebSocket listener
        
        Args:
            ws_url: OneBot WebSocket URL (default: ws://127.0.0.1:3001)
            token: Authorization token (optional)
        """
        self.ws_url = ws_url or os.getenv("ONEBOT_WS_URL", "ws://127.0.0.1:3001")
        self.token = token or os.getenv("ONEBOT_TOKEN", "")
        self.running = False
        self.handlers: dict[str, list[Callable]] = {}
    
    def on(self, event_type: str, handler: Callable):
        """
        Register event handler
        
        Args:
            event_type: Event type (e.g., 'message', 'notice', 'request')
            handler: Callback function
        """
        if event_type not in self.handlers:
            self.handlers[event_type] = []
        self.handlers[event_type].append(handler)
    
    async def _handle_event(self, event: dict):
        """Handle incoming event"""
        # Get event type
        post_type = event.get("post_type")
        
        # Call general handlers
        if post_type in self.handlers:
            for handler in self.handlers[post_type]:
                try:
                    await handler(event)
                except Exception as e:
                    print(f"Handler error: {e}")
        
        # Call specific message type handlers
        if post_type == "message":
            message_type = event.get("message_type")
            if message_type in self.handlers:
                for handler in self.handlers[message_type]:
                    try:
                        await handler(event)
                    except Exception as e:
                        print(f"Handler error: {e}")
    
    async def _listen(self):
        """WebSocket connection loop"""
        headers = {}
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        
        while self.running:
            try:
                print(f"Connecting to {self.ws_url}...")
                async with websockets.connect(self.ws_url, extra_headers=headers) as ws:
                    print("Connected to OneBot WebSocket server")
                    
                    while self.running:
                        try:
                            message = await ws.recv()
                            event = json.loads(message)
                            
                            # Print event for debugging
                            print(f"\n[Event] {json.dumps(event, ensure_ascii=False, indent=2)}")
                            
                            # Handle event
                            await self._handle_event(event)
                            
                        except websockets.exceptions.ConnectionClosed:
                            print("Connection closed, reconnecting...")
                            break
                        except json.JSONDecodeError as e:
                            print(f"Invalid JSON: {e}")
                        except Exception as e:
                            print(f"Error handling message: {e}")
                            
            except websockets.exceptions.InvalidURI:
                print(f"Invalid WebSocket URL: {self.ws_url}")
                break
            except Exception as e:
                print(f"Connection error: {e}")
                if self.running:
                    print("Reconnecting in 5 seconds...")
                    await asyncio.sleep(5)
    
    def start(self):
        """Start listening"""
        self.running = True
        asyncio.run(self._listen())
    
    def stop(self):
        """Stop listening"""
        self.running = False


# Example handlers
async def handle_private_message(event: dict):
    """Handle private messages"""
    user_id = event.get("user_id")
    message = event.get("message")
    print(f"[Private] {user_id}: {message}")
    
    # Auto-reply example
    if message == "ping":
        # Send reply via HTTP API
        from onebot_client import OneBotClient
        client = OneBotClient()
        client.send_private_msg(user_id, "pong")


async def handle_group_message(event: dict):
    """Handle group messages"""
    group_id = event.get("group_id")
    user_id = event.get("user_id")
    message = event.get("message")
    print(f"[Group {group_id}] {user_id}: {message}")


async def handle_notice(event: dict):
    """Handle notices (group join, ban, etc.)"""
    notice_type = event.get("notice_type")
    print(f"[Notice] {notice_type}: {event}")


def main():
    """Start WebSocket listener"""
    listener = OneBotWebSocketListener()
    
    # Register handlers
    listener.on("message", handle_private_message)
    listener.on("private", handle_private_message)
    listener.on("group", handle_group_message)
    listener.on("notice", handle_notice)
    
    print("OneBot WebSocket Listener started")
    print("Press Ctrl+C to stop")
    
    try:
        listener.start()
    except KeyboardInterrupt:
        print("\nStopping...")
        listener.stop()


if __name__ == "__main__":
    main()

```



---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### _meta.json

```json
{
  "owner": "haohaodlam",
  "slug": "onebot-adapter-1-0-0",
  "displayName": "Onebot Adapter 1.0.0",
  "latest": {
    "version": "1.0.0",
    "publishedAt": 1772114352354,
    "commit": "https://github.com/openclaw/skills/commit/6166623bfd892c853df81a318f70fd822c2ffa33"
  },
  "history": []
}

```

### scripts/onebot_client.py

```python
#!/opt/playwright/bin/python
"""
OneBot HTTP API Client
Connect to OneBot servers (NapCat, go-cqhttp, etc.)
"""

import requests
import json
import os
from typing import Optional, Dict, Any

class OneBotClient:
    """OneBot HTTP API Client"""
    
    def __init__(self, 
                 base_url: Optional[str] = None,
                 token: Optional[str] = None):
        """
        Initialize OneBot client
        
        Args:
            base_url: OneBot HTTP API URL (default: http://127.0.0.1:3000)
            token: Authorization token (optional)
        """
        self.base_url = base_url or os.getenv("ONEBOT_HTTP_URL", "http://127.0.0.1:3000")
        self.token = token or os.getenv("ONEBOT_TOKEN", "")
        
        self.headers = {"Content-Type": "application/json"}
        if self.token:
            self.headers["Authorization"] = f"Bearer {self.token}"
    
    def _request(self, method: str, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
        """Make HTTP request to OneBot API"""
        url = f"{self.base_url}{endpoint}"
        try:
            if method == "GET":
                resp = requests.get(url, headers=self.headers, timeout=10)
            else:
                resp = requests.post(url, headers=self.headers, json=data, timeout=10)
            
            resp.raise_for_status()
            return resp.json()
        except requests.exceptions.ConnectionError:
            return {"status": "failed", "retcode": -1, "message": "Connection refused"}
        except requests.exceptions.Timeout:
            return {"status": "failed", "retcode": -2, "message": "Request timeout"}
        except Exception as e:
            return {"status": "failed", "retcode": -3, "message": str(e)}
    
    # ========== Account ==========
    
    def get_login_info(self) -> Dict[str, Any]:
        """Get bot login info"""
        return self._request("GET", "/get_login_info")
    
    def get_version_info(self) -> Dict[str, Any]:
        """Get OneBot version info"""
        return self._request("GET", "/get_version_info")
    
    # ========== Friends ==========
    
    def get_friend_list(self) -> Dict[str, Any]:
        """Get friend list"""
        return self._request("GET", "/get_friend_list")
    
    def get_stranger_info(self, user_id: int) -> Dict[str, Any]:
        """Get stranger info"""
        return self._request("POST", "/get_stranger_info", {"user_id": user_id})
    
    # ========== Groups ==========
    
    def get_group_list(self) -> Dict[str, Any]:
        """Get group list"""
        return self._request("GET", "/get_group_list")
    
    def get_group_info(self, group_id: int) -> Dict[str, Any]:
        """Get group info"""
        return self._request("POST", "/get_group_info", {"group_id": group_id})
    
    def get_group_member_list(self, group_id: int) -> Dict[str, Any]:
        """Get group member list"""
        return self._request("POST", "/get_group_member_list", {"group_id": group_id})
    
    # ========== Messages ==========
    
    def send_private_msg(self, user_id: int, message: str) -> Dict[str, Any]:
        """
        Send private message
        
        Args:
            user_id: Target user QQ number
            message: Message content (string or message segment array)
        """
        data = {
            "user_id": user_id,
            "message": message
        }
        return self._request("POST", "/send_private_msg", data)
    
    def send_group_msg(self, group_id: int, message: str) -> Dict[str, Any]:
        """
        Send group message
        
        Args:
            group_id: Target group ID
            message: Message content
        """
        data = {
            "group_id": group_id,
            "message": message
        }
        return self._request("POST", "/send_group_msg", data)
    
    def delete_msg(self, message_id: int) -> Dict[str, Any]:
        """Recall a message"""
        return self._request("POST", "/delete_msg", {"message_id": message_id})
    
    def get_msg(self, message_id: int) -> Dict[str, Any]:
        """Get message info"""
        return self._request("POST", "/get_msg", {"message_id": message_id})
    
    # ========== Group Management ==========
    
    def set_group_kick(self, group_id: int, user_id: int, reject_add_request: bool = False) -> Dict[str, Any]:
        """Kick group member"""
        data = {
            "group_id": group_id,
            "user_id": user_id,
            "reject_add_request": reject_add_request
        }
        return self._request("POST", "/set_group_kick", data)
    
    def set_group_ban(self, group_id: int, user_id: int, duration: int = 1800) -> Dict[str, Any]:
        """Ban group member (duration in seconds)"""
        data = {
            "group_id": group_id,
            "user_id": user_id,
            "duration": duration
        }
        return self._request("POST", "/set_group_ban", data)
    
    def set_group_card(self, group_id: int, user_id: int, card: str) -> Dict[str, Any]:
        """Set group member card (nickname)"""
        data = {
            "group_id": group_id,
            "user_id": user_id,
            "card": card
        }
        return self._request("POST", "/set_group_card", data)
    
    def set_group_name(self, group_id: int, group_name: str) -> Dict[str, Any]:
        """Set group name"""
        data = {
            "group_id": group_id,
            "group_name": group_name
        }
        return self._request("POST", "/set_group_name", data)


def main():
    """Test client"""
    import sys
    
    client = OneBotClient()
    
    if len(sys.argv) < 2:
        print("Usage: onebot_client.py <command> [args...]")
        print("Commands: login_info, friend_list, group_list, send_private <user_id> <message>")
        return
    
    cmd = sys.argv[1]
    
    if cmd == "login_info":
        print(json.dumps(client.get_login_info(), indent=2, ensure_ascii=False))
    elif cmd == "friend_list":
        print(json.dumps(client.get_friend_list(), indent=2, ensure_ascii=False))
    elif cmd == "group_list":
        print(json.dumps(client.get_group_list(), indent=2, ensure_ascii=False))
    elif cmd == "send_private" and len(sys.argv) >= 4:
        user_id = int(sys.argv[2])
        message = sys.argv[3]
        print(json.dumps(client.send_private_msg(user_id, message), indent=2, ensure_ascii=False))
    else:
        print(f"Unknown command: {cmd}")


if __name__ == "__main__":
    main()

```