Skip to main content

Advanced: Context Manager

Prefer Decorators

For most agents, the decorator pattern is simpler and covers 90% of use cases. Use WaxellContext only when you need explicit lifecycle control -- batch loops, multi-step orchestration across functions, conditional context creation, or multiple runs in a single function.

See Decorator vs Context Manager for a decision guide.

WaxellContext is a context manager that gives you explicit control over run lifecycle, LLM call recording, step tracking, and mid-execution policy checks.

It works as both async with (for async code) and plain with (for sync code).

Async Usage​

from waxell_observe import WaxellContext

async with WaxellContext(agent_name="research-agent") as ctx:
    result = await run_research(query)

    ctx.record_llm_call(model="gpt-4o", tokens_in=300, tokens_out=150)
    ctx.record_step("research", output={"sources": 5})
    ctx.set_result({"answer": result})

Sync Usage​

from waxell_observe import WaxellContext

with WaxellContext(agent_name="batch-processor") as ctx:
    result = process_data(input_data)

    ctx.record_llm_call(model="gpt-4o", tokens_in=300, tokens_out=150)
    ctx.record_step("process", output={"items": 42})
    ctx.set_result({"output": result})

The sync path uses native __enter__ / __exit__ with synchronous HTTP calls — ContextVars are set in the calling thread, so auto-instrumentation works correctly.

When to use sync vs async

Use with (sync) for batch processing scripts, CLI tools, ETL pipelines, and any code that doesn't use async/await. Use async with for async web servers, async agent frameworks, and code that's already async.

Convenience Aliases​

waxell.context and waxell.session are re-exports for the cleanest one-liner usage:

import waxell_observe as waxell

# `waxell.context(...)` is an alias for `WaxellContext(...)`
with waxell.context(agent_name="my-agent") as ctx:
    ...

# `waxell.session(...)` sets session_id / user_id for any run opened inside the
# block — useful for thread-less frameworks (pydantic-ai, smolagents, raw SDK
# loops) where the framework has no native thread concept.
with waxell.session(session_id=thread_id, user_id=email):
    agent.run_sync(user_message)  # auto-instrumented run inherits session/user

You can also generate a random session id with waxell.generate_session_id() (returns sess_<16 hex chars>).

Lifecycle​

On entering the context:

  1. Policies are checked (if enforce_policy=True)
  2. A new execution run is started on the control plane

On exiting the context:

  1. Buffered LLM calls are flushed to the control plane
  2. Buffered steps are flushed to the control plane
  3. The run is completed with success or error status

Enhanced Context Options​

Session and User Tracking​

Group related runs into sessions and track end-user identity:

with WaxellContext(
    agent_name="my-chatbot",
    session_id="session-abc-123",  # Group related runs
    user_id="user-456",            # Track end-user
) as ctx:
    # Your LLM calls here
    response = call_llm(prompt)

Tags and Metadata​

Add structured metadata to runs for filtering and analysis:

with WaxellContext(agent_name="my-agent") as ctx:
    ctx.set_tag("environment", "production")
    ctx.set_tag("pipeline", "rag-v2")
    ctx.set_metadata("retrieval_count", 5)
    ctx.set_metadata("model_version", "gpt-4-turbo")

    # Your LLM calls here

Recording Scores​

Capture quality metrics and user feedback:

with WaxellContext(agent_name="my-agent") as ctx:
    response = call_llm(prompt)

    # Numeric score (0-1 range)
    ctx.record_score(
        name="relevance",
        value=0.92,
        data_type="numeric",
        comment="Highly relevant to the query"
    )

    # Boolean score
    ctx.record_score(
        name="contains_hallucination",
        value=False,
        data_type="boolean"
    )

    # Categorical score
    ctx.record_score(
        name="tone",
        value="professional",
        data_type="categorical"
    )

Recording Steps​

Track sub-operations within a run:

with WaxellContext(agent_name="rag-pipeline") as ctx:
    # Step 1: Retrieval
    docs = retrieve_documents(query)
    ctx.record_step("retrieval", output={"doc_count": len(docs)})

    # Step 2: Generation
    response = generate_response(query, docs)
    ctx.record_step("generation", output={"response_length": len(response)})

Constructor Parameters​

ParameterTypeDefaultDescription
agent_namestr(required)Name for this agent in the control plane
workflow_namestr"default"Workflow name for grouping runs
inputsdict | NoneNoneInput data to record with the run
metadatadict | NoneNoneArbitrary metadata to attach to the run
clientWaxellObserveClient | NoneNonePre-configured client. If None, creates a new one using current configuration
enforce_policyboolTrueCheck policies on context entry
session_idstr""Session ID for grouping related runs
user_idstr""End-user ID for per-user tracking and analytics
user_groupstr""User group for authorization policies (e.g., "enterprise", "free")
end_user_idstr""Sub-user identity for end-user budget / rate-limit / suspension policy handlers. Wired into metadata["tenant_sub_user_id"]
interaction_modestr | NoneNoneOne of "auto", "interactive", "autonomous". None/"auto" is inferred at run start: a session_id (chat turn / multi-turn thread) means interactive; no session means autonomous
mid_execution_governancebool | NoneNoneFlush data and check governance after each governance-bearing record_* call. None resolves to True unless WAXELL_DISABLE_MID_EXECUTION_GOVERNANCE=1 is set
auto_groundingboolFalseAuto-bridge retrieval scores to grounding governance
on_policy_blockCallable | NoneNoneCallback for policy blocks. Receives PolicyViolationError, returns ApprovalDecision. Built-in: prompt_approval, auto_approve, auto_deny

Recording Methods​

record_llm_call​

Record an LLM API call. All parameters are keyword-only.

ctx.record_llm_call(
    model="gpt-4o",
    tokens_in=500,
    tokens_out=200,
    cost=0.0,              # Optional: auto-estimated if 0.0
    task="summarize",      # Optional: label for this call
    prompt_preview="...",  # Optional: first N chars of prompt
    response_preview="...",  # Optional: first N chars of response
    duration_ms=350,       # Optional: call duration in milliseconds
    provider="openai",     # Optional: inferred from model name if empty
)
ParameterTypeDefaultDescription
modelstr(required)Model name (e.g., "gpt-4o", "claude-sonnet-4")
tokens_inint(required)Input/prompt token count
tokens_outint(required)Output/completion token count
costfloat0.0Cost in USD. If 0.0, automatically estimated using built-in model pricing
taskstr""A label describing this LLM call's purpose
prompt_previewstr""Preview of the prompt text
response_previewstr""Preview of the response text
duration_msint | NoneNoneLLM call duration in milliseconds
providerstr""Provider name (e.g., "openai", "anthropic"). If empty, inferred from model name

LLM calls are buffered in memory and flushed to the control plane when the context exits.

record_step​

Record a named execution step.

ctx.record_step("extract_entities", output={"count": 12})
ParameterTypeDefaultDescription
step_namestr(required)Name identifying this step
outputdict | NoneNoneOptional output data for the step

Steps are automatically numbered in order of recording. Like LLM calls, they are buffered and flushed on context exit.

set_result​

Set the final result for the run.

ctx.set_result({"answer": "The capital of France is Paris.", "confidence": 0.95})
ParameterTypeDefaultDescription
resultdict(required)Result data to include when the run is completed

Call this before the context exits. If not called, the run completes with an empty result.

check_policy / check_policy_sync​

Perform a mid-execution policy check. This is useful for long-running agents that should re-validate policies between steps.

# Async
policy = await ctx.check_policy()

# Sync
policy = ctx.check_policy_sync()
if policy.blocked:
    print(f"Blocked: {policy.reason}")
    # Handle the block (e.g., stop processing)
elif policy.action == "warn":
    print(f"Warning: {policy.reason}")
    # Continue but log the warning

Returns a PolicyCheckResult with:

  • action -- one of "allow", "block", "warn", "throttle"
  • reason -- human-readable explanation
  • metadata -- additional policy data
  • allowed -- property, True if action is "allow" or "warn"
  • blocked -- property, True if action is "block" or "throttle"

record_score​

Record a quality score or feedback metric for the current run.

ctx.record_score(
    name="relevance",
    value=0.92,
    data_type="numeric",
    comment="Highly relevant to the query",
)
ParameterTypeDefaultDescription
namestr(required)Score name (e.g., "relevance", "accuracy", "thumbs_up")
valuefloat | str | bool(required)Score value. Type depends on data_type
data_typestr"numeric"One of "numeric", "categorical", "boolean"
commentstr""Optional free-text comment

Scores are buffered and flushed to the control plane when the context exits.

set_tag​

Set a searchable tag on the current run. Tags become OTel span attributes and are queryable in Grafana TraceQL.

ctx.set_tag("environment", "production")
ctx.set_tag("pipeline", "rag-v2")
ParameterTypeDefaultDescription
keystr(required)Tag name (alphanumeric, underscores, hyphens)
valuestr(required)Tag value (string only)

set_metadata​

Set arbitrary metadata on the current run. Complex values are JSON-serialized.

ctx.set_metadata("retrieval_count", 5)
ctx.set_metadata("model_version", "gpt-4-turbo")
ParameterTypeDefaultDescription
keystr(required)Metadata key
valueAny(required)Any JSON-serializable value

Behavior Tracking​

Track agent behaviors beyond LLM calls and steps. These methods buffer data as spans and flush on context exit.

record_tool_call​

Record a tool or function call.

ctx.record_tool_call(
    name="web_search",
    input={"query": "latest news"},
    output={"results": [...]},
    duration_ms=250,
    status="ok",
    tool_type="api",
)
ParameterTypeDefaultDescription
namestr(required)Tool name (e.g., "web_search", "database_query")
inputdict | str""Tool input parameters
outputdict | str""Tool output/result
duration_msint | NoneNoneExecution time in milliseconds
statusstr"ok""ok" or "error"
tool_typestr"function"Classification: "function", "api", "database", "retriever"
errorstr""Error message if status is "error"

record_retrieval​

Record a RAG document retrieval.

ctx.record_retrieval(
    query="How does the billing system work?",
    documents=[{"id": "doc1", "title": "Billing FAQ", "score": 0.92}],
    source="pinecone",
    duration_ms=120,
    top_k=5,
    scores=[0.92, 0.87, 0.81],
)
ParameterTypeDefaultDescription
querystr(required)The retrieval query string
documentslist[dict](required)Retrieved documents (e.g., [{id, title, score, snippet}])
sourcestr""Data source name (e.g., "pinecone", "elasticsearch")
duration_msint | NoneNoneRetrieval time in milliseconds
top_kint | NoneNoneNumber of documents requested
scoreslist[float] | NoneNoneRelevance scores for each retrieved document

record_decision​

Record a decision or routing point.

ctx.record_decision(
    name="route_to_agent",
    options=["billing", "technical", "general"],
    chosen="billing",
    reasoning="User mentioned invoice and payment",
    confidence=0.95,
)
ParameterTypeDefaultDescription
namestr(required)Decision name (e.g., "route_to_agent", "select_model")
optionslist[str](required)Available choices
chosenstr(required)The selected option
reasoningstr""Why this option was chosen
confidencefloat | NoneNoneConfidence score (0.0-1.0)
metadatadict | NoneNoneAdditional context
instrumentation_typestr"manual"How this decision was captured: "manual", "decorator", or "auto"

record_reasoning​

Record a reasoning or chain-of-thought step.

ctx.record_reasoning(
    step="evaluate_sources",
    thought="Source A is more recent but Source B has higher authority",
    evidence=["Source A: 2024", "Source B: cited 500 times"],
    conclusion="Use Source B as primary, Source A as supplement",
)
ParameterTypeDefaultDescription
stepstr(required)Reasoning step name
thoughtstr(required)The reasoning text/thought process
evidencelist[str] | NoneNoneSupporting evidence or references
conclusionstr""Conclusion reached at this step

record_retry​

Record a retry or fallback event.

ctx.record_retry(
    attempt=2,
    reason="Rate limited by OpenAI",
    strategy="fallback",
    original_error="429 Too Many Requests",
    fallback_to="claude-sonnet-4",
    max_attempts=3,
)
ParameterTypeDefaultDescription
attemptint(required)Current attempt number (1-based)
reasonstr(required)Why a retry/fallback occurred
strategystr"retry""retry", "fallback", or "circuit_break"
original_errorstr""The error that triggered the retry
fallback_tostr""Name of fallback target (model, agent, tool)
max_attemptsint | NoneNoneMaximum attempts configured

record_policy_check​

Record a policy evaluation result as a governance span.

ctx.record_policy_check(
    policy_name="budget-limit",
    action="warn",
    category="budget",
    reason="Approaching 80% of daily budget",
    phase="mid_execution",
)
ParameterTypeDefaultDescription
policy_namestr(required)Name of the policy evaluated
actionstr(required)Evaluation result: "allow", "warn", "block", etc.
categorystr""Policy category (e.g., "budget", "rate-limit")
reasonstr""Reason for the action (empty for allow)
duration_msfloat0Evaluation time in milliseconds
phasestr"pre_execution""pre_execution", "mid_execution", or "post_execution"
priorityint100Policy priority (lower = evaluated first)

Conversation & Human Interaction​

For interactive agents (chat, REPL, ticketing) — make user input and agent output visible in the trace alongside LLM calls. These also bump conversation counters, so context-management / recursion-bound policies fire mid-run.

record_user_message​

Record an inbound user message. Always lands at position=0.

ctx.record_user_message("What's the weather?")
ParameterTypeDefaultDescription
contentstr""The user's message text
messagestr""Alias for content; content wins if both set
metadatadict | NoneNoneOptional extra context (channel, user_id, ...)

record_agent_response​

Record an outbound agent response (distinct from the raw LLM call payload).

ctx.record_agent_response("It's sunny in Paris today.")
ParameterTypeDefaultDescription
contentstr(required)The agent's response text shown to the user
metadatadict | NoneNoneOptional extra context (citations, confidence, ...)

record_human_interaction​

Record a completed human-in-the-loop interaction as a single IO span. Use this when you already have the prompt, response, and timing — for the streaming / context-manager variants use ctx.input(...) or ctx.human_turn(...) (below).

ctx.record_human_interaction(
    prompt="Approve refund?",
    response="yes",
    channel="slack",
    action="approval",
    elapsed_ms=12500,
)
ParameterTypeDefaultDescription
promptstr""What was shown to the human
responsestr""What the human replied
channelstr"terminal"Where it happened ("terminal", "slack", "ui", "webhook", ...)
actionstr""Interaction kind ("confirmation", "input", "approval", ...)
elapsed_msfloat | int | NoneNoneHow long the human took to respond
metadatadict | NoneNoneArbitrary extra context

ctx.input​

Drop-in for the built-in input() — auto-captures prompt + response + wait time as a human_turn span. Strips ANSI escape codes before recording.

answer = ctx.input("Approve? (y/n): ")

ctx.human_turn​

Context manager for non-terminal channels (Slack, webhooks, UI dialogs) where the wait shape varies:

with ctx.human_turn(prompt="Review PR", channel="github") as turn:
    result = wait_for_webhook()
    turn.set_response(result)

record_communication​

Record an outbound communication (Slack, email, SMS, etc.) for communication governance policies (allowed channels, message limits, disclaimer rules).

ctx.record_communication(
    channel="slack",
    recipient="#general",
    body="Deploy completed",
)
ParameterTypeDefaultDescription
channelstr(required)"slack", "email", "sms", ...
recipientstr""Target ("#general", "user@acme.com", ...)
bodystr""Message body
subjectstr""Subject line (email, etc.)
metadatadict | NoneNoneExtra context

HITL Pause​

ctx.pause​

Mark a human-in-the-loop pause — wrap any blocking wait so the trace renders a ⏸ pause marker with wait duration + reason. Framework-agnostic.

with ctx.pause(reason="awaiting_approval"):
    decision = wait_for_human()   # blocks

reason mirrors runtime PausedReason values: awaiting_user, awaiting_approval, awaiting_signal, awaiting_timer, awaiting_child.

ctx.mark_resumed​

Mark that this run resumed after a cross-process pause (e.g. a LangGraph interrupt() answered by a separate resume invocation). Emits a post_resume phase marker.

ctx.mark_resumed(from_run_id="run_abc123")

Memory​

Record agent memory reads and writes — emits kind=memory spans and feeds memory governance + the run's Memory tab + memory-state replay.

remember_episodic​

Write an episodic memory (named-slot KV).

ctx.remember_episodic(slot_name="deal_findings", data={"company": "Acme"})
ParameterTypeDefaultDescription
slot_namestr(required)Named slot (e.g. "deal_findings")
dataAny(required)Value to store (dict / list / scalar)
scope_keystr | NoneNoneDefaults to agent:session
ttl_secondsint86400TTL on the Memory tab write
max_itemsint | NoneNoneCap on items in the slot
memory_typestr | NoneNoneOptional sub-classification

remember_semantic​

Write a semantic fact (free-text, embedded for similarity search).

ctx.remember_semantic(
    slot_name="customer_preferences",
    content="Acme prefers monthly invoicing in EUR.",
)
ParameterTypeDefaultDescription
slot_namestr(required)Named slot
contentstr(required)Fact text
scope_keystr | NoneNoneDefaults to agent:session
importancefloat0.70–1 weight
tagslist | NoneNoneSearchable tags
fact_keystr | NoneNoneStable de-dup key
source_toolstr""Where the fact came from
content_embeddinglist | NoneNonePre-computed embedding (falls back to OpenAI text-embedding-3-small if OPENAI_API_KEY is set)

record_memory_recall​

Record a memory read (search/lookup) — the read side of the memory signal, analogous to grounding. Surfaces hit-rate + relevance per turn.

ctx.record_memory_recall(
    query="What did Acme say about pricing?",
    tier="episodic",
    store="mem0",
    results_count=3,
    top_score=0.84,
    results=hits,
)
ParameterTypeDefaultDescription
querystr""Recall query
tierstr"episodic""episodic" or "semantic"
storestr""Memory store name ("mem0", "zep", "letta", ...)
results_countint00 ⇒ miss; >0 ⇒ hit
top_scorefloat | NoneNoneBest match relevance (0–1)
slot_namestr""Named slot if applicable
resultslist | NoneNoneRecalled items — bounded-captured into the memory snapshot for replay

Prompt Lineage​

record_prompt_use​

Stamp the run with the registry prompt name + version it used. This is the lineage link that ties eval + replay to the exact prompt version. get_prompt calls this automatically; use it directly when you fetch a prompt some other way.

ctx.record_prompt_use(name="customer_intake", version=7, label="prod")
ParameterTypeDefaultDescription
namestr(required)Registry prompt name
versionint0Pinned version (0 means latest at fetch time)
labelstr""Label resolved against ("prod", "staging")
content_hashstr""Content hash of the rendered prompt

Approval Lifecycle​

record_approval_request​

Record that an approval workflow was initiated after a policy block. Call after catching PolicyViolationError when the policy metadata indicates approval is required.

ctx.record_approval_request(
    action_type="delete",
    approvers=["security@acme.com"],
    timeout_minutes=30,
    reason="Mass deletion above 100 records",
)
ParameterTypeDefaultDescription
action_typestr(required)Action awaiting approval ("delete", "refund", ...)
approverslist | NoneNoneEmails / group names
timeout_minutesfloat | NoneNoneApproval window
reasonstr""Why approval is needed
metadatadict | NoneNoneExtra context

record_approval_response​

Record the outcome of an approval request. Call after the human decides or the timeout expires.

ctx.record_approval_response(
    action_type="delete",
    decision="approved",
    approver="security@acme.com",
    elapsed_seconds=420,
)
ParameterTypeDefaultDescription
action_typestr(required)Action that was awaiting approval
decisionstr(required)"approved", "denied", or "timeout"
approverstr""Who decided
elapsed_secondsfloat | NoneNoneTime from block to decision
metadatadict | NoneNoneExtra context

Governance Signals​

Buffer state the controlplane's policy handlers read from conversation_state. All are additive — call as the agent works.

record_data_access​

ctx.record_data_access(source="postgres", operation="read", records=42)
ParameterTypeDefaultDescription
sourcestr(required)Data source ("postgres", "s3", "redis", ...)
operationstr"read""read" or "write"
recordsint0Number of records accessed

record_network_request​

ctx.record_network_request(url="https://api.stripe.com/v1/charges")
ParameterTypeDefaultDescription
urlstr(required)URL or domain accessed

record_scope_impact​

Running totals of the run's blast radius. Each call increments.

ctx.record_scope_impact(records_modified=10, transaction_total=499.99)
ParameterTypeDefaultDescription
records_modifiedint0Records updated
records_deletedint0Records deleted
files_changedint0Files written
transaction_totalfloat0.0Dollar value of transactions
api_writesint0External API write operations

Incremental Flush​

For long-running contexts (REPLs, chat sessions, batch loops) where you want buffered data visible in the UI — and governance evaluated — before the context exits.

# async
await ctx.flush()

# sync
ctx.flush_sync()

Each call sends only data buffered since the last flush. Safe to call any number of times. Both raise PolicyViolationError if a governance-checked send (LLM calls, steps) trips a block policy mid-run.

Properties​

PropertyTypeDescription
run_idstrThe run ID from the control plane, or "" if the run has not started

Error Handling​

If an exception occurs inside the context, the run is automatically completed with status="error" and the error message. The exception is not suppressed -- it propagates normally:

# Async
try:
    async with WaxellContext(agent_name="my-agent") as ctx:
        raise ValueError("Something went wrong")
except ValueError:
    pass  # Run was completed with status="error"

# Sync
try:
    with WaxellContext(agent_name="my-agent") as ctx:
        raise ValueError("Something went wrong")
except ValueError:
    pass  # Run was completed with status="error"

If flushing telemetry to the control plane fails (e.g., network error), the failure is logged as a warning but does not interfere with your agent's execution.

Policy Enforcement on Entry​

When enforce_policy=True, policies are checked before the run starts. If the policy result is block or throttle, a PolicyViolationError is raised and no run is created:

# Canonical import — also exported from waxell_observe.errors
from waxell_observe import PolicyViolationError

# Works identically with both async and sync context managers
try:
    with WaxellContext(
        agent_name="my-agent",
        enforce_policy=True,
    ) as ctx:
        ...
except PolicyViolationError as e:
    print(f"Blocked: {e}")
    print(f"Action: {e.policy_result.action}")

When to Use Context Manager vs Decorator​

Choose WaxellContext over @observe when you need:

  • Multi-step orchestration -- wrap complex logic that spans multiple functions
  • Mid-execution policy checks -- re-validate policies between steps
  • Explicit input/metadata control -- pass structured inputs and metadata at context creation
  • Multiple runs in one function -- start and complete several runs in sequence
  • Conditional observability -- only create a context under certain conditions
  • Synchronous code -- batch scripts, CLI tools, and ETL pipelines that don't use async

Example of multiple runs (sync -- natural fit for batch processing):

def batch_process(items: list[str]):
    for item in items:
        with WaxellContext(
            agent_name="batch-processor",
            inputs={"item": item},
        ) as ctx:
            result = process_item(item)
            ctx.record_llm_call(model="gpt-4o-mini", tokens_in=50, tokens_out=30)
            ctx.set_result({"output": result})

The same pattern works with async with for async code:

async def batch_process(items: list[str]):
    for item in items:
        async with WaxellContext(
            agent_name="batch-processor",
            inputs={"item": item},
        ) as ctx:
            result = await process_item(item)
            ctx.record_llm_call(model="gpt-4o-mini", tokens_in=50, tokens_out=30)
            ctx.set_result({"output": result})

Full Example (Async)​

from waxell_observe import WaxellObserveClient, WaxellContext

WaxellObserveClient.configure(
    api_url="https://acme.waxell.dev",
    api_key="wax_sk_...",
)

async def run_pipeline(query: str) -> dict:
    async with WaxellContext(
        agent_name="research-pipeline",
        workflow_name="deep-research",
        inputs={"query": query},
        metadata={"version": "2.1"},
        enforce_policy=True,
    ) as ctx:
        # Step 1: Search
        sources = await search(query)
        ctx.record_step("search", output={"source_count": len(sources)})

        # Step 2: Synthesize
        synthesis = await synthesize(query, sources)
        ctx.record_llm_call(
            model="claude-sonnet-4",
            tokens_in=2000,
            tokens_out=500,
            task="synthesize",
        )
        ctx.record_step("synthesize", output={"length": len(synthesis)})

        # Mid-execution policy check
        policy = await ctx.check_policy()
        if policy.blocked:
            ctx.set_result({"error": "Policy blocked continuation"})
            return {"error": policy.reason}

        # Step 3: Refine
        final = await refine(synthesis)
        ctx.record_llm_call(
            model="gpt-4o",
            tokens_in=800,
            tokens_out=300,
            task="refine",
        )
        ctx.record_step("refine")

        result = {"answer": final, "sources": len(sources)}
        ctx.set_result(result)
        return result

Full Example (Sync)​

from waxell_observe import WaxellObserveClient, WaxellContext

WaxellObserveClient.configure(
    api_url="https://acme.waxell.dev",
    api_key="wax_sk_...",
)

def process_tickets(tickets: list[dict]) -> list[dict]:
    results = []
    for ticket in tickets:
        with WaxellContext(
            agent_name="ticket-processor",
            workflow_name="support-pipeline",
            inputs={"ticket_id": ticket["id"], "subject": ticket["subject"]},
            enforce_policy=True,
        ) as ctx:
            ctx.set_tag("priority", ticket["priority"])

            # Step 1: Classify
            category = classify_ticket(ticket)
            ctx.record_llm_call(model="gpt-4o-mini", tokens_in=200, tokens_out=10, task="classify")
            ctx.record_step("classify", output={"category": category})

            # Step 2: Generate response
            response = generate_response(ticket, category)
            ctx.record_llm_call(model="gpt-4o", tokens_in=500, tokens_out=200, task="respond")
            ctx.record_step("respond", output={"length": len(response)})

            # Mid-execution policy check (sync variant)
            policy = ctx.check_policy_sync()
            if policy.blocked:
                ctx.set_result({"error": policy.reason})
                results.append({"ticket_id": ticket["id"], "error": policy.reason})
                continue

            ctx.record_score("response_quality", 0.9)
            result = {"ticket_id": ticket["id"], "category": category, "response": response}
            ctx.set_result(result)
            results.append(result)

    return results

Conversation State​

WaxellContext automatically tracks conversation metrics from LLM calls:

  • ctx.conversation_turns — number of user turns in the conversation
  • ctx.context_utilization — context window usage as a percentage (0-100%)
  • ctx.message_count — total messages in the LLM context

These properties are read-only and updated automatically when auto-instrumentation records LLM calls.

Manual Recording​

For agents not using auto-instrumented LLM providers, drive the counters explicitly with ctx.record_user_message(...) / ctx.record_agent_response(...) (see Conversation & Human Interaction above). These methods create IO spans that appear in the trace timeline alongside LLM calls and tool invocations. See Conversation Tracking for full details.

Module-level Helpers​

Every method above has a module-level shorthand that operates on the currently active WaxellContext (resolved via ContextVar) — no-op outside a run. Use these from deeply-nested tool bodies, callbacks, or third-party adapters that the decorator can't pass ctx to.

import waxell_observe as waxell

# Access the active context (returns None outside a run)
ctx = waxell.get_current_context()   # or: waxell.get_context()

# Recording — no ctx argument needed
waxell.score("relevance", 0.92)
waxell.tag("environment", "prod")
waxell.metadata("retrieval_count", 5)
waxell.step("classify", output={"category": "billing"})
waxell.decide("route", chosen="research", options=["direct", "research"])
waxell.reason("evaluate", thought="Source B has higher authority")
waxell.retrieve(query=q, documents=docs, source="pinecone")
waxell.retry(attempt=2, reason="Rate limited", strategy="fallback")

# Memory + prompt lineage
waxell.remember_episodic("deal_findings", data={"company": "Acme"})
waxell.remember_semantic("preferences", "Acme prefers monthly invoicing.")
waxell.recall(query=q, tier="episodic", results_count=3, top_score=0.84)
waxell.prompt_use("customer_intake", version=7, label="prod")

# Conversation + HITL
waxell.user_message("What's the weather?")
waxell.agent_response("Sunny in Paris.")
waxell.communication(channel="slack", recipient="#ops", body="deploy ok")
answer = waxell.input("Approve? (y/n): ")        # drop-in for input()
with waxell.human_turn(prompt="Review PR", channel="github") as turn:
    turn.set_response(wait_for_webhook())
with waxell.pause(reason="awaiting_approval"):
    decision = wait_for_human()

# Approval lifecycle
waxell.approval_request(action_type="delete", approvers=["sec@acme.com"])
waxell.approval_response(action_type="delete", decision="approved")

# Incremental flush
await waxell.flush()       # async
waxell.flush_sync()        # sync

Span Cap (long-running agents)​

A long-running or autonomous agent can emit unboundedly many spans. To prevent the in-process buffer from ballooning, each run is capped at 50 000 spans (root + first 49 999 — the tail is dropped, a WARNING is logged, and the flushed run carries a _span_cap_hit summary so truncation is visible).

Tune via env var:

export WAXELL_MAX_SPANS_PER_RUN=100000   # raise the cap
export WAXELL_MAX_SPANS_PER_RUN=0        # disable the cap entirely

Typical batch loops won't hit this. If you do, prefer wrapping each iteration in its own WaxellContext so each batch item is a discrete run with its own budget — see the batch examples above.

Next Steps​