Quickstart

Create your first Flow AI runtime in five to ten minutes.

This quickstart builds a minimal coordinator and specialist, runs the native runtime with the deterministic testing interpreter, and prints the event stream. It does not require provider credentials.

Before you begin

• Install Python 3.11 or newer.
• Use an environment with access to the private preview package.
• No Anthropic, OpenAI, or other provider API key is required for this example.

Install

pip install flowai-harness

What you build

You will build one runtime with:

• A tenant identity for runtime-owned state.
• A coordinator that receives the user request.
• A specialist that can be routed to by the coordinator.
• A deterministic no-network mock response for local testing.

Create hello_flowai.py

Save the following as hello_flowai.py. It is one complete, runnable script:

import asyncio

from flowai_harness import (
    TestingConfig,
    create_runtime,
    define_coordinator,
    define_runtime,
    define_specialist,
    define_tenant,
)

async def main() -> None:
    tenant = define_tenant("acme", "v1")

    specialist = define_specialist(
        name="greeter",
        model="claude-haiku-4-5",
        prompt="You greet the user politely.",
    )
    coordinator = define_coordinator(
        name="hello_coordinator",
        model="claude-sonnet-4-6",
        routes=["greeter"],
        prompt="Route greeting requests to the greeter specialist.",
    )

    runtime_spec = define_runtime(
        tenant=tenant,
        agents=[coordinator, specialist],
        providers={"anthropic": {"apiKey": "unused"}},
    )

    runtime = create_runtime(
        runtime_spec,
        testing=TestingConfig(mock_response="hello from the Rust runtime"),
    )

    async for event in runtime.query("Say hello", thread_id="thread-1"):
        print(event)

asyncio.run(main())

Run it

python hello_flowai.py

Expected output

The runtime prints a short stream of event dictionaries. Identifiers such as toolInvocationId differ on every run, but the shape looks like this:

{'agentName': 'hello_coordinator', 'state': 'call', 'toolInvocationId': 'inv-1955407c-815d-4f47-a49c-99f719900160', 'type': 'tool-agent'}
{'type': 'step-start'}
{'text': 'Received: Say hello\n\n', 'type': 'text'}
{'text': 'hello from the Rust runtime', 'type': 'text'}
{'data': {'hadTimeout': False, 'phases': {'llmCalls': 1, 'llmTimeMs': 0, 'subAgentTimeMs': 0, 'toolTimeMs': 0}, 'retryCount': 0, 'toolTimings': [], 'totalDurationMs': 0}, 'type': 'data-latency-summary'}
{'finishReason': 'stop', 'type': 'finish', 'usage': {'cacheCreationInputTokens': 0, 'cacheReadInputTokens': 0, 'completionTokens': 25, 'promptTokens': 50, 'totalTokens': 75}}
{'agentName': 'hello_coordinator', 'state': 'result', 'toolInvocationId': 'inv-1955407c-815d-4f47-a49c-99f719900160', 'type': 'tool-agent'}

What happened

• define_tenant("acme", "v1") created the tenant identity that keys all runtime-owned state.
• define_specialist and define_coordinator built two validated Pydantic agent specs; the coordinator routes greeting requests to the specialist. Coordinators with routes=[...] receive the built-in call_agent tool by default; you do not need to list toolkits=["agents"].
• define_runtime assembled the specs into a RuntimeSpec, including the provider declaration that every agent model resolves against.
• create_runtime(..., testing=TestingConfig(...)) selected the deterministic testing interpreter, so no API key or network access was needed.
• runtime.query(...) streamed events from the embedded Rust runtime, ending with the mock response text.

The testing interpreter returns a fixed response, so routing and specialist logic are stubbed: the events show hello_coordinator, not greeter. You are verifying wiring here, not behavior — swap in a live interpreter to see routing and specialists actually fire.

Common errors

Next steps

• Build the full Acme scenario-planning tutorial.
• Read the Concepts section for the mental model.
• Browse the Guides for task-focused how-tos.