SURFACE ROUTING — read first

This skill hand-rolls subagent orchestration via raw Anthropic API calls. A managed runtime now does the same job. Which one to use depends on your surface:

• In Claude Code (incl. CCotw): use the native runtime, NOT this skill. If you can invoke /deep-research, trigger a run with the workflow keyword, set /effort ultracode, or spawn Task subagents — do that instead. The runtime gives 16-concurrent / 1000-agent ceilings, an approval gate, adversarial cross-review, and in-session resume that this skill would otherwise reimplement badly. Dynamic workflows shipped in research preview (Claude Code v2.1.154+, 2026).
• In claude.ai chat or the bare API (no workflow runtime): use this skill. Parallel API instances over httpx is the only fan-out path here. Proceed below.

Discriminator: do you have a native subagent/Task tool or a workflow command? Yes → native. No → this skill. Never reimplement the runtime where it already exists.

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

invoke_claude()

Single synchronous invocation with full control:

invoke_claude(
    prompt: str | list[dict],
    model: str = "claude-sonnet-4-6",
    system: str | list[dict] | None = None,
    max_tokens: int = 4096,
    temperature: float = 1.0,
    streaming: bool = False,
    cache_system: bool = False,
    cache_prompt: bool = False,
    messages: list[dict] | None = None,
    **kwargs
) -> str

Parameters:

• prompt: The user message (string or list of content blocks)
• model: Claude model to use (default: claude-sonnet-4-6)
• system: Optional system prompt (string or list of content blocks)
• max_tokens: Maximum tokens in response (default: 4096)
• temperature: Randomness 0-1 (default: 1.0)
• streaming: Enable streaming response (default: False)
• cache_system: Add cache_control to system prompt (requires 1024+ tokens, default: False)
• cache_prompt: Add cache_control to user prompt (requires 1024+ tokens, default: False)
• messages: Pre-built messages list for multi-turn (overrides prompt)
• **kwargs: Additional API parameters (top_p, top_k, etc.)

Returns: Response text as string

Note: Caching requires minimum 1,024 tokens per cache breakpoint. Cache lifetime is 5 minutes (refreshed on use).

invoke_parallel()

Concurrent invocations using lightweight workflow pattern:

invoke_parallel(
    prompts: list[dict],
    model: str = "claude-sonnet-4-6",
    max_tokens: int = 4096,
    max_workers: int = 5,
    shared_system: str | list[dict] | None = None,
    cache_shared_system: bool = False
) -> list[str]

Parameters:

• prompts: List of dicts with 'prompt' (required) and optional 'system', 'temperature', 'cache_system', 'cache_prompt', etc.
• model: Claude model for all invocations
• max_tokens: Max tokens per response
• max_workers: Max concurrent API calls (default: 5, max: 10)
• shared_system: System context shared across ALL invocations (for cache efficiency)
• cache_shared_system: Add cache_control to shared_system (default: False)

Returns: List of response strings in same order as prompts

Note: For optimal cost savings, put large common context (1024+ tokens) in shared_system with cache_shared_system=True. Firs