agent-protocol – OpenClaw Skill

---

name: agent-protocol description: Agent-to-agent communication protocol. Enables skills to communicate via events, build workflow chains, and orchestrate without human intervention. version: 1.0.0

Agent-to-Agent Protocol

A foundational communication layer for Clawdbot skills and agents.

Enable your agents to talk to each other, build automated workflows, and orchestrate complex multi-step tasks without human intervention.

Vision

Research-Agent finds article
    ↓ publishes "research.found"
Summary-Agent subscribes to events
    ↓ generates digest
    ↓ publishes "summary.ready"
Notification-Agent subscribes
    ↓ posts to Telegram/Discord

Architecture

1. Event Bus (File-based Message Passing)

• Agents publish events to ~/.clawdbot/events/
• Events are JSON files with schema validation
• Persistent, debuggable, auditable
• Automatic cleanup of processed events

2. Workflow Engine

• Define pipelines in JSON or YAML
• Conditional routing based on event data
• Error handling, retries, fallbacks
• Cron integration for scheduled execution

3. Shared Context

• Agents read/write to shared memory space
• Context passing between workflow steps
• State persistence across agent invocations

4. Agent Registry

• Discover available agents/skills
• Capability advertisement
• Permission management

Core Concepts

Events

Events are the fundamental unit of communication:

{
  "event_id": "evt_20260128_001",
  "event_type": "research.article_found",
  "timestamp": "2026-01-28T23:00:00Z",
  "source_agent": "research-agent",
  "payload": {
    "title": "ETH 2.0 Upgrade Complete",
    "url": "https://example.com/article",
    "importance": 9,
    "summary": "Major Ethereum upgrade..."
  },
  "metadata": {
    "session_id": "main",
    "requires_action": true
  }
}

Workflows

Workflows define how agents respond to events:

{
  "workflow_id": "research-to-telegram",
  "name": "Research → Summary → Notification",
  "trigger": {
    "event_type": "research.article_found",
    "conditions": {
      "payload.importance": { "gte": 7 }
    }
  },
  "steps": [
    {
      "agent": "summary-agent",
      "action": "summarize",
      "input": "{{payload}}",
      "output_event": "summary.ready"
    },
    {
      "agent": "notification-agent",
      "action": "notify",
      "input": "{{previous.summary}}",
      "channels": ["telegram"]
    }
  ]
}

Quick Start

1. Installation

cd /root/clawd/skills/agent-protocol
python3 scripts/setup.py

2. Start Event Bus

python3 scripts/event_bus.py start

3. Publish Your First Event

python3 scripts/publish.py \
  --type "test.hello" \
  --source "my-agent" \
  --payload '{"message": "Hello, world!"}'

4. Subscribe to Events

python3 scripts/subscribe.py \
  --types "test.hello" \
  --handler "./my_handler.py"

5. Define a Workflow

cp examples/simple-workflow.json config/workflows/my-workflow.json
python3 scripts/workflow_engine.py --validate

Event Types (Conventions)

Standard Event Types

• research.article_found - Research agent found relevant content
• research.topic_suggested - New research topic suggested
• summary.ready - Summary generated
• analytics.insight - Personal analytics insight
• sports.goal_scored - Sports ticker goal event
• sports.match_started - Match started
• notification.sent - Notification delivered
• workflow.started - Workflow execution started
• workflow.completed - Workflow completed
• workflow.failed - Workflow failed

Event Naming Convention

<domain>.<action_past_tense>

• Use lowercase, underscores
• Domain: broad category (research, sports, notification)
• Action: what happened (article_found, goal_scored)

Workflow Examples

Example 1: Research → Notification

{
  "workflow_id": "eth-news-alert",
  "trigger": {
    "event_type": "research.article_found",
    "conditions": {
      "payload.keywords": { "contains": ["ethereum", "ETH"] },
      "payload.importance": { "gte": 8 }
    }
  },
  "steps": [
    {
      "agent": "notification-agent",
      "action": "send_telegram",
      "input": {
        "message": "🚨 Important ETH News!\n{{payload.title}}\n{{payload.url}}"
      }
    }
  ]
}

Example 2: Sports Goal → TTS Announcement

{
  "workflow_id": "goal-announcement",
  "trigger": {
    "event_type": "sports.goal_scored",
    "conditions": {
      "payload.team": { "eq": "Barcelona" }
    }
  },
  "steps": [
    {
      "agent": "tts-agent",
      "action": "announce",
      "input": {
        "text": "Goal for Barcelona! {{payload.scorer}} scores! {{payload.score}}"
      }
    }
  ]
}

Example 3: Daily Analytics → Research Topics

{
  "workflow_id": "analytics-to-research",
  "trigger": {
    "event_type": "analytics.daily_report",
    "schedule": "0 9 * * *"
  },
  "steps": [
    {
      "agent": "analytics-agent",
      "action": "generate_insights",
      "output_event": "analytics.insights_ready"
    },
    {
      "agent": "research-agent",
      "action": "suggest_topics",
      "input": "{{previous.insights}}",
      "conditions": {
        "previous.insights.count": { "gte": 3 }
      }
    }
  ]
}

Commands

Event Bus

# Start the event bus daemon
python3 scripts/event_bus.py start

# Check status
python3 scripts/event_bus.py status

# Stop
python3 scripts/event_bus.py stop

# View recent events
python3 scripts/event_bus.py tail --count 20

Publishing Events

# Publish event (JSON payload)
python3 scripts/publish.py \
  --type "research.article_found" \
  --source "research-agent" \
  --payload '{"title": "Article", "url": "..."}'

# Publish from file
python3 scripts/publish.py --file event.json

# Publish with priority
python3 scripts/publish.py \
  --type "alert.urgent" \
  --priority high \
  --payload '{"message": "Critical alert!"}'

Subscribing to Events

# Subscribe to event types
python3 scripts/subscribe.py \
  --types "research.*,sports.goal_scored" \
  --handler "./handlers/my_handler.py"

# Subscribe with filter
python3 scripts/subscribe.py \
  --types "research.*" \
  --filter '{"importance": {"gte": 8}}' \
  --handler "./handlers/important_only.py"

# List active subscriptions
python3 scripts/subscribe.py --list

Workflow Management

# Validate workflows
python3 scripts/workflow_engine.py --validate

# Run workflow engine (processes workflows)
python3 scripts/workflow_engine.py --run

# Test specific workflow
python3 scripts/workflow_engine.py --test eth-news-alert

# List workflows
python3 scripts/workflow_engine.py --list

# Enable/disable workflow
python3 scripts/workflow_engine.py --enable research-to-telegram
python3 scripts/workflow_engine.py --disable research-to-telegram

Agent Registry

# Register your agent
python3 scripts/registry.py register \
  --name "my-agent" \
  --capabilities "summarize,notify" \
  --events "research.article_found"

# List available agents
python3 scripts/registry.py list

# Query agents by capability
python3 scripts/registry.py query --capability "summarize"

Integration with Existing Skills

Sports Ticker Integration

Modify sports-ticker/scripts/live_monitor.py to publish events:

from agent_protocol import publish_event

# After detecting a goal:
publish_event(
    event_type="sports.goal_scored",
    source="sports-ticker",
    payload={
        "team": team_name,
        "scorer": player_name,
        "opponent": opponent,
        "score": f"{home_score}-{away_score}",
        "minute": clock
    }
)

Research Agent Integration

from agent_protocol import publish_event

# After finding an article:
publish_event(
    event_type="research.article_found",
    source="research-agent",
    payload={
        "title": article_title,
        "url": article_url,
        "importance": calculate_importance(article),
        "summary": snippet
    }
)

Personal Analytics Integration

from agent_protocol import publish_event

# Daily insights:
publish_event(
    event_type="analytics.insight",
    source="personal-analytics",
    payload={
        "type": "productivity",
        "insight": "Your focus time increased 20% this week",
        "recommendations": ["Schedule deep work in morning"]
    }
)

Security & Permissions

Permission Model

{
  "agent": "research-agent",
  "permissions": {
    "publish": ["research.*"],
    "subscribe": ["summary.*", "notification.*"],
    "workflows": ["research-to-telegram"]
  }
}

Sandboxing

• Agents can only publish to their designated event types
• Subscriptions require explicit permission
• Workflows are validated before execution

Configuration

Main Config: config/protocol.json

{
  "event_bus": {
    "storage_path": "~/.clawdbot/events",
    "retention_days": 7,
    "max_event_size_kb": 512
  },
  "workflow_engine": {
    "enabled": true,
    "poll_interval_seconds": 30,
    "max_concurrent_workflows": 5
  },
  "registry": {
    "agents_path": "~/.clawdbot/agents/registry.json"
  },
  "security": {
    "require_permissions": true,
    "audit_log": true
  }
}

Advanced Features

1. Conditional Routing

{
  "steps": [
    {
      "condition": {
        "payload.importance": { "gte": 9 }
      },
      "then": { "agent": "urgent-notifier" },
      "else": { "agent": "standard-notifier" }
    }
  ]
}

2. Parallel Execution

{
  "steps": [
    {
      "parallel": [
        { "agent": "telegram-notifier" },
        { "agent": "discord-notifier" },
        { "agent": "email-notifier" }
      ]
    }
  ]
}

3. Error Handling

{
  "steps": [
    {
      "agent": "external-api",
      "retry": {
        "max_attempts": 3,
        "backoff_seconds": 5
      },
      "on_error": {
        "agent": "error-logger",
        "continue": true
      }
    }
  ]
}

4. Scheduled Workflows

{
  "trigger": {
    "schedule": "0 9 * * *",
    "event_type": "cron.daily_run"
  }
}

Monitoring & Debugging

Event Log

All events are logged to ~/.clawdbot/events/log/

# View event log
tail -f ~/.clawdbot/events/log/events.log

# Search events
python3 scripts/query.py --type "research.*" --since "1 hour ago"

Workflow Execution Log

# View workflow executions
python3 scripts/workflow_engine.py --history

# Inspect failed workflow
python3 scripts/workflow_engine.py --inspect <workflow_id>

Metrics

# Show event statistics
python3 scripts/metrics.py

# Output:
# Total events published: 1,234
# Event types: 15
# Active subscriptions: 8
# Workflows executed: 456
# Average workflow duration: 2.3s

Best Practices

1. Event Design

   • Keep payloads small and focused
   • Include enough context for handlers
   • Use consistent naming conventions

2. Workflow Design

   • Keep workflows simple and focused
   • Use descriptive names
   • Test thoroughly before enabling

3. Error Handling

   • Always define error handlers
   • Log errors for debugging
   • Use retries for transient failures

4. Performance

   • Avoid high-frequency events
   • Clean up old events regularly
   • Monitor workflow execution times

5. Security

   • Validate event payloads
   • Use permission system
   • Audit sensitive operations

Python API

from agent_protocol import (
    publish_event,
    subscribe,
    create_workflow,
    register_agent
)

# Publish event
publish_event(
    event_type="my.event",
    source="my-agent",
    payload={"key": "value"}
)

# Subscribe to events
@subscribe(["research.*"])
def handle_research(event):
    print(f"Got research event: {event['payload']}")

# Create workflow programmatically
workflow = create_workflow(
    workflow_id="my-workflow",
    trigger={"event_type": "my.trigger"},
    steps=[
        {"agent": "processor", "action": "process"}
    ]
)

# Register agent
register_agent(
    name="my-agent",
    capabilities=["process", "notify"],
    event_types=["my.event"]
)

JavaScript API

const { publishEvent, subscribe, createWorkflow } = require('./scripts/protocol.js');

// Publish event
await publishEvent({
  eventType: 'my.event',
  source: 'my-agent',
  payload: { key: 'value' }
});

// Subscribe
subscribe(['research.*'], (event) => {
  console.log('Got event:', event);
});

// Create workflow
await createWorkflow({
  workflowId: 'my-workflow',
  trigger: { eventType: 'my.trigger' },
  steps: [
    { agent: 'processor', action: 'process' }
  ]
});

Roadmap

• Visual workflow builder (web UI)
• WebSocket support for real-time events
• Cross-instance event relay (multi-bot networks)
• AI-powered workflow suggestions
• Event replay and debugging tools
• Performance profiling
• GraphQL query API for events

Contributing

This skill is part of Clawdbot's core infrastructure. Contributions welcome!

License

MIT

---

Built with 🦎 by Robby

Agent-to-Agent Protocol

Enable your Clawdbot skills to communicate and orchestrate workflows automatically.

What Is This?

The Agent Protocol allows Clawdbot skills to:

• Publish events when something interesting happens
• Subscribe to events from other skills
• Chain workflows across multiple agents
• Automate complex tasks without human intervention

Real-World Examples

1. Research → Summary → Notification

Research Agent finds article about ETH
    ↓ publishes "research.article_found"
Summary Agent sees event
    ↓ generates digest
    ↓ publishes "summary.ready"
Telegram Notifier sees event
    ↓ sends notification

2. Sports Goal → TTS Announcement

Sports Ticker detects Barcelona goal
    ↓ publishes "sports.goal_scored"
TTS Agent sees event
    ↓ announces via speaker
"Goal for Barcelona! Messi scores!"

3. Daily Analytics → Research Topics

Cron triggers at 9 AM
    ↓ Personal Analytics generates insights
    ↓ publishes "analytics.insights_ready"
Research Agent sees event
    ↓ suggests topics based on insights

Quick Start

1. Installation

cd /root/clawd/skills/agent-protocol
cp config.example.json config/protocol.json
chmod +x scripts/*.py

2. Test the Event Bus

# Publish a test event
python3 scripts/publish.py \
  --type "test.hello" \
  --source "my-skill" \
  --payload '{"message": "Hello from agent protocol!"}'

# Check event bus status
python3 scripts/event_bus.py status

3. Create Your First Workflow

# Copy an example workflow
cp examples/simple-workflow.json config/workflows/my-workflow.json

# Edit it for your needs
nano config/workflows/my-workflow.json

# Validate
python3 scripts/workflow_engine.py --validate

4. Start the Workflow Engine

# Run once to process pending events
python3 scripts/workflow_engine.py --run

# Or run as daemon (polls every 30 seconds)
python3 scripts/workflow_engine.py --daemon

Integration Guide

From Python Skills

import sys
from pathlib import Path

# Add agent-protocol to path
sys.path.insert(0, str(Path(__file__).parent.parent / "agent-protocol" / "scripts"))

from publish import publish_event

# Publish event
publish_event(
    event_type="my_skill.something_happened",
    source_agent="my-skill",
    payload={
        "key": "value",
        "importance": 8
    }
)

From JavaScript Skills

const { publishEvent } = require('../agent-protocol/scripts/protocol.js');

// Publish event
await publishEvent({
  eventType: 'my_skill.something_happened',
  source: 'my-skill',
  payload: {
    key: 'value',
    importance: 8
  }
});

From Shell Scripts

python3 /root/clawd/skills/agent-protocol/scripts/publish.py \
  --type "shell.command_completed" \
  --source "backup-script" \
  --payload '{"files": 42, "size_mb": 1024}'

Workflow Definition

Workflows are JSON files in config/workflows/:

{
  "workflow_id": "unique-id",
  "name": "Human-readable name",
  "enabled": true,
  "trigger": {
    "event_type": "research.article_found",
    "conditions": {
      "payload.importance": { "gte": 7 }
    }
  },
  "steps": [
    {
      "agent": "summary-agent",
      "action": "summarize",
      "input": {
        "url": "{{payload.url}}"
      },
      "output_event": "summary.ready"
    },
    {
      "agent": "notification-agent",
      "action": "notify",
      "input": {
        "message": "{{payload.title}}\n\n{{previous.summary}}"
      }
    }
  ]
}

Event Types (Standard Conventions)

Research

• research.article_found - Article discovered
• research.topic_suggested - New topic suggested

Sports

• sports.goal_scored - Goal/point scored
• sports.match_started - Match/game started
• sports.match_ended - Match/game finished

Analytics

• analytics.insight - Insight generated
• analytics.daily_report - Daily report ready

Notifications

• notification.sent - Notification delivered
• notification.failed - Notification failed

Workflow

• workflow.started - Workflow execution started
• workflow.completed - Workflow completed
• workflow.failed - Workflow failed

Custom

Use format: <domain>.<action_past_tense>

• Lowercase, underscores only
• Domain: broad category
• Action: what happened

Advanced Features

Conditional Routing

{
  "condition": {
    "payload.importance": { "gte": 9 }
  },
  "then": { "agent": "urgent-handler" },
  "else": { "agent": "normal-handler" }
}

Parallel Execution

{
  "parallel": [
    { "agent": "telegram-notifier" },
    { "agent": "discord-notifier" },
    { "agent": "email-notifier" }
  ]
}

Error Handling

{
  "agent": "external-api",
  "retry": {
    "max_attempts": 3,
    "backoff_seconds": 5
  },
  "on_error": {
    "agent": "error-logger",
    "continue": true
  }
}

Cron Integration

Add to Clawdbot cron jobs:

{
  "id": "workflow-engine",
  "name": "Agent Protocol Workflow Engine",
  "schedule": "*/5 * * * *",
  "command": "cd /root/clawd/skills/agent-protocol && python3 scripts/workflow_engine.py --run",
  "enabled": true
}

Or run as daemon:

nohup python3 scripts/workflow_engine.py --daemon > /tmp/workflow-engine.log 2>&1 &

Monitoring

View Recent Events

python3 scripts/event_bus.py tail --count 50

Check Workflow Status

python3 scripts/workflow_engine.py --list

View Logs

tail -f ~/.clawdbot/events/log/events.log
tail -f ~/.clawdbot/events/log/workflows/engine.log

Architecture

File-Based Event Bus

• Events stored in ~/.clawdbot/events/queue/
• Processed events move to processed/ or failed/
• Simple, debuggable, persistent
• No daemon required (optional workflow engine daemon)

Workflow Engine

• Polls event queue (default: every 30 seconds)
• Matches events to workflow triggers
• Executes workflow steps sequentially or in parallel
• Handles errors and retries

Directory Structure

~/.clawdbot/events/
  ├── queue/           # Pending events
  ├── processed/       # Successfully processed
  ├── failed/          # Failed processing
  ├── log/
  │   ├── events.log   # Event bus log
  │   └── workflows/   # Workflow execution logs
  └── subscriptions.json

/root/clawd/skills/agent-protocol/
  ├── config/
  │   ├── protocol.json
  │   └── workflows/   # Workflow definitions
  ├── scripts/
  │   ├── event_bus.py
  │   ├── publish.py
  │   ├── subscribe.py
  │   ├── workflow_engine.py
  │   └── protocol.js
  └── examples/

Security

• Events validate size limits (default: 512 KB)
• Audit logging tracks all publishes
• Future: Permission system for agents

Performance

• Lightweight: file-based, no database
• Scalable: handles thousands of events
• Configurable: adjust polling intervals
• Cleanup: auto-delete old events (default: 7 days)

Troubleshooting

Events Not Processing

# Check queue
ls -la ~/.clawdbot/events/queue/

# Check workflow engine is running
ps aux | grep workflow_engine

# Validate workflows
python3 scripts/workflow_engine.py --validate

Handler Errors

# Check failed events
ls -la ~/.clawdbot/events/failed/

# View logs
tail -f ~/.clawdbot/events/log/workflows/engine.log

Contributing

This is core infrastructure for Clawdbot. Contributions welcome!

Ideas for future features:

• WebSocket support for real-time events
• Visual workflow builder (web UI)
• Cross-instance event relay (multi-bot networks)
• GraphQL query API
• Event replay and debugging tools

License

MIT

---

Built with 🦎 for Clawdbot