Build a parallel sub-agent fan-out

This tutorial builds a parallel sub-agent composition: one parent agent calls one composable lambda, and that lambda submits three calls to one delegate configuration before collecting any result. Each delegate call runs in its own ephemeral session. For specialist-agent variants and failure handling, see Call other tools from lambda tools.

What you will build

The script below:

1. Creates a tool-less delegate agent.
2. Creates a lambda with a private sub_agent tool configuration.
3. Creates a parent agent that exposes the lambda with a 300-second execution budget.
4. Creates a parent session and sends one request.
5. Prints the visible parent-session tool calls and the lambda result.
6. Deletes the parent agent, delegate agent, and lambda tool.

The parent session exposes only the lambda call. The lambda-owned sub-agent calls produce no parent-session tool events.

Prerequisites

• A Vectara account.
• An API key that can create agents and tools and configure the parent agent to invoke the delegate agent.
• Access to the gpt-4o model.
• Python 3.10 or later with the requests package.

Install the client dependency and export the API key:

pip install requests
export VECTARA_API_KEY="YOUR_API_KEY"

Run the complete example

Save the following as parallel_subagent_fanout.py, then run python parallel_subagent_fanout.py.

import json
import os
import time

import requests

BASE_URL = "https://api.vectara.io"
API_KEY = os.environ["VECTARA_API_KEY"]
HEADERS = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json",
}


def api_request(method: str, path: str, payload: dict | None = None) -> dict:
    response = requests.request(
        method,
        f"{BASE_URL}{path}",
        headers=HEADERS,
        json=payload,
        timeout=360,
    )
    if not response.ok:
        raise RuntimeError(
            f"{method} {path} failed with {response.status_code}: {response.text}"
        )
    return response.json() if response.content else {}


suffix = str(int(time.time()))
delegate_key = f"parallel_delegate_{suffix}"
parent_key = f"parallel_parent_{suffix}"
tool_name = f"parallel_fanout_{suffix}"
tool_id = None
delegate_created = False
parent_created = False

fanout_code = '''def process(question: str, fan_out: int) -> dict:
    """Sends the same question to this tool's own sub_agent configuration several times in parallel and collects every answer.

    Each invocation is submitted with `tool.delegate.submit()` before any result is awaited, so the delegate sub-agents run concurrently in their own
    ephemeral sessions. A failed invocation is counted instead of voiding its siblings' answers.

    Args:
        question: The question to pass verbatim to each delegate sub-agent.
        fan_out: How many parallel delegate invocations to run.

    Returns:
        A dict with every answer in submission order, the number of distinct sub-agent sessions that produced them, and the count of failed invocations.
    """
    handles = [tool.delegate.submit(message=question) for _ in range(int(fan_out))]
    answers = []
    session_keys = set()
    failures = 0
    for handle in handles:
        try:
            result = handle.result()
            answers.append(result["sub_agent_response"])
            session_keys.add(result["session_key"])
        except tool.ToolError:
            failures += 1
    return {"answers": answers, "distinct_sessions": len(session_keys), "failures": failures}
'''

try:
    delegate = api_request(
        "POST",
        "/v2/agents",
        {
            "key": delegate_key,
            "name": f"Parallel delegate {suffix}",
            "description": "Tool-less delegate used by the parallel fan-out tutorial.",
            "model": {"name": "gpt-4o"},
            "first_step_name": "main",
            "steps": {
                "main": {
                    "instructions": [
                        {
                            "type": "inline",
                            "name": "main-instruction",
                            "template": "You are a test agent. Follow the user's instructions exactly and answer as tersely as possible.",
                        }
                    ],
                    "output_parser": {"type": "default"},
                }
            },
            "tool_configurations": {},
            "enabled": True,
        },
    )
    delegate_created = True
    print(f"Created delegate agent: {delegate['key']}")

    lambda_tool = api_request(
        "POST",
        "/v2/tools",
        {
            "type": "lambda",
            "language": "python",
            "name": tool_name,
            "title": "Parallel Delegate Composer",
            "description": "Fans one question out to parallel sub-agent delegations and returns every answer.",
            "code": fanout_code,
            "tool_configurations": {
                "delegate": {
                    "type": "sub_agent",
                    "sub_agent_configuration": {
                        "agent_key": delegate_key,
                        "session_mode": "ephemeral",
                    },
                }
            },
        },
    )
    tool_id = lambda_tool["id"]
    print(f"Created lambda tool: {tool_id}")

    parent = api_request(
        "POST",
        "/v2/agents",
        {
            "key": parent_key,
            "name": f"Parallel fan-out parent {suffix}",
            "description": "Calls one lambda that fans out sub-agent delegations.",
            "model": {"name": "gpt-4o"},
            "first_step_name": "main",
            "steps": {
                "main": {
                    "instructions": [
                        {
                            "type": "inline",
                            "name": "main-instruction",
                            "template": "When the user asks you something, call the ask_fan_out tool once, with question set to the user's message passed through verbatim and fan_out set to 3. The tool returns a JSON object with an answers array. Reply with exactly the first element of answers and nothing else.",
                        }
                    ],
                    "output_parser": {"type": "default"},
                }
            },
            "tool_configurations": {
                "ask_fan_out": {
                    "type": "lambda",
                    "tool_id": tool_id,
                    "max_execution_time_seconds": 300,
                }
            },
            "enabled": True,
        },
    )
    parent_created = True
    print(f"Created parent agent: {parent['key']}")

    session = api_request(
        "POST",
        f"/v2/agents/{parent_key}/sessions",
        {
            "name": "Parallel fan-out tutorial",
            "description": "Runs one three-way sub-agent fan-out.",
        },
    )
    session_key = session["key"]
    print(f"Created parent session: {session_key}")

    run = api_request(
        "POST",
        f"/v2/agents/{parent_key}/sessions/{session_key}/events",
        {
            "type": "input_message",
            "messages": [
                {
                    "type": "text",
                    "content": "Ask the delegates to reply with exactly: OMEGA",
                }
            ],
            "stream_response": False,
        },
    )

    visible_tool_calls = [
        event["tool_configuration_name"]
        for event in run["events"]
        if event["type"] == "tool_input"
    ]
    lambda_outputs = [
        event["tool_output"]
        for event in run["events"]
        if event["type"] == "tool_output"
        and event["tool_configuration_name"] == "ask_fan_out"
    ]

    print("Visible parent-session tool calls:")
    print(json.dumps(visible_tool_calls, indent=2))
    print("Fan-out lambda output:")
    print(json.dumps(lambda_outputs, indent=2))
finally:
    if parent_created:
        response = requests.delete(
            f"{BASE_URL}/v2/agents/{parent_key}", headers=HEADERS, timeout=60
        )
        print(f"Delete parent agent: {response.status_code}")
    if delegate_created:
        response = requests.delete(
            f"{BASE_URL}/v2/agents/{delegate_key}", headers=HEADERS, timeout=60
        )
        print(f"Delete delegate agent: {response.status_code}")
    if tool_id is not None:
        response = requests.delete(
            f"{BASE_URL}/v2/tools/{tool_id}", headers=HEADERS, timeout=60
        )
        print(f"Delete lambda tool: {response.status_code}")

Verify the orchestration

The visible_tool_calls list should contain only ask_fan_out. The lambda's private delegate calls do not produce events in the parent session.

For the request above, when all three delegate calls succeed, the lambda output contains three OMEGA answers, distinct_sessions is 3, and failures is 0. Because the sub-agent configuration uses session_mode: ephemeral, each submitted call creates a separate sub-agent session.

The script submits every handle before its result loop. Moving handle.result() into the submission loop would serialize the calls.

Understand the limits

One lambda execution currently runs at most eight submitted calls concurrently; additional submits queue. All in-code invocations in the hosting run also share the run-level concurrency and total-call budgets. For the authoritative limits, timeout behavior, failure handling, and specialist-agent variant, see Run composed calls in parallel.

For all sub-agent session modes and artifact sharing, see Sub-agents.