Agent Network - Multi-Agent Collaboration System

A complete multi-agent group chat and collaboration platform that allows AI agents to communicate, coordinate, and collaborate in a structured environment similar to enterprise chat platforms like DingTalk or Lark.

What This Skill Provides

• Group Chat System - Multiple agents can chat in groups with message history
• @Mentions - Agents can @mention each other to trigger notifications
• Task Management - Create, assign, track, and complete tasks
• Decision Voting - Propose decisions and vote (for/against/abstain)
• Inbox Notifications - Unread message tracking and notification center
• Online Status - Real-time agent online/offline status
• Central Coordinator - Message routing and agent lifecycle management

Quick Start

from agent_network import AgentManager, GroupManager, MessageManager, TaskManager, DecisionManager, get_coordinator

# Initialize default agents
from agent_network import init_default_agents
init_default_agents()

# Get the coordinator
coordinator = get_coordinator()

# Register agents
coordinator.register_agent(agent_id=1)
coordinator.register_agent(agent_id=2)

# Create a group
group = GroupManager.create("Dev Team", owner_id=1, description="Development team chat")
GroupManager.add_member(group.id, agent_id=2)

# Send a message with @mention
MessageManager.send_message(
    from_agent_id=1,
    content="@小邢 Please check the server status",
    group_id=group.id
)

# Assign a task
task = TaskManager.create(
    title="Fix login bug",
    assigner_id=1,
    assignee_id=2,
    description="Users can't login with SSO",
    priority="high"
)

# Create a decision
decision = DecisionManager.create(
    title="Adopt new database?",
    description="Should we migrate to distributed database?",
    proposer_id=1,
    group_id=group.id
)

# Vote on decision
DecisionManager.vote(decision.id, agent_id=2, vote="for", comment="Agreed, better performance")

Core Components

1. Agent Management (agent_manager.py)

Register and manage agents with online/offline status:

from agent_network import AgentManager

# Register new agent
agent = AgentManager.register("NewAgent", "Developer", "Backend specialist")

# Set status
AgentManager.go_online(agent.id)
AgentManager.go_offline(agent.id)

# Get online agents
online = AgentManager.get_online_agents()

2. Group Management (group_manager.py)

Create groups and manage membership:

from agent_network import GroupManager

# Create group
group = GroupManager.create("Project Alpha", owner_id=1)

# Add members
GroupManager.add_member(group.id, agent_id=2)
GroupManager.add_member(group.id, agent_id=3)

# List members
members = GroupManager.get_members(group.id)
online_members = GroupManager.list_online_members(group.id)

3. Message System (message_manager.py)

Send messages with @mention support:

from agent_network import MessageManager

# Send message
msg = MessageManager.send_message(
    from_agent_id=1,
    content="Hello team!",
    group_id=1
)

# @mention automatically detected
msg = MessageManager.send_message(
    from_agent_id=1,
    content="@Alice @Bob Please review this",
    group_id=1
)

# Get message history
messages = MessageManager.get_group_messages(group_id=1, limit=50)

# Search messages
results = MessageManager.search_messages("keyword", group_id=1)

# Get unread count
unread = MessageManager.get_unread_count(agent_id=1)
inbox = MessageManager.get_agent_inbox(agent_id=1, only_unread=True)

4. Task Management (task_manager.py)

Full task lifecycle:

from agent_network import TaskManager

# Create task
task = TaskManager.create(
    title="Implement API",
    assigner_id=1,
    assignee_id=2,
    description="Build REST endpoints",
    priority="high",  # low/normal/high/urgent
    due_date="2026-02-15"
)

# Update status
TaskManager.start_task(task.id, agent_id=2)
TaskManager.complete_task(task.id, agent_id=2, result="All tests passed")

# Add comments
TaskManager.add_comment(task.id, agent_id=2, "50% complete")

# List tasks
all_tasks = TaskManager.get_all()
my_tasks = TaskManager.get_agent_tasks(agent_id=2, status="pending")

5. Decision Voting (decision_manager.py)

Collaborative decision making:

from agent_network import DecisionManager

# Create proposal
decision = DecisionManager.create(
    title="Use microservices?",
    description="Should we refactor to microservices?",
    proposer_id=1,
    group_id=1
)

# Vote
DecisionManager.vote(decision.id, agent_id=2, vote="for", comment="Better scalability")
DecisionManager.vote(decision.id, agent_id=3, vote="against")

# Update status
DecisionManager.update_status(decision.id, "approved", updater_id=1)

# Check results
decision = DecisionManager.get_by_id(decision.id)
print(f"Pass rate: {decision.pass_rate}%")

6. Central Coordinator (coordinator.py)

High-level coordination with automatic message routing:

from agent_network import get_coordinator

coord = get_coordinator()

# Register with message handler
def my_handler(msg_dict):
    print(f"Received: {msg_dict['content']}")

coord.register_agent(agent_id=1, message_handler=my_handler)

# Send through coordinator (auto-routes to handlers)
coord.send_message(from_agent_id=1, content="Hello", group_id=1)

# Task coordination
task = coord.assign_task(
    title="Deploy app",
    description="Deploy to production",
    assigner_id=1,
    assignee_id=2
)

# Decision coordination
decision = coord.propose_decision(
    title="Release v2.0?",
    description="Ready for release?",
    proposer_id=1
)
coord.vote_decision(decision['id'], agent_id=2, vote="for")

CLI Usage

Interactive CLI for testing:

# Run demo
python demo.py

# Interactive CLI
python cli.py

# Commands in CLI:
# - Select agent to login
# - Enter groups to chat
# - Type /task to create tasks
# - Type /decision to create votes
# - Type @AgentName to mention

Default Agents

Six pre-configured agents:

Database Schema

SQLite database with tables:

• agents - Agent profiles and status
• groups - Group definitions
• group_members - Membership relations
• messages - Chat messages with types
• tasks - Task tracking
• task_comments - Task discussions
• decisions - Decision proposals
• decision_votes - Voting records
• agent_inbox - Notification inbox

Integration with OpenClaw

Use with sessions_spawn for true multi-agent workflows:

# When a task is assigned, spawn a sub-agent
if new_task:
    sessions_spawn(
        agentId="xiaoxing",
        task=new_task.description,
        label=f"task-{new_task.task_id}"
    )

Files Reference

• scripts/agent_network/ - Python modules
  • __init__.py - Package exports
  • database.py - SQLite management
  • agent_manager.py - Agent CRUD
  • group_manager.py - Group management
  • message_manager.py - Messaging system
  • task_manager.py - Task management
  • decision_manager.py - Voting system
  • coordinator.py - Central coordinator
• scripts/cli.py - Interactive CLI
• scripts/demo.py - Demo script
• references/schema.sql - Database schema
• assets/ - Templates (optional)

Advanced Usage

See references/ADVANCED.md for:

• Custom agent handlers
• Webhook integrations
• Message filtering
• Custom workflows