OpenAI Agents SDK

uv run python <<'EOF'
import json
from agent import build_agent

print(json.dumps(
    [
        {"name": t.name, "description": t.description, "parameters": t.params_json_schema}
        for t in build_agent().tools
    ],
    indent=2,
))
EOF

fastapi>=0.115
uvicorn[standard]>=0.32
httpx>=0.27
openai-agents>=0.0.7

import json
from typing import Any

import httpx
from agents import Agent, Runner, function_tool
from agents.items import MessageOutputItem, ToolCallItem, ToolCallOutputItem
from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()


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 _build_agent(proxy_url: str, run_token: str) -> Agent:
    @function_tool
    def get_order(order_id: str) -> dict:
        return _proxy_call(proxy_url, run_token, "get_order", {"order_id": order_id})

    @function_tool
    def refund_order(order_id: str) -> dict:
        return _proxy_call(proxy_url, run_token, "refund_order", {"order_id": order_id})

    return Agent(
        name="orders-triage",
        instructions=(
            "You triage refund requests. Look up orders before deciding. "
            "Return the final outcome as a single sentence."
        ),
        tools=[get_order, refund_order],
        model="gpt-5",
    )


def _serialize_messages(run_items) -> list[dict]:
    out: list[dict] = []
    for item in run_items:
        if isinstance(item, MessageOutputItem):
            raw = item.raw_item
            content_parts = getattr(raw, "content", None) or []
            text = "".join(
                getattr(p, "text", "") for p in content_parts if getattr(p, "type", "") == "output_text"
            )
            out.append({"role": "assistant", "content": text or None})
        elif isinstance(item, ToolCallItem):
            raw = item.raw_item
            args = getattr(raw, "arguments", None)
            try:
                parsed_args = json.loads(args) if isinstance(args, str) else args
            except (TypeError, ValueError):
                parsed_args = args
            out.append({
                "role": "assistant",
                "content": None,
                "tool_calls": [{
                    "id": getattr(raw, "call_id", None) or getattr(raw, "id", ""),
                    "name": getattr(raw, "name", ""),
                    "arguments": parsed_args,
                }],
            })
        elif isinstance(item, ToolCallOutputItem):
            raw = item.raw_item
            out.append({
                "role": "tool",
                "tool_call_id": (
                    raw.get("call_id") if isinstance(raw, dict) else getattr(raw, "call_id", "")
                ),
                "content": str(item.output) if item.output is not None else None,
            })
    return out


@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 ""
    agent = _build_agent(proxy_url, run_token)
    result = await Runner.run(agent, user_instruction)

    usage = getattr(result, "context_wrapper", None)
    return {
        "final_response": str(result.final_output),
        "messages": [
            {"role": "user", "content": user_instruction},
            *_serialize_messages(result.new_items),
        ],
        "metadata": {
            "model": "gpt-5",
            "total_input_tokens": getattr(usage, "input_tokens", None) if usage else None,
            "total_output_tokens": getattr(usage, "output_tokens", None) if usage else None,
        },
    }

from pipelines.odyssey.adapters.openai_agents import forward_run_result_events

forward_run_result_events(result)  # best-effort