Skip to main content

Run Agents in Parallel

client.compilations.run_instruction(...) dispatches a fresh execution and returns a new run immediately - it does not block until the run finishes. That makes fan-out simple: take one frozen compilation, start one run per row or item, and wait on each run separately.

Runs execute in the FlyMy.AI cloud, so your script only dispatches and polls.

Freeze first

Parallel batches should run from a frozen compilation, not from live agent chats. The frozen instruction replays a fixed plan, so every item goes through the same deterministic pipeline - cheaper and faster than re-planning per item. See Freeze & Re-run.

Fan out with a worker pool​

One thread per in-flight run; run_instruction_and_wait dispatches the run and polls (~2s) until it reaches a terminal status.

from concurrent.futures import ThreadPoolExecutor
from flymyai import AgentClient

client = AgentClient(api_key="fly-***")
COMPILATION_ID = 42  # from client.agents.compile_from_run(...)

rows = [
    {"item_id": "lead-001", "website_url": "https://example.com"},
    {"item_id": "lead-002", "website_url": "https://example.org"},
    # ... one dict per item, matching the agent's input_schema
]

def process(row: dict) -> dict:
    result = client.compilations.run_instruction_and_wait(
        COMPILATION_ID,
        variables=row,
        timeout=600,
    )
    return {"row": row, "status": result.status, "output": result.output,
            "error": result.error, "run_id": result.id}

with ThreadPoolExecutor(max_workers=10) as pool:
    results = list(pool.map(process, rows))

done = [r for r in results if r["status"] == "completed"]
failed = [r for r in results if r["status"] != "completed"]
print(f"{len(done)} completed, {len(failed)} failed")

Prefer asyncio? AsyncAgentClient exposes the same methods - wrap the call in a task per row and bound concurrency with asyncio.Semaphore(10).

Retry failed runs​

Each item is an independent run, so retry only the failures - re-dispatch with the same variables:

for r in failed:
    retry = client.compilations.run_instruction_and_wait(
        COMPILATION_ID, variables=r["row"], timeout=600,
    )
    if retry.status == "completed":
        done.append({"row": r["row"], "output": retry.output, "run_id": retry.id})

Check what the batch cost​

Billing is per-use - each run bills its real cost. From the FlyMy MCPs gateway, get_execution_price returns the billed cost of one run and list_execution_prices returns costs across recent runs, so you can total a batch or compute cost per item.

Practical notes​

  • Idempotent variables - include a stable identifier per item (like item_id above) in variables, and make any side-effecting step keyed on it. A retried run then updates the same record instead of creating a duplicate.
  • Modest concurrency - start with ~10 in-flight runs. Confirm outputs and cost per item look right on a small batch before raising it.
  • Retries - treat failed as retryable and inspect error on runs that fail twice; the cause is often a bad input row, not the pipeline.
  • Timeouts - run_instruction_and_wait raises TimeoutError if a run outlives timeout; catch it in process() if a single slow item should not sink the batch.

Next steps​

  • Runs - run lifecycle, polling, and freeze semantics
  • Python SDK - run_instruction, run_instruction_and_wait, and the lower-level building blocks