Subagent#

Subagent tool — spawn, monitor, and coordinate child agents.

Extracted from a single 1100-line module into a package for maintainability.

Package structure:

types.py — Data classes and module-level state (Subagent, ReturnType, etc.)
hooks.py — Completion notification system (LOOP_CONTINUE hook)
api.py — Public API (subagent, subagent_status, subagent_wait, etc.)
batch.py — Batch execution (BatchJob, subagent_batch, subagent_parallel, subagent_pipeline)
execution.py — Execution backends (thread, subprocess, process monitoring)

Instructions

You can create, check status, wait for, and read logs from subagents.

Subagents support a "fire-and-forget-then-get-alerted" pattern:
- Call subagent() to start an async task (returns immediately)
- Continue with other work
- Receive completion messages via the LOOP_CONTINUE hook
- Optionally use subagent_wait() for explicit synchronization

Key features:
- Agent profiles: Use profile names as agent_id for automatic profile detection
- model="provider/model": Override parent's model (route cheap tasks to faster models)
- use_subprocess=True: Run subagent in subprocess for output isolation
- use_acp=True: Run subagent via ACP protocol (supports any ACP-compatible agent)
- acp_command="claude-code-acp": Use a different ACP agent (default: gptme-acp)
- isolation="worktree": Run subagent in a git worktree for filesystem isolation (preferred over isolated=True)
- workdir="/path/to/dir": Set the working directory for the subagent (defaults to cwd)
- redact_secrets=True (default): Redact API keys, tokens, and passwords from workspace context
- context_window=0: Minimal context — only agent identity + tools, no workspace files (strongest isolation)
- context_window=N: Limit workspace context to at most N messages
- subagent_parallel(tasks, timeout, max_concurrent=N, budget=SubagentBudget(...)): Fan out N subagents and wait for all — returns ordered list of results. Use max_concurrent to cap simultaneous agents (extras queue). Use budget= to gate new spawns once a token ceiling is hit.
- subagent_pipeline(items, *stages, timeout): Multi-stage fan-out with no barrier between stages — item A advances to stage 2 while item B is still in stage 1; each stage callable receives (item_prompt, prev_result) and returns the next stage's prompt
- subagent_batch(): Start multiple subagents and return a BatchJob for explicit synchronization
- subagent_cancel(): Cancel a running subagent (SIGTERM for subprocess, marks result for threads)
- subagent_continue(agent_id, message): Reuse a finished non-isolated child for related follow-up work. Prefer this over spawning a new child when its prior findings or decisions matter: the child keeps its full conversation/session context, avoiding a repeated briefing and improving continuity (thread, subprocess, or ACP).
- subagent_steer(agent_id, message): Inject a steering message into a RUNNING subagent's conversation — redirect, clarify, or course-correct mid-run without restarting. Works for thread-mode and subprocess-mode subagents. Distinct from subagent_reply() which only works on finished clarification_needed subagents.
- subagent_wait_any(agent_ids, timeout): Wait for the first of N subagents to complete — returns (agent_id, result). Useful for race/hedging patterns.
- subagent_reply(agent_id, reply): Answer a clarification request and re-spawn the subagent (for subagents that already stopped with clarification_needed status)
- Hook-based notifications: Completions (and clarification requests) delivered as system messages

## Token Budget and Concurrency Control

Use ``SubagentBudget`` to coordinate token spend across a fleet of subagents.
Passing the same budget object to multiple ``subagent_parallel()`` calls
accumulates spend across iterations — the canonical dynamic fan-out loop:

```python
budget = SubagentBudget(total=500_000)   # 500k output tokens
results = []
while not budget.exhausted():
    batch = next_batch()   # your function to get the next chunk of work
    if not batch:
        break
    batch_results = subagent_parallel(batch, budget=budget, max_concurrent=4)
    results.extend(r for r in batch_results if r["status"] == "success")
    # items skipped after budget exhaustion have status="budget_exceeded"
```

Each result dict has ``input_tokens`` / ``output_tokens`` fields so you can
inspect per-agent spend. Budget is output-token-only (the expensive marginal
cost). Agents already running when the budget hits zero are allowed to finish
normally — only *new* spawns are blocked.

``max_concurrent`` limits how many agents run simultaneously; excess tasks are
queued (never dropped) and start as slots free up. It composes with ``budget``:
both caps are enforced.

## Context Isolation

Subagents do NOT inherit the parent's conversation history — they always start
with a fresh context. What subagents DO inherit (in context_mode="full"):

- Workspace files listed in gptme.toml [prompt] files (e.g. AGENTS.md, README)
- Dynamic context_cmd output (if configured in gptme.toml)
- User-level config files from ~/.config/gptme

This means secrets stored in workspace config files or produced by context_cmd
can reach the subagent. Secret patterns (API_KEY, TOKEN, PASSWORD, etc.) are
redacted by default (redact_secrets=True). Pass redact_secrets=False to disable
if legitimate config values are incorrectly redacted.

### Controlling context depth with context_window

Limit how much workspace context flows to the subagent. Reach for these when
you need tighter control over what the subagent sees:

- `context_window=None` (default): The subagent sees your full workspace (files,
  tools, recent conversation). Best for tasks that benefit from maximum awareness.
- `context_window=0`: **Strongest isolation** — the subagent gets only agent
  identity and tool descriptions, no workspace files or context_cmd output. Use
  this when the subagent handles sensitive data (secrets, prompts) that should
  not leak into verification or analysis tasks. The subagent knows only what
  you explicitly tell it in the task prompt.
- `context_window=N`: Limits workspace to at most N context messages. Useful when
  the default is too bloated but you still want the subagent to see some workspace
  history — trim without fully isolating.

`context_window=0` is equivalent to `context_mode="selective", context_include=["agent", "tools"]`
but is a simpler one-parameter alternative. Only applies to thread-mode subagents.

## Agent Profiles for Subagents

Use profiles to create specialized subagents with appropriate capabilities.
When agent_id matches a profile name, the profile is auto-applied:
- explorer: Read-only analysis (tools: read)
- researcher: Web research without file modification (tools: browser, read)
- developer: Full development capabilities (all tools)
- verifier: Critical review & validation (tools: read, shell, ipython, chats)
- isolated: Restricted processing for untrusted content (tools: read, ipython)
- computer-use: Visual UI testing specialist (tools: computer, vision, ipython, shell)
- browser-use: Web interaction and testing specialist (tools: browser, screenshot, vision, shell) — supports interactive browsing (open_page, click, fill, scroll) and one-shot reads

Example: `subagent("explorer", "Explore codebase")`
With model override: `subagent("researcher", "Find docs", model="openai/gpt-4o-mini")`
Computer-use example: `subagent("computer-use", "Click the Submit button, wait for the modal, and screenshot the result")`
Browser-use example: `subagent("browser-use", "Open localhost:5173, fill the chat input, click send, and report the result")`

Use subagent_read_log() to inspect a subagent's conversation log for debugging.

## Structured Delegation Template

For complex delegations, use this 7-section template for clear task handoff:

TASK: [What the subagent should do]
EXPECTED OUTCOME: [Specific deliverable - format, structure, quality bars]
REQUIRED SKILLS: [What capabilities the subagent needs]
REQUIRED TOOLS: [Specific tools the subagent should use]
MUST DO: [Non-negotiable requirements]
MUST NOT DO: [Explicit constraints and forbidden actions]
CONTEXT: [Background info, dependencies, related work]

Example prompt using the template:
'''
TASK: Implement the user authentication feature
EXPECTED OUTCOME: auth.py with login/logout endpoints, passing tests
REQUIRED SKILLS: Python, FastAPI, JWT tokens
REQUIRED TOOLS: save, shell (for pytest)
MUST DO: Use bcrypt for password hashing, return proper HTTP status codes
MUST NOT DO: Store plaintext passwords, skip input validation
CONTEXT: This is for the gptme server API, see existing endpoints in server.py
'''

## Clarification Requests

When a subagent ends with a ``clarify`` block, it signals that it needs more
information from the parent before it can continue:

```clarify
Which output format should I use: JSON or CSV?
```

The parent receives a hook notification:
  ❓ Subagent 'X' needs clarification: Which output format should I use: JSON or CSV?
  Call subagent_reply('X', '<your answer>') to continue.

Use ``subagent_reply(agent_id, reply)`` to answer and re-spawn the subagent.
The re-spawned subagent receives the original prompt plus the Q&A so it can
complete the task without losing context.

Examples

Executor Mode (single task)

User	compute fib 13 using a subagent
Assistant	Starting a subagent to compute the 13th Fibonacci number. ipython subagent("fib-13", "compute the 13th Fibonacci number")
System	Subagent started successfully.
Assistant	Now we need to wait for the subagent to finish the task. ipython subagent_wait("fib-13")
System	{"status": "success", "result": "The 13th Fibonacci number is 233"}.

Planner Mode (multi-task delegation)

User	implement feature X with tests
Assistant	I'll use planner mode to delegate implementation and testing to separate subagents. ipython subtasks = [ {{"id": "implement", "description": "Write implementation for feature X"}}, {{"id": "test", "description": "Write comprehensive tests"}}, ] subagent("feature-planner", "Feature X adds new functionality", mode="planner", subtasks=subtasks)
System	Planner spawned 2 executor subagents.
Assistant	Now I'll wait for both subtasks to complete. ipython subagent_wait("feature-planner-implement")
System	{"status": "success", "result": "Implementation complete in feature_x.py"}. ipython subagent_wait("feature-planner-test")
System	{"status": "success", "result": "Tests complete in test_feature_x.py, all passing"}. ### Context Modes

Full Context (default)

User

analyze this codebase

Assistant

I'll use full context mode for comprehensive analysis.ipython
subagent("analyze", "Analyze code quality and suggest improvements", context_mode="full")

Selective Context (choose specific components)

User

write tests using pytest

Assistant

I'll use subprocess mode so selective context can include workspace files without inheriting the full parent context.ipython
subagent("tests", "Write pytest tests for the calculate function", context_mode="selective", context_include=["workspace"], use_subprocess=True)

Subprocess Mode (output isolation)

User

run a subagent without output mixing with parent

Assistant

I'll use subprocess mode for better output isolation.ipython
subagent("isolated", "Compute complex calculation", use_subprocess=True)

System

Subagent started in subprocess mode.

Workspace-Aware Subagent (explicit workdir)

User

I just cd'd into /path/to/project which has a gptme.toml — spawn a subagent there

Assistant

I'll use workdir to make the subagent operate in that workspace and load its config.ipython
subagent("project", "Add feature X", workdir="/path/to/project", use_subprocess=True)

System

Subagent started in subprocess mode (workdir=/path/to/project).

ACP Mode (multi-harness support)

User

delegate this task to a Claude Code agent

Assistant

I'll use ACP mode to run this via a different agent harness.ipython
subagent("claude-task", "Analyze and refactor the auth module", use_acp=True, acp_command="claude-code-acp")

System

Started subagent "claude-task" in ACP mode.

List Subagents (observability)

User

what subagents are currently running?

Assistant

I'll use subagent_list to check running agents.ipython
subagent_list()

System

Listing 2 subagents::
  - analyze (running, 42s) -- "Analyze the codebase architecture..."
  - fib-13 (success, 120s) -- "compute the 13th Fibonacci number"

Parallel Fan-out (wait for all, ordered results)

User

implement, test, and document a feature in parallel and collect all results

Assistant

I'll use subagent_parallel to fan out all tasks and wait for them together.ipython
tasks = [
    ("impl", "Implement the user authentication feature"),
    ("test", "Write tests for authentication"),
    ("docs", "Document the authentication API"),
]
results = subagent_parallel(tasks, timeout=300)
for (agent_id, _), result in zip(tasks, results):
    print(f"{{agent_id}}: {{result['status']}} — {{result['result'][:60]}}")

System

impl: success — Authentication feature implemented in auth.py
test: success — 12 tests added, all passing
docs: success — API documented in docs/auth.md

Budget-Aware Parallel (fleet cap + token budget)

User

audit 20 modules but stop if we spend more than 300k output tokens

Assistant

I'll use SubagentBudget with max_concurrent to cap spend and concurrency.ipython
from gptme.tools.subagent import SubagentBudget, subagent_parallel

budget = SubagentBudget(total=300_000)   # 300k output tokens
modules = [f"module_{i}" for i in range(20)]
tasks = [(m, f"Audit {m} for security issues") for m in modules]

# At most 4 agents run simultaneously; extras queue.

# New spawns are blocked once 300k output tokens are consumed.
results = subagent_parallel(tasks, max_concurrent=4, budget=budget)

for (agent_id, _), r in zip(tasks, results):
    status = r["status"]
    tokens = r.get("output_tokens") or 0
    print(f"{agent_id}: {status} ({tokens} tokens)")

print(f"Total output tokens used: {budget.spent()}")

System

module_0: success (1200 tokens)
module_1: success (980 tokens)
...
module_14: budget_exceeded (0 tokens)
Total output tokens used: 298430

Dynamic Fan-out Loop (respawn until budget exhausted)

User

keep running subagents on a work queue until we hit 500k tokens

Assistant

I'll loop over batches with a shared budget — the loop exits automatically when tokens run out.ipython
from gptme.tools.subagent import SubagentBudget, subagent_parallel

budget = SubagentBudget(total=500_000)
work_queue = [("task-" + str(i), f"Process item {i}") for i in range(100)]
all_results = []

while work_queue and not budget.exhausted():
    batch, work_queue = work_queue[:5], work_queue[5:]
    batch_results = subagent_parallel(batch, budget=budget, max_concurrent=4)
    all_results.extend(r for r in batch_results if r["status"] == "success")
    # Status "budget_exceeded" means remaining budget was hit mid-batch

print(f"Completed {len(all_results)} tasks, spent {budget.spent()} output tokens")

System

Completed 34 tasks, spent 499821 output tokens

Pipeline (multi-stage fan-out, no barrier between stages)

User	review these files in two stages — first find issues, then verify each finding
Assistant	I'll use subagent_pipeline so file B's review starts while file A's verification is running. ipython results = subagent_pipeline( [("auth", "Review auth.py for bugs"), ("db", "Review db.py for bugs")], # Stage 0: review each file lambda item, _: item, # Stage 1: adversarially verify the review findings lambda item, prev: "Verify these findings, keep only real bugs: " + prev, timeout=300, ) # auth advances to stage 1 as soon as its stage 0 finishes, # while db may still be in stage 0. for (prefix, _), stage_results in zip([("auth", ...), ("db", ...)], results): print(f"{prefix}: {stage_results[-1]['status']}")
System	auth-s0 done → auth-s1 started; db-s0 done → db-s1 started
System	auth: success, db: success

Batch Execution (fire-and-forget with explicit sync)

User

start tasks in background and continue working

Assistant

I'll use subagent_batch to start tasks in the background. Completion hooks will notify me.ipython
job = subagent_batch([
    ("impl", "Implement the user authentication feature"),
    ("test", "Write tests for authentication"),
])

# Do other work while subagents run — hook notifications arrive automatically:

# "✅ Subagent 'impl' completed: ..."

# Or explicitly wait for all when needed:

results = job.wait_all(timeout=300)

System

Started batch of 2 subagents: ['impl', 'test']

Fire-and-Forget with Hook Notifications

User	start a subagent and continue working
Assistant	I'll spawn a subagent. Completion will be delivered via the LOOP_CONTINUE hook. ipython subagent("compute-demo", "Compute pi to 100 digits") # I can continue with other work now # When the subagent completes, I'll receive a system message like: # "✅ Subagent 'compute-demo' completed: pi = 3.14159..."
System	Started subagent "compute-demo"
System	✅ Subagent 'compute-demo' completed: pi = 3.14159265358979...

Profile-Based Subagents (auto-detected from agent_id)

User

explore this codebase and summarize the architecture

Assistant

I'll use the explorer profile for a read-only analysis.ipython
subagent("explorer", "Analyze the codebase architecture and summarize key patterns")

System

Subagent started successfully.

Profile with Model Override

User

research best practices for error handling

Assistant

I'll spawn a researcher subagent with a faster model for web research.ipython
subagent("researcher", "Research error handling best practices in Python", model="openai/gpt-4o-mini")

System

Subagent started successfully.

Structured Delegation Template

User

implement a robust auth feature

Assistant

I'll use the structured delegation template for clear task handoff.ipython
subagent("auth-impl", "TASK: Implement JWT auth | OUTCOME: auth.py with tests | MUST: bcrypt, validation | MUST NOT: plaintext passwords")

System

Subagent started successfully.

Isolated Subagent (Worktree)

User

implement a feature without affecting my working directory

Assistant

I'll run the subagent in an isolated git worktree so it won't modify your files.ipython
subagent("feature-impl", "Implement the new caching layer in cache.py", isolated=True)

System

Subagent started successfully.

Context-Isolated Subagent (no workspace context)

User

verify this output without exposing our workspace secrets to the subagent

Assistant

I'll use context_window=0 so the subagent only sees what I explicitly give it in the prompt, with no workspace files or secrets inherited.ipython
subagent("verifier", "Check that the output file has no syntax errors", context_window=0)

System

Subagent started successfully.

Steer a Running Subagent (in-flight course correction)

User

redirect a running subagent without restarting it

Assistant

I'll use subagent_steer() to inject a new instruction into the running subagent's conversation.ipython
subagent("researcher", "Research Python web frameworks and their performance")

# ... later, after checking progress ...

subagent_steer("researcher", "Focus only on async frameworks — skip synchronous ones like Flask/Django")

System

Steering message queued for subagent 'researcher'. It will be injected into the subagent's conversation on its next loop iteration.

class gptme.tools.subagent.BatchJob

Manages a batch of subagents for parallel execution.

Note: With the hook-based notification system, the orchestrator will receive completion messages automatically via the LOOP_CONTINUE hook. This class provides additional utilities for explicit synchronization when needed.

__init__(agent_ids: list[str], results: dict[str, ~gptme.tools.subagent.types.ReturnType] = <factory>, output_schema: type | dict | None = None, budget: ~gptme.tools.subagent.types.SubagentBudget | None = None, _lock: ~_thread.allocate_lock = <factory>) → None

get_completed() → dict[str, dict]

Get results of completed subagents so far.

When the BatchJob was created with an output_schema (via subagent_batch(output_schema=...)) the results are automatically parsed through _parse_result() before being returned, matching the behaviour of wait_all().

is_complete() → bool: Check if all subagents have completed.

total_tokens() → dict[str, int | None]

Return aggregated token counts across all completed subagents.

Sums input_tokens and output_tokens from each completed result. Any subagent whose log has no usage metadata contributes None to its part — the aggregate is None when no completed subagent has token data, otherwise it is the sum of available counts.

Returns:: Dict with keys "input_tokens" and "output_tokens". Values are integers (sum of available counts) or None when no usage metadata was found in any completed subagent’s log.

Example:

job = subagent_batch([("a", "task A"), ("b", "task B")])
results = job.wait_all()
stats = job.total_tokens()
print(f"Tokens used: {stats['input_tokens']} in / {stats['output_tokens']} out")

wait_all(timeout: int = 300, cancel_on_failure: bool = False) → dict[str, dict]

Wait for all subagents to complete concurrently.

Uses a thread pool to wait for all subagents simultaneously, so the wall-clock time is bounded by the slowest agent, not the sum of all agent times.

Parameters:

timeout – Maximum seconds to wait for all subagents
cancel_on_failure –
When True, cancel all remaining running subagents as soon as the first failure or timeout is detected. Cancelled agents are marked with status="failure" and result="Cancelled due to sibling failure" in the returned dict. For subprocess-mode agents this sends SIGTERM (fast); for thread-mode agents it marks the result immediately while the background thread continues until its next natural checkpoint.

Note: regardless of this flag, any agents still running when the overall timeout expires are always cancelled so subprocess/ACP agents do not keep running after the caller has received timed-out results.

Returns:

Dict mapping agent_id to status dict. When output_schema is set, the "result" value is the parsed/validated object rather than a raw JSON string.

wait_any(timeout: int = 300) → tuple[str, dict]

Wait for the first subagent to complete and return its result.

Useful for speculative/hedging patterns: spawn N subagents and take whichever finishes first, then cancel the rest.

Parameters:: timeout – Maximum seconds to wait for any agent to complete.
Returns:: Tuple of (agent_id, result_dict) for the first agent that completes. When output_schema is set on the BatchJob the result is automatically parsed.
Raises:: TimeoutError – If no agent completes within timeout seconds.

Example:

job = subagent_batch([
    ("attempt-fast", "Try the quick approach for task X"),
    ("attempt-thorough", "Try the thorough approach for task X"),
])
first_id, result = job.wait_any(timeout=120)
print(f"{first_id} finished first: {result['status']}")
# Cancel the remaining agent
from gptme.tools.subagent import subagent_cancel
for aid in job.agent_ids:
    if aid != first_id:
        subagent_cancel(aid)

class gptme.tools.subagent.ReturnType

ReturnType(status: Literal[‘running’, ‘success’, ‘failure’, ‘clarification_needed’, ‘timeout’, ‘budget_exceeded’, ‘cancelled’], result: str | dict[str, object] | None = None, input_tokens: int | None = None, output_tokens: int | None = None)

__init__(status: Literal['running', 'success', 'failure', 'clarification_needed', 'timeout', 'budget_exceeded', 'cancelled'], result: str | dict[str, object] | None = None, input_tokens: int | None = None, output_tokens: int | None = None) → None

class gptme.tools.subagent.Subagent

Represents a running or completed subagent.

Supports both thread-based (default) and subprocess-based execution modes. Subprocess mode provides better output isolation.

Communication Model:

Parent sends prompt, child executes independently
Results retrieved after completion via status()/subagent_wait()
Subagents can use the clarify code block to signal ambiguity; the parent receives a hook notification and can call subagent_reply() to re-spawn with the question answered.

__init__(agent_id: str, prompt: str, thread: ~threading.Thread | None, logdir: ~pathlib.Path, model: str | None, context_mode: ~typing.Literal['full', 'selective'] = 'full', context_include: list[str] | None = None, profile: str | None = None, output_schema: type | dict | None = None, use_acp: bool = False, process: ~subprocess.Popen | None = None, execution_mode: ~typing.Literal['thread', 'subprocess', 'acp'] = 'thread', acp_command: str | None = None, acp_session_id: str | None = None, workdir: ~pathlib.Path | None = None, base_workdir: ~pathlib.Path | None = None, isolated: bool = False, worktree_path: ~pathlib.Path | None = None, repo_path: ~pathlib.Path | None = None, isolation_mode: ~typing.Literal['worktree'] | None = None, timeout: int = 1800, role: ~typing.Literal['general', 'explore', 'implement', 'verify'] | None = None, redact_secrets: bool = True, context_window: int | None = None, context_turns: int | None = None, started_at: float = <factory>, max_time: float | None = None, parent_logdir: ~pathlib.Path | None = None, prompt_queue_closed: ~threading.Event = <factory>) → None

is_running() → bool: Check if the subagent is still running.

class gptme.tools.subagent.SubagentBudget

Fleet-wide output-token budget tracker. Thread-safe.

Pass a shared instance to subagent_parallel() or subagent_pipeline() to gate new agent spawns when the budget is exhausted. Agents that are already running when the budget hits zero are allowed to complete normally — only new spawns are blocked.

Tracks output tokens only (the expensive marginal cost), matching the Claude Code Workflow budget.spent() semantics.

Example:

from gptme.tools.subagent import subagent_parallel, SubagentBudget

budget = SubagentBudget(total=200_000)   # 200k output tokens
results = subagent_parallel(tasks, budget=budget)
# Items spawned after the budget was exhausted have status="budget_exceeded"

Dynamic loop pattern (accumulate until budget runs out):

budget = SubagentBudget(total=500_000)
findings = []
while not budget.exhausted():
    batch_results = subagent_parallel(next_batch, budget=budget)
    findings.extend(r["result"] for r in batch_results if r["status"] == "success")

__init__(total: int | None = None) → None

exhausted() → bool: Return True when a finite budget has been fully consumed.

record(output_tokens: int) → None: Add output_tokens to the spent counter.

remaining() → float: Return remaining token budget, or float('inf') when total is None.

spent() → int: Return total output tokens spent so far.

class gptme.tools.subagent.SubtaskDef: Definition of a subtask for planner mode.

gptme.tools.subagent.get_current_agent_id() → str | None

Return the agent_id of the currently running subagent, or None.

Thread-mode subagents: set via _create_subagent_thread (thread-local). Subprocess-mode subagents: set via GPTME_SUBAGENT_AGENT_ID env var.

gptme.tools.subagent.notify_completion(agent_id: str, status: Literal['running', 'success', 'failure', 'clarification_needed', 'timeout', 'budget_exceeded', 'cancelled'], summary: str) → None

Add a subagent completion to the notification queue.

Called by the monitor thread when a subagent finishes. The queued notification will be delivered via the subagent_completion hook during the next LOOP_CONTINUE cycle.

Parameters:

agent_id – The subagent’s identifier
status – “success” or “failure”
summary – Brief summary of the result

gptme.tools.subagent.notify_progress(agent_id: str, message: str) → None

Add a subagent progress update to the notification queue.

Called by the progress tool when a subagent sends an intermediate update. The parent’s LOOP_CONTINUE hook delivers it as a system message so the orchestrator can react without blocking on subagent_wait().

Note: For thread-mode subagents the progress tool calls this directly (same process). For subprocess-mode subagents, _poll_subprocess_progress reads from the file channel and calls this function on behalf of the child process.

Parameters:

agent_id – The subagent’s identifier
message – Progress update message

gptme.tools.subagent.subagent(agent_id: str, prompt: str, mode: Literal['executor', 'planner'] = 'executor', subtasks: list[SubtaskDef] | None = None, execution_mode: Literal['parallel', 'sequential'] = 'parallel', context_mode: Literal['full', 'selective'] = 'full', context_include: list[str] | None = None, output_schema: type | dict | None = None, use_subprocess: bool | None = None, use_acp: bool = False, acp_command: str = 'gptme-acp', profile: str | None = None, model: str | None = None, isolated: bool | None = None, isolation: Literal['worktree'] | None = None, timeout: int = 1800, role: Literal['general', 'explore', 'implement', 'verify'] | None = None, redact_secrets: bool = True, context_window: int | None = None, max_time: float | None = None, context_turns: int | None = None, workdir: str | Path | None = None)

Starts an asynchronous subagent. Returns None immediately.

Subagent completions are delivered via the LOOP_CONTINUE hook, enabling a “fire-and-forget-then-get-alerted” pattern where the orchestrator can continue working and get notified when subagents finish.

Profile auto-detection: If agent_id matches a known profile name (e.g. “explorer”, “researcher”, “developer”, “verifier”) or a common role alias (“explore”→”explorer”, “research”→”researcher”, “impl”/”dev”→”developer”, “verify”→”verifier”), the profile is applied automatically — no need to pass profile separately.

Role-based defaults (role parameter):

"explore": Defaults profile to explorer (read-only analysis)
"implement": Defaults profile to developer (full capability)
"verify": Defaults profile to verifier plus use_subprocess=True and isolated=True (read-only validation in isolation)

Explicit arguments override role defaults.

Parameters:

agent_id – Unique identifier for the subagent. If it matches a known profile name (or a common alias like impl/dev), that profile is auto-applied (unless profile is explicitly set to something else).
prompt – Task prompt for the subagent (used as context for planner mode)
mode – “executor” for single task, “planner” for delegating to multiple executors
subtasks – List of subtask definitions for planner mode (required when mode=”planner”)
execution_mode – “parallel” (default) runs all subtasks concurrently, “sequential” runs subtasks one after another. Only applies to planner mode.
context_mode – Controls what context is shared with the subagent: - “full” (default): Share complete context (agent identity, tools, workspace) - “selective”: Share only specified context components (requires context_include)
context_include – For selective mode, list of context components to include: - Thread mode supports “agent” and “tools” - Subprocess mode also supports “workspace”, which maps to the CLI’s “files” context Legacy subprocess values like “files”, “cmd”, and “all” are still accepted.
use_subprocess – If True, run subagent in subprocess for output isolation. Subprocess mode captures stdout/stderr separately from the parent.
use_acp – If True, run subagent via ACP (Agent Client Protocol). This enables multi-harness support — the subagent can be any ACP-compatible agent (gptme, Claude Code, Cursor, etc.). Requires the acp package: pip install ‘gptme[acp]’.
acp_command – ACP agent command to invoke (default: “gptme-acp”). Only used when use_acp=True. Can be any ACP-compatible CLI.
profile – Agent profile name to apply. Profiles provide: - System prompt customization (behavioral hints) - Tool access restrictions (which tools the subagent can use) - Behavior rules (read-only, no-network, etc.) Use ‘gptme-util profile list’ to see available profiles. Built-in profiles: default, explorer, researcher, developer, verifier, isolated, computer-use, browser-use. If not set, auto-detected from agent_id when it matches a profile name.
model – Model to use for the subagent. Overrides parent’s model. Useful for routing cheap tasks to faster/cheaper models.
isolated – If True, run the subagent in a git worktree for filesystem isolation. The subagent gets its own copy of the repository and can modify files without affecting the parent. The worktree is automatically cleaned up after the subagent completes. Falls back to a temporary directory if not in a git repo. Prefer isolation="worktree" for new code — it is the string-based API equivalent and enables smarter cleanup behaviour.
isolation –
String-based isolation mode. Use "worktree" to create a temporary git worktree for the subagent, giving it an isolated copy of the repository to work in. On completion:
- No local changes: worktree directory and branch are removed automatically (zero cleanup needed).
- Local changes exist: the branch is preserved and its name is reported in the result so the orchestrator can inspect or merge it. The working-tree directory is still removed.
Falls back to a temporary directory when not in a git repository. Equivalent to isolated=True but adds smart cleanup behaviour.
timeout – Maximum seconds before the subprocess monitor kills the subagent (default 1800 = 30 min). Only applies to subprocess mode.
redact_secrets –
If True (default), scrub common secret patterns from workspace context messages before they are passed to the subagent. Redacts values from lines where the variable name matches patterns like API_KEY, TOKEN, PASSWORD, PRIVATE_KEY, etc.

Note: subagents do NOT inherit the parent’s conversation history — they always start with a fresh context containing only the task prompt and workspace context (files from gptme.toml [prompt] files, and context_cmd output when context_mode=”full”). This option sanitizes that inherited workspace context.

Only applies to thread-mode subagents (subprocess and ACP modes run as a separate gptme process and handle their own context). Set to False to disable redaction if legitimate config values are being incorrectly redacted.
context_window –
Limit workspace context messages passed to the subagent. Controls how much of the workspace context (files from gptme.toml [prompt] files, context_cmd output) is shared with the subagent.
- None (default): no limit — full workspace context is shared.
- 0: minimal context — only agent identity and tools; no workspace files or context_cmd output. Equivalent to context_mode="selective", context_include=["agent", "tools"].
- N > 0: at most N workspace context messages are passed.
Use context_window=0 when the subagent does not need the parent workspace configuration (e.g. a verification task that should only see what the orchestrator explicitly tells it).

Only applies to thread-mode subagents; has no effect in subprocess or ACP modes (which build their own context as a separate process).
max_time –
Wall-clock time limit in seconds. When set, a watchdog timer marks the subagent result as "timeout" after max_time seconds and delivers a timeout status notification via the LOOP_CONTINUE hook. In subprocess mode the child process is terminated. In thread mode the background thread is not force-stopped; callers see the cached timeout result immediately while the thread continues until it finishes naturally. Defaults to None (no limit).

Use this for defensive orchestration (prevent a stuck subagent from blocking the parent) or hard time budgets in autonomous sessions. max_time=None is fully backwards-compatible — no change in behavior.
context_turns –
Number of recent parent conversation turns to forward to the subagent as context. A “turn” starts at a user message and includes all subsequent assistant and tool-result (system) messages until the next user message, so the total message count per turn varies with the number of tool calls. The messages are injected as a system message so the subagent understands what the parent has been doing without confusing its own conversation flow.
- None (default): no parent context forwarded (current behavior).
- N > 0: forward the last N turns from the parent’s active log.
The parent log is fetched automatically from the currently active LogManager (set by the chat loop via ContextVar). This works when subagent() is called from within the ipython tool during a running chat session.

Use this when the subagent needs awareness of what the parent has already done (e.g. “the parent tried A and B, now try C”) or when the task prompt alone doesn’t provide enough context.

Only applies to thread-mode subagents; has no effect in subprocess or ACP modes.
workdir –
Working directory for the subagent. Defaults to the current working directory (Path.cwd()) when None.

Use this when you want the subagent to operate in a specific directory — for example, when a cd into a project with a gptme.toml triggers workspace detection and you want the subagent to load that workspace’s config:
```
subagent("impl", "Add feature X", workdir="/path/to/project",
         use_subprocess=True)
```
In subprocess mode the subagent process starts with this as its cwd, so it picks up the gptme.toml from that directory. In thread mode the workspace context (files, context_cmd) is loaded relative to this path.

Returns:

Starts asynchronous execution.

In executor mode, starts a single task execution. In planner mode, starts execution of all subtasks using the specified execution_mode.

Executors use the complete tool to signal completion with a summary. The full conversation log is available at the logdir path.

Return type:

None

gptme.tools.subagent.subagent_batch(tasks: list[tuple[str, str]], use_subprocess: bool = False, use_acp: bool = False, acp_command: str = 'gptme-acp', model: str | None = None, profile: str | None = None, isolated: bool = False, output_schema: type | dict | None = None, workdir: str | Path | None = None, context_turns: int | None = None, context_window: int | None = None, redact_secrets: bool = True, budget: SubagentBudget | None = None) → BatchJob

Start multiple subagents in parallel and return a BatchJob to manage them.

This is a convenience function for fire-and-gather patterns where you want to run multiple independent tasks concurrently.

With the hook-based notification system, completion messages are delivered automatically via the LOOP_CONTINUE hook. The BatchJob provides additional utilities for explicit synchronization when needed.

Parameters:

tasks – List of (agent_id, prompt) tuples
use_subprocess – If True, run subagents in subprocesses for output isolation
use_acp – If True, run subagents via ACP protocol
acp_command – ACP agent command (default: “gptme-acp”)
model – Model override applied to every subagent.
profile – Agent profile name applied to every subagent.
isolated – If True, run each subagent in its own git worktree so file edits don’t conflict between agents or with the parent.
output_schema – Optional Pydantic model class. When set, subagents are instructed to return JSON matching the schema in their complete block. Results are automatically parsed when wait_all() is called — the "result" value in each returned dict will be the parsed/validated object rather than a raw JSON string, matching the behaviour of subagent_parallel(output_schema=...).
workdir – Working directory passed to every subagent. Useful when running subagents against a specific project directory.
context_turns – Number of recent parent conversation turns to forward to each subagent as context prefix. Pass None (default) to use no parent context.
context_window – Limit workspace context messages passed to each subagent. Pass 0 for strongest isolation (subagent sees only agent identity and tools, no workspace files). Pass None (default) for the full inherited workspace context. Only applies to thread-mode subagents.
redact_secrets – If True (default), redact secrets from workspace context passed to subagents. Pass False only if you need subagents to see config values that are incorrectly flagged as secrets.

Returns:

A BatchJob instance for managing the parallel subagents. The BatchJob provides wait_all(timeout) to wait for completion, is_complete() to check status, and get_completed() for partial results.

Example:

job = subagent_batch([
    ("impl", "Implement feature X"),
    ("test", "Write tests for feature X"),
    ("docs", "Document feature X"),
])
# Orchestrator continues with other work...
# Completion messages delivered via LOOP_CONTINUE hook:
#   "✅ Subagent 'impl' completed: Feature implemented"
#   "✅ Subagent 'test' completed: 5 tests added"
#
# Or explicitly wait for all if needed:
results = job.wait_all(timeout=300)

gptme.tools.subagent.subagent_cancel(agent_id: str) → str

Cancel a running subagent.

For subprocess-mode subagents, writes a cancel op to logdir/control.jsonl so the agent’s cooperative checkpoint can exit cleanly before SIGTERM arrives, then sends SIGTERM (and SIGKILL after 5s) as escalation.

For thread-mode subagents, writes the cancel op and marks the result cache so callers don’t block. The thread stops at its next STEP_PRE checkpoint and releases its concurrency slot.

ACP-mode subagents keep today’s cache-mark-only behavior (no control file).

Parameters:: agent_id – The subagent to cancel
Returns:: A human-readable status message

gptme.tools.subagent.subagent_continue(agent_id: str, message: str) → None

Continue a finished subagent in its existing conversation.

Thread and subprocess children append the follow-up to their original gptme conversation log. ACP children reload their durable ACP session before the prompt. Isolated children cannot be continued after their workspace cleanup.

Parameters:

agent_id – The completed subagent to continue.
message – Follow-up instruction for the child.

gptme.tools.subagent.subagent_list() → list[dict]

Returns a list of all subagents with their current status.

Each entry contains: - agent_id: The subagent identifier - status: running/success/failure/clarification_needed - model: The model used (or None) - execution_mode: thread/subprocess/acp - elapsed_s: Seconds since the subagent started (from started_at timestamp) - prompt_preview: First 100 characters of the prompt

Useful for: - Interactive sessions: “what’s running right now?” - Orchestrators deciding whether to spawn more agents - Debugging runaway subagent fans

gptme.tools.subagent.subagent_parallel(tasks: list[tuple[str, str]], timeout: int = 300, max_concurrent: int | None = None, use_subprocess: bool = False, use_acp: bool = False, acp_command: str = 'gptme-acp', model: str | None = None, profile: str | None = None, isolated: bool = False, isolation: Literal['worktree'] | None = None, output_schema: type | dict | None = None, workdir: str | Path | None = None, context_turns: int | None = None, context_window: int | None = None, redact_secrets: bool = True, cancel_on_failure: bool = False, budget: SubagentBudget | None = None) → list[dict]

Fan out N subagents in parallel, wait for all, return results as an ordered list.

This is the simplest way to run independent tasks concurrently and collect all results. Unlike subagent_batch(), this function blocks until every subagent has finished (or timed out) and returns the results in the same order as the input tasks.

Waits for all subagents concurrently — wall-clock time is bounded by the slowest agent, not the sum of all agent times.

Parameters:

tasks – List of (agent_id, prompt) tuples. Each agent_id must be unique within this call.
timeout – Maximum seconds to wait for all subagents to finish. Agents that exceed this deadline are reported with status "timeout".
max_concurrent – Maximum number of subagents to run at the same time. When set, excess tasks are queued and spawned as earlier agents complete — this never raises an error, it only limits concurrency. Pass None (default) to spawn all tasks simultaneously. Useful for large fan-outs where running too many agents at once would exhaust system resources or API rate limits.
use_subprocess – If True, run each subagent in a subprocess for output isolation. Subprocess mode captures stdout/stderr separately and supports hard-kill on timeout.
use_acp – If True, run each subagent via the ACP protocol.
acp_command – ACP agent command (default: “gptme-acp”). Only used when use_acp=True.
model – Model override applied to every subagent. Pass None to inherit the parent’s model.
profile – Agent profile name applied to every subagent (e.g. "explorer", "developer", "verifier").
isolated – If True, run each subagent in its own git worktree so file edits don’t conflict between agents or with the parent. Prefer isolation="worktree" for new code.
isolation – String-based isolation mode. Use "worktree" to give each subagent its own git worktree. On completion, worktrees with no local changes are auto-removed; worktrees with commits ahead of HEAD have their branch preserved (reported in the result) for the caller to inspect or merge.
output_schema – Optional Pydantic model class. When set, subagents are instructed to return valid JSON matching the schema in their complete block. Results are automatically parsed: on success the "result" value is the parsed/validated object (a dict for Pydantic models) rather than a raw JSON string. A "parse_error" key is added to any result that cannot be parsed.
workdir – Working directory passed to every subagent. Useful when running subagents against a specific project directory.
context_turns – Number of recent parent conversation turns to forward to each subagent as context prefix. Pass None (default) to use no parent context.
context_window – Limit workspace context messages passed to each subagent. Pass 0 for strongest isolation (subagent sees only agent identity and tools, no workspace files). Pass None (default) for the full inherited workspace context. Only applies to thread-mode subagents.
redact_secrets – If True (default), scrub common secret patterns from workspace context before passing it to subagents.
cancel_on_failure –
When True, cancel all remaining running subagents as soon as the first failure or timeout is detected. Cancelled agents are reported with status="failure" and result="Cancelled due to sibling failure". For subprocess-mode agents this sends SIGTERM (fast); for thread-mode agents the result is marked immediately while the background thread finishes naturally.

Use this for defensive fan-out orchestration where one failing subagent means the overall task has failed and continuing would waste resources. Equivalent to calling subagent_cancel() manually after subagent_wait_any() detects a failure, but handled automatically inside the parallel wait.
budget – Optional shared token budget. When set, each agent’s output tokens are recorded after it completes, and any agent whose spawn is attempted after the budget is exhausted is returned immediately with status="budget_exceeded" without being started. Agents already running when the budget hits zero are allowed to finish normally. Pass a SubagentBudget to share a budget across multiple subagent_parallel() calls (dynamic fan-out loop).

Returns:

List of result dicts in the same order as tasks. Each dict has "status" ("success" / "failure" / "timeout" / "budget_exceeded") and "result" (parsed object when output_schema is set, else the summary text from the subagent’s complete block).

Example:

# Process three independent tasks in parallel
results = subagent_parallel([
    ("researcher", "Research the top 5 Python async frameworks"),
    ("coder",      "Implement a basic async HTTP client"),
    ("tester",     "Write pytest tests for an async HTTP client"),
])
for (agent_id, _), result in zip(tasks, results):
    print(f"{agent_id}: {result['status']} — {result['result'][:80]}")

# With worktree isolation for concurrent file edits (string API)
results = subagent_parallel(
    [("fix-a", "Fix bug in module A"), ("fix-b", "Fix bug in module B")],
    isolation="worktree",
)

# Fail-fast: cancel the fleet when the first subagent fails
results = subagent_parallel(
    [("verifier-a", "Verify output A"), ("verifier-b", "Verify output B")],
    cancel_on_failure=True,
)
if any(r["status"] == "failure" for r in results):
    print("Verification failed — remaining agents were cancelled")

# With structured output (Pydantic model)
from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    score: int
    issues: list[str]

results = subagent_parallel(
    [("a1", "Analyze module A"), ("a2", "Analyze module B")],
    output_schema=AnalysisResult,
)
for r in results:
    if r["status"] == "success":
        analysis = r["result"]  # already a validated dict
        print(f"Score: {analysis['score']}, Issues: {analysis['issues']}")

# Budget-aware dynamic loop (spawn until budget runs out)
from gptme.tools.subagent import SubagentBudget
budget = SubagentBudget(total=200_000)
while not budget.exhausted():
    results = subagent_parallel(next_batch, budget=budget)
    # items skipped after budget exhaustion have status="budget_exceeded"

# Fleet cap: run at most 4 agents at a time, queueing the rest
results = subagent_parallel(
    [("task-1", "..."), ("task-2", "..."), ("task-3", "..."), ("task-4", "..."),
     ("task-5", "..."), ("task-6", "...")],
    max_concurrent=4,
)
# task-5 and task-6 start only after earlier slots free up

gptme.tools.subagent.subagent_pipeline(items: list[tuple[str, str]], *stages: Callable[[str, str], str], timeout: float = 600, use_subprocess: bool = False, use_acp: bool = False, acp_command: str = 'gptme-acp', model: str | None = None, profile: str | None = None, isolated: bool = False, output_schema: type | dict | None = None, workdir: str | Path | None = None, context_turns: int | None = None, context_window: int | None = None, redact_secrets: bool = True, budget: SubagentBudget | None = None) → list[list[dict]]

Process items through multiple stages with no barrier between stages.

Each item is processed through all stages sequentially. Items at different stages run concurrently — item A can be in stage 2 while item B is still in stage 1. This is the “pipeline” pattern as opposed to repeated subagent_parallel() calls which add a full barrier between stages.

Wall-clock time is bounded by the slowest single-item chain, not the sum of the slowest per-stage.

Parameters:

items – List of (agent_id_prefix, initial_prompt) tuples.
*stages – Callables of the form stage(item_prompt, prev_result) -> str where item_prompt is the original item prompt and prev_result is the raw result text from the previous stage (empty string for the first stage). Each callable returns the prompt to use for the next subagent in the chain.
timeout – Maximum seconds to wait for the entire pipeline to finish.
use_subprocess – If True, run each subagent in a subprocess.
use_acp – If True, run each subagent via the ACP protocol.
acp_command – ACP agent command (default: “gptme-acp”). Only used when use_acp=True.
model – Model override applied to every subagent.
profile – Agent profile name applied to every subagent.
isolated – If True, run each subagent in its own git worktree.
output_schema – Optional Pydantic model class. When set, each final-stage subagent is instructed to return JSON matching the schema and results are automatically parsed.
workdir – Working directory passed to every subagent.
context_turns – Number of recent parent turns to forward to each subagent.
context_window – Limit workspace context messages passed to each subagent. Pass 0 for strongest isolation (subagent sees only agent identity and tools, no workspace files). Pass None (default) for the full inherited workspace context. Only applies to thread-mode subagents.
redact_secrets – If True (default), redact secrets from workspace context.

Returns:

List of lists of result dicts. results[i][j] is the result dict for item i at stage j. Each dict has "status" and "result" keys (plus "input_tokens" / "output_tokens" when available). When output_schema is set, the final-stage "result" value is the parsed/validated object rather than a raw JSON string.

Example:

# Two-stage review pipeline: find issues, then verify each finding
results = subagent_pipeline(
    [("file-auth", "Review auth.py"), ("file-db", "Review db.py")],
    # Stage 0: review
    lambda item, _: f"Review this file for bugs: {item}",
    # Stage 1: verify each review finding
    lambda item, prev: (
        f"Adversarially verify each finding in this review:\n{prev}\n"
        f"Original file to review: {item}"
    ),
)
# file-auth advances to stage 1 as soon as its stage 0 completes,
# while file-db may still be in stage 0.
for (prefix, _), stage_results in zip(items, results):
    final = stage_results[-1]
    print(f"{prefix}: {final['status']} — {final['result'][:80]}")

# With isolated worktrees so concurrent file edits don't conflict
results = subagent_pipeline(
    [("impl-a", "Implement feature A"), ("impl-b", "Implement feature B")],
    lambda item, _: item,
    lambda item, prev: f"Write tests for: {prev}",
    isolated=True,
)

gptme.tools.subagent.subagent_read_log(agent_id: str, max_messages: int = 50, include_system: bool = False, message_filter: str | None = None) → str

Read the conversation log of a subagent.

Parameters:

agent_id – The subagent to read logs from
max_messages – Maximum number of messages to return
include_system – Whether to include system messages
message_filter – Filter messages by role (user/assistant/system) or None for all

Returns:

Formatted log output showing the conversation

gptme.tools.subagent.subagent_reply(agent_id: str, reply: str) → None

Re-spawn a subagent that requested clarification.

When a subagent ends with a clarify block, it stops and asks the parent a question. Call this function with your answer to re-start the subagent. The new run receives the original prompt plus an appended Q&A block so it has full context.

Parameters:

agent_id – The subagent that raised the clarification request.
reply – Your answer to the subagent’s question.

gptme.tools.subagent.subagent_status(agent_id: str) → dict: Returns the status of a subagent.

gptme.tools.subagent.subagent_steer(agent_id: str, message: str) → str

Inject a steering message into a running subagent’s conversation.

The message is queued via the subagent’s logdir prompt-queue with a steer flag. The subagent’s STEP_PRE checkpoint hook drains steer-flagged messages at each step boundary so the very next LLM call in the same agentic turn sees the guidance immediately — no need to wait for the current tool chain to finish. Works for thread-mode and subprocess-mode subagents.

This is distinct from subagent_reply(), which re-spawns a subagent that has already stopped with a clarification_needed status. Use this function to steer a subagent that is still actively running.

Parameters:

agent_id – The running subagent to steer.
message – The guidance to inject. This will appear as a user turn in the subagent’s conversation on its next loop iteration.

Returns:

A human-readable confirmation message.

Raises:

ValueError – If no subagent with agent_id is found, or if the subagent has already finished (use subagent_reply() for clarification-needed subagents).
NotImplementedError – If the subagent is running in ACP mode, which does not expose a logdir channel for steering.

Note

Delivery is guaranteed for both thread-mode and subprocess-mode subagents:

Thread mode: prompt_queue_closed is a Python event set inside the subagent thread immediately when chat() returns.
Subprocess mode: chat() writes a "prompt-queue-closed" sentinel file to the logdir in its finally block — before the process exits — giving a child-side signal that closes the drain boundary precisely. The parent-side prompt_queue_closed event (set after _monitor_subprocess returns) is a second-level catch.

Any steer attempted after the sentinel/event is set raises ValueError.

Example:

subagent("researcher", "Research Python async frameworks")
# ... the researcher is going off-track ...
subagent_steer("researcher", "Focus only on frameworks with >5k GitHub stars")

gptme.tools.subagent.subagent_wait(agent_id: str, timeout: int = 60, max_result_chars: int = 2000) → dict

Waits for a subagent to finish.

Parameters:

agent_id – The subagent to wait for
timeout – Maximum seconds to wait (default 60)
max_result_chars – Truncate result text to this many characters (default 2000). Long subagent outputs are truncated to keep the parent’s context clean. Call subagent_read_log(agent_id) to read the full output.

Returns:

Status dict with ‘status’ and ‘result’ keys

gptme.tools.subagent.subagent_wait_any(agent_ids: list[str], timeout: int = 300) → tuple[str, dict]

Wait for the first of the given subagents to complete.

Useful for speculative/hedging patterns: spawn N subagents and take whichever finishes first, then cancel the rest with subagent_cancel().

Parameters:

agent_ids – List of agent IDs to wait on.
timeout – Maximum seconds to wait for any agent to complete.

Returns:

Tuple of (agent_id, result_dict) for the first agent that completes. result_dict has "status" ("success" / "failure" / "clarification_needed") and "result" keys.

Raises:

ValueError – If agent_ids is empty.
TimeoutError – If no agent completes within timeout seconds.

Example:

# Race pattern: take whichever approach finishes first
subagent("fast", "Quick attempt at task X")
subagent("thorough", "Thorough attempt at task X")
first_id, result = subagent_wait_any(["fast", "thorough"], timeout=120)
print(f"{first_id} won the race: {result['status']}")
# Cancel the slower agent
for aid in ("fast", "thorough"):
    if aid != first_id:
        subagent_cancel(aid)

Subagent Isolation Contract#

When spawning a subagent you need to know exactly what it inherits from the parent and what it starts fresh. There are four dimensions:

1. Workspace config loading

Thread mode (default, use_subprocess=False):: The subagent inherits the parent’s already-assembled workspace context — the [prompt] files from gptme.toml and the context_cmd output as they were loaded for the parent session. It does not re-read from the subagent’s working directory, so a subdirectory with its own gptme.toml will not be picked up automatically.
Subprocess mode (use_subprocess=True):: Spawns a fresh gptme process with workdir as the CWD, which naturally loads that directory’s gptme.toml. Use this when you want subagents to pick up directory-local workspace config.

Fine-grained control:

context_mode="selective" + context_include — share only specific components ("agent", "tools", "workspace") instead of the full workspace.

Behavior by mode:

Thread: fully supported — filters the inherited context to the specified components.

Subprocess: context_mode is ignored (the child loads its own workspace from gptme.toml); context_include=["workspace"] maps to the --context files CLI flag to include workspace files. Other context_include values are ignored in this mode.

ACP: both parameters are ignored.
context_window=N — limit how many inherited context messages are forwarded (0 = none, None = all). Thread mode only; ignored in subprocess and ACP.
context_turns=N — forward the last N turns of the parent conversation. Thread mode only; ignored in subprocess and ACP.

2. Tool and state inheritance

By default the subagent starts with the same tool list as the parent (both threads share the same initial snapshot; contextvars are thread-isolated so the parent’s tool state cannot be mutated by the subagent).

Three ways to restrict tools:

profile="explorer" (or any built-in profile) — applies a tool allowlist at spawn time. Built-in profiles: explorer (read-only), researcher, developer (full), verifier (read-only); see Agent Profiles. Note: role="verify" forces use_subprocess=True and isolated=True in addition to the verifier profile.
isolated=True — runs the subagent in a git worktree so filesystem writes don’t affect the parent repo. The worktree is auto-cleaned after completion.
redact_secrets=True (default) — scrubs common secret patterns (API keys, tokens, passwords) from workspace context messages before they reach the subagent. Thread-mode only; has no effect in subprocess or ACP modes (the child process’s own gptme.toml controls its secret handling).

Signal tools are loaded regardless of allowlist so the subagent can communicate back. Thread-mode subagents get complete, clarify, and progress. Subprocess subagents get complete and clarify; progress is not loaded because it depends on the parent’s in-process notification queue.

3. Cancellation and timeout

max_time (seconds) — a watchdog timer that marks the subagent result as "timeout" after the specified duration and delivers a timeout status notification. In subprocess mode the child process is terminated. In thread mode the background thread is not force-stopped; callers see the cached timeout result immediately while the thread continues until it finishes naturally.
timeout (default 1800 s) — subprocess monitor kills the child process after this many seconds. Only applies in subprocess mode.
The parent does not block waiting for subagents. Completion is delivered via the LOOP_CONTINUE hook, which re-enters the parent’s loop with a notification message.

4. Child transcript and result delivery

Subagents always start with a fresh conversation — they do not inherit the parent’s message history by default. The result/transcript lifecycle:

context_turns=N — the parent’s last N turns are prepended to the subagent’s conversation as context.
On completion the subagent calls the complete signal tool with a summary; this is queued back to the parent via the LOOP_CONTINUE hook.
subagent_read_log(agent_id) — retrieve the full child transcript from the parent after the subagent completes.
subagent_status(agent_id) — poll completion/error state without waiting.

Fan-out and Parallel Execution#

Two helpers make it easy to run multiple independent tasks concurrently:

subagent_parallel(tasks, ...)

Fan out N subagents in parallel and block until all complete. Returns results in the same order as the input tasks. Wall-clock time is bounded by the slowest agent, not their sum. Use this for straightforward parallel delegation where the parent needs all results before continuing:

results = subagent_parallel([
    ("researcher", "Research async Python frameworks"),
    ("coder",      "Implement a basic async HTTP client"),
    ("tester",     "Write pytest tests for an async HTTP client"),
])

Key parameters: isolated=True (each agent gets its own git worktree), output_schema (structured output — see below), model, profile, context_turns, workdir.

subagent_batch(tasks, ...)

Non-blocking variant. Launches all subagents and returns a BatchJob object immediately so the parent can continue working while agents run. Call job.wait_all() later to collect results. Useful when the parent has its own work to interleave:

job = subagent_batch([
    ("a", "..."),
    ("b", "..."),
])
# ... parent does other work ...
results = job.wait_all()

subagent_pipeline(items, *stages, ...)

Staged fan-out without a barrier between stages. Each item is processed through all stages in order, but items at different stages run concurrently — item A advances to stage 2 as soon as its stage-1 subagent completes, while item B may still be in stage 1. Wall-clock time is bounded by the slowest single-item chain, not the sum of the slowest per stage.

This is more efficient than repeated subagent_parallel() calls (which add a full synchronisation barrier between stages) when items are independent. Each stage is a callable stage(item_prompt, prev_result) -> next_prompt:

items = [("auth", "Review auth.py"), ("db", "Review db.py")]
results = subagent_pipeline(
    items,
    # Stage 0: review
    lambda item, _: f"Find bugs in this file: {item}",
    # Stage 1: verify — runs on auth while db is still in stage 0
    lambda item, prev: f"Adversarially verify these findings:\n{prev}",
)
# results[i][j] — result for item i at stage j
for (prefix, _), stage_results in zip(items, results):
    print(f"{prefix}: {stage_results[-1]['result'][:80]}")

Set isolated=True so concurrent file-editing subagents each get their own git worktree.

subagent_wait_any(agent_ids, ...)

Return the first of the given subagents to complete. Useful for speculative / hedging patterns: spawn N subagents racing on the same task and take whichever finishes first, then cancel the rest:

subagent("fast",     "Quick attempt at task X")
subagent("thorough", "Thorough attempt at task X")
first_id, result = subagent_wait_any(["fast", "thorough"], timeout=120)
print(f"{first_id} won: {result['status']}")
for aid in ("fast", "thorough"):
    if aid != first_id:
        subagent_cancel(aid)

agent_ids is the list of IDs to wait on. Raises TimeoutError if no agent completes within timeout seconds (default 300).

Structured Output (output_schema)#

Both subagent_parallel() and subagent_batch() accept an output_schema parameter (a Pydantic model class). When set, each subagent is instructed to return valid JSON matching the schema inside its complete block. Results are automatically parsed and validated — the "result" value in each result dict is the parsed/validated object rather than a raw string:

from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    score: int
    issues: list[str]

results = subagent_parallel(
    [("a1", "Analyze module A"), ("a2", "Analyze module B")],
    output_schema=AnalysisResult,
)
for r in results:
    if r["status"] == "success":
        analysis = r["result"]  # already a validated dict
        print(f"Score: {analysis['score']}")

The output_schema parameter is also available on the low-level subagent() call for single-agent structured output.

Token Budget Tracking#

subagent_wait() and BatchJob.wait_all() include token usage in their result dicts:

result = subagent_wait("my-agent")
# result["input_tokens"]  — tokens consumed by the subagent's prompts
# result["output_tokens"] — tokens generated by the subagent

This lets the parent track cumulative cost across a fleet of delegated tasks and gate further spawning when a budget limit is reached.

Subagent

Contents

Subagent#

Subagent Isolation Contract#

Fan-out and Parallel Execution#

Structured Output (output_schema)#

Token Budget Tracking#