Accessing Agents 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"]}}

import requests
import os

# Server root, e.g. https://yourserver.seeq.site
base_url = os.environ["SEEQ_SERVER_URL"]
headers = { "Content-Type": "application/json" }

spy.client.add_authorization_header(headers)

import uuid

chat_id = str(uuid.uuid4())

body = {
    "prompt": "Find the Area A temperature signal and summarize the last week of data.",
    "agentType": "aq",
    "chat_id": chat_id,
    "stream": True,
    "streamTools": True,   # emit tool-call updates on the thinking stream
    "asyncStream": False,  # synchronous: the POST response streams the answer text
}

with requests.post(f"{base_url}/genai/llm/chat", json=body, headers=headers, stream=True) as r:
    r.raise_for_status()
    answer = []
    for chunk in r.iter_content(chunk_size=None, decode_unicode=True):
        answer.append(chunk)
      
print("".join(answer))

import uuid

chat_id = str(uuid.uuid4())

body = {
    "prompt": "Make all temperature signals blue",
    "agentType": "actions",
    "chat_id": chat_id,
    "stream": True,
    "streamTools": True,   # emit tool-call updates on the thinking stream
    "asyncStream": False,  # synchronous: the POST response streams the answer text
}

workbook_id = "0F1B12F8-882F-E8A0-9666-4A3F74ECBED3"
worksheet_id = "0F1B12F8-88C9-7590-B3F4-835469EA7C95"

headers.update(
  {
    "x-sq-origin": "workbench",
    "x-sq-origin-url": f"/workbook/{workbook_id}/worksheet/{worksheet_id}"
  }
)

with requests.post(f"{base_url}/genai/llm/chat", json=body, headers=headers, stream=True) as r:
    r.raise_for_status()
    answer = []
    for chunk in r.iter_content(chunk_size=None, decode_unicode=True):
        answer.append(chunk)
      
print("".join(answer))

body["asyncStream"] = True
resp = requests.post(f"{base_url}/genai/llm/chat", json=body, headers=headers)
resp.raise_for_status()
bot_message_id = resp.json()["bot_message_id"]

# 1) The answer text
answer = requests.get(
    f"{base_url}/genai/llm/chat/stream/{chat_id}",
    params={"bot_message_id": bot_message_id},
    headers=headers,
    stream=True,
)

# 2) The tool-call / progress updates (see next section)
thinking = requests.get(
    f"{base_url}/genai/llm/chat/stream/thinking/{chat_id}",
    params={"bot_message_id": bot_message_id},
    headers=headers,
    stream=True,
)

import json
import threading

def tail_thinking():
    """Print tool-call / progress updates as they arrive."""
    with requests.get(
        f"{base_url}/genai/llm/chat/stream/thinking/{chat_id}",
        params={"bot_message_id": bot_message_id},
        headers=headers,
        stream=True,
    ) as r:
        r.raise_for_status()
        for line in r.iter_lines(decode_unicode=True):
            if not line:
                continue
            event = json.loads(line)
            print(f"[{event.get('type')}] {event.get('description') or event.get('title') or ''}")

def tail_answer(out: list):
    """Accumulate (and echo) the answer text."""
    with requests.get(
        f"{base_url}/genai/llm/chat/stream/{chat_id}",
        params={"bot_message_id": bot_message_id},
        headers=headers,
        stream=True,
    ) as r:
        r.raise_for_status()
        for chunk in r.iter_content(chunk_size=None, decode_unicode=True):
            out.append(chunk)
            print(chunk, end="")

answer_parts: list[str] = []
thinking_thread = threading.Thread(target=tail_thinking)
answer_thread = threading.Thread(target=tail_answer, args=(answer_parts,))
thinking_thread.start()
answer_thread.start()
thinking_thread.join()
answer_thread.join()

answer_text = "".join(answer_parts)

import json

with requests.get(
    f"{base_url}/genai/llm/chat/stream/thinking/{chat_id}",
    params={"bot_message_id": bot_message_id},
    headers=headers,
    stream=True,
) as r:
    r.raise_for_status()
    for line in r.iter_lines(decode_unicode=True):
        if not line:
            continue
        event = json.loads(line)
        match event.get("type"):
            case "thinking_section_start":
                # May be an id-only placeholder; the title arrives in a later update
                print(f"\n== {event.get('title', event.get('section_id', ''))} ==")
            case "thinking_item_start":
                print(f"  -> {event.get('name', '')}: {event.get('description', '')}")
            case "thinking_item_complete":
                print(f"  ✓ {event.get('description', '')}")
            case "thinking_item_error":
                print(f"  ✗ {event.get('description', '')}: {event.get('error', '')}")
            case "thinking_section_complete":
                status = "with errors" if event.get("error") else "ok"
                print(f"== section done ({status}) ==")

with open("process_data.csv", "rb") as f:
    upload = requests.post(
        f"{base_url}/api/markdown/links",
        headers={"sq-auth": spy.client.auth_token},
        files={"file": ("process_data.csv", f, "text/csv")},
    )
upload.raise_for_status()
# Response contains the stored link, e.g. ".../api/markdown/links/<uuid>.csv"
file_url = upload.json()["link"]
file_id = file_url.rsplit("/", 1)[-1]  # "<uuid>.csv"

body = {
    "prompt": "Analyze the attached CSV and describe any anomalies.",
    "agentType": "aq",
    "chat_id": chat_id,
    "streamTools": True,
    "attachments": [
        {
            "id": file_id,                            # file id with extension
            "url": f"/api/markdown/links/{file_id}",
            "type": "csv",                            # "image" | "csv" | "pdf"
            "mimeType": "text/csv",
            "name": "process_data.csv",
            # Optional but recommended for CSVs: first ~20 rows so the agent
            # can see the shape of the data without a tool round-trip
            "csvPreview": csv_preview_text,
        }
    ],
}