Metadata-Version: 2.5
Name: intended
Version: 0.3.0
Summary: Python SDK for Intended — The Authority Runtime for AI Agents (incl. physical-AI helpers)
Project-URL: Homepage, https://intended.so
Project-URL: Documentation, https://intended.so/developers/docs
Project-URL: Repository, https://github.com/intended-so/intended-python
License-Expression: Apache-2.0
Requires-Python: >=3.9
Requires-Dist: httpx>=0.24
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == 'dev'
Description-Content-Type: text/markdown

# intended

Python SDK for Intended — the intent verification runtime for autonomous agents.

Evaluate every AI agent action against your organization's policies before execution. Fail-closed by default: if the authority service is unreachable, all actions are denied.

## Installation

```bash
pip install intended
```

## Quick Start

### Synchronous

```python
from intended import IntendedSdk

client = IntendedSdk(api_key="intended_live_...", tenant_id="tenant-a")

# If the action is escalated, this waits (up to 10 minutes) for a human decision.
decision = client.authorize("deploy to production", actor="ci-agent", system="github")

if decision.allowed:
    print(f"Approved — token: {decision.token}")
elif decision.escalated:
    print(f"Still awaiting human approval: intent {decision.intent_id}")
else:
    print(f"Denied: {decision.reason}")
```

### Asynchronous

```python
from intended.client import AsyncIntendedClient

async with AsyncIntendedClient(api_key="intended_live_...", tenant_id="tenant-a") as client:
    decision = await client.authorize("scale cluster to 100 nodes", actor="auto-scaler")
    if decision.allowed:
        await scale_cluster(100)
```

## MCP Integration

Wrap any MCP tool function with Intended authority checks:

```python
from intended.mcp import IntendedMCPMiddleware

middleware = IntendedMCPMiddleware(api_key="intended_live_...", tenant_id="tenant-a")

@middleware.protect
def execute_sql(query: str) -> str:
    # Only runs if Intended approves
    return db.execute(query)

# Or check manually
decision = middleware.check("execute_sql", {"query": "DROP TABLE users"})
print(decision.decision)  # "DENY"
```

## LangChain Integration

Guard any LangChain tool with Intended:

```python
from intended.langchain import IntendedToolGuard
from langchain.tools import ShellTool

guard = IntendedToolGuard(api_key="intended_live_...", tenant_id="tenant-a")

shell = ShellTool()
safe_shell = guard.wrap(shell)

# Now shell.run() checks with INTENDED before executing
agent = create_react_agent(llm, tools=[safe_shell])
```

## Argument values and policies

The integrations send a tool call as two parts:

- the action text: the tool name and the argument names only, for example
  `Call MCP tool "execute_sql" with args: query` (MCP) or
  `Call LangChain tool "delete_database" with args: target` (LangChain);
- `arguments`: the argument values. Intended evaluates them in memory and
  stores only a sha256 digest.

Argument values are never sent in the action text or in `metadata`. To make a
policy depend on a value, use an `args.<path>` condition (for example
`args.target EQ staging`), not a match on the action text. When you call
`authorize()` directly, pass tool-call values the same way:

```python
client.authorize(
    'Call tool "execute_sql" with args: query',
    actor="my-agent",
    arguments={"query": "select 1"},
)
```

## PydanticAI Integration

Use as a decorator on PydanticAI tool functions:

```python
from intended.pydantic_ai import IntendedPydanticGuard
from pydantic_ai import Agent

guard = IntendedPydanticGuard(api_key="intended_live_...", tenant_id="tenant-a")

agent = Agent('openai:gpt-4o')

@agent.tool
@guard.protect
async def deploy(env: str) -> str:
    return f"Deployed to {env}"
```

## AuthorityDecision

Every call to `authorize()` returns an `AuthorityDecision`:

| Field | Type | Description |
|---|---|---|
| `decision` | `str` | `"ALLOW"`, `"DENY"`, or `"ESCALATE"` |
| `intent_id` | `str` | Unique ID of the evaluated intent |
| `risk_score` | `int` | 0-100 risk assessment |
| `confidence` | `float` | Confidence in the risk assessment |
| `reason` | `str` | Human-readable explanation |
| `token` | `str \| None` | Authority token for approved actions |
| `escalation_id` | `str \| None` | ID for tracking escalated actions |
| `expired` | `bool` | `True` when an escalation expired before an approver decided (`decision` is `"DENY"`) |
| `outcome` | `DecisionOutcome \| None` | The wait result, when `authorize()` waited on an escalation |

Convenience properties: `decision.allowed`, `decision.denied`, `decision.escalated`.

## Escalations: what your agent gets back

When a policy escalates an action to a human approver, `authorize()` waits for
the approver's decision by default and returns the final answer:

| What happened | `authorize()` returns |
|---|---|
| Approver approved | `decision="ALLOW"`, `allowed=True`, `token` set (the single-use token minted at approval), same `intent_id` |
| Approver rejected, or the escalation was canceled | `decision="DENY"`, `reason` from the approver |
| No approver decided before the escalation expired | `decision="DENY"`, `expired=True` |
| Still undecided after `wait_timeout_seconds` | `decision="ESCALATE"`, `reason` says it is still awaiting approval, with `intent_id` and `escalation_id` |

Options on `authorize()` (sync and async):

- `on_escalate="wait"` (default) or `"return"`. `"return"` returns the
  `ESCALATE` decision immediately, as versions before 0.3.0 did.
- `wait_timeout_seconds=600` (10 minutes) bounds the wait.

To keep waiting later, or to wait on an intent from another process, call
`wait_for_decision`:

```python
outcome = client.wait_for_decision(decision.intent_id, timeout_seconds=600)
if outcome.allowed:
    run_action(token=outcome.token)
elif outcome.pending:
    ...  # still undecided; outcome.timed_out is True
else:
    print(outcome.status, outcome.reason)  # DENY or EXPIRED
```

`wait_for_decision` long-polls `GET /v1/decisions/{intent_id}/wait` (needs the
`intent:read` scope) and returns a `DecisionOutcome` with `status` (`ALLOW`,
`DENY`, `PENDING` or `EXPIRED`), `reason`, `decided_by`, `decided_at`,
`token`, `token_expires_at`, `receipt_url`, `escalation_id`, `expires_at`,
`timed_out` and `error`. It retries rate limits (429), server errors and
network errors with backoff until the deadline, and stops at once on 401, 403
or 404 with a `DENY` result whose `error` names the problem.

Treat `EXPIRED` exactly like `DENY` (`outcome.denied` is `True` for both). Only
`ALLOW` permits the action; the SDK never reports `allowed` for anything else.

The framework guards (`IntendedMCPMiddleware`, `IntendedToolGuard`,
`IntendedOpenAIGuard`, `IntendedPydanticGuard`) wait the same way: an approved
tool call runs; anything else raises, and a call that is still pending says it
is awaiting human approval in Intended and gives the intent id. Each guard
accepts `on_escalate` and `wait_timeout_seconds`. The async guards
(PydanticAI, async LangChain tools) wait in a worker thread so the event loop
is not blocked.

## Business Intent APIs

The Python SDK also exposes the business-intent compiler and runtime feedback surfaces:

```python
from intended import IntendedClient

client = IntendedSdk(api_key="intended_live_...", tenant_id="tenant-a", api_url="http://localhost:3101")

compiled = client.compile_business_intent_and_plan(
    "Shift the nightly ETL window to 3am",
    workflow_family="enterprise-operations",
    target_system="airflow",
    metadata={"environment": "production"},
)

print(compiled["prediction"]["canonicalIntent"])
print(compiled["intentObject"]["process"]["stages"])

object_evals = client.get_business_intent_object_evals()
print(object_evals["report"]["summary"]["passedCases"])

feedback = client.capture_business_intent_feedback(
    input_text="Shift the nightly ETL window to 3am",
    predicted_intent="ops.batch.schedule",
    workflow_family="enterprise-operations",
    outcome="accepted",
    confidence=0.93,
    target_system="airflow",
    execution_status="succeeded",
    verification_status="verified",
)

evals = client.get_business_intent_feedback_evals(workflow_family="enterprise-operations")
print(evals["evals"]["verification"]["verificationRate"])

summary = client.get_business_intent_feedback_summary(workflow_family="enterprise-operations")
print(summary["summary"]["approvalPredictionHits"])

report = client.get_business_intent_feedback_report(workflow_family="enterprise-operations")
print(report["report"]["metadataDrift"])
```

Available helper methods:
- `compile_business_intent()`
- `compile_business_intent_and_plan()`
- `capture_business_intent_feedback()`
- `get_business_intent_object_evals()`
- `get_business_intent_feedback_summary()`
- `get_business_intent_feedback_report()`
- `get_business_intent_feedback_evals()`
- `get_business_intent_feedback_backlog()`
- `review_business_intent_feedback()`
- `promote_business_intent_feedback()`

Example review and promote flow:

```python
backlog = client.get_business_intent_feedback_backlog(workflow_family="enterprise-operations")

for record in backlog["backlog"]:
    client.review_business_intent_feedback(record["id"], "approved")

promotion = client.promote_business_intent_feedback("enterprise-operations")
print(promotion["snapshotVersion"])
print(promotion["snapshotPath"])
```

## Fail-Closed Behavior

If the INTENDED API is unreachable or returns an error, the SDK returns a DENY decision with `risk_score=100` and `confidence=0`. This ensures that network issues or misconfigurations never result in unauthorized actions.

## License

Apache-2.0
