Runs and Steps

A run is one execution of your agent. A step is one decision inside it — a thought, a tool call, a result. client.run() returns a context manager that opens the run on entry, accumulates step/token/cost totals, and closes it on exit.

with client.run("research-agent", input="what is keelwave") as run:
    run.step("think", content="deciding which tool to call")
    run.tool_call("web_search", input={"q": "keelwave"}, output={"hits": 3})
    run.set_output("done")

AsyncRun has the same surface as Run; step() and tool_call() are coroutines, while set_output(), mark_loop(), and check_fingerprint() stay synchronous.

Opening a run

client.run(
    agent_name: str,
    *,
    input: str | None = None,
    metadata: dict | None = None,
) -> Run

run() only constructs the object. Nothing is sent until you enter the with block.

Lifecycle

On __enter__ the run posts to /v1/ingest/agent/runs, stores the returned id and timestamp, and publishes itself as the current run in a ContextVar.

On __exit__ it posts to /v1/ingest/agent/runs/{id}/finish with the totals it accumulated, and clears the ContextVar. Status is derived from whether the block raised:

The run does not swallow the exception — it records the failure and re-raises. duration_ms is measured with time.monotonic() across the block.

Properties

run.id and run.timestamp are available once the run is open. Reading either before entering the with block raises RuntimeError.

with client.run("research-agent") as run:
    print(run.id)         # server-assigned run id
    print(run.timestamp)  # partition timestamp, needed by the loops endpoint

Recording steps

step

run.step(
    step_type: str,
    content: str | None = None,
    *,
    tokens: int | None = None,
    cost_usd: float | None = None,
    metadata: dict | None = None,
) -> None

step_type is free-form. The SDK itself emits "think" (from the provider adapters) and "tool_call"; "result" and any custom type of your own work equally well. Each call increments step_index and adds tokens / cost_usd into the run totals.

run.step("think", content="the query returned nothing useful", tokens=412)
run.step("result", content="answer assembled from 3 sources")

tool_call

run.tool_call(
    tool_name: str,
    *,
    input: dict,
    output: Any,
    ok: bool = True,
    tokens: int | None = None,
    cost_usd: float | None = None,
    latency_ms: int | None = None,
    metadata: dict | None = None,
) -> None

Emits a step with step_type="tool_call". input must be a dict. output accepts anything — a dict is sent as-is, anything else is wrapped as {"value": str(output)}.

run.tool_call(
    "web_search",
    input={"q": "keelwave"},
    output={"results": [...]},
    ok=True,
    latency_ms=182,
)

set_output

run.set_output(output: str) -> None

Sets the run's final output. It is sent with the finish call, so the last write before the block exits wins.

Loop detection

Repeating the same tool call with the same arguments is the classic stuck-agent signature. check_fingerprint() hashes tool_name plus the canonically-encoded input with SHA-256 and remembers the step index where each fingerprint was first seen. The first duplicate calls mark_loop() with that original index.

run.check_fingerprint(tool_name: str, tool_input: dict) -> None
run.mark_loop(step_index: int | None = None) -> None

Once marked, the finish call carries loop_detected=True and loop_step_index.

with client.run("looper") as run:
    for _ in range(3):
        run.tool_call("search", input={"q": "same"}, output={"results": []})
        run.check_fingerprint("search", {"q": "same"})

You can also call mark_loop() yourself when your own heuristic decides the agent is stuck:

if turns > MAX_TURNS:
    run.mark_loop()

The ambient run

get_current_run() returns the Run or AsyncRun currently inside its with block, or None.

from keelwave import get_current_run

run = get_current_run()
if run is not None:
    run.step("note", content="reached the summarise phase")

This is what lets the decorators and provider adapters attach to the right run without you threading a run object through every function. It is backed by a ContextVar, which is per-task under asyncio — concurrent asyncio.gather branches each get their own snapshot, so nested runs do not leak into each other.

Runs opened inside a with block nest: the inner run becomes current until it exits, then the outer one is restored.

Full example

Loop detection end to end, from the SDK's own examples/looping_agent.py:

import os
from keelwave import Keelwave, get_current_run

keelwave_client = Keelwave(
    api_key=os.environ["KEELWAVE_API_KEY"],
    endpoint=os.environ.get("KEELWAVE_ENDPOINT", "http://localhost:8080"),
)


@keelwave_client.observe(name="web_search", step_type="tool_call")
def web_search(q: str, max_results: int = 3) -> dict:
    return {"results": search(q, max_results=max_results)}


@keelwave_client.agent(name="looping-agent")
def run_loop(question: str) -> str:
    run = get_current_run()
    for _ in range(5):
        # @observe emits the tool_call step and checks the fingerprint;
        # the duplicate on turn 2 marks the run as a loop.
        web_search(q=question, max_results=3)
    return run.id if run else ""

Steps are buffered server-side before they land in the query tables, so if you read /v1/projects/{projectID}/agent/runs/{runID}/loops straight after a run finishes, give the flush a moment.