Metadata-Version: 2.4
Name: intencion
Version: 0.7.0
Summary: Durable, dependency-free Python SDK for capturing AI-agent runs and shipping them to Intencion.
Project-URL: Homepage, https://intencion.io
Project-URL: Repository, https://github.com/las7/intent-analysis
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
```

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

```python
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("...")

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

```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 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. To group a multi-call task into one run tree, wrap it (see [Traces](#traces)).

## Capturing content (output + tool I/O)

By default the SDK captures **metadata only**, capture is private by default. Opt in to content with `capture_content=True`:

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

With it on, auto-instrumented model calls fold their **reply text** onto the run as `output_text`, and `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=...)`. All captured content is run through redaction before send. (Streamed responses capture metadata only.)

Redaction catches the built-in PII patterns (emails, cards, SSNs, phone numbers). Model outputs and tool returns can carry other sensitive data (names, account details, secrets), so when you enable content capture, 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 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:

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

```python
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; 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 conversation (grouped by time 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` | `False`                          | Also capture model output text + `run.tool()` return values (redacted). |
| `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, 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)

# 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
