Triggering a Custom Agent from Data Lab

import os

from seeq import sdk  # spy is pre-imported in Data Lab

# if you are running outside Data Lab, you'll need to install and import the seeq.spy package and authenticate first

api_client = spy.client

APPSERVER_JSON = "application/vnd.seeq.v1+json"

def base_url(api_client: sdk.ApiClient) -> str:
    """spy's client host is '<server>/api'; the /genai routes hang off the server root."""
    if os.environ.get("SEEQ_SERVER_URL"):
        return os.environ["SEEQ_SERVER_URL"].rstrip("/")

    host = (getattr(api_client, "host", None) or "").rstrip("/")
    return host[: -len("/api")] if host.endswith("/api") else host

def auth_headers(api_client: sdk.ApiClient) -> dict[str, str]:
    headers = {
        "Accept": APPSERVER_JSON,
        "Content-Type": APPSERVER_JSON,
        "x-sq-origin": "datalab",
        "x-sq-origin-label": "custom-agent-trigger",
    }
    api_client.add_authorization_header(headers)
    csrf = getattr(api_client, "csrf_token", None)

    if isinstance(csrf, str) and csrf:
        headers["x-sq-csrf"] = csrf
    return headers

SEEQ = base_url(api_client)
HEADERS = auth_headers(api_client)

agents_api = sdk.CustomAgentsApi(api_client)

# All agents you can see
print("All agents:")
for agent in agents_api.get_custom_agents():
    print(agent.id, "\t", agent.name)

# Only agents that have a published production version (what you usually want to trigger)
published = agents_api.get_custom_agents(versioned=True, production="set")
print("\nPublished agents:")
for agent in published:
    print(agent.id, "\t", agent.name)

{
  "customAgentId": "…",        // required (mutually exclusive with templateId)
  "templateId": "…",           // use instead of customAgentId to run a built-in template
  "versionId": "…",            // optional: pin to a published version
  "prompt": "…",               // the user message handed to the agent
  "inputs": {"node-id": "…"},  // optional: override a node's stored instruction for this run
  "chatId": "…",               // optional: associate the run with a chat thread
  "options": {
    "stream": true,            // stream the response body (default true)
    "streamTools": true,       // emit tool/progress events (default true)
    "interleaveThinking": true // mix those events into the text body as JSON (default true)
  }
}

import asyncio
import httpx

async def run_agent(agent_id: str, prompt: str, *, version_id: str | None = None) -> str:
    body = {
        "customAgentId": agent_id,
        "prompt": prompt,
        "options": {
            "stream": True,
            "streamTools": False,
            "interleaveThinking": False,  # keep the body pure text
        },
    }
  
    if version_id:
        body["versionId"] = version_id

    # Runs can take minutes; only bound the connect phase.
    timeout = httpx.Timeout(600.0, connect=30.0)  # 10 minutes read, 30s connect
    pieces: list[str] = []
    url = f"{SEEQ}/genai/agent-workflows/runs"

    async with httpx.AsyncClient(timeout=timeout) as client:
        async with client.stream("POST", url, json=body, headers=HEADERS) as response:
            if response.status_code >= 400:
                await response.aread()
                raise RuntimeError(f"Run start failed ({response.status_code}): {response.text}")
            run_id = response.headers.get("X-Workflow-Run-Id")
            print(f"run {run_id} started")
            async for chunk in response.aiter_text():
                pieces.append(chunk)
                print(chunk, end="", flush=True)  # live output

    return "".join(pieces)

answer = await run_agent("<agent-id>", "Summarize last week's compressor downtime.")

import requests

url = f"{SEEQ}/genai/agent-workflows/runs"

with requests.post(url, json=body, headers=HEADERS, stream=True, timeout=(30, 600)) as response:
    response.raise_for_status()
    run_id = response.headers.get("X-Workflow-Run-Id")
    for chunk in response.iter_content(chunk_size=None, decode_unicode=True):
        print(chunk, end="", flush=True)

import asyncio
import json
import httpx

async def _read_thinking(client: httpx.AsyncClient, run_id: str, printer: "ProgressPrinter") -> None:
    """Consume the JSONL progress stream for a run."""
    url = f"{SEEQ}/genai/agent-workflows/runs/{run_id}/stream/thinking"
    async with client.stream("GET", url, headers=HEADERS) as response:
        buffer = ""
        async for chunk in response.aiter_text():
            buffer += chunk
            while "\n" in buffer:
                line, buffer = buffer.split("\n", 1)
                line = line.strip()
                if not line:
                    continue
                try:
                    event = json.loads(line)
                except json.JSONDecodeError:
                    continue  # partial line; the next chunk completes it

                printer.handle(event)


class ProgressPrinter:
    """Renders the thinking stream, collapsing each section/item to a single line.
    Placeholder events (ids only, no label) are skipped; each section and item prints
    once, when its label arrives.
    """
    def __init__(self, show_reasoning: bool = False) -> None:
        self._seen: set[tuple] = set()
        self._show_reasoning = show_reasoning

    def handle(self, event: dict) -> None:
        kind = event.get("type", "")
        if kind == "workflow_node_start":
            print(f"\n▶ node {event.get('node_id')} started", flush=True)
        elif kind == "workflow_node_complete":
            print(f"\n✔ node {event.get('node_id')} finished", flush=True)
        elif kind == "workflow_node_error":
            print(f"\n✘ node {event.get('node_id')}: {event.get('error')}", flush=True)
        elif kind == "workflow_warning":
            print(f"\n⚠ {event.get('code')}: {event.get('message')}", flush=True)
        elif kind in ("workflow_complete", "workflow_error"):
            print(f"\n== {kind} ==", flush=True)
        elif kind == "thinking_section_start":
            self._once(("section", event.get("section_id")), event.get("title"), "  ┌ ")
        elif kind == "thinking_item_start":
            if event.get("item_type") == "reasoning_summary" and not self._show_reasoning:
                return  # the model's own reasoning; label is always blank
            key = ("item", event.get("section_id"), event.get("item_id"))
            label = event.get("name") or event.get("description")
            self._once(key, label, "  … ")
        elif kind == "thinking_item_error":
            print(f"\n  ✘ {event.get('description')}: {event.get('error')}", flush=True)

    def _once(self, key: tuple, label: str | None, prefix: str) -> None:
        label = (label or "").strip()
        if not label or label == "N/A":
            return  # placeholder - the labeled event follows
        if key in self._seen:
            return  # already printed this section/item
        self._seen.add(key)
        print(f"\n{prefix}{label}", flush=True)

async def run_agent_with_progress(agent_id: str, prompt: str) -> str:
    body = {
        "customAgentId": agent_id,
        "prompt": prompt,
        "options": {
            "stream": True,
            "streamTools": True,  # required for thinking events to be emitted
            "interleaveThinking": False,  # …but keep them out of the text body
        },
    }

    timeout = httpx.Timeout(600, connect=30.0)
    pieces: list[str] = []
    url = f"{SEEQ}/genai/agent-workflows/runs"

    async with httpx.AsyncClient(timeout=timeout) as client:
        async with client.stream("POST", url, json=body, headers=HEADERS) as response:
            if response.status_code >= 400:
                await response.aread()
                raise RuntimeError(f"Run start failed ({response.status_code}): {response.text}")
            run_id = response.headers["X-Workflow-Run-Id"]
            thinking = asyncio.create_task(_read_thinking(client, run_id, ProgressPrinter()))
            async for chunk in response.aiter_text():
                pieces.append(chunk)
                print(chunk, end="", flush=True)
            await thinking
    return "".join(pieces)

import time
import requests

body = {
    "customAgentId": agent_id,
    "prompt": prompt,
    "options": {"stream": False, "streamTools": True},
}

start_url = f"{SEEQ}/genai/agent-workflows/runs"
start = requests.post(start_url, json=body, headers=HEADERS, timeout=60)
start.raise_for_status()  # 202 Accepted
run_id = start.json()["runId"]
poll_url = f"{SEEQ}/genai/agent-workflows/runs/{run_id}"

TERMINAL = {"finished", "error", "canceled"}

while True:
    status = requests.get(poll_url, headers=HEADERS, timeout=30).json()
    if status["state"] in TERMINAL:
        break
    time.sleep(2)

print("state:", status["state"])

for node_id, output in status["nodeOutputs"].items():
    print(f"\n--- {node_id} ---\n{output.get('text', '')}")

if status["state"] == "error":
    print("error:", status.get("error"))

requests.post(f"{SEEQ}/genai/agent-workflows/runs/{run_id}/cancel", headers=HEADERS, timeout=30)

{"detail": {"code": "UNREACHABLE_NODES", "message": "…", "nodeIds": ["node-3"]}}