Python Quickstart

Install the Python SDK, wrap your agent, and see runs in the dashboard in five minutes.

Instrument a real Anthropic tool-calling agent and see its decision trace, token usage, and any loops in the dashboard.

Prerequisites

A running keelwave server. Follow Self-hosting with Docker. The API and dashboard both listen on http://localhost:8080.
An API key. make seed in core/ creates a dev project and prints a plaintext kw_... key. The server only stores its SHA-256 hash, so copy it when it is printed.
Python 3.10 or newer.

Keep the kw_... key out of source control. Every example below reads it from the environment.

Install

pip install keelwave

The Anthropic SDK is an optional dependency — the adapter imports it lazily, so install it only if you use that adapter:

pip install anthropic

Set your environment

export KEELWAVE_API_KEY=kw_...              # from `make seed`
export KEELWAVE_ENDPOINT=http://localhost:8080
export ANTHROPIC_API_KEY=sk-ant-...

KEELWAVE_API_KEY and KEELWAVE_ENDPOINT are a convention used by the examples, not magic names — the SDK reads nothing from the environment on its own. You pass both to the constructor explicitly.

Initialize the client

import os
from keelwave import Keelwave

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

endpoint defaults to http://localhost:8080. Check the connection before you go further:

print(client.health())

Instrument the agent

Three pieces do the work:

client.wrap_anthropic(...) returns a drop-in proxy around anthropic.Anthropic. Every messages.create call is recorded as an LLM trace with model, tokens, latency, and status — and any thinking blocks in the response become think steps automatically.
@client.observe turns a plain function into a tool_call step. Its input is fingerprinted, so a repeated call with identical arguments flags the run as looping.
@client.agent opens a run before the function body and closes it after, recording completed / failed, duration, and the totals. Steps emitted inside it link to the run through a ContextVar, so you never pass a run object around.

Save this as agent.py:

import json
import os

import anthropic
from keelwave import Keelwave

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

claude = client.wrap_anthropic(
    anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
)

TOOLS = [
    {
        "name": "web_search",
        "description": "Search the web. Returns title/url/snippet for top results.",
        "input_schema": {
            "type": "object",
            "properties": {"q": {"type": "string", "description": "search query"}},
            "required": ["q"],
        },
    },
]


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


@client.agent(name="research-agent")
def run_agent(question: str) -> str:
    MAX_TURNS = 8
    messages = [{"role": "user", "content": question}]

    for _ in range(MAX_TURNS):
        resp = claude.messages.create(
            model="claude-opus-5",
            max_tokens=2048,
            tools=TOOLS,
            messages=messages,
        )

        text = "\n".join(
            b.text for b in resp.content if getattr(b, "type", None) == "text"
        )
        tool_uses = [b for b in resp.content if getattr(b, "type", None) == "tool_use"]

        if not tool_uses:
            return text

        tool_results = []
        for tu in tool_uses:
            result = web_search(q=tu.input["q"])
            tool_results.append(
                {
                    "type": "tool_result",
                    "tool_use_id": tu.id,
                    "content": json.dumps(result),
                }
            )

        messages.append({"role": "assistant", "content": resp.content})
        messages.append({"role": "user", "content": tool_results})

    raise RuntimeError(f"agent exceeded {MAX_TURNS} turns")


if __name__ == "__main__":
    print(run_agent("what changed in our churn numbers last week?"))

The Python SDK is not fail-soft everywhere. @client.observe, client.span(), and the provider adapters swallow emit failures into a warnings.warn, so a keelwave outage cannot break them. But @client.agent opens and closes a real run, and a failure on those calls — an unreachable server, a bad key — propagates out of your decorated function. Handle it if that matters; see errors.

See it in the dashboard

Open http://localhost:8080 and sign in. The user make seed creates has no password and is not email-verified, so it cannot sign in — register an account through the dashboard for UI access, as described in Docker.

Runs — your research-agent run appears with its status, step count, token total, cost, and duration, alongside the project-wide completion rate and loop rate.
Run detail — click the run for its step timeline in order: each think step emitted from the model's thinking blocks, each web_search tool call with its input, output, and success flag.
Tools — web_search shows up in the tool registry with its success rate and latency.

If the same search runs twice with identical arguments, the second call matches the first fingerprint and the run is flagged as looping, with the step index where the repetition began.

Without decorators

If decorators don't fit, open the run yourself. The context manager POSTs the run on entry, sends each step as it happens, and finishes the run on exit — including marking it failed when the block raises.

with client.run("research-agent", input=question) as run:
    run.step("think", content="need to search first")
    run.tool_call("web_search", input={"q": question}, output={"results": []})
    run.set_output("done")

Async

AsyncKeelwave mirrors the sync client. Its constructor takes the same api_key and endpoint; client.run(...) returns an AsyncRun used with async with, run.step(...) and run.tool_call(...) are awaitable, and @client.agent / @client.observe wrap async def functions.

import os
from keelwave import AsyncKeelwave

client = AsyncKeelwave(
    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")
async def web_search(q: str) -> dict:
    return {"results": await search(q)}


@client.agent(name="research-agent")
async def run_agent(question: str) -> str:
    results = await web_search(q=question)
    return str(results["results"])

Or open the run explicitly:

async with client.run("research-agent", input=question) as run:
    await run.step("think", content="need to search first")
    await run.tool_call("web_search", input={"q": question}, output={"results": []})

Python Quickstart

Prerequisites

Install

Set your environment

Initialize the client

Instrument the agent

Run it

See it in the dashboard

Without decorators

Async

Next steps

Python SDK reference

Adapters

Loop detection

Cost tracking

On this page