Multi-Agent - mAIvn Docs

#Multi-Agent Swarms

A swarm is a team of specialist agents that hand work to each other. You describe who does what, who depends on whom, and which member writes the final answer. The runtime handles the coordination.

#What a Swarm Is

Think of a swarm the way you would think of a small project team. One person kicks things off, others contribute their specialty, and one person is responsible for the finished deliverable. A Swarm models exactly that:

Each member is an Agent with a focused role (research, analysis, writing, review).
The runtime selects members from the roster according to the request and declared dependencies. List order does not designate an entry agent.
Members hand work to each other through declared dependencies, so an analyst can wait for the researcher's output before starting.
One member (or one swarm-level tool) is designated to produce the final output the caller receives.

Why use a swarm instead of one big agent? Splitting work across specialists keeps each agent's instructions and tools narrow and easy to reason about, lets independent work run in parallel, and makes multi-step workflows explicit instead of buried in one long prompt.

A swarm is an Agent sibling. Both Agent and Swarm are scopes. They share tool registration, events, batching, scheduling, and final-tool support. The Agent-only .structured_output(Model) builder is not available on Swarm.

#Building a Swarm

Create the member agents first, then group them. The first member can supply the swarm's default client; that does not require it to execute first.

Python

from maivn import Agent, Swarm
from maivn.messages import HumanMessage

researcher = Agent(
    name='researcher',
    description='Finds and gathers information on a topic',
    system_prompt='You research topics thoroughly and cite what you find.',
    api_key='your-api-key',
)

analyst = Agent(
    name='analyst',
    description='Turns research into insights',
    system_prompt='You analyze findings and extract the key takeaways.',
    api_key='your-api-key',
)

writer = Agent(
    name='writer',
    description='Writes the final article',
    system_prompt='You write clear, well-structured content.',
    api_key='your-api-key',
    use_as_final_output=True,  # this member owns the final answer
)

swarm = Swarm(
    name='content_team',
    description='Researches a topic and writes an article about it',
    agents=[researcher, analyst, writer],
)

Every member needs credentials the same way a standalone Agent does: pass an api_key or a shared client to each agent.

#Managing members after construction

You can also build a swarm incrementally:

Python

swarm = Swarm(name='content_team', agents=[])

swarm.add_agent(researcher)
swarm.add_agent(analyst)
swarm.add_agent(writer)

Inspect or look up members at any time:

Python

for agent in swarm.list_agents():
    print(f'{agent.name}: {agent.description}')

# Look up a member from the public roster.
member = next((item for item in swarm.list_agents() if item.name == 'writer'), None)

The current SDK exposes add_agent and list_agents, but no get_agent or get_swarm method. Retain application references when you need the parent or a particular member. A deployed agent identity defaults to its name.

#Member Registration with Dependencies

Adding an agent to the list registers it as a member. To declare how members depend on each other, use the swarm.member decorator builder. It registers the agent and attaches dependency metadata to the generated agent-invocation tool, the same vocabulary you already use for tools, applied at the member level.

Python

from maivn import Agent, Swarm, depends_on_tool

swarm = Swarm(name='content_team')

# A swarm-level dependency tool
@swarm.toolify(description='Load editorial context')
def load_context() -> dict:
    return {'audience': 'technical leaders'}

# Register a member whose invocation depends on a swarm tool's output
@depends_on_tool(load_context, arg_name='context')
def build_researcher() -> Agent:
    return Agent(name='researcher', api_key='your-api-key')

researcher = swarm.member(build_researcher)

# Register a member that depends on another member's output
writer = swarm.member.depends_on_agent(
    researcher,
    arg_name='research_notes',
)(Agent(name='writer', api_key='your-api-key', use_as_final_output=True))

Keep the concrete Agent reference when later member dependencies need it.

The member builder is chainable and supports the full dependency vocabulary:

Builder method	What it does
`swarm.member.depends_on_agent`	Wait for another member and inject its output
`swarm.member.depends_on_tool`	Wait for a tool and inject its result before the member runs
`swarm.member.depends_on_await_for`	Enforce ordering without injecting any data
`swarm.member.depends_on_reevaluate`	Insert a planning boundary before this member runs
`swarm.member.depends_on_interrupt`	Collect user input before the member is invoked

Python

from maivn import Agent, Swarm

reviewer = swarm.member.depends_on_agent(
    writer, arg_name='draft',
).depends_on_await_for(
    researcher, timing='after',
)(Agent(name='reviewer', api_key='your-api-key'))

Private data on members. Private-data dependencies are not attached directly to member agents. Put the secret access behind a swarm-level tool and have the member depend on that tool. See the cross-links below for the shared dependency vocabulary. It is intentionally not re-explained here.

#Designated Final Output

A swarm produces one final answer. You decide which member owns it.

#One member owns the answer

Set use_as_final_output=True on exactly one member. That agent's output becomes the user-facing response.

Python

writer = Agent(
    name='writer',
    api_key='your-api-key',
    use_as_final_output=True,
)

At most one member across the swarm may carry this flag. If two do, validation fails when you invoke the swarm.

#A swarm-scope final tool

Instead of designating an agent, you can define a single typed report at the swarm level with final_tool=True. The swarm coordinates its members and emits this structured object as the result:

Python

from pydantic import BaseModel

@swarm.toolify(final_tool=True)
class TeamReport(BaseModel):
    """Combined output from the whole team."""
    research_summary: str
    analysis_insights: list[str]
    final_article: str

#When you need both

If multiple members own final tools, designate one member with use_as_final_output=True. A swarm-level final tool cannot be combined with a member carrying that flag. Member-local final tools can still produce intermediate results when the swarm owns its final output.

Validation runs when the swarm is invoked. Common errors:

Two agents both marked use_as_final_output=True.
Multiple members own final tools without a designated final-output member.
A swarm-level final tool conflicts with a designated final-output member.

#Forcing structured output per member

A member can be required to return a typed report every time it is deployed. Set force_final_tool=True on that Agent and register its report model with final_tool=True. This per-member flag is honored when the member runs inside a swarm.

Python

report_agent = Agent(
    name='report_agent',
    api_key='your-api-key',
    force_final_tool=True,
)

@report_agent.toolify(final_tool=True)
class MemberReport(BaseModel):
    headline: str
    bullets: list[str]

#Running a Swarm

A swarm is invoked the same four ways an agent is: invoke, stream, ainvoke, and astream. They accept a single message or a sequence of messages.

Python

# Synchronous, returns a InvokeResponse
response = swarm.invoke(HumanMessage(content='Write about AI trends in 2026'))

print(response.response)   # final assistant text
print(response.result)     # structured / final-tool output, when present

To request a structured final result, pass force_final_tool=True. The swarm resolves it against the use_as_final_output member or the swarm-scope final tool:

Python

response = swarm.invoke(
    HumanMessage(content='Write about AI trends'),
    force_final_tool=True,
)
report = response.result

Continue a conversation across calls with a thread_id:

Python

response = swarm.invoke(
    HumanMessage(content='Now make it shorter'),
    thread_id='article-session-123',
)

#Keyword-only config

Swarm config kwargs are keyword-only. As with Agent.invoke(), every option after messages on Swarm.invoke() / stream() / ainvoke() / astream() must be passed by keyword. Call them as swarm.invoke(messages, force_final_tool=True), never positionally.

The keyword options include model, reasoning, force_final_tool, stream_response, thread_id, verbose, metadata, memory_config, and system_tools_config. metadata is accepted but ignored. orchestration_config is accepted as a compatibility value but is not sent to the current runtime. Scope-level defaults set on the Swarm(...) constructor are merged with per-call overrides.

#Streaming and async

stream() yields StreamEvent values as the swarm executes, including any status-message events emitted by the runtime.

Python

for event in swarm.stream(
    HumanMessage(content='Run the multi-agent workflow'),
):
    handle(event)

The async variants mirror the synchronous ones:

Python

response = await swarm.ainvoke(HumanMessage(content='Plan the launch'))

async for event in swarm.astream(HumanMessage(content='Run the workflow')):
    handle(event)

For filtered, higher-level event reporting, wrap any invocation with events() (inherited from the shared scope):

Python

response = swarm.events(
    include=['enrichment', 'agent', 'model'],
    on_event=lambda payload: post_to_ui(payload),
).invoke(HumanMessage(content='Run the multi-agent workflow'))

#How coordination works

Conceptually, the runtime reads each member's declared dependencies and runs the team accordingly: members with no dependencies start first (and run in parallel when they are independent), and a member that depends on another waits for that member's output before it begins. The designated final-output member runs with the upstream results available and produces the answer the caller receives. You declare the data flow; the runtime sequences the work.

Python

from maivn import Agent, Swarm, depends_on_agent

# data_collector and trend_analyzer are independent -> they can run in parallel
# synthesizer depends on both -> it waits for them, then produces the final output
data_collector = Agent(name='data_collector', api_key='your-api-key')
trend_analyzer = Agent(name='trend_analyzer', api_key='your-api-key')
synthesizer = Agent(name='synthesizer', api_key='your-api-key', use_as_final_output=True)

@synthesizer.toolify()
@depends_on_agent(data_collector, arg_name='data')
@depends_on_agent(trend_analyzer, arg_name='trends')
def synthesize(data: dict, trends: dict) -> dict:
    return {'synthesis': 'Combined insights'}

swarm = Swarm(
    name='analysis_team',
    agents=[data_collector, trend_analyzer, synthesizer],
)

#Shared Tools and Config Across Members

Tools registered on the swarm belong to the parent invocation. Member agents carry their own registered tools; they do not automatically inherit all parent tools. Register a utility on each member that needs to call it, or use a swarm-level tool as an explicit member dependency.

Python

swarm = Swarm(name='team', agents=[agent1, agent2])

@swarm.toolify(description='Utility for the parent invocation')
def shared_utility(data: dict) -> dict:
    return {'processed': True}

agent1.add_tool(shared_utility)
agent2.add_tool(shared_utility)

A Swarm also accepts the active typed configuration objects an Agent does, applied as defaults for every run and overridable per invocation:

memory_config — default memory behavior for the swarm.
system_tools_config — allowlists and approval controls for system tools.

SessionOrchestrationConfig is still accepted for compatibility, but the current SDK does not send it to the runtime. It cannot currently create a supervised repair loop or impose a cycle limit. Model required passes with explicit member or tool dependencies, and have the application start another invocation when a postcondition fails.

use_as_final_output=True identifies the member that produces the user-facing answer. A member can still appear more than once in runtime events; key UI cards by assignment or invocation ID rather than agent name.

#Complete Example

This example uses fixture research; replace it with your own data source. Each member has a final tool so the dependency receives the expected dictionary.

Python

from pydantic import BaseModel
from maivn import Agent, Swarm, depends_on_agent
from maivn.messages import HumanMessage

researcher = Agent(
    name='researcher',
    description='Researches topics and finds relevant information',
    system_prompt='You are a research specialist. Gather information and cite sources.',
    api_key='your-api-key',
)

@researcher.toolify(description='Return sample research for the topic', final_tool=True)
def research_topic(topic: str) -> dict:
    return {
        'topic': topic,
        'findings': [f'Finding about {topic}'],
        'sources': ['source1.com', 'source2.com'],
    }

writer = Agent(
    name='writer',
    description='Writes articles based on research',
    system_prompt='You are a content writer. Turn research into an article.',
    api_key='your-api-key',
    use_as_final_output=True,
)

@writer.toolify(description='Write an article from research', final_tool=True)
@depends_on_agent(researcher, arg_name='research')
def write_article(research: dict, style: str = 'professional') -> dict:
    return {
        'title': f"Article about {research['topic']}",
        'body': f"Based on research: {research['findings']}",
        'style': style,
    }

content_team = Swarm(
    name='content_team',
    description='A team that researches and writes articles',
    agents=[researcher, writer],
)

response = content_team.events().invoke(
    HumanMessage(content='Write an article about quantum computing'),
    force_final_tool=True,
)

print(response.result)

#Best Practices

Give each member one clear job. Overlapping responsibilities make coordination ambiguous; narrow roles keep prompts and tools focused.
Keep dependency chains short. Prefer a shallow graph (A -> B -> C) over a long pipeline; independent members run in parallel and finish sooner.
Designate the final output explicitly whenever more than one scope could produce it.
Use event tracing while developing. swarm.events().invoke(...) surfaces what each member is doing and makes coordination easy to follow.

#See Also

Tools — how to define and register the tools your members use.
Dependencies — the shared dependency vocabulary (depends_on_agent, depends_on_tool, depends_on_await_for, depends_on_reevaluate, depends_on_interrupt) used by both tools and swarm.member.
Structured Output — typed results via final_tool + force_final_tool and the structured_output() fast path.

#Next steps

Swarm API — the full Swarm class reference.
Agent API — the Agent class your members are built from.