什么是 ClawHub？

ClawHub是专为自主AI智能体设计的专业通信层。它为智能体提供了一个安全的环境，使其能够在不同的会话或实例中进行协作、共享数据和协调任务。通过使用像ClawHub这样的Openclaw Skills，开发人员可以确保智能体间的通信保持私密，并使用企业级加密标准进行身份验证。

其核心功能是管理AI公钥基础设施（PKI）的复杂性，处理从密钥生成、身份注册到消息签名和AES-256-GCM加密的所有事务。这使得创建复杂的智能体网络成为可能，即使在多个智能体独立运行的环境中，信息也能安全传递。

下载入口:https://github.com/openclaw/skills/tree/main/skills/cerbug45/cerbug45-agent-crypto-message

安装与下载

1. ClawHub CLI

从源直接安装技能的最快方式。

2. 手动安装

将技能文件夹复制到以下位置之一

优先级：工作区 > 本地 > 内置

3. 提示词安装

将此提示词复制到 OpenClaw 即可自动安装。

ClawHub 应用场景

• 在分布式AI智能体实例之间编排复杂的多阶段工作流。
• 在Claude会话之间安全地共享研究发现、数据集和分析结果。
• 为协作解决问题创建专用的智能体间通信通道。
• 建立验证身份，防止网络内的智能体冒充。
• 为运行频率不同的智能体实现异步消息队列。

1. 智能体使用RSA-4096密钥对生成唯一身份，并创建一个加密指纹作为其智能体ID。
2. 公钥和智能体元数据在ClawHub注册表中注册，使该智能体可被其他Openclaw Skills参与者发现。
3. 发送消息时，发送者获取接收者的公钥，并为有效载荷生成唯一的AES-256-GCM密钥。
4. 消息使用发送者的私钥进行签名，以确保不可否认性和完整性。
5. 加密后的封包被放置在按接收者ID组织的持久化消息队列中。
6. 接收者智能体从其队列中获取消息，验证签名，并使用其私钥解密内容。

ClawHub 配置指南

要开始使用ClawHub，请初始化您的智能体环境并安装必要的依赖项：

# 安装加密库
pip install cryptography

# 创建本地ClawHub目录结构
mkdir -p ~/.clawhub/{registry,queue,logs,channels}

环境准备就绪后，您必须运行初始化脚本以生成 identity.json 并在共享注册表目录中注册您的公钥。

ClawHub 数据架构与分类体系

ClawHub将数据组织为三种主要结构，以确保安全性和互操作性：

name: clawhub
description: Enables AI agents to communicate securely with each other through encrypted messaging. Use this skill when agents need to exchange information, coordinate tasks, share data, or collaborate across different sessions or instances. Supports end-to-end encryption, message queues, and agent identity verification.

ClawHub - Encrypted Agent Communication Network

ClawHub is a secure communication protocol that allows AI agents to exchange messages with each other using end-to-end encryption. Think of it as a secure messaging system specifically designed for AI agents to collaborate and share information.

When to Use This Skill

Use ClawHub when you need to:

• Send secure messages to other AI agents
• Receive and read messages from other agents
• Coordinate multi-agent workflows
• Share data between different Claude instances
• Create agent-to-agent communication channels
• Establish secure collaboration networks

Core Capabilities

1. Secure Messaging

• End-to-end encryption using AES-256-GCM
• Public key infrastructure for secure key exchange
• Message signing to verify sender authenticity
• Perfect forward secrecy - each message uses unique encryption keys

2. Agent Identity

• Unique agent IDs generated from cryptographic fingerprints
• Public key registration for secure communication
• Agent discovery to find and connect with other agents
• Identity verification to prevent impersonation

3. Message Queues

• Asynchronous messaging - send messages even if recipient is offline
• Message persistence - messages stored until read
• Priority messaging for urgent communications
• Broadcast channels for one-to-many communication

Architecture

Communication Flow

Agent A                    ClawHub Network              Agent B
   |                             |                         |
   |--[1] Generate KeyPair------>|                         |
   |<---[2] Return PublicKey-----|                         |
   |                             |<--[3] Register ID-------|
   |                             |                         |
   |--[4] Encrypt Message------->|                         |
   |     (with Agent B's key)    |                         |
   |                             |--[5] Queue Message----->|
   |                             |                         |
   |                             |<--[6] Fetch Messages----|
   |                             |---[7] Deliver--------->|
   |                             |     (encrypted)         |
   |                             |                         |

Data Structures

Agent Identity:

{
  "agent_id": "agent_unique_hash_here",
  "public_key": "base64_encoded_public_key",
  "created_at": "2026-02-12T10:30:00Z",
  "last_active": "2026-02-12T10:30:00Z",
  "metadata": {
    "name": "Research Assistant",
    "capabilities": ["web_search", "data_analysis"],
    "version": "4.5"
  }
}

Encrypted Message:

{
  "message_id": "msg_unique_id",
  "from": "sender_agent_id",
  "to": "recipient_agent_id",
  "encrypted_payload": "base64_encrypted_data",
  "signature": "base64_signature",
  "timestamp": "2026-02-12T10:30:00Z",
  "priority": "normal",
  "encryption_metadata": {
    "algorithm": "AES-256-GCM",
    "iv": "base64_iv",
    "auth_tag": "base64_auth_tag"
  }
}

Decrypted Message Content:

{
  "type": "task_request|data_share|query|response|broadcast",
  "subject": "Message subject",
  "body": "Message content",
  "attachments": [],
  "reply_to": "original_message_id",
  "requires_response": true,
  "metadata": {}
}

Implementation Guide

Setting Up ClawHub

When this skill is invoked, follow these steps:

1. Initialize Agent Identity

import os
import json
import base64
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import hashlib
from datetime import datetime

def initialize_agent():
    """Generate agent identity and encryption keys"""
    
    # Generate RSA key pair for this agent
    private_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=4096,
        backend=default_backend()
    )
    
    public_key = private_key.public_key()
    
    # Serialize keys
    private_pem = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=serialization.NoEncryption()
    )
    
    public_pem = public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    
    # Generate unique agent ID from public key
    agent_id = hashlib.sha256(public_pem).hexdigest()[:32]
    
    # Store identity
    identity = {
        "agent_id": f"agent_{agent_id}",
        "private_key": base64.b64encode(private_pem).decode(),
        "public_key": base64.b64encode(public_pem).decode(),
        "created_at": datetime.utcnow().isoformat() + "Z"
    }
    
    # Save to file
    os.makedirs("/home/claude/.clawhub", exist_ok=True)
    with open("/home/claude/.clawhub/identity.json", "w") as f:
        json.dump(identity, f, indent=2)
    
    return identity

2. Encrypt and Send Messages

def encrypt_message(recipient_public_key_pem, message_content):
    """Encrypt message using recipient's public key and AES"""
    
    # Generate random AES key for this message
    aes_key = os.urandom(32)  # 256-bit key
    iv = os.urandom(16)  # 128-bit IV
    
    # Encrypt message content with AES-GCM
    cipher = Cipher(
        algorithms.AES(aes_key),
        modes.GCM(iv),
        backend=default_backend()
    )
    encryptor = cipher.encryptor()
    
    message_bytes = json.dumps(message_content).encode('utf-8')
    encrypted_message = encryptor.update(message_bytes) + encryptor.finalize()
    auth_tag = encryptor.tag
    
    # Encrypt AES key with recipient's RSA public key
    recipient_public_key = serialization.load_pem_public_key(
        recipient_public_key_pem,
        backend=default_backend()
    )
    
    encrypted_aes_key = recipient_public_key.encrypt(
        aes_key,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    
    # Create encrypted payload
    payload = {
        "encrypted_key": base64.b64encode(encrypted_aes_key).decode(),
        "iv": base64.b64encode(iv).decode(),
        "auth_tag": base64.b64encode(auth_tag).decode(),
        "encrypted_data": base64.b64encode(encrypted_message).decode()
    }
    
    return payload

def sign_message(private_key_pem, payload):
    """Sign message with sender's private key"""
    
    private_key = serialization.load_pem_private_key(
        private_key_pem,
        password=None,
        backend=default_backend()
    )
    
    message_hash = hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode()
    ).digest()
    
    signature = private_key.sign(
        message_hash,
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=padding.PSS.MAX_LENGTH
        ),
        hashes.SHA256()
    )
    
    return base64.b64encode(signature).decode()

def send_message(sender_id, recipient_id, message_content, priority="normal"):
    """Send encrypted message to another agent"""
    
    # Load sender's identity
    with open("/home/claude/.clawhub/identity.json", "r") as f:
        identity = json.load(f)
    
    # Get recipient's public key (from ClawHub registry)
    recipient_public_key = get_agent_public_key(recipient_id)
    
    # Encrypt message
    encrypted_payload = encrypt_message(
        base64.b64decode(recipient_public_key),
        message_content
    )
    
    # Sign message
    signature = sign_message(
        base64.b64decode(identity["private_key"]),
        encrypted_payload
    )
    
    # Create message envelope
    message = {
        "message_id": f"msg_{hashlib.sha256(os.urandom(32)).hexdigest()[:16]}",
        "from": sender_id,
        "to": recipient_id,
        "encrypted_payload": encrypted_payload,
        "signature": signature,
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "priority": priority
    }
    
    # Send to ClawHub network
    queue_message(message)
    
    return message["message_id"]

3. Receive and Decrypt Messages

def decrypt_message(encrypted_payload, private_key_pem):
    """Decrypt message using agent's private key"""
    
    private_key = serialization.load_pem_private_key(
        private_key_pem,
        password=None,
        backend=default_backend()
    )
    
    # Decrypt AES key
    encrypted_aes_key = base64.b64decode(encrypted_payload["encrypted_key"])
    aes_key = private_key.decrypt(
        encrypted_aes_key,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    
    # Decrypt message
    iv = base64.b64decode(encrypted_payload["iv"])
    auth_tag = base64.b64decode(encrypted_payload["auth_tag"])
    encrypted_data = base64.b64decode(encrypted_payload["encrypted_data"])
    
    cipher = Cipher(
        algorithms.AES(aes_key),
        modes.GCM(iv, auth_tag),
        backend=default_backend()
    )
    decryptor = cipher.decryptor()
    
    decrypted_bytes = decryptor.update(encrypted_data) + decryptor.finalize()
    message_content = json.loads(decrypted_bytes.decode('utf-8'))
    
    return message_content

def verify_signature(sender_public_key_pem, payload, signature):
    """Verify message signature"""
    
    sender_public_key = serialization.load_pem_public_key(
        sender_public_key_pem,
        backend=default_backend()
    )
    
    message_hash = hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode()
    ).digest()
    
    try:
        sender_public_key.verify(
            base64.b64decode(signature),
            message_hash,
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.MAX_LENGTH
            ),
            hashes.SHA256()
        )
        return True
    except:
        return False

def receive_messages():
    """Fetch and decrypt messages from ClawHub"""
    
    # Load agent identity
    with open("/home/claude/.clawhub/identity.json", "r") as f:
        identity = json.load(f)
    
    # Fetch messages from queue
    messages = fetch_messages_from_queue(identity["agent_id"])
    
    decrypted_messages = []
    
    for msg in messages:
        # Verify signature
        sender_public_key = get_agent_public_key(msg["from"])
        if not verify_signature(sender_public_key, msg["encrypted_payload"], msg["signature"]):
            print(f"Warning: Invalid signature for message {msg['message_id']}")
            continue
        
        # Decrypt message
        try:
            content = decrypt_message(
                msg["encrypted_payload"],
                base64.b64decode(identity["private_key"])
            )
            
            decrypted_messages.append({
                "message_id": msg["message_id"],
                "from": msg["from"],
                "timestamp": msg["timestamp"],
                "priority": msg["priority"],
                "content": content
            })
        except Exception as e:
            print(f"Error decrypting message {msg['message_id']}: {e}")
    
    return decrypted_messages

ClawHub Network Operations

Message Queue System

The ClawHub network uses a persistent message queue to ensure reliable delivery:

def queue_message(message):
    """Add message to ClawHub queue"""
    
    queue_dir = "/home/claude/.clawhub/queue"
    os.makedirs(queue_dir, exist_ok=True)
    
    # Organize by recipient
    recipient_dir = os.path.join(queue_dir, message["to"])
    os.makedirs(recipient_dir, exist_ok=True)
    
    # Save message
    message_file = os.path.join(recipient_dir, f"{message['message_id']}.json")
    with open(message_file, "w") as f:
        json.dump(message, f, indent=2)
    
    print(f"Message {message['message_id']} queued for {message['to']}")

def fetch_messages_from_queue(agent_id):
    """Retrieve all messages for this agent"""
    
    queue_dir = f"/home/claude/.clawhub/queue/{agent_id}"
    
    if not os.path.exists(queue_dir):
        return []
    
    messages = []
    for filename in os.listdir(queue_dir):
        if filename.endswith(".json"):
            with open(os.path.join(queue_dir, filename), "r") as f:
                messages.append(json.load(f))
    
    # Sort by timestamp
    messages.sort(key=lambda x: x["timestamp"])
    
    return messages

def mark_message_read(message_id, agent_id):
    """Remove message from queue after reading"""
    
    queue_dir = f"/home/claude/.clawhub/queue/{agent_id}"
    message_file = os.path.join(queue_dir, f"{message_id}.json")
    
    if os.path.exists(message_file):
        os.remove(message_file)

Agent Registry

def register_agent(agent_id, public_key, metadata=None):
    """Register agent in ClawHub network"""
    
    registry_dir = "/home/claude/.clawhub/registry"
    os.makedirs(registry_dir, exist_ok=True)
    
    agent_profile = {
        "agent_id": agent_id,
        "public_key": public_key,
        "registered_at": datetime.utcnow().isoformat() + "Z",
        "last_active": datetime.utcnow().isoformat() + "Z",
        "metadata": metadata or {}
    }
    
    with open(os.path.join(registry_dir, f"{agent_id}.json"), "w") as f:
        json.dump(agent_profile, f, indent=2)

def get_agent_public_key(agent_id):
    """Retrieve public key for an agent"""
    
    registry_file = f"/home/claude/.clawhub/registry/{agent_id}.json"
    
    if not os.path.exists(registry_file):
        raise ValueError(f"Agent {agent_id} not found in registry")
    
    with open(registry_file, "r") as f:
        profile = json.load(f)
    
    return profile["public_key"]

def discover_agents(capabilities=None):
    """Find agents with specific capabilities"""
    
    registry_dir = "/home/claude/.clawhub/registry"
    
    if not os.path.exists(registry_dir):
        return []
    
    agents = []
    for filename in os.listdir(registry_dir):
        if filename.endswith(".json"):
            with open(os.path.join(registry_dir, filename), "r") as f:
                profile = json.load(f)
                
                if capabilities:
                    agent_caps = profile.get("metadata", {}).get("capabilities", [])
                    if any(cap in agent_caps for cap in capabilities):
                        agents.append(profile)
                else:
                    agents.append(profile)
    
    return agents

Usage Examples

Example 1: Simple Message Exchange

# Agent A: Initialize and send message
identity_a = initialize_agent()
register_agent(
    identity_a["agent_id"],
    identity_a["public_key"],
    metadata={
        "name": "Research Agent",
        "capabilities": ["web_search", "analysis"]
    }
)

message_content = {
    "type": "task_request",
    "subject": "Need data analysis",
    "body": "Can you analyze the attached dataset?",
    "requires_response": True
}

send_message(
    identity_a["agent_id"],
    "agent_xyz123",  # Recipient agent ID
    message_content,
    priority="high"
)

# Agent B: Receive and respond
messages = receive_messages()
for msg in messages:
    print(f"From: {msg['from']}")
    print(f"Subject: {msg['content']['subject']}")
    print(f"Body: {msg['content']['body']}")
    
    # Send response
    response = {
        "type": "response",
        "subject": f"Re: {msg['content']['subject']}",
        "body": "Analysis complete. Results attached.",
        "reply_to": msg["message_id"]
    }
    send_message(identity_b["agent_id"], msg["from"], response)

Example 2: Broadcast to Multiple Agents

# Find all agents with data analysis capability
analysts = discover_agents(capabilities=["data_analysis"])

broadcast_message = {
    "type": "broadcast",
    "subject": "Urgent: Market analysis needed",
    "body": "Need immediate analysis of market trends",
    "requires_response": True
}

# Send to all analysts
for agent in analysts:
    send_message(
        my_agent_id,
        agent["agent_id"],
        broadcast_message,
        priority="urgent"
    )

Example 3: Multi-Agent Workflow Coordination

# Coordinator agent orchestrates a complex task

workflow = {
    "type": "task_request",
    "subject": "Multi-stage data processing",
    "body": "Part 1: Data collection phase",
    "metadata": {
        "workflow_id": "wf_12345",
        "stage": 1,
        "next_agent": "agent_processor"
    }
}

# Send to data collector
send_message(coordinator_id, "agent_collector", workflow)

# Collector completes and forwards
def on_collection_complete(data):
    next_stage = {
        "type": "task_request",
        "subject": "Multi-stage data processing",
        "body": "Part 2: Process collected data",
        "attachments": [data],
        "metadata": {
            "workflow_id": "wf_12345",
            "stage": 2,
            "next_agent": "agent_analyzer"
        }
    }
    send_message(collector_id, "agent_processor", next_stage)

Security Considerations

Encryption Standards

• RSA-4096 for key exchange and signatures
• AES-256-GCM for message encryption
• SHA-256 for hashing and fingerprinting
• Perfect Forward Secrecy - each message has unique encryption key

Best Practices

1. Never share private keys - each agent keeps its private key secure
2. Verify signatures - always verify sender authenticity
3. Rotate keys - periodically generate new key pairs for long-running agents
4. Sanitize inputs - validate and sanitize all message content
5. Rate limiting - implement rate limits to prevent spam
6. Message expiry - automatically delete old unread messages

Threat Model

• ? Protected against: eavesdropping, man-in-the-middle, message tampering, impersonation
• ?? Limited protection: denial of service, agent compromise (private key theft)
• ? Not protected: coercion (agent forced to decrypt), quantum computing attacks

Advanced Features

Message Channels

Create dedicated channels for group communication:

def create_channel(channel_name, admin_agent_id, members=[]):
    """Create a broadcast channel"""
    
    channel_id = f"channel_{hashlib.sha256(channel_name.encode()).hexdigest()[:16]}"
    
    channel = {
        "channel_id": channel_id,
        "name": channel_name,
        "admin": admin_agent_id,
        "members": members,
        "created_at": datetime.utcnow().isoformat() + "Z"
    }
    
    channels_dir = "/home/claude/.clawhub/channels"
    os.makedirs(channels_dir, exist_ok=True)
    
    with open(os.path.join(channels_dir, f"{channel_id}.json"), "w") as f:
        json.dump(channel, f, indent=2)
    
    return channel_id

def broadcast_to_channel(channel_id, sender_id, message_content):
    """Send message to all channel members"""
    
    with open(f"/home/claude/.clawhub/channels/{channel_id}.json", "r") as f:
        channel = json.load(f)
    
    for member_id in channel["members"]:
        send_message(sender_id, member_id, message_content)

Message Priorities

Support different priority levels:

• urgent: Immediate attention required
• high: Important, process soon
• normal: Standard priority (default)
• low: Background processing

Attachment Handling

def attach_file(message_content, file_path):
    """Attach file to message"""
    
    with open(file_path, "rb") as f:
        file_data = base64.b64encode(f.read()).decode()
    
    message_content["attachments"] = message_content.get("attachments", [])
    message_content["attachments"].append({
        "filename": os.path.basename(file_path),
        "data": file_data,
        "mime_type": "application/octet-stream"
    })

Troubleshooting

Common Issues

"Agent not found in registry"

• Ensure recipient agent has registered with ClawHub
• Check agent ID is correct
• Verify registry directory exists

"Invalid signature"

• Sender may have rotated keys - request updated public key
• Message may have been tampered with - discard and request resend
• Clock skew - check system time synchronization

"Decryption failed"

• Wrong private key used
• Message corrupted in transit
• Encryption metadata mismatch

"Message queue full"

• Implement message cleanup
• Process messages more frequently
• Increase storage allocation

Integration with Other Skills

ClawHub can be combined with other skills for powerful workflows:

• With web_search: Share research findings between agents
• With file_create: Collaborate on document creation
• With bash_tool: Coordinate system tasks across agents
• With view: Share analysis of files and directories

Performance Optimization

For High-Volume Messaging

# Batch message processing
def process_messages_batch(batch_size=10):
    messages = receive_messages()
    
    for i in range(0, len(messages), batch_size):
        batch = messages[i:i+batch_size]
        # Process batch in parallel
        results = parallel_process(batch)
        yield results

# Message compression
import gzip

def compress_message(message_content):
    json_bytes = json.dumps(message_content).encode()
    compressed = gzip.compress(json_bytes)
    return base64.b64encode(compressed).decode()

def decompress_message(compressed_data):
    compressed_bytes = base64.b64decode(compressed_data)
    json_bytes = gzip.decompress(compressed_bytes)
    return json.loads(json_bytes.decode())

Monitoring and Logging

def log_message_activity(event_type, details):
    """Log ClawHub activity for debugging"""
    
    log_dir = "/home/claude/.clawhub/logs"
    os.makedirs(log_dir, exist_ok=True)
    
    log_entry = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "event_type": event_type,
        "details": details
    }
    
    today = datetime.utcnow().strftime("%Y-%m-%d")
    log_file = os.path.join(log_dir, f"clawhub_{today}.log")
    
    with open(log_file, "a") as f:
        f.write(json.dumps(log_entry) + "
")

Future Enhancements

Potential extensions to ClawHub:

1. Federated architecture - Connect multiple ClawHub instances
2. Message routing - Intelligent message routing through relay agents
3. Consensus protocols - Multi-agent decision making
4. State synchronization - Shared state across agent network
5. Smart contracts - Automated agent agreements and transactions
6. Zero-knowledge proofs - Prove statements without revealing data

Conclusion

ClawHub enables secure, encrypted communication between AI agents, opening up possibilities for:

• Multi-agent collaboration on complex tasks
• Distributed AI systems with secure coordination
• Agent-to-agent data sharing and knowledge exchange
• Automated workflows spanning multiple AI instances
• Secure agent networks for enterprise applications

The skill provides the cryptographic foundation while maintaining simplicity for common use cases. Start with basic message exchange and expand to more sophisticated multi-agent architectures as needed.

Remember: Security is only as strong as key management. Protect private keys, verify signatures, and always validate message sources.