Skip to main content

Overview

Multi-agent systems involve multiple AI agents working together to solve complex problems. While other frameworks require you to find contrived ways to have agents spawn agents, Agentica makes this trivial: pass a magic function or an agent to another magic function or agent. That’s it. No special orchestration layers, no graphs, no composition primitives or DSL, no message-passing protocols, no global state management. It’s just code.

When to Use Multi-Agent Systems

Multi-agent systems shine when:
  • Tasks can be parallelized: Multiple independent sub-tasks may be worked on simultaneously
  • Specialization helps: Different agents with different capabilities or contexts tackle different aspects
  • Complex workflows: The problem naturally breaks down into distinct phases or responsibilities
  • Iterative refinement: One agent’s output becomes another agent’s input

The Core Concept

Because Agentica allows you to pass any runtime value into scope—functions, objects, classes, SDKs—this naturally includes:
  • The spawn() function itself
  • Existing agents
  • Magic functions
  • Custom orchestration logic
When you pass these to an agent or magic function, they can use them just like anything else in scope. No schemas, no special handling. It’s just code.

Pattern: Basic Composition

Let’s start with the simplest possible multi-agent pattern: calling one magic function from another.
from agentica import magic
from dataclasses import dataclass

@dataclass
class CodeAnalysis:
    bugs: list[str]
    style_issues: list[str]
    performance_issues: list[str]

# Two separate AI functions, each specialized
@magic()
async def analyze_code(code: str) -> CodeAnalysis:
    """Analyze code for bugs, style issues, and performance problems."""
    ...

@magic()
async def fix_code(code: str, analysis: CodeAnalysis) -> str:
    """Fix the code based on the analysis."""
    ...

# Compose them together
analysis = await analyze_code(code)
fixed = await fix_code(code, analysis)
This is a multi-agent system: two separate AI invocations, each specialized for its task, composed together. However the “workflow” here is very boring — it’s entirely serial, with no conditionality. Let’s take a look at what introducing control flow is like.

Pattern: External Control Flow

In this pattern, you orchestrate the agents from your code. You decide when to call agents, what tasks to give them, and how to combine their results. You write the control flow: loops, conditionals, error handling—all the logic that coordinates the agents. This is useful when:
  • The workflow involves iteration, conditionals, or complex branching
  • You want explicit control over execution order and conditions
  • You need to inject custom logic between agent calls
  • You’re building a specific, repeatable process with clear control flow

Example: Parallelization, Iteration, and Conditional Logic

from agentica import magic

... # other logic and classes

@magic()
async def evaluate_candidate(cand: Candidate) -> EvaluationResult:
    """Evaluate if candidate meets our technical and cultural requirements."""
    ...

@magic()
async def create_interview_plan(cand: Candidate, eval: EvaluationResult) -> InterviewPlan:
    """Create a tailored interview plan focusing on the candidate's strengths and gaps."""
    ...

@magic()
async def create_rejection_letter(cand: Candidate, eval: EvaluationResult) -> RejectionLetter:
    """Generate a polite rejection letter with constructive feedback."""
   ...

async def pipeline(candidates: list[Candidates]):
    # Evaluate all candidates in parallel
    evaluations = await asyncio.gather(*[evaluate_candidate(c) for c in candidates])

    to_be_interviewed: dict[Candidate, InterviewPlan] = dict()
    to_be_rejected: dict[Candidate, RejectionLetter] = dict()

    # Branch based on results
    async def pick_candidate(candidate: Candidate, evaluation: EvaluationResult):
        if evaluation.is_qualified:
            to_be_interviewed[candidate] = await create_interview_plan(candidate, evaluation)
        else:
            to_be_rejected[candidate] = await create_rejection_letter(candidate, evaluation)

    async with asyncio.TaskGroup() as tg:
        for candidate, evaluation in zip(candidates, evaluations):
            tg.create_task(pick_candidate(candidate, evaluation))

    # Do something with the data
    ...
Notice how you write the parallelization, the conditional branching, the iteration logic. The AI agents are just functions you call—the control flow is yours. This is just regular programming with AI as building blocks.

Pattern: Coordinator + Workers

Now here’s where Agentica’s design truly shines. Instead of you controlling the flow, let an agent control it. A coordinator agent manages multiple worker agents, each handling independent sub-tasks. This is useful when:
  • The number of sub-tasks isn’t known in advance
  • The workflow needs to adapt based on intermediate results
  • You want the AI to decide how to parallelize work
  • Sub-tasks benefit from specialization
  • The problem is too complex for fixed orchestration logic

Most Constrained: Passing Specific Agent Objects

Let’s start with the most constrained approach: pass pre-configured agent objects with specific capabilities.
from agentica import spawn

# Create specialized agents with permanent tool access
data_analyst = await spawn(
    premise="You are a data analyst",
    scope={"query_database": query_database}    # ← Functions at spawn time
)
market_analyst = await spawn(
    premise="You are a market analyst",
    scope={"get_market_data": get_market_data}  # ← Functions at spawn time
)

coordinator = await spawn(premise="You are a research coordinator")

report = await coordinator(
    ComprehensiveReport,             # ← The rigid class defining the report type
    f"Research {topic} using the data analyst and market analyst",
    data_analyst=data_analyst,       # ← Pass agent objects
    market_analyst=market_analyst    # ← Just like anything else
)
The coordinator can now invoke the specialist agents, pass data between them, and decide when to use which one—all dynamically based on the task.

Least Constrained: Passing spawn Itself

For maximum flexibility, pass spawn itself so the coordinator can create any agents it needs:
from agentica import spawn

... # other logic and classes

async def web_search(query: str) -> list[SearchResult]: ...

# Create a coordinator with the ability to spawn
coordinator = await spawn(premise="You are a research coordinator who can spawn sub-agents")

# Give it spawn and tools - it decides how many sub-agents it needs
report = await coordinator(
    ResearchReport,       # ← The rigid return type defining the report
    f"Research '{topic}' thoroughly using sub-agents for different aspects",
    spawn=spawn,          # ← The magic happens here
    web_search=web_search # ← A function that may be passed to sub-agents
)
What just happened? The coordinator agent can now:
  • Decide how many sub-agents it needs
  • Spawn them dynamically using spawn()
  • Assign them specific tasks
  • Wait for their results
  • Synthesize everything into a final report
You didn’t write any of that orchestration logic. The agent figured it out.
For a richer case of this pattern, see the Deep Research example.

Best Practices

1. Choose the Right Pattern

  • Basic composition: For simple sequential AI operations
  • External control flow: For predictable processes with iteration and conditionals
  • Coordinator + workers: For dynamic, adaptive workflows where agents orchestrate other agents

2. Be Clear About Responsibilities

Give each agent a focused premise that defines its role:
researcher = await spawn(
    premise="You are a researcher who finds and verifies facts. Do not write content."
)

writer = await spawn(
    premise="You are a writer who creates engaging content from facts. Do not do research."
)

3. Use Appropriate Scope

Only pass what each agent needs:
# Good: Focused scope
researcher = await spawn()
result = await researcher(ResearchReport, "Research X", web_search=web_search)

# Avoid: Too much scope
result = await researcher(
    ResearchReport,
    "Research X",
    web_search=web_search,
    database=db,
    api_client=client  # Researcher doesn't need these
)

Why This Works in Agentica

In traditional AI frameworks, multi-agent systems require:
  • Message-passing protocols
  • State management systems
  • Complex orchestration layers
  • Schema definitions for inter-agent communication
In Agentica, none of that is needed because:
  1. Everything is just scope: Agents, functions, objects—they’re all just values you can pass around
  2. RPC handles communication: The framework automatically manages calls between your runtime and the sandbox
  3. Proxies enable references: Agents can pass complex objects without serialization
  4. REPL enables agency: Agents can write actual code to orchestrate other agents, not just call predefined tools
The result: multi-agent systems that feel like regular programming, not framework wrangling.

Next Steps