Multi-Agent Collaboration

Build multi-agent systems with AgenticX.

Multi-Agent Collaboration

Overview

AgenticX is designed from the ground up for multi-agent systems. Multiple agents can collaborate on complex tasks through delegation, parallel execution, and structured communication protocols.

Agent Teams

python

1from agenticx.runtime import AgentTeamManager
2from agenticx import Agent
3from agenticx.llms import OpenAIProvider
4
5llm = OpenAIProvider(model="gpt-4o")
6
7# Define team members
8researcher = Agent(id="researcher", name="Researcher", role="Information Gatherer",
9                   goal="Find accurate information", organization_id="team")
10analyst = Agent(id="analyst", name="Analyst", role="Data Analyst",
11                goal="Analyze and interpret data", organization_id="team")
12writer = Agent(id="writer", name="Writer", role="Content Writer",
13               goal="Produce clear written content", organization_id="team")
14
15# Team manager handles concurrency, session isolation, and agent lifecycle
16team = AgentTeamManager(agents=[researcher, analyst, writer], max_concurrency=3)

Meta-Agent Pattern

The Meta-Agent acts as a CEO/project manager, dispatching work to specialized sub-agents:

1User Request
2    ↓
3Meta-Agent (analyzes, plans, delegates)
4    ↓
5┌───────────────────────────────┐
6│  Researcher │ Analyst │ Writer │  ← Sub-agents running concurrently
7└───────────────────────────────┘
8    ↓
9Meta-Agent (aggregates, synthesizes)
10    ↓
11Final Response to User

The Meta-Agent maintains an active snapshot of all running sub-agents and injects their status into its system prompt each turn.

A2A Communication Protocol

Agents can communicate directly using the A2A (Agent-to-Agent) protocol:

python

1from agenticx.protocols.a2a import A2AClient, A2AServer, AgentCard
2
3# Publish an agent as an A2A service
4card = AgentCard(
5    agent_id="researcher",
6    skills=["web_search", "document_analysis"],
7    endpoint="http://localhost:8001"
8)
9server = A2AServer(agent=researcher, card=card)
10server.start()
11
12# Another agent calls it
13client = A2AClient()
14result = client.invoke_skill(
15    agent_id="researcher",
16    skill="web_search",
17    params={"query": "latest AI papers"}
18)

Parallel Execution

Run multiple agents simultaneously:

python

1from agenticx.flow import ParallelExecutor
2
3executor = ParallelExecutor(max_workers=4)
4
5tasks = [
6    (researcher, Task(description="Research topic A")),
7    (researcher, Task(description="Research topic B")),
8    (researcher, Task(description="Research topic C")),
9]
10
11results = executor.run_all(tasks)

Human-in-the-Loop

Pause agent execution to get human approval:

python

1from agenticx.runtime import HumanInTheLoop
2
3hitl = HumanInTheLoop(
4    trigger_on=["tool_call:delete_file", "tool_call:send_email"],
5    timeout_seconds=300,
6    default_action="reject"  # auto-reject if no human response
7)
8
9executor = AgentExecutor(agent=agent, llm=llm, human_in_the_loop=hitl)

Session Isolation

Each agent team run is isolated by owner_session_id, preventing cross-contamination between concurrent sessions. The global registry allows looking up agent status across sessions for monitoring purposes.