Sub-Agents & Parallel Execution

from harness.types.agents import AgentDef

# Define a custom agent type
security_agent = AgentDef(
    name="security",
    description="Security-focused code reviewer",
    model=None,           # None = inherit from parent
    tools=("Read", "Glob", "Grep"),
    system_prompt="You are a security expert. Identify vulnerabilities only.",
    max_turns=50,
    read_only=True,
)

import asyncio
import harness
from harness.agents.manager import AgentManager

async def main():
    # AgentManager needs a provider and tools —
    # normally these come from the engine internally.
    # For standalone use, you can get them from a running session.

    # Simple example using harness.run() with sub-agents via the SDK
    async for msg in harness.run(
        "Explore the src/ directory and summarize the architecture",
        provider="anthropic",
    ):
        if isinstance(msg, harness.ToolUse) and msg.name == "Task":
            print(f"Sub-agent spawned: {msg.args.get('prompt', '')[:50]}...")
        elif isinstance(msg, harness.Result):
            print(f"Done: {msg.text[:200]}")

asyncio.run(main())

from harness.agents.manager import AgentManager

# Create manager (provider and tools come from your setup)
manager = AgentManager(
    provider=my_provider,
    tools=my_tools,
    cwd="/path/to/project",
)

# Spawn a single sub-agent
result = await manager.spawn(
    "explore",
    "Find all API endpoint definitions in this codebase",
)
print(result)  # Returns the agent's response text

# team_sequential.py
import asyncio
from harness.agents.manager import AgentManager

async def sequential_workflow(manager: AgentManager):
    # Step 1: Explore the codebase
    print("Step 1: Exploring...")
    exploration = await manager.spawn(
        "explore",
        "Map out the authentication system: find all files, classes, and functions involved",
    )
    print(f"Found: {exploration[:100]}...")

    # Step 2: Plan changes based on exploration
    print("Step 2: Planning...")
    plan = await manager.spawn(
        "plan",
        f"Based on this exploration:\n{exploration}\n\nPlan how to add OAuth2 support",
    )
    print(f"Plan: {plan[:100]}...")

    # Step 3: Implement (uses general agent with write access)
    print("Step 3: Implementing...")
    implementation = await manager.spawn(
        "general",
        f"Implement this plan:\n{plan}",
    )
    print(f"Done: {implementation[:100]}...")

    return implementation

# team_parallel.py
import asyncio
from harness.agents.manager import AgentManager

async def parallel_exploration(manager: AgentManager):
    # Spawn 3 explore agents in parallel
    results = await manager.spawn_parallel([
        ("explore", "Find all database models and their relationships"),
        ("explore", "Find all API routes and their handlers"),
        ("explore", "Find all test files and their coverage"),
    ])

    db_models, api_routes, test_coverage = results

    print(f"DB Models: {db_models[:100]}...")
    print(f"API Routes: {api_routes[:100]}...")
    print(f"Test Coverage: {test_coverage[:100]}...")

    # Now plan based on all three findings
    combined = f"""
    Database Models: {db_models}
    API Routes: {api_routes}
    Test Coverage: {test_coverage}
    """

    plan = await manager.spawn(
        "plan",
        f"Based on this analysis:\n{combined}\n\nPlan a comprehensive refactoring",
    )
    return plan

# team_workflow.py — full team workflow with parallel exploration
import asyncio
import time
import harness

async def team_workflow():
    start = time.time()

    # Phase 1: Parallel exploration (3 agents at once)
    print("=== Phase 1: Parallel Exploration ===")
    phase1_start = time.time()

    async for msg in harness.run(
        """Run three parallel explorations:
        1. Find all Python files with security vulnerabilities
        2. Find all files missing type hints
        3. Find all TODO/FIXME comments

        Use sub-agents to explore in parallel, then summarize all findings.""",
        provider="anthropic",
        model="claude-sonnet-4-20250514",
        permission_mode="plan",  # Read-only for safety
    ):
        if isinstance(msg, harness.TextMessage) and not msg.is_partial:
            print(msg.text)
        elif isinstance(msg, harness.Result):
            phase1_time = time.time() - phase1_start
            print(f"\nPhase 1 complete: {phase1_time:.1f}s")
            print(f"  Turns: {msg.turns}, Tools: {msg.tool_calls}")
            exploration_result = msg.text

    # Phase 2: Planning (single agent)
    print("\n=== Phase 2: Planning ===")
    async for msg in harness.run(
        f"Based on these findings:\n{exploration_result}\n\nCreate a prioritized action plan",
        provider="anthropic",
        permission_mode="plan",
    ):
        if isinstance(msg, harness.Result):
            plan_result = msg.text
            print(f"Plan created: {len(plan_result)} chars")

    # Phase 3: Review (single agent)
    print("\n=== Phase 3: Review ===")
    async for msg in harness.run(
        f"Review this plan for completeness and risks:\n{plan_result}",
        provider="anthropic",
        permission_mode="plan",
    ):
        if isinstance(msg, harness.Result):
            total_time = time.time() - start
            print(f"\nTotal time: {total_time:.1f}s")
            print(f"Total cost: ${msg.total_cost:.4f}")

asyncio.run(team_workflow())

# Start REPL
harness

# Use /team to decompose and parallelize
> /team Add comprehensive test coverage for the auth module
# Harness automatically:
# 1. Spawns an explore agent to find existing tests
# 2. Spawns an explore agent to find auth module files
# 3. Plans test cases
# 4. Implements tests