Adapters

Adapters wrap a provider's client so every model call is recorded to keelwave's ai_traces — model, provider, token usage, latency, errors — and linked to the active run automatically. Your calling code does not change.

import anthropic

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

with client.run("research-agent", input="what is keelwave") as run:
    resp = claude.messages.create(
        model="claude-opus-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": "what is keelwave"}],
    )

The wrapped client is a proxy: only the intercepted method is replaced, and every other attribute delegates to the real client. Both providers are optional dependencies — imports happen inside the wrapper, so installing keelwave does not pull them in.

Anthropic

client.wrap_anthropic(client: Any, *, provider: str = "anthropic")

Wraps anthropic.Anthropic (or anthropic.AsyncAnthropic on AsyncKeelwave). Only .messages.create is intercepted.

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"])
)

with client.run("research-agent", input="what is keelwave") as run:
    resp = claude.messages.create(
        model="claude-opus-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": "what is keelwave"}],
    )
    text = "\n".join(
        b.text for b in resp.content if getattr(b, "type", None) == "text"
    )
    run.set_output(text)

Each call records model, status, provider, input_tokens, output_tokens, latency_ms, error_message, request_id, the active agent_run_id, and metadata carrying tool_choice, stop_reason, and num_tools_in_request.

Thinking blocks become steps. If a response contains thinking content blocks and a run is active, the adapter emits a think step per block with the block's text and the call's total tokens. No extra code required.

Non-Anthropic endpoints

provider defaults to "anthropic". Override it when calling an Anthropic-compatible endpoint so the dashboard attributes traces correctly:

claude = client.wrap_anthropic(
    anthropic.Anthropic(
        api_key=os.environ["ANTHROPIC_API_KEY"],
        base_url=os.environ["ANTHROPIC_ENDPOINT"],
    ),
    provider="deepseek",
)

parse_response

For reading a response yourself, parse_response pulls the universal observability fields off an Anthropic response.

from keelwave import parse_anthropic_response

parsed = parse_anthropic_response(resp)
parsed.request_id      # str | None  (resp.id)
parsed.input_tokens    # int | None
parsed.output_tokens   # int | None
parsed.total_tokens    # int | None
parsed.stop_reason     # str | None
parsed.raw             # the original response
parsed.tokens_total()  # input + output, falling back to total_tokens, else 0

It deliberately does not flatten content blocks — walk parsed.raw.content yourself for text, thinking, and tool-use blocks.

OpenAI

client.wrap_openai(client: Any, *, provider: str = "openai")

Wraps openai.OpenAI (or openai.AsyncOpenAI on AsyncKeelwave). Only .chat.completions.create is intercepted.

import os
import openai
from keelwave import Keelwave

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

gpt = client.wrap_openai(openai.OpenAI(api_key=os.environ["OPENAI_API_KEY"]))

with client.run("triage-agent", input="classify this bug report") as run:
    resp = gpt.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "classify this bug report"}],
    )
    run.set_output(resp.choices[0].message.content or "")

Token field names are normalised to keelwave's vocabulary: usage.prompt_tokens becomes input_tokens, usage.completion_tokens becomes output_tokens, and usage.total_tokens is passed through. Metadata carries tool_choice, finish_reason, and num_tools_in_request.

Reasoning becomes a think step. If choices[0].message carries reasoning (o1/o3-style) or reasoning_content (DeepSeek-style) and a run is active, the adapter emits one think step with that text.

OpenAI-compatible endpoints

Override provider for compatible endpoints such as Groq, Together, Azure, OpenRouter, or DeepSeek:

gpt = client.wrap_openai(
    openai.OpenAI(
        api_key=os.environ["OPENAI_API_KEY"],
        base_url=os.environ["OPENAI_ENDPOINT"],
    ),
    provider="deepseek",
)

parse_response

from keelwave import parse_openai_response

parsed = parse_openai_response(resp)
parsed.request_id      # str | None  (resp.id)
parsed.input_tokens    # int | None  (usage.prompt_tokens)
parsed.output_tokens   # int | None  (usage.completion_tokens)
parsed.total_tokens    # int | None  (usage.total_tokens)
parsed.finish_reason   # str | None  (choices[0].finish_reason)
parsed.raw             # the original response
parsed.tokens_total()  # total_tokens, else input + output, else 0

Message content and tool calls are not extracted — walk parsed.raw.choices[0].message for those.

pydantic-ai

The pydantic-ai adapter is not a client proxy. It runs a pydantic_ai.Agent inside a keelwave run, subscribing to the agent's event stream so every tool invocation becomes a tool_call step.

Import it from its module — it is not re-exported from the package root:

from keelwave.adapters.pydantic_ai import instrument

instrument

The high-level entry point. It opens a run for you, or reuses one if a run is already active.

async def instrument(
    keelwave_client: Keelwave | AsyncKeelwave,
    agent: Agent[Any, Any],
    user_prompt: str | None = None,
    *,
    agent_name: str | None = None,
    deps: Any = None,
    **run_kwargs: Any,
) -> Any

import os
from pydantic_ai import Agent

from keelwave import Keelwave
from keelwave.adapters.pydantic_ai import instrument

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

agent = Agent("openai:gpt-4o", name="bug-triage")


@agent.tool_plain
def read_file(path: str) -> str:
    return open(path).read()


result = await instrument(client, agent, "find bugs in checkout.py")
print(result.output)

instrument is a coroutine and must be awaited, even with a sync Keelwave client. It returns pydantic-ai's AgentRunResult unchanged.

The run name is agent_name if given, else the agent's own name, else "pydantic-ai-agent". Extra keyword arguments are forwarded to agent.run(...).

Reusing an active run

If a run is already open — including one opened by @client.agent — instrument attaches to it instead of opening a new one, so the pydantic-ai steps land in the same trace as your other instrumentation.

@client.agent(name="triage-agent")
def triage(task: str) -> str:
    result = asyncio.run(instrument(client, agent, task))
    return str(result.output)

run_with_steps

Use the low-level helpers when you already have a run open and want explicit control.

async def run_with_steps(
    run: Run,
    agent: Agent[Any, Any],
    user_prompt: str | None = None,
    *,
    deps: Any = None,
    **run_kwargs: Any,
) -> Any

async def async_run_with_steps(
    run: AsyncRun,
    agent: Agent[Any, Any],
    user_prompt: str | None = None,
    *,
    deps: Any = None,
    **run_kwargs: Any,
) -> Any

from keelwave.adapters.pydantic_ai import run_with_steps

with client.run("bug-triage", input="find bugs") as run:
    result = await run_with_steps(run, agent, "find bugs in checkout.py")

Pick run_with_steps for a sync Run and async_run_with_steps for an AsyncRun. Both are coroutines.

What gets recorded

Both helpers install a pydantic-ai event_stream_handler that watches for FunctionToolCallEvent and FunctionToolResultEvent. Each matched pair emits one run.tool_call() step with the tool name, its arguments, and the result truncated to 2000 characters. ok is False when the result part is flagged as an error.

After the agent finishes, token usage from result.usage (input plus output) is added to the run's total.

Step-emit failures raise a warnings.warn rather than an exception, so a tracing problem never fails the agent run.