Multi-Agent

Agent Team coordinates a leader agent and multiple teammate agents working in parallel to complete complex tasks.

Agent Team is experimental. APIs may change between releases.

Agent Team is NexAU’s model for parallel task coordination. A leader agent analyzes a user request, spawns teammate agents from pre-configured role templates, creates tasks on a shared task board, and assigns them to teammates. Teammates work independently and communicate through a message bus. All agents stream their output through a single SSE connection, with each event tagged by the originating agent.

Core concepts

Leader

The coordinator. Spawns teammates, creates and assigns tasks, monitors progress, and calls finish_team when the work is done.

Teammates

Workers spawned from pre-configured role templates (candidates). Each runs in a loop, waiting for task assignments or messages.

Task board

A shared, database-backed list of tasks. Tasks move through pending → in_progress → completed. Dependencies between tasks are tracked automatically.

Message bus

Persistent point-to-point and broadcast messaging between agents. The leader can message individual teammates or broadcast to all.

Quick start

1. Define agent configs

   Create YAML configs for the leader and each candidate role. The leader receives team coordination tools automatically — you only need to define its domain tools.

   leader_agent.yaml

   type: agent
   name: team_leader
   system_prompt: ./systemprompt_leader.md
   max_iterations: 200
   llm_config:
     model: ${env.LLM_MODEL}
     api_key: ${env.LLM_API_KEY}
     stream: true
   tools:
     - name: read_file
       yaml_path: ./tools/read_file.tool.yaml
       binding: nexau.archs.tool.builtin.file_tools:read_file
   stop_tools: [ask_user]
   

   builder_agent.yaml

   type: agent
   name: builder
   system_prompt: ./systemprompt_builder.md
   max_iterations: 80
   llm_config:
     model: ${env.LLM_MODEL}
     api_key: ${env.LLM_API_KEY}
     stream: true
   tools:
     - name: read_file
       yaml_path: ./tools/read_file.tool.yaml
       binding: nexau.archs.tool.builtin.file_tools:read_file
     - name: write_file
       yaml_path: ./tools/write_file.tool.yaml
       binding: nexau.archs.tool.builtin.file_tools:write_file
   
2. Register the team and start the server

   from nexau.archs.main_sub.config import AgentConfig
   from nexau.archs.session.orm import InMemoryDatabaseEngine
   from nexau.archs.transports.http import HTTPConfig, SSETransportServer
   
   leader_config = AgentConfig.from_yaml("leader_agent.yaml")
   rfc_writer_config = AgentConfig.from_yaml("rfc_writer_agent.yaml")
   builder_config = AgentConfig.from_yaml("builder_agent.yaml")
   
   engine = InMemoryDatabaseEngine()
   server = SSETransportServer(
       engine=engine,
       config=HTTPConfig(host="0.0.0.0", port=8000),
       default_agent_config=leader_config,
   )
   
   registry = server.team_registry
   if registry is not None:
       registry.register_config(
           "default",
           leader_config=leader_config,
           candidates={
               "rfc_writer": rfc_writer_config,
               "builder": builder_config,
           },
       )
   
   server.run()
   
3. Send a request

   curl -X POST http://localhost:8000/team/stream \
     -H "Content-Type: application/json" \
     -d '{"user_id": "u1", "session_id": "s1", "message": "Build a TODO app"}'
   

   The response is an SSE stream. Each event is a TeamStreamEnvelope tagged with the originating agent:

   data: {"team_id":"team_abc","agent_id":"leader-001","role_name":"leader","event":{"type":"TEXT_MESSAGE_CONTENT","delta":"Let me plan this..."}}
   
   data: {"team_id":"team_abc","agent_id":"builder-1","role_name":"builder","event":{"type":"TEXT_MESSAGE_CONTENT","delta":"Writing the API..."}}
   

Passing runtime variables

You can inject variables into the leader and all spawned teammates via the variables field:

curl -X POST http://localhost:8000/team/stream \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "u1",
    "session_id": "s1",
    "message": "Build a TODO app",
    "variables": {
      "template": {"date": "2026-03-04", "username": "alice"},
      "runtime_vars": {"api_key": "sk-..."},
      "sandbox_env": {"WORKSPACE": "/tmp/project"}
    }
  }'

Field	Description
template	Jinja2 variables injected into system prompts (e.g. {{ date }}).
runtime_vars	Values accessible via agent_state.get_variable() but not injected into prompts.
sandbox_env	Environment variables injected into the shared sandbox.

Team tools

Team tools are injected automatically. You don’t define them in agent YAML files.

Leader tools

Tool	Description
spawn_teammate	Instantiate a new teammate from a candidate role
remove_teammate	Remove an idle teammate
create_task	Add a task to the shared task board
claim_task	Assign a task to a specific teammate
update_task_status	Update task status (pending → in_progress → completed)
release_task	Unassign a task
list_tasks	List tasks with optional status filter
list_teammates	List all teammates and their status
message	Send a point-to-point message to a teammate
broadcast	Send a message to all teammates
finish_team	End the team run and return a summary

Teammate tools

Teammates receive a subset: message, broadcast, list_teammates, list_tasks, claim_task, update_task_status, and release_task. Teammates cannot spawn other teammates, create tasks, or end the team run.

Direct Python usage

You can use AgentTeam directly without starting an HTTP server — useful for scripts, notebooks, or embedding team coordination in your own application.

Non-streaming

import asyncio

from nexau.archs.main_sub.config import AgentConfig
from nexau.archs.main_sub.team.agent_team import AgentTeam
from nexau.archs.session.orm import InMemoryDatabaseEngine
from nexau.archs.session.session_manager import SessionManager

leader_config = AgentConfig.from_yaml("leader_agent.yaml")
builder_config = AgentConfig.from_yaml("builder_agent.yaml")
rfc_writer_config = AgentConfig.from_yaml("rfc_writer_agent.yaml")

engine = InMemoryDatabaseEngine()
session_manager = SessionManager(engine=engine)

team = AgentTeam(
    leader_config=leader_config,
    candidates={
        "rfc_writer": rfc_writer_config,
        "builder": builder_config,
    },
    engine=engine,
    session_manager=session_manager,
    user_id="user_1",
    session_id="session_1",
)

result = asyncio.run(team.run(message="Build a TODO app"))
print(result)

run() blocks until the leader calls finish_team, then returns the leader’s final response as a string. You don’t need to call initialize() separately.

Streaming

async def main() -> None:
    # ... same setup as above ...

    async for envelope in team.run_streaming(message="Build a TODO app"):
        print(f"[{envelope.role_name}:{envelope.agent_id}] {envelope.event}")

asyncio.run(main())

run_streaming() returns an async generator of TeamStreamEnvelope objects — the same format used by the HTTP SSE endpoint. You can also pass an on_envelope callback for side effects like persistence:

envelopes: list[TeamStreamEnvelope] = []

async for envelope in team.run_streaming(
    message="Build a TODO app",
    on_envelope=envelopes.append,
):
    pass  # or process each envelope in real time

# envelopes now contains the full event history

Both run() and run_streaming() accept a variables argument. Pass a ContextValue to inject runtime variables:

from nexau.archs.main_sub.context_value import ContextValue

variables = ContextValue(
    template={"date": "2026-03-04", "username": "alice"},
    runtime_vars={"api_key": "sk-..."},
    sandbox_env={"WORKSPACE": "/tmp/project"},
)

result = asyncio.run(team.run(message="Build a TODO app", variables=variables))

Session persistence

By default, AgentTeam uses InMemoryDatabaseEngine, which does not survive process restarts. For production or long-running workflows, swap it for SQLDatabaseEngine:

from nexau.archs.session.orm import SQLDatabaseEngine

engine = SQLDatabaseEngine.from_url("sqlite+aiosqlite:///team.db")

Team state — tasks, messages, and team membership — is persisted and automatically restored on the next run() call with the same user_id and session_id.