Metadata-Version: 2.4
Name: intencion
Version: 0.9.0
Summary: Durable, dependency-free Python SDK for capturing AI-agent runs and shipping them to Intencion.
Project-URL: Homepage, https://intencion.io
Author: Intencion
License: MIT
License-File: LICENSE
Keywords: agent-analytics,agent-outcomes,agents,ai,ai-agents,analytics,intencion,llm,observability,product-analytics,telemetry
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.8
Description-Content-Type: text/markdown

# intencion

Durable, dependency-free Python SDK for capturing AI-agent runs and shipping them to [Intencion](https://intencion.io). Pure stdlib, Python 3.8+, non-blocking background transport.

## Install

```bash
pip install intencion
```

## Quickstart: the whole integration is one line

Wrap your model client once. From then on every call is captured automatically, with the model, token usage, latency, and outcome filled in for you, and the intent inferred on the server:

```python
from openai import OpenAI
import intencion

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

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

That is the entire setup: an API key plus one line. Anthropic and Gemini work the same way with `instrument_anthropic` and `instrument_gemini`, and each call's intent is inferred on the server (for example `order_status`). See [Auto-instrumentation](#auto-instrumentation-zero-per-call-code) for streaming, framework-built clients, and pinning a fixed intent.

### Let your AI assistant wire it up

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.

## Concepts: run, step, session, trace

A **run** is one unit of agent work: one goal, with an input, an outcome, and the steps taken to get there. A **step** is one action inside a run, a single model call or tool call. Runs and steps nest: the run is the goal, its steps are the moves.

A **session** and a **trace** are two independent ways to group runs. A session gathers the runs of one continuous span of agent work — a chat, a long task, a background job — by session id and (optional) user. A trace gathers the runs of one task into a causal tree, a parent run with its sub-agent runs underneath. They sit on separate axes, so a run carries a session and a trace at once, and one session can hold many traces.

For a single request handled in one unit of work, one run and its steps are the whole picture. Sessions and traces start to matter once a session spans many runs or a task fans out into sub-agent runs. See [Sessions](#sessions) and [Traces](#traces).

## Group work into runs (for richer signal)

Auto-instrumentation captures one run per model call. To group an agent's work into a meaningful unit (one user turn, one task), wrap it in `intencion.run(...)`. Model calls inside the block fold in as steps on that run. **You don't name the intent.** Leave it off and Intencion infers a real, per-run label from the input:

```python
import intencion

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

with intencion.run(input=user_msg, user="u_123",     # intent inferred ("auto")
                   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("...")

# decorator form
@intencion.trace(capture_input=True)
def classify(msg): ...

result = intencion.flush()                    # force send queued runs
if result.dropped:                            # confirm it landed, don't assume
    ...                                       # something is misconfigured
```

Pass `intent=` explicitly only when you want deterministic grouping under a label you control (e.g. `intent="checkout"`). Don't hardcode a single constant like `intent="chat-turn"` on every run, since that buckets distinct work under one label; leaving it as `"auto"` gives a real intent per run. If the wrapped code raises, the run is recorded as `failure` and the exception is re-raised unchanged.

Runs are batched and sent on a background worker, so you normally never touch the transport. Two things make delivery observable instead of silent: **`flush()` returns a `FlushResult` (`sent` / `dropped` / `queued`)** so you can confirm telemetry actually reached the server, and a **rejected API key (or other auth/config error) prints a one-time warning** even with `debug=False`, so a bad key never drops every run quietly. (Using capture before `init()` warns once too.)

## 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):

```python
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. Caught tool errors are a reliability signal, not an outcome.** If the block exits normally but a `run.tool()` step errored (and you recovered), the error is recorded on the step and surfaced as a reliability signal; the run stays `success` on the goal axis. To treat an errored step as a failure, opt in with a resolver: `classify_outcome=lambda run: "failure" if run.has_errored_steps else None`.

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

```python
intencion.init(
    api_key="in_pk_...",
    classify_outcome=lambda run: "failure" if not run.steps
                     else "failure" 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:

```python
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()` → `confirm_outcome` → `classify_outcome` → return/raise default (a returned run is `success`).

**5. Built-in heuristics: label the common failures with no code.** `confirm_outcome_from_heuristics()` returns a ready-made `confirm_outcome` with stable reason codes. By default it flags `empty_output` (an empty/whitespace string answer) and `no_results` (an empty collection like `[]` or `{"hits": []}`). Still deterministic, still no judge model:

```python
import intencion
from intencion import confirm_outcome_from_heuristics

intencion.init(api_key="in_pk_...", confirm_outcome=confirm_outcome_from_heuristics())
```

Not every run is a conversational answer, so the answer-shaped checks are conservative: refusal detection is opt-in (`refusal_phrases=True`, or pass your own list), and `missing_output=True` also flags a run that returned nothing at all. Scope it to the intents that are conversational with `intents=["support", "chat"]` (or a `skip` callable), point `get_text` at a nested answer field when the result isn't a plain string (e.g. `lambda r: r["message"]["content"]`), and set `results_keys` for your own empty-collection keys.

**Labeling failures.** Any resolver can return `{"outcome": ..., "reason": ...}` instead of a bare outcome string; on a `"failure"` the `reason` becomes the run's `failure_reason`, so failures group by mode on the dashboard:

```python
confirm_outcome=lambda run: {"outcome": "failure", "reason": "no_results"}
    if isinstance(run.last_result, list) and len(run.last_result) == 0 else None,
```

## 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:

```python
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.

```python
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:

```python
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 auto-instrumented** (roadmap): stacks that don't call the official SDK — the Vercel AI SDK, CrewAI/LiteLLM, and raw `boto3` Bedrock. Capture these today by wrapping calls in `intencion.run()` and recording steps with `run.tool()`; for stacks that emit OpenTelemetry (e.g. the Vercel AI SDK) you can alternatively stream their OTel export into the OpenTelemetry ingest endpoint. To group a multi-call task into one run tree, wrap it (see [Traces](#traces)).

## Capturing content (output, tool I/O, request context)

By default the SDK captures **full content** so a failing run is diagnosable in-product: the prompt and **reply text**, reconstructed tool-call `input`/`output`, and the **request context** the model ran with — its **system prompt**, **tool definitions**, and **sampling params** (temperature, max_tokens, …). All content is run through PII + secret redaction before send. Switch to metadata-only capture with `capture_content=False`:

```python
intencion.init(api_key="in_pk_...", capture_content=False)
```

With content capture on (the default), auto-instrumented model calls fold their reply text onto the run as `output_text` (streamed replies are reassembled from chunks), tool calls become steps with their `input`/`output`, and the system prompt, tool definitions, and request params are stored on the run. `run.tool(name, kind, fn)` records the tool's **return value** as the step's `output`. You can always attach step I/O explicitly: `run.step(name=..., input=..., output=...)`.

Redaction catches the built-in PII + secret patterns (emails, cards, SSNs, phone numbers, API keys, tokens). Captured content can carry other sensitive data (names, account details), so supply a `redactor` or `redact_patterns` to scrub what's specific to your domain.

## Backfill existing logs (no re-instrumentation)

Already have OpenAI / Anthropic logs on disk or in a warehouse? Import them as runs without touching your app, so you see value before changing any code. Each log becomes one run: input, model, tokens, and tool-call steps are extracted with the **same parser** live capture uses:

```python
# one log dict, or a list. `request` is the body you sent; `response` is what you got back.
intencion.import_openai([
    {"request": saved_request, "response": saved_response,
     "intent": "support", "id": log_id, "user": user, "session": session},
])
intencion.import_anthropic({"request": request, "response": response, "id": log_id})

# provider-agnostic (CSV/JSONL/your own shape):
intencion.import_runs([
    {"intent": "checkout", "input": msg, "model": "gpt-4o", "outcome": "success", "steps": [...]},
])

intencion.flush()
```

Put a stable `id` on a record (the provider's `chatcmpl-…` id or your own conversation id) and re-importing is idempotent: the id is the server's dedupe key. Import bypasses sampling (an explicit backfill is always captured) and is redacted like any other run.

## Custom redaction

Redaction is on by default (`redact=True`). To plug in your own scrubbing, pass a `redactor` (fully replaces the built-in email/card/SSN/phone patterns) or `redact_patterns` (extra `(pattern, replacement)` applied after the built-ins). Preview exactly what would be removed before sending real traffic:

```python
intencion.init(api_key="in_pk_...", redact_patterns=[(r"\bacct_\w+\b", "<ACCT>")])

intencion.preview_redaction("ref acct_777 for jane@example.com")
# {"redacted": "ref <ACCT> for <EMAIL>", "matches": [{"value": "acct_777", "replacement": "<ACCT>"}, ...]}
```

## Sessions

Tie a whole session together — one continuous span of agent work (a chat, a long task, a background job). 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:

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

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

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

### Example: a multi-turn chat agent

Sessions aren't chat-specific, but a chat agent is the clearest example. Wrap the **session** in `intencion.session(...)` and each **turn** in `intencion.agent_turn(...)`. 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. No per-turn intent: each turn is labeled automatically from its input, so distinct asks land as distinct intents instead of one bucket:

```python
def handle_turn(conversation_id: str, user_id: str, message: str) -> str:
    with intencion.session(conversation_id, user=user_id):
        with intencion.agent_turn(input=message) as run:   # intent inferred per turn
            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; error recorded on the step
```

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

## Traces

A **trace** groups the runs of one task into a causal tree. A multi-agent task reads as a parent run with its sub-agent runs nested underneath, so a failure pins to the exact sub-agent that caused it. A session is one continuous span of work (grouped by session id and user); a trace is one task (grouped by cause and effect). The two are independent axes.

Nesting is automatic. An `intencion.run(...)` opened inside another becomes its child: the two share a `trace_id`, and the child carries the parent's id as `parent_run_id`.

```python
with intencion.run(intent="research_task"):         # the task: a trace root
    with intencion.run(intent="search"):
        ...                                           # a sub-agent run
    with intencion.run(intent="summarize"):
        ...                                           # another sub-agent run
```

Auto-instrumented model calls inside a run stay **steps** on that run; only an explicit `intencion.run(...)` becomes a child run. To group sibling runs that aren't nested, wrap them in `intencion.task(...)` (named `task` because `intencion.trace` is the run decorator):

```python
with intencion.task():                   # both runs below are roots of one trace
    with intencion.run(intent="plan"):
        ...
    with intencion.run(intent="act"):
        ...

# or join a trace propagated across a process boundary:
with intencion.task(incoming_trace_id):
    with intencion.run(intent="step"):
        ...
```

A run with no enclosing trace and no nesting stands alone, exactly as before. OpenTelemetry exporters get the same shape automatically: a trace whose spans carry agent/chain boundaries (an `intencion.intent`, an OpenInference `AGENT`/`CHAIN` span, or `intencion.run_boundary`) splits into the matching run tree. Traces show up under **Traces** in the dashboard.

## 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:

```python
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.   |
| `redact`         | `True`                            | Scrub PII (emails, cards, SSNs, phones) before send. |
| `redactor`       | `None`                            | Custom `str -> str` that fully replaces the built-in PII patterns. |
| `redact_patterns` | `None`                           | Extra `[(pattern, replacement), ...]` applied after the built-ins. |
| `capture_content` | `True`                           | Capture content (redacted): reply text, tool I/O, system prompt, tool definitions, request params. Set `False` for metadata-only. |
| `confirm_outcome` | `None`                           | Goal-level resolver `lambda run: Outcome \| {"outcome", "reason"} \| None`; sees `run.last_result` / `run.produced_output` (set via `run.set_result(...)`). Runs before `classify_outcome`. A `reason` becomes the run's `failure_reason`. |
| `classify_outcome` | `None`                          | Structural resolver returning an outcome string, a `{"outcome", "reason"}` dict, or `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`:

```python
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

```python
intencion.init(api_key, endpoint=None, flush_interval=5.0, max_batch=100,
               max_queue=1000, sample_rate=1.0, disabled=False, debug=False,
               redact=True, redactor=None, redact_patterns=None,
               confirm_outcome=None, classify_outcome=None)

# Build a deterministic confirm_outcome (see Outcomes)
intencion.confirm_outcome_from_heuristics(empty_output=True, missing_output=False,
               no_results=True, refusal_phrases=False, outcome_for_refusal="failure",
               get_text=None, intents=None, skip=None)

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

intencion.trace(intent=None, 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)

# Backfill existing logs (no app changes); see "Backfill existing logs"
intencion.import_openai(logs)        # one dict or a list of {"request","response",...}
intencion.import_anthropic(logs)
intencion.import_runs(records)        # provider-agnostic {"intent","input","steps",...}

# Dry-run what redaction would scrub before sending
intencion.preview_redaction(text)     # {"redacted": ..., "matches": [...]}

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)`, `set_tokens(tokens_in, tokens_out)`, `set_model(model)`, and the `has_errored_steps` property.

## License

MIT. See [LICENSE](./LICENSE).

https://intencion.io
