> ## Documentation Index > Fetch the complete documentation index at: https://docs.cosmon.com/llms.txt > Use this file to discover all available pages before exploring further. # Guide > Run attended or unattended, answer the agent's questions and confirmations with handlers, stream the work, attach files, and run batches. ## One call: `run(...).text()` `client.run(...)` starts a one-shot run and returns a **`Run`**; `.text()` drives it to completion and returns the run's **transcript** — the agent's answer with each tool step inline as a `▶ ` line, paragraph-separated. Pass `live=True` to also print the run while you wait (answer tokens to stdout, tool steps and `[agent]` speaker labels to stderr); the return value stays the authoritative, loss-free version. For a follow-up that builds on a previous turn, use a [session](#multi-turn-sessions). ```python theme={null} from cosmon_agent_sdk import Client with Client() as client: print(client.run("Open beam.inp and describe the model", agent="abaqus").text()) # or watch it work while it runs: # client.run("...", agent="abaqus").text(live=True) ``` SDK runs don't appear in the app's chat list or search — they're managed entirely through the SDK. The app UI shows only its own conversations. ## Multi-turn sessions `client.run(...)` is one-shot: each call is an independent conversation. To keep context across turns — so a follow-up like *"now double the load"* builds on the previous run — open a **session**. Every `run` on it reuses one conversation, and the agent remembers the earlier turns. ```python theme={null} from cosmon_agent_sdk import Client with Client() as client: with client.session(agent="solidworks") as chat: chat.run("Mesh bracket.sldprt").text() chat.run("Now double the load and re-run").text() # remembers the mesh # leaving the `with` block ends the session; the conversation persists ``` * **Entry agent, then follow the work.** Pass `agent=` to `session(...)` to pick the agent for the first turn. Later `run`s omit it and continue with the agent in control — so if the first turn **hands off** to another agent, the next turn keeps going with that agent (the same behaviour as the app). * **Cleanup.** Leaving the `with` block (or calling `chat.close()`) ends the session — the conversation **persists** server-side (hidden from the app) and can be picked up again later with `client.session(id=chat.id)`. Call `chat.delete()` to remove it entirely, or open the session with `ephemeral=True` to have it deleted automatically on close. Both are idempotent. The same applies to one-shots: `client.run(..., ephemeral=True)` deletes the conversation when the run finishes. * **Finding old conversations.** `client.chats()` lists what your scripts left on the server — `id`, `title`, `updated_at` — so a persisted conversation is never orphaned even if you lost the id (`cosmon chats` from the shell). * **Handlers still apply.** `on_question` / `on_confirm` on the client work inside a session exactly as for a one-shot run. `AsyncClient` mirrors it — `async with client.session() as chat:` and `await chat.run(...).text()`. ## Attended vs unattended One rule decides whether the agent can stop to ask you something: **the handlers you pass**. There's no mode flag, and each handler controls its own mechanism: | You pass | Questions (`ask_user`) | Confirmation gates | | ------------------ | ------------------------------------------- | ------------------------------------- | | neither | not offered — the agent proceeds on its own | auto: approve safe, decline dangerous | | `on_question` only | your handler | auto: approve safe, decline dangerous | | `on_confirm` only | not offered (question-free run) | your handler — **every** gate | | both | your handler | your handler — **every** gate | Registering `on_confirm` is opting in to being asked: the app forwards every confirmation gate to it, including ones the app's own "Confirm tool calls" toggle would auto-approve for UI users. Omitting a handler never silently declines — the unwired mechanism falls back to the safe auto-rule. The handlers are called as the agent needs input, while `.text()` runs: ```python theme={null} from cosmon_agent_sdk import Client, Question, Confirmation def answer(question: Question) -> str: # ask the human; question.options lists any offered choices ([] if free-form) return input(f"{question.prompt} {question.options or ''}\n> ") def approve(confirmation: Confirmation) -> bool: return input(f"{confirmation.summary} — approve? [y/N] ").strip().lower() == "y" with Client(on_question=answer, on_confirm=approve) as client: print(client.run("Run a static structural study on bracket.sldprt", agent="solidworks").text()) ``` A few things follow from this: * **Safe by default.** The auto-rule never rubber-stamps a destructive op — dangerous confirmations are declined unless *your* `on_confirm` approves them. `on_confirm=lambda c: True` approves **everything, dangerous included** — that's a deliberate, explicit choice (the CLI's `--auto-approve`), not something you can fall into by omission. * **Handlers may block on a human — for up to \~2 minutes.** They run in a worker thread, so an `input()` prompt sitting open doesn't stall the client's heartbeats or cancellation. But the agent won't wait forever: after the window (`Question.expires_in` / `Confirmation.expires_in`, \~120s — the same countdown the app draws as a bar) a question auto-dismisses, a confirmation is declined, and the SDK stops waiting on your handler. Show the window to your human like the example does. * **`.text()` never stalls.** Handlers answer out of band, so a run always drives to completion — it never blocks or errors waiting for input. Under the hood this is MCP **elicitation**: passing a handler makes the client declare the elicitation capability (the app runs the agent interactively); passing none declares no capability (the app runs it unattended). It's a capability, not a flag on the wire. ## Streaming the run **Iterate** the run when you need the events **as data** — `.text(live=True)` is just the terminal rendering of this same stream. Route them to a GUI, a websocket, metrics, or `break` mid-stream to cancel the run based on what you see. Each iteration yields a typed event: `ToolCall` (a tool-call boundary, `finished=False` on start and `True` on finish) or `TextChunk` (the answer, as it streams). The `TextChunk`s reconstruct the same answer `.text()` returns. Both event types carry `agent` — which agent produced the event. A run can span a **handoff chain** (one agent delegating to another mid-run), and the `agent` field is how multi-agent output stays attributable. ```python theme={null} import sys from cosmon_agent_sdk import Client, TextChunk, ToolCall with Client() as client: speaker = "" for event in client.run("Mesh and solve the model", agent="abaqus"): if event.agent and event.agent != speaker: speaker = event.agent print(f"[{speaker}]", file=sys.stderr) if isinstance(event, TextChunk): print(event.text, end="", flush=True) elif isinstance(event, ToolCall) and not event.finished: print(f"▶ {event.summary}", file=sys.stderr) ``` (`.text(live=True)` prints exactly this layout for you.) Observe vs respond: iterating **observes** what the agent does. Answering what the agent **needs** — `ask_user` / `confirm_action` — is the `on_question` / `on_confirm` hooks above, because the agent blocks waiting for a returned value. The two compose: stream a run *and* have it interactive at the same time. ### Images a tool produces When a tool produces an image for the user — a CAD screenshot, a plotted result — it rides the `ToolCall`'s **finish** event as `event.images`, a tuple of image `data:` URIs (empty on the start event). Decode and save them as they arrive: ```python theme={null} import base64, re from pathlib import Path from cosmon_agent_sdk import Client, ToolCall with Client() as client: for event in client.run("Screenshot the open part", agent="solidworks"): if isinstance(event, ToolCall) and event.finished: for uri in event.images: m = re.match(r"^data:image/\w+;base64,(.+)$", uri) if m: Path("shot.png").write_bytes(base64.b64decode(m[1])) ``` Each `data:` image is also embedded in the run's final result as an MCP image block, so plain MCP clients see it too — `.text()` is unaffected (it joins only text). See the [Images from tools example](/sdk/examples/images) for the full snippet. ## Capping a run Set `timeout` on the client (seconds) so a stuck run can't hang forever. The agent can work for many minutes, so the default is generous (10 minutes), and there is no server-side cap — a long solve runs as long as you're willing to wait. When the timeout fires it doesn't just give up waiting: it **cancels the run** server-side, so nothing keeps working (or holding the one-run-at-a-time guard) behind your back. ```python theme={null} with Client(timeout=7200) as client: # a 2-hour solve? fine. client.run("Run a mesh-convergence study on bracket.sldprt", agent="solidworks").text() ``` ## Attaching files Hand specific files to a run with `attachments`. Relative paths resolve against **your working directory** (the SDK absolutizes them before sending), and a missing file raises `CosmonError` immediately — before the agent runs. The agent reads them alongside your prompt: ```python theme={null} client.run( "Compare the attached revision against the current model and list what changed.", agent="solidworks", attachments=["C:/Projects/brackets/bracket_v7.sldprt"], ) ``` ## Unattended batches No handlers, a `timeout`, and decisions baked into the prompt — leave it running: ```python theme={null} from cosmon_agent_sdk import Client, CosmonError files = ["beam1.sldprt", "beam2.sldprt", "beam3.sldprt"] with Client(timeout=600) as client: # unattended; 10-minute cap per run for path in files: try: answer = client.run( f"Run a static structural study on {path} with a 10mm mesh seed " f"and report the peak von Mises stress.", agent="solidworks", ).text() print(f"{path}: {answer}") except CosmonError as err: print(f"{path}: failed — {err}") ``` ## Triggering runs from your own code The SDK is a plain library — no daemon of its own — so you call it from whatever fires in your workflow: a file watcher, a webhook, a cron job, a CI step, a queue consumer. Behind an HTTP endpoint, use the async client so one process can drive several runs at once: ```python theme={null} from fastapi import FastAPI from cosmon_agent_sdk import AsyncClient app = FastAPI() @app.post("/analyze") async def analyze(part_path: str) -> dict[str, str]: async with AsyncClient() as client: answer = await client.run( "Run a static structural study and report peak von Mises.", agent="solidworks", attachments=[part_path], ).text() return {"result": answer} ``` The one requirement: the Nexus app must be **running and signed in** on that machine — the SDK rides it for authentication and the live CAD/CAE connection. ## Errors vs. in-run failures Errors come in two channels: **opening the client** (`CosmonConnectionError`, `CosmonAuthError`, `CosmonProtocolError`) versus **during a run** (`CosmonRunError`, `CosmonTimeoutError`, or a dropped `CosmonConnectionError`). All subclass `CosmonError`, so a single `except CosmonError` catches everything; catch the subclasses to branch. In a batch or pipeline the one worth handling explicitly is `CosmonRunError` — in-run failures (including provider errors like *out of credits*) surface here, and its **`code`** lets you branch without parsing the message: ```python theme={null} try: client.run(prompt, agent="solidworks").text() except CosmonRunError as err: if err.code == "busy": time.sleep(30) # another run owns the CAD instance — retry later else: raise ``` See the [Python reference](/sdk/python-reference#errors) for every error, when each is raised, and the full list of `CosmonRunError.code` values. ## Next steps The full API surface — clients, runs, interaction handlers, the async twin, streaming, and errors. The `cosmon` CLI — `doctor`, `run`, and `chats` from the shell.