Python from scratch

uv run python -c 'import json; from app import TOOLS; print(json.dumps(TOOLS, indent=2))'

fastapi>=0.115
uvicorn[standard]>=0.32
httpx>=0.27

import asyncio
import json
import os
from functools import lru_cache
from typing import Any

import httpx
from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()


@lru_cache(maxsize=1)
def _api_key() -> str:
    # Lazy so the server still boots (and answers the ping probe) when
    # OPENAI_API_KEY isn't set yet. Reading it at import would crash
    # uvicorn before Test connection can ever succeed.
    return os.environ["OPENAI_API_KEY"]


MODEL = "gpt-5"
MAX_TURNS = 10

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order",
            "description": "Look up an order by id.",
            "parameters": {
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "refund_order",
            "description": "Refund an order.",
            "parameters": {
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
        },
    },
]


def _proxy_call(proxy_url: str, run_token: str, tool_name: str, args: dict) -> Any:
    url = f"{proxy_url.rstrip('/')}/tools/{tool_name}"
    with httpx.Client(timeout=120.0) as c:
        r = c.post(url, json=args, headers={"Authorization": f"Bearer {run_token}"})
    r.raise_for_status()
    return r.json()["response"]


def _llm(messages: list[dict]) -> dict:
    with httpx.Client(timeout=60.0) as c:
        r = c.post(
            "https://api.openai.com/v1/chat/completions",
            json={"model": MODEL, "messages": messages, "tools": TOOLS},
            headers={"Authorization": f"Bearer {_api_key()}"},
        )
    r.raise_for_status()
    return r.json()


def _run_loop(proxy_url: str, run_token: str, user_instruction: str) -> dict:
    """Synchronous multi-turn loop. Runs on a worker thread (see dispatch)."""
    messages: list[dict] = [
        {"role": "system", "content": "Triage refund requests; look up orders before deciding."},
        {"role": "user", "content": user_instruction},
    ]
    trace: list[dict] = list(messages)

    final_response = ""
    in_tokens = 0
    out_tokens = 0

    for _ in range(MAX_TURNS):
        completion = _llm(messages)
        usage = completion.get("usage") or {}
        in_tokens += int(usage.get("prompt_tokens", 0))
        out_tokens += int(usage.get("completion_tokens", 0))

        choice = completion["choices"][0]["message"]
        messages.append(choice)
        trace.append({
            "role": "assistant",
            "content": choice.get("content"),
            "tool_calls": [
                {"id": tc["id"], "name": tc["function"]["name"], "arguments": tc["function"]["arguments"]}
                for tc in choice.get("tool_calls") or []
            ] or None,
        })

        tool_calls = choice.get("tool_calls") or []
        if not tool_calls:
            final_response = choice.get("content") or ""
            break

        for tc in tool_calls:
            args = json.loads(tc["function"]["arguments"] or "{}")
            try:
                result = _proxy_call(proxy_url, run_token, tc["function"]["name"], args)
                content = json.dumps(result)
            except httpx.HTTPError as exc:
                content = json.dumps({"error": str(exc)})

            tool_message = {"role": "tool", "tool_call_id": tc["id"], "content": content}
            messages.append(tool_message)
            trace.append(tool_message)

    return {
        "final_response": final_response or "(no final response)",
        "messages": trace,
        "metadata": {
            "model": MODEL,
            "total_input_tokens": in_tokens,
            "total_output_tokens": out_tokens,
        },
    }


@app.post("/dispatch")
async def dispatch(
    request: Request,
    x_pipelines_run_token: str | None = Header(default=None),
    x_pipelines_odyssey_proxy_url: str | None = Header(default=None),
):
    payload = await request.json()
    proxy_url = payload.get("odyssey_proxy_url") or x_pipelines_odyssey_proxy_url
    run_token = x_pipelines_run_token
    if not proxy_url or not run_token:
        raise HTTPException(400, "missing proxy URL or run token")

    user_instruction = (payload.get("input") or {}).get("user_instruction") or ""

    # Offload the blocking httpx loop to a worker thread so a single
    # multi-turn run doesn't freeze the event loop and serialize every
    # other concurrent dispatch (concurrency_cap is load-bearing).
    return await asyncio.to_thread(_run_loop, proxy_url, run_token, user_instruction)

return {"final_response": final_response or "(no final response)"}