Quickstart with Python

pip install httpx

import json
import os
import httpx

ENGINE_URL = os.environ.get("ENGINE_URL", "http://localhost:8000")
ENGINE_KEY = os.environ["ENGINE_KEY"]

def execute(message: str, task_id: str) -> None:
    """Stream a response from the Engine and print text deltas."""
    headers = {
        "X-Engine-Key": ENGINE_KEY,
        "Content-Type": "application/json",
    }
    payload = {"message": message, "task_id": task_id}

    with httpx.Client(timeout=None) as client:
        with client.stream("POST", f"{ENGINE_URL}/execute",
                           json=payload, headers=headers) as resp:
            resp.raise_for_status()
            for line in resp.iter_lines():
                if not line or not line.startswith("data:"):
                    continue
                data = json.loads(line[len("data:"):].strip())
                if "text" in data:
                    print(data["text"], end="", flush=True)
            print()  # final newline

if __name__ == "__main__":
    execute("Say hello in one short sentence.", task_id="demo-001")

export ENGINE_URL=http://localhost:8000
export ENGINE_KEY=dev-engine-key

python engine_quickstart.py

import asyncio
import json
import os
import httpx

ENGINE_URL = os.environ.get("ENGINE_URL", "http://localhost:8000")
ENGINE_KEY = os.environ["ENGINE_KEY"]

async def execute(message: str, task_id: str) -> None:
    headers = {
        "X-Engine-Key": ENGINE_KEY,
        "Content-Type": "application/json",
    }
    payload = {"message": message, "task_id": task_id}

    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("POST", f"{ENGINE_URL}/execute",
                                 json=payload, headers=headers) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if not line or not line.startswith("data:"):
                    continue
                data = json.loads(line[len("data:"):].strip())
                if "text" in data:
                    print(data["text"], end="", flush=True)
            print()

if __name__ == "__main__":
    asyncio.run(execute("Say hello in one short sentence.", task_id="demo-001"))

def handle_event(event_type: str, data: dict) -> None:
    if event_type == "text_delta":
        print(data.get("text", ""), end="", flush=True)
    elif event_type == "tool_call":
        print(f"\n[tool call: {data['tool']}]", flush=True)
    elif event_type == "tool_result":
        print(f"\n[tool result: {data.get('output', '')[:60]}...]", flush=True)
    elif event_type == "hitl_request":
        print(f"\n[engine asks: {data['question']}]", flush=True)
    elif event_type == "thread_lifecycle":
        print(f"\n[{data['phase']}]", flush=True)

event_type = None
for line in resp.iter_lines():
    if line.startswith("event:"):
        event_type = line[len("event:"):].strip()
    elif line.startswith("data:"):
        data = json.loads(line[len("data:"):].strip())
        handle_event(event_type, data)
        event_type = None

execute("Repeat what you just said in French.", task_id="demo-001")