Decorators

@client.agent wraps a function in a run. @client.observe records each call to a function as a step on that run. Together they trace an agent without a single with block.

import os
from keelwave import Keelwave

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


@client.observe(name="web_search", step_type="tool_call")
def web_search(q: str) -> dict:
    return {"results": [f"result for: {q}"]}


@client.agent(name="research-agent")
def run_agent(task: str) -> str:
    results = web_search(q=task)
    return f"Found: {results['results'][0]}"


answer = run_agent("what is keelwave")

That produces one run named research-agent with input "what is keelwave", a tool_call step for web_search with its arguments, result, and latency, and the returned string as the run output.

Use AsyncKeelwave and decorate async def functions for the async equivalent; the decorators on AsyncKeelwave produce async wrappers.

@agent

client.agent(fn=None, *, name: str | None = None)

Opens a run when the function is called and closes it when it returns or raises. name defaults to the function's __name__.

Both forms work:

@client.agent
def my_agent(task: str) -> str: ...


@client.agent(name="research-agent")
def my_agent(task: str) -> str: ...

Input capture. The first positional argument becomes the run's input; failing that, an input= keyword argument is used. Values are stringified and truncated to 500 characters.

Output capture. A non-None return value is stringified (again truncated to 500 characters) and set as the run's output.

Errors. The decorator does not catch your exceptions. An exception propagates, and the underlying run records status="failed" with termination_reason="error".

@observe

client.observe(fn=None, *, name: str | None = None, step_type: str = "tool_call")

Instruments a single function. name defaults to the function's __name__.

@client.observe
def web_search(q: str) -> dict: ...


@client.observe(name="search", step_type="tool_call")
def web_search(q: str) -> dict: ...

What it emits depends on where the call happens.

Inside a run

With the default step_type="tool_call", it calls run.tool_call() with the function name, its arguments, its return value, a success flag, and the measured latency — then calls run.check_fingerprint(), so duplicate calls with identical arguments trip loop detection automatically.

With any other step_type, it calls run.step() instead, using str(result) as content and {"fn": ..., "latency_ms": ...} as metadata. No fingerprint check runs in this mode.

@client.observe(step_type="retrieval")
def fetch_docs(query: str) -> list[str]: ...

Outside a run

If no run is active, @observe falls back to client.ingest_ai() with model=<function name> and provider="observe", recording status, latency, any error message, and the serialised input/output in metadata. Decorated helpers therefore still produce data when called standalone.

Argument serialisation

Positional arguments are collected under an args key, keyword arguments by their own names, and every value is stringified and truncated to 500 characters. Return values that are already dicts are sent as-is; anything else becomes {"value": ...}. When the function raises, the recorded output is {"error": ...} with the message truncated to 500 characters.

client.span

span() is a context manager for instrumenting a block of code rather than a whole function.

client.span(step_type: str = "span", *, name: str | None = None)

with client.span("retrieval", name="fetch_docs") as span:
    docs = fetch(query)
    span.set(content=f"fetched {len(docs)} docs")

set() records what the step should report and can be called any number of times inside the block:

span.set(
    content: str | None = None,
    *,
    tokens: int | None = None,
    metadata: dict | None = None,
)

On exit the span emits one run.step() of the given step_type, with latency_ms measured across the block added to metadata, plus span_name when name was given. If no run is active, nothing is emitted. As with @observe, emit failures become warnings, never exceptions.

Combining with adapters

Decorators and provider adapters compose — the adapters attach LLM traces to the same ambient run the decorators opened:

import anthropic

claude = client.wrap_anthropic(anthropic.Anthropic())


@client.observe(name="web_search", step_type="tool_call")
def web_search(q: str) -> dict: ...


@client.agent(name="research-agent")
def run_agent(question: str) -> str:
    resp = claude.messages.create(
        model="claude-opus-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": question}],
    )
    ...

See adapters.