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

pip install intended

Quick Start

Synchronous

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

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:

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:

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:

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:

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:

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:

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:

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