Swarm Multi-Agent Orchestration

from isa_agent_sdk import Agent
from isa_agent_sdk.agents import Swarm, SwarmOrchestrator, SwarmAgent, SwarmRunResult, SwarmState

from isa_agent_sdk.agents import Agent, Swarm
 
swarm = Swarm([
    Agent("researcher", tools=["web_search"], handoff_to="writer"),
    Agent("writer", complete=True),
])
 
result = await swarm.run("Research quantum computing and write a summary")
print(result.text)           # Clean text (directives stripped)
print(result.final_agent)    # "writer"
print(result.handoff_trace)  # [{"from": "researcher", "to": "writer", ...}]

swarm = Swarm([
    Agent("researcher",
          tools=["web_search"],
          system_prompt="You are a research specialist. Gather accurate facts.",
          handoff_to="writer"),
    Agent("writer",
          system_prompt="You are a technical writer. Write clear summaries.",
          complete=True),
])

swarm = Swarm([agent1, agent2, agent3], dag=True)
result = await swarm.run("Complex multi-step task")

from isa_agent_sdk import Agent, ISAAgentOptions
from isa_agent_sdk.agents import SwarmOrchestrator, SwarmAgent
 
researcher = SwarmAgent(
    agent=Agent("researcher", ISAAgentOptions(
        system_prompt="You are a research agent. Gather facts only.",
    )),
    description="Research and information gathering",
)
 
writer = SwarmAgent(
    agent=Agent("writer", ISAAgentOptions(
        system_prompt="You are a technical writer. Write polished summaries.",
    )),
    description="Writing and documentation",
)
 
swarm = SwarmOrchestrator(
    agents=[researcher, writer],
    entry_agent="researcher",
    max_handoffs=10,
)
 
result = await swarm.run("Research quantum computing and write a summary")
print(result.text)           # Clean text (directives stripped)
print(result.final_agent)    # "writer"
print(result.handoff_trace)  # [{"from": "researcher", "to": "writer", "reason": "..."}]

## Agent Collaboration

You are part of a multi-agent team. The following peer agents are available:
- **writer**: Writing and documentation

### Handoff Protocol
When you determine that another agent is better suited to continue,
end your response with:
[HANDOFF: agent_name] reason for handoff

When your part of the task is fully complete, end your response with:
[COMPLETE]

async for msg in swarm.stream("Research and write about AI"):
    if msg.metadata.get("event") == "swarm_agent_start":
        print(f"\n--- {msg.metadata['agent']} started ---")
    elif msg.metadata.get("event") == "swarm_handoff":
        print(f"\n--- handoff: {msg.metadata['from_agent']} -> {msg.metadata['to_agent']} ---")
    elif msg.is_text:
        agent = msg.metadata.get("agent", "?")
        print(f"[{agent}] {msg.content}", end="")

result = await swarm.run_dag([
    {
        "id": "research",
        "title": "Research Python",
        "description": "Gather key facts about Python.",
        "agent": "researcher",
    },
    {
        "id": "write_summary",
        "title": "Write Summary",
        "description": "Write a summary based on the research.",
        "agent": "writer",
        "depends_on": ["research"],
    },
])
 
# Both agents ran in dependency order
print(result.agent_outputs["researcher:research"].text)
print(result.agent_outputs["writer:write_summary"].text)

result = await swarm.run_dag([
    # Wavefront 0: these two run concurrently (different agents)
    {"id": "fact_python", "title": "Python", "description": "One fact about Python.", "agent": "researcher"},
    {"id": "fact_rust", "title": "Rust", "description": "One fact about Rust.", "agent": "analyst"},
 
    # Wavefront 1: runs after both above complete
    {"id": "combine", "title": "Combine", "description": "Combine into comparison.",
     "agent": "writer", "depends_on": ["fact_python", "fact_rust"]},
])

Wavefront 0: [fact_python (researcher), fact_rust (analyst)]  <- concurrent
Wavefront 1: [combine (writer)]                               <- sequential, gets both results

SwarmOrchestrator(
    agents,           # List[SwarmAgent] or Dict[str, Agent]
    entry_agent,      # str — first agent to run (default: first in list)
    max_handoffs,     # int — safety cap (default: 10)
    shared_state,     # Dict — initial shared state passed across agents
)

swarm = SwarmOrchestrator(
    agents={
        "researcher": Agent("researcher", ISAAgentOptions(...)),
        "writer": Agent("writer", ISAAgentOptions(...)),
    }
)

@dataclass
class SwarmRunResult:
    text: str                              # Final clean text (directives stripped)
    messages: List[AgentMessage]           # All messages from all agents
    final_agent: str                       # Agent that produced the final response
    handoff_trace: List[Dict[str, str]]    # [{"from", "to", "reason", "step"}]
    shared_state: Dict[str, Any]           # Accumulated shared state
    agent_outputs: Dict[str, AgentRunResult]  # Per-agent results

# String prompt — handoff section appended
Agent("a", ISAAgentOptions(system_prompt="You are a researcher."))
 
# SystemPromptConfig with preset — handoff added to append
Agent("a", ISAAgentOptions(
    system_prompt=SystemPromptConfig(preset="reason", append="Custom instructions.")
))
 
# SystemPromptConfig with replace — handoff added to replace text
Agent("a", ISAAgentOptions(
    system_prompt=SystemPromptConfig(replace="Full custom prompt.")
))

swarm = SwarmOrchestrator(
    agents=[...],
    shared_state={"project": "my-project"},
)
 
result = await swarm.run("do something")
print(result.shared_state)  # {"project": "my-project", ...merged from agents...}

python -m pytest tests/test_swarm.py -v  # 33 unit tests

# From isA_Agent directory with dev.env loaded
source deployment/environments/dev.env
python tests/test_swarm_e2e.py --live