Agent Network

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

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:

| Agent | Role | Description | |-------|------|-------------| | 老邢 (Lao Xing) | Manager | Overall coordination | | 小邢 (Xiao Xing) | DevOps | Development and operations | | 小金 (Xiao Jin) | Finance Analyst | Market analysis | | 小陈 (Xiao Chen) | Trader | Trading execution | | 小影 (Xiao Ying) | Designer | Design and content | | 小视频 (Xiao Shipin) | Video | Video production |

Database Schema

SQLite database with tables:

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

Advanced Usage

See references/ADVANCED.md for:

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")