intencion 0.5.0

Durable, dependency-free Python SDK for capturing AI-agent runs and shipping them to Intencion.

intencion

Durable, dependency-free Python SDK for capturing AI-agent runs and shipping them to Intencion. Pure stdlib, Python 3.8+, non-blocking background transport.

Install

pip install intencion

Instrument with your AI assistant

The fastest path: point your editor's AI (Claude, Cursor, …) at this README plus your agent file and ask:

Instrument this agent with the intencion package: init() once, auto-instrument the model client, wrap each user turn in intencion.run and the whole conversation in intencion.session(conversation_id, user=user_id), record tool calls with run.tool, and flush() before the process exits. Keep the diff minimal.

Everything below is enough context for that to one-shot.

Quickstart

import intencion

intencion.init(api_key="in_pk_...")           # call once at startup

with intencion.run(intent="support", input=user_msg, user="u_123",
                   session="s_1", model="gpt-4o") as run:
    run.step(name="lookup_order", tool="db", status="success", ms=42)
    result = my_agent(user_msg)               # your agent work
    # outcome defaults to "success"; override with run.fail("...")/run.abandon()

# decorator form
@intencion.trace(intent="classify")
def classify(msg): ...

intencion.flush()                             # force send queued runs

If the wrapped code raises, the run is recorded as failure and the exception is re-raised unchanged.

Outcomes

Outcomes are deterministic — no model judges success. A run(...) block that exits normally is success; one that raises is failure. But agents usually catch their errors and reply anyway, so "the block returned" is not "the user was helped." Three things stop failures from being silently counted as success:

1. run.tool(...) — record a tool call without forgetting its status. It times the call, marks the step success or (on raise) error with the message, and returns the value (or re-raises):

with intencion.run(intent="refund_request", input=msg) as run:
    order = run.tool("lookup_order", "orders-db", lambda: lookup_order(oid))
    refund = run.tool("issue_refund", "payments", lambda: issue_refund(order))
    # the tool kind is optional: run.tool("lookup_order", lambda: lookup_order(oid))

2. degraded — inferred from errored steps. If the block exits normally but a step errored (and you didn't set an outcome), the run is recorded degraded, not success, so a caught tool error always shows up. Disable with init(..., infer_outcome_from_steps=False). An explicit run.ok() still wins (use it when the agent recovered).

3. Declarative classification. Centralize outcome logic with a global classify_outcome resolver instead of scattering run.fail() calls:

intencion.init(
    api_key="in_pk_...",
    classify_outcome=lambda run: "abandoned" if not run.steps
                     else "degraded" if run.has_errored_steps else None,
)

4. confirm_outcome — close the "successful read still didn't help" gap. A global resolver asked "was the user's goal actually met?". Unlike classify_outcome (structural), it can inspect the run's business result, which you feed in with run.set_result(...) (the context-manager API has no return value), so you can downgrade a run whose tools all succeeded but whose result was empty, deterministically, with no judge model:

intencion.init(
    api_key="in_pk_...",
    # a search run that returned zero hits didn't meet the goal
    confirm_outcome=lambda run: "failure"
        if isinstance(run.last_result, dict) and run.last_result.get("hits") == [] else None,
)

with intencion.run(intent="search") as run:
    hits = run.tool("query", "search-index", lambda: search(q))
    run.set_result({"hits": hits})   # confirm_outcome sees run.last_result

Precedence: explicit ok()/fail()/abandon() → confirm_outcome → classify_outcome → degraded from errored steps → return/raise default.

Auto-instrumentation (zero per-call code)

Wrap your OpenAI or Anthropic client once and every model call is captured automatically — model, token usage, latency, and outcome — with no run.step(...) calls:

from openai import OpenAI
import intencion

intencion.init(api_key="in_pk_...")
client = intencion.instrument_openai(OpenAI())   # the whole integration

# Just use the client. A run shows up in Intencion for every call.
client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "where is my order?"}],
)

• Calls made inside an intencion.run(...) block become steps on that run, and their model + token usage are folded into it.
• Calls made outside a run emit a standalone one-call run. Its intent defaults to "auto", which the server infers into a real label (e.g. order_status) from the input.
• Sync, async (AsyncOpenAI / AsyncAnthropic), and streaming calls are all supported; iteration is transparent.

client = intencion.instrument_anthropic(Anthropic())
# Pin a fixed intent, or skip prompt capture:
client = intencion.instrument_openai(OpenAI(), intent="support", capture_input=False)

Patching is at the class level, so it covers every client instance — including the ones agent frameworks (LangChain, the OpenAI Agents SDK, LlamaIndex, Instructor) build internally. You can pass a client, or call with no argument to patch the installed package directly:

intencion.instrument_openai()       # patches the openai package (covers framework-built clients)
intencion.instrument_anthropic()    # patches the anthropic package

It instruments create, parse (structured outputs), and the stream() helper, across sync/async. instrument_* is idempotent and never raises; enable debug=True logging to see which methods were patched (it warns loudly if it found nothing — so a miss isn't silent).

For streamed OpenAI chat completions, the call is always captured, but token counts arrive only if you pass stream_options={"include_usage": True} (an OpenAI requirement). Anthropic streaming and OpenAI Responses streaming capture tokens with no extra flag.

Gemini is covered too: intencion.instrument_gemini(client) patches google-genai's models.generate_content / generate_content_stream (sync + async) at the class level.

Not yet covered natively (roadmap): stacks that don't call the official SDK — the Vercel AI SDK (enable its OpenTelemetry export to stream into the OpenTelemetry ingest endpoint), CrewAI/LiteLLM, and raw boto3 Bedrock. And without an intencion.run(...) wrapper, a multi-call agent task is recorded as several runs rather than one trace.

Sessions

Tie a whole conversation together. Every run created inside an intencion.session(...) block — an intencion.run(...) or an auto-instrumented call — inherits the session (and optional user) and is grouped by session_id, with no plumbing:

with intencion.session("conv_123", user="u_42"):
    client.chat.completions.create(...)   # session_id = conv_123
    client.chat.completions.create(...)   # same session

# imperative form for request handlers where wrapping a block isn't convenient:
intencion.set_session("conv_123", user="u_42")
intencion.clear_session()

Nested sessions override; an explicit session= or user= on intencion.run(...) still wins over the ambient session.

Multi-turn conversations

For a chat agent, wrap the conversation in intencion.session(...) and each turn in intencion.run(...). The model calls + tool calls inside fold into steps under that one run, so an N-message conversation is N runs grouped by session_id — one run per turn, in order:

def handle_turn(conversation_id: str, user_id: str, message: str) -> str:
    with intencion.session(conversation_id, user=user_id):
        with intencion.run(intent="auto", input=message) as run:
            while True:
                resp = client.messages.create(model=MODEL, tools=tools, messages=messages)  # captured as a step
                if resp.stop_reason != "tool_use":
                    return final_text
                for call in tool_uses(resp):
                    run.tool(call.name, "tool", lambda: exec_tool(call))   # tool step; errored -> degraded

Call handle_turn(...) for each message with the SAME conversation_id.

Short-lived processes

The worker flushes on an interval, on atexit, and on SIGTERM/SIGINT. For a script, a serverless function, or any process that exits quickly, call flush() before the process ends to ensure queued runs are sent:

intencion.flush()      # block until queued runs are sent (or timeout)
intencion.shutdown()   # flush + stop the worker thread

Configuration

Option	Default	Meaning
api_key	— (required)	Sent as Authorization: Bearer <api_key>.
endpoint	https://intencion.io/api/ingest	Ingest URL.
flush_interval	5.0	Seconds between timed flushes.
max_batch	100	Max runs per request (hard-capped at 500).
max_queue	1000	Bounded queue size; drop-oldest when full.
sample_rate	1.0	Fraction of runs captured (0.0 to 1.0).
disabled	False	Disable all capture.
debug	False	Enable debug logging on the intencion logger.
infer_outcome_from_steps	True	Infer degraded when a run exits normally with an errored step.
confirm_outcome	None	Goal-level resolver lambda run: Outcome|None; sees run.last_result / run.produced_output (set via run.set_result(...)). Runs before classify_outcome.
classify_outcome	None	Structural resolver lambda run: "success"|"failure"|"abandoned"|"degraded"|None for un-set outcomes.

To validate capture locally without the real endpoint, point init(endpoint=...) at a tiny local HTTP server and inspect the POSTed { "events": [run, ...] } body. Each run carries intent_label (your intent; stays "auto" if you let the server infer it), session_id, user_ref, steps (with per-step status/error), outcome, tokens_in/out, and latency_ms:

import http.server, json, threading
events = []
class H(http.server.BaseHTTPRequestHandler):
    def do_POST(self):
        body = self.rfile.read(int(self.headers["Content-Length"]))
        events.extend(json.loads(body)["events"])
        self.send_response(200); self.end_headers(); self.wfile.write(b"{}")
    def log_message(self, *a): pass
srv = http.server.HTTPServer(("127.0.0.1", 8799), H)
threading.Thread(target=srv.serve_forever, daemon=True).start()
intencion.init(api_key="test", endpoint="http://127.0.0.1:8799/api/ingest", flush_interval=0.1)
# ... run your agent, then intencion.flush(); assert events[0]["outcome"] == ...

API

intencion.init(api_key, endpoint=None, flush_interval=5.0, max_batch=100,
               max_queue=1000, sample_rate=1.0, disabled=False, debug=False,
               infer_outcome_from_steps=True, classify_outcome=None)

intencion.run(intent, input=None, user=None, session=None, model=None)
# use as a context manager (with statement)

intencion.trace(intent, user=None, session=None, model=None, capture_input=False)
# use as a function decorator

intencion.flush(timeout=None)
intencion.shutdown(timeout=2.0)

# Auto-instrument a provider client — every call is captured automatically
intencion.instrument_openai(client, intent="auto", capture_input=True)
intencion.instrument_anthropic(client, intent="auto", capture_input=True)
intencion.instrument_gemini(client, intent="auto", capture_input=True)

intencion.current_run()   # the run in scope inside a run() block, or None

A run object exposes: step(name, status="success", tool=None, ms=None, error=None), tool(name, tool=None, fn=...) (runs fn, records the step + status, returns its value), ok(), fail(reason=None), abandon(reason=None), set_tokens(tokens_in, tokens_out), set_model(model), and the has_errored_steps property.

License

MIT. See LICENSE.

https://intencion.io