Orchestrating Agents

SURFACE ROUTING — read first

Fan-out has three possible engines. Which exist depends on where you are running.

Pick the engine before writing any orchestration code.

| Engine | claude.ai | Cowork | Claude Code / CCotw |

|---|:---:|:---:|:---:|

| Native subagents (Agent / Task / Workflow) | ✗ | ✓ | ✓ |

| Gemini via CF AI Gateway (invoking-gemini) | ✓ | ✓ | ✓ |

| This skill's httpx fan-out (raw Anthropic API) | ✓ | last resort | last resort |

Primary discriminator — check the tool list, not the filesystem. If an Agent,

Task, or Workflow tool is callable, native subagents exist. That single fact

decides the row. Everything below is elaboration.

If native subagents exist (Cowork, Claude Code, CCotw)

Use them. Do not hand-roll from this skill. The managed runtime gives

16-concurrent / 1000-agent ceilings, an approval gate, adversarial cross-review,

and in-session resume — all of which this skill would reimplement worse. Route

model and effort per agent-routing (calibrated on 300 measured Haiku calls);

do not re-derive that here.

Cowork adds one option Claude Code doesn't: subagents can be declared rather

than spawned ad hoc, as agents/*.md in a plugin — frontmatter name,

description, model, effort, maxTurns, tools, disallowedTools,

skills, memory, background, isolation: worktree. They appear as

plugin-name:agent-name. Note hooks, mcpServers, and permissionMode are

refused in plugin agents for security, so a declared agent inherits the session's

MCP connections and cannot bring its own.

Reach back into this skill on those surfaces only for what the runtime lacks:

inter-agent messaging (AgentPool), stall detection, or a long-lived

ConversationThread.

If native subagents do NOT exist (claude.ai chat and project sessions)

Two engines, and Gemini is the default — see subagent-delegation-protocol

in ops. Use this skill's httpx fan-out when you specifically want Claude-family

output, multi-turn threads with cached history, or inter-agent messaging.

Gemini via Cloudflare — available on every surface

Even where native subagents exist, Gemini is the right call for

mechanical-but-large work (extractions, ports, boilerplate, schema transforms)

and for a genuinely independent second opinion in a judge panel — a different

model family fails differently, which is the whole point of a panel.

Call mechanics live in invoking-gemini; do not duplicate them here. Three

things that bite:

• Pass the explicit model string gemini-3.6-flash. The flash alias still

resolves to 3.5 until that plugin's model table regenerates.

• thinking_level is a string in {minimal, low, medium, high}, default

medium. Set minimal for mechanical generation or the model silently spends

its output budget reasoning — symptom is an empty or truncated response.

• Credentials come from the CF AI Gateway config, BYOK. Requests route through

the gateway rather than Google directly.

Non-negotiable on all surfaces

Review is not delegable. Diff security- and protocol-critical paths

line-by-line against source, run syntax/lint checks, live-test whatever is

network-testable. Delegated output ships only after your own review, regardless

of which model produced it or which engine ran it.

Cross-model review tools (challenge, verify_patch) keep their own model

config, often deliberately a Claude. This routing does not silently repoint them.

Orchestrating Agents

This skill enables programmatic API invocations for advanced workflows including parallel processing, task delegation, and multi-agent analysis using the Anthropic API.

When to Use This Skill

Primary use cases:

• Parallel sub-tasks: Break complex analysis into simultaneous independent streams
• Multi-perspective analysis: Get 3-5 different expert viewpoints concurrently
• Delegation: Offload specific subtasks to specialized API instances
• Recursive workflows: Orchestrator coordinating multiple API instances
• High-volume processing: Batch process multiple items concurrently

Trigger patterns:

• "Parallel analysis", "multi-perspective review", "concurrent processing"
• "Delegate subtasks", "coordinate multiple agents"
• "Run analyses from different perspectives"
• "Get expert opinions from multiple angles"

Quick Start

Single Invocation

import sys
sys.path.append('/home/user/claude-skills/orchestrating-agents/scripts')
from claude_client import invoke_claude

response = invoke_claude(
    prompt="Analyze this code for security vulnerabilities: ...",
    model="claude-sonnet-4-6"
)
print(response)

Parallel Multi-Perspective Analysis

from claude_client import invoke_parallel

prompts = [
    {
        "prompt": "Analyze from security perspective: ...",
        "system": "You are a security expert"
    },
    {
        "prompt": "Analyze from performance perspective: ...",
        "system": "You are a performance optimization expert"
    },
    {
        "prompt": "Analyze from maintainability perspective: ...",
        "system": "You are a software architecture expert"
    }
]

results = invoke_parallel(prompts, model="claude-sonnet-4-6")

for i, result in enumerate(results):
    print(f"\n=== Perspective {i+1} ===")
    print(result)

Parallel with Shared Cached Context (Recommended)

For parallel operations with shared base context, use caching to reduce costs by up to 90%:

from claude_client import invoke_parallel

# Large context shared across all sub-agents (e.g., codebase, documentation)
base_context = """
<codebase>
...large codebase or documentation (1000+ tokens)...
</codebase>
"""

prompts = [
    {"prompt": "Find security vulnerabilities in the authentication module"},
    {"prompt": "Identify performance bottlenecks in the API layer"},
    {"prompt": "Suggest refactoring opportunities in the database layer"}
]

# First sub-agent creates cache, subsequent ones reuse it
results = invoke_parallel(
    prompts,
    shared_system=base_context,
    cache_shared_system=True  # 90% cost reduction for cached content
)

Multi-Turn Conversation with Auto-Caching

For sub-agents that need multiple rounds of conversation:

from claude_client import ConversationThread

# Create a conversation thread (auto-caches history)
agent = ConversationThread(
    system="You are a code refactoring expert with access to the codebase",
    cache_system=True
)

# Turn 1: Initial analysis
response1 = agent.send("Analyze the UserAuth class for issues")
print(response1)

# Turn 2: Follow-up (reuses cached system + turn 1)
response2 = agent.send("How would you refactor the login method?")
print(response2)

# Turn 3: Implementation (reuses all previous context)
response3 = agent.send("Show me the refactored code")
print(response3)

Streaming Responses

For real-time feedback from sub-agents:

from claude_client import invoke_claude_streaming

def show_progress(chunk):
    print(chunk, end='', flush=True)

response = invoke_claude_streaming(
    "Write a comprehensive security analysis...",
    callback=show_progress
)

Parallel Streaming

Monitor multiple sub-agents simultaneously:

from claude_client import invoke_parallel_streaming

def agent1_callback(chunk):
    print(f"[Security] {chunk}", end='', flush=True)

def agent2_callback(chunk):
    print(f"[Performance] {chunk}", end='', flush=True)

results = invoke_parallel_streaming(
    [
        {"prompt": "Security review: ..."},
        {"prompt": "Performance review: ..."}
    ],
    callbacks=[agent1_callback, agent2_callback]
)

Interruptible Operations

Cancel long-running parallel operations:

from claude_client import invoke_parallel_interruptible, InterruptToken
import threading
import time

token = InterruptToken()

# Run in background
def run_analysis():
    results = invoke_parallel_interruptible(
        prompts=[...],
        interrupt_token=token
    )
    return results

thread = threading.Thread(target=run_analysis)
thread.start()

# Interrupt after 5 seconds
time.sleep(5)
token.interrupt()

Core Functions

| Function | Module | Purpose |

|---|---|---|

| invoke_claude() | core | Single synchronous invocation, full parameter control |

| invoke_parallel() | core | Concurrent invocations, results in input order |

| invoke_claude_streaming() | core | Single invocation, token-by-token callback |

| invoke_parallel_streaming() | core | Concurrent invocations with per-agent stream callbacks |

| invoke_parallel_interruptible() | core | Concurrent invocations cancellable mid-flight |

| ConversationThread | core | Stateful multi-turn thread with cached history |

| StallDetector | core | Flags agents idle beyond a timeout |

| TaskTracker | task_state | Tracks task status across an orchestration run |

| invoke_with_retry() | orchestration | Single invocation with backoff on transient errors |

| invoke_parallel_managed() | orchestration | Concurrency-limited parallel run with retry, stall hooks, reconciliation |

Full signatures, parameters, and worked examples for each:

references/function-reference.md.

Example Workflows

See references/workflows.md for detailed examples including:

• Multi-expert code review
• Parallel document analysis
• Recursive task delegation
• Advanced Agent SDK delegation patterns
• Prompt caching workflows

Execute Mode (Default Sub-Agent Prompt)

For autonomous sub-agents that should execute without asking questions:

from claude_client import invoke_claude, EXECUTE_MODE

response = invoke_claude(
    prompt="Review auth.py for SQL injection vulnerabilities",
    system=f"You are a security expert.\n\n{EXECUTE_MODE}"
)

EXECUTE_MODE encodes these principles (adapted from OpenAI Codex):

• Make assumptions instead of asking questions; state them briefly
• Think ahead: what else might be needed?
• Report failures with what you tried and what you'll do next
• Summarize deliverables and how to validate them

Agent Pool (Named Agents with Messaging)

For workflows where multiple agents need to communicate:

from agent_pool import AgentPool

pool = AgentPool(
    shared_system="You are reviewing the auth module of a web app.",
    max_depth=3,    # prevent recursive spawn explosion
    max_agents=10,
)

# Spawn named agents with roles
pool.spawn("security", system=f"Focus on vulnerabilities.\n\n{pool.EXECUTE_MODE}")
pool.spawn("perf", system=f"Focus on performance.\n\n{pool.EXECUTE_MODE}")

# Run turns (pending inter-agent messages auto-injected)
sec_result = pool.run("security", "Review the login flow")

# Agent-to-agent messaging
pool.send("security", to="perf",
          content="Auth does N+1 queries in the session check loop",
          trigger_turn=True)  # auto-runs perf with this context

# Broadcast to all agents
pool.broadcast("security", "Auth uses bcrypt cost=12, 200ms per hash")

# Query pool state
pool.agents()           # ["security", "perf"]
pool.agent_info("perf") # {name, depth, children, pending_messages, turns}

Spawn Reservation (Atomic Agent Creation)

For complex workflows where agent creation might fail:

from agent_pool import AgentPool

pool = AgentPool(shared_system="Code review team")

# Reservation pattern: name is reserved, rolled back on exception
with pool.reserve("analyst", parent="lead") as res:
    res.configure(system="You analyze code complexity.", model="claude-opus-4-6")
    # If configure or any other work raises, the name is released
# Agent "analyst" is now live

# Depth limits prevent unbounded recursion
pool.spawn("sub-analyst", parent="analyst")  # depth=2, OK
pool.spawn("sub-sub", parent="sub-analyst")  # depth=3, raises ValueError

When to Use AgentPool vs invoke_parallel

| Pattern | Use When |

|---------|----------|

| invoke_parallel() | Independent tasks, no inter-agent communication needed |

| AgentPool | Agents need to share findings, build on each other's work, or have parent/child relationships |

| invoke_parallel_managed() | Independent tasks with retry, stall detection, concurrency limits |

Setup

Prerequisites:

• Install anthropic library:

   uv pip install anthropic

• Configure the API key as a file the shell reads directly — never as

something a tool call returns.

On claude.ai the project's files are mounted at /mnt/project, so the key can

be sourced without ever entering context:

   set -a; . /mnt/project/ANTHROPIC.env 2>/dev/null; set +a

⚠️ Do not use project_read to fetch a credential, on any surface. Small

docs are returned *inline*, so the key lands in the transcript — verified

2026-07-30: the documented "large text is written to a local file" branch does

not fire even at 64 KB. In Cowork there is no /mnt/project mount at all and

no safe read path, so the key must arrive by a route the shell can read

(synced skill directory, or fetched by a script from the CF config store).

Writing is safe in both directions — project_write with local_path keeps

contents out of context — but reading is not.

Get your API key: https://console.anthropic.com/settings/keys

Installation check:

python3 -c "import anthropic; print(f'✓ anthropic {anthropic.__version__}')"

Error Handling

The module provides comprehensive error handling:

from claude_client import invoke_claude, ClaudeInvocationError

try:
    response = invoke_claude("Your prompt here")
except ClaudeInvocationError as e:
    print(f"API Error: {e}")
    print(f"Status: {e.status_code}")
    print(f"Details: {e.details}")
except ValueError as e:
    print(f"Configuration Error: {e}")

Common errors:

• API key missing: Add ANTHROPIC_API_KEY.txt to project knowledge (see Setup above)
• Rate limits: Reduce max_workers or add delays
• Token limits: Reduce prompt size or max_tokens
• Network errors: Automatic retry with exponential backoff

Prompt Caching

For detailed caching workflows and best practices, see references/workflows.md.

Performance Considerations

Token efficiency:

• Parallel calls use more tokens but save wall-clock time
• Use prompt caching for shared context (90% cost reduction)
• Use concise system prompts to reduce overhead
• Consider token budgets when setting max_tokens

Rate limits:

• Anthropic API has per-minute rate limits
• Default max_workers=5 is safe for most tiers
• Adjust based on your API tier and rate limits

Cost management:

• Each invocation consumes API credits
• Monitor usage in Anthropic Console
• Use smaller models (haiku) for simple tasks
• Use prompt caching for repeated context (90% savings)
• Cache lifetime: 5 minutes, refreshed on each use

Best Practices

• Use parallel invocations for independent tasks only
• Don't parallelize sequential dependencies
• Each parallel task should be self-contained

• Set appropriate system prompts
• Define clear roles/expertise for each instance
• Keeps responses focused and relevant

• Handle errors gracefully
• Always wrap invocations in try-except
• Provide fallback behavior for failures

• Test with small batches first
• Verify prompts work before scaling
• Check token usage and costs

• Consider alternatives
• Not all tasks benefit from multiple instances
• Sometimes sequential with context is better

Token Efficiency

Loading this skill costs roughly 2k tokens. On surfaces with native subagents the

routing table at the top is usually all you need — read it, spawn natively, and

skip the rest of the file.

See Also

Routing companions — read these before choosing an engine:

• agent-routing skill — model + effort selection for native subagents

(Haiku/Sonnet/Opus, cascades, verifier gates). Calibrated on measured data.

Applies to Cowork and Claude Code; explicitly not to claude.ai.

• invoking-gemini skill — call mechanics for the CF AI Gateway path, model

table, and thinking_level semantics.

• subagent-delegation-protocol (ops config) — why Gemini is the claude.ai

default, the Sonnet fallback config, and the non-delegable-review rule.

This skill's own internals:

• references/function-reference.md - Full signatures for every function this skill exposes
• references/api-reference.md - Anthropic API details: models, rate limits, caching
• references/workflows.md - Worked orchestration examples
• Anthropic API Docs - Official documentation