Metadata-Version: 2.4
Name: intencion
Version: 0.5.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("...")/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):

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

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

```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()/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:

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

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

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

```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,
               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](./LICENSE).

https://intencion.io
