> ## Documentation Index
> Fetch the complete documentation index at: https://dev.haico.gr/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Recipes

> A runnable two-turn co-construction loop in Python, and a token check for CI.

## A full loop in Python

Runs two turns on one conversation and reads back what the agent built. Requires only `httpx`.

```python theme={null}
import json
import os

import httpx

BASE = "https://haico.gr/api"
TOKEN = os.environ["HAICO_TOKEN"]
HEADERS = {"Authorization": f"Bearer {TOKEN}"}


def run_turn(client: httpx.Client, query: str, thread_id: str | None = None):
    """Send one message to the agent and consume the SSE stream.

    Returns the thread id, the reasoning steps, and any artifacts.
    """
    steps, artifacts = [], []

    with client.stream(
        "POST",
        f"{BASE}/query/stream_steps/sse",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"query": query, "thread_id": thread_id},
        timeout=None,          # the agent may think for a while
    ) as response:
        response.raise_for_status()

        for line in response.iter_lines():
            # Only "data:" lines carry events. Lines starting with ":" are
            # keep-alive comments. The event kind is the "type" field inside
            # the JSON, not an SSE "event:" line.
            if not line.startswith("data:"):
                continue

            data = json.loads(line[len("data:"):].strip())
            kind = data.get("type")

            if kind == "conversation_info":
                thread_id = data["thread_id"]
            elif kind == "step":
                steps.append(data["step"])
            elif kind == "artifact":
                artifacts.append(data["artifact"])
            elif kind == "error":
                raise RuntimeError(f"agent turn failed: {data['error']}")
            elif kind == "complete":
                break

    return thread_id, steps, artifacts


with httpx.Client() as client:
    # Confirm the token works before spending anything.
    me = client.get(f"{BASE}/auth/me", headers=HEADERS)
    me.raise_for_status()
    print("authenticated as", me.json()["username"])

    thread_id, steps, artifacts = run_turn(
        client, "Draft a one-paragraph summary of the Apollo programme."
    )
    print(f"thread {thread_id}: {len(steps)} steps, {len(artifacts)} artifacts")

    # A second turn on the same thread continues the co-construction.
    thread_id, *_ = run_turn(client, "Now make it two sentences shorter.", thread_id)

    # Read back the shared workspace. The document endpoint returns JSON null
    # (with HTTP 200) until the agent has actually written one, so a turn that
    # only answered in chat leaves this empty.
    document = client.get(f"{BASE}/workspace/{thread_id}/document", headers=HEADERS).json()
    print(document["content"] if document else "(no document written yet)")
```

## Verifying a token in CI

Tokens are created interactively by design, but everything after that can be scripted. A typical
setup stores one long-lived `read,write` token as an encrypted secret and checks it before the
pipeline does anything expensive:

```bash theme={null}
status=$(curl -s -o /dev/null -w "%{http_code}" \
  https://haico.gr/api/auth/me -H "Authorization: Bearer $HAICO_TOKEN")

case "$status" in
  200) echo "token OK" ;;
  401) echo "HAICO_TOKEN is invalid, revoked, or expired" >&2; exit 1 ;;
  429) echo "rate limited; retry later" >&2; exit 1 ;;
  *)   echo "unexpected status $status" >&2; exit 1 ;;
esac
```

Because every token carries a visible `expires_at`, a scheduled job can warn you before one lapses
rather than letting you discover it through a failed run.