Python SDK

The armoriq_sdk package for Python agents. ArmorIQClient, for_user, from_config, sessions, exception handling, and the plan-token-invoke loop.

The Python SDK ships as armoriq-sdk on PyPI. One package gives you the CLI (armoriq) and the library (armoriq_sdk).

pipx install armoriq-sdk     # CLI on $PATH
# or
pip install armoriq-sdk      # into your project venv

Public API surface

From armoriq_sdk you import:

from armoriq_sdk import (
    # Core
    ArmorIQClient,
    ArmorIQSession,
    SessionOptions,
    SessionMode,
    EnforceResult,
    ReportOptions,

    # Plan helpers
    build_plan_from_tool_calls,
    default_tool_name_parser,
    hash_tool_calls,

    # Exceptions
    ArmorIQException,
    InvalidTokenException,
    IntentMismatchException,
    MCPInvocationException,
    TokenExpiredException,
    DelegationException,
    ConfigurationException,
    PolicyBlockedException,
    PolicyHoldException,

    # Models
    IntentToken,
    PlanCapture,
    MCPInvocation,
    MCPInvocationResult,
    DelegationRequest,
    DelegationResult,
)

`ArmorIQClient`

The main entry point. One client per service; cheap to hold as a module-level singleton.

Minimum setup

from armoriq_sdk import ArmorIQClient

client = ArmorIQClient()  # reads ~/.armoriq/credentials.json

That's the complete setup after armoriq login. Every other constructor argument is optional and auto-resolves.

Constructor

ArmorIQClient(
    api_key: Optional[str] = None,           # resolves env → credentials file
    user_id: Optional[str] = None,           # auto-defaults to "__sdk_multiuser__"
    agent_id: Optional[str] = None,          # auto-defaults to "__sdk_multiuser__"
    iap_endpoint: Optional[str] = None,      # auto-resolves to platform default
    proxy_endpoint: Optional[str] = None,
    backend_endpoint: Optional[str] = None,
    proxy_endpoints: Optional[Dict[str, str]] = None,  # per-MCP overrides
    context_id: Optional[str] = None,
    timeout: float = 30.0,
    max_retries: int = 3,
    verify_ssl: bool = True,
    use_production: bool = True,
    mcp_credentials: Optional[Mapping] = None,
)

Only api_key really matters - and even that resolves from the env var (ARMORIQ_API_KEY) or the credentials file if you omit it.

`from_config()` - load an armoriq.yaml

For config-as-code teams. Scaffold the YAML with armoriq init, then:

client = ArmorIQClient.from_config("armoriq.yaml")

The YAML drives identity, environment, MCP list, policy, and intent settings. See the YAML reference for every field.

The API key authenticates your service. An email identifies the end user your service is acting on behalf of. Together they drive audit trails, per-user policy enforcement, and delegation flows.

Option A - per-call `user_email`

The simplest path. Pass user_email on every invoke():

client = ArmorIQClient()

plan = client.capture_plan(
    llm="gpt-4o",
    prompt="Book a SFO→JFK flight",
    plan={
        "goal": "Book flight",
        "steps": [
            {"action": "book_flight", "mcp": "travel-mcp", "params": {...}},
        ],
    },
)
token = client.get_intent_token(plan, validity_seconds=300)

result = client.invoke(
    mcp="travel-mcp",
    action="book_flight",
    intent_token=token,
    params={"from": "SFO", "to": "JFK"},
    user_email="alice@customer.com",   # ← end user
)

The audit log row for this invocation is tagged with alice@customer.com; policies that scope access by email (or by the membership that email maps to) are evaluated against her.

Option B - session mode for long-running conversations

For multi-turn chat agents that make many tool calls, use for_user(email) to get a ArmorIQUserScope, then start a session. The session carries the user email across every subsequent call automatically.

client = ArmorIQClient()
alice = client.for_user("alice@customer.com")
session = alice.start_session(SessionOptions(mode="sdk"))

token = session.start_plan([
    {"action": "search_hotels", "mcp": "travel-mcp"},
    {"action": "book_hotel",    "mcp": "travel-mcp"},
])

decision = session.check("search_hotels", {"city": "Paris"})
if decision.allowed:
    ...  # forward to MCP

Option C - pre-auth check without invoking

resolve_user(email) fetches the user's membership, role, applicable policies, and approver chain - without triggering an MCP call. Useful for pre-authorization screens.

context = client.resolve_user("alice@customer.com")
# context = {"user": {...}, "policies": [...], "approverChain": [...]}

Cached 300 seconds per email to keep it cheap.

The plan → token → invoke loop

Every tool call follows the same three-step flow.

  Your agent                 ArmorIQ SDK
      │                           │
      │  capture_plan ──────────► │  structures the plan, computes
      │                           │  canonical hash (Merkle root)
      │                           │
      │  get_intent_token ──────► │  calls IAP with plan hash,
      │                           │  receives signed JWT
      │                           │
      │  invoke ─────────────────►│  routes through the proxy,
      │                           │  proxy verifies token + policy,
      │                           │  forwards to MCP, returns result

`capture_plan(llm, prompt, plan, metadata=None)`

Capture the LLM's reasoning and the planned tool calls:

plan = client.capture_plan(
    llm="gpt-4o",
    prompt="Book flights and hotels for a Paris trip",
    plan={
        "goal": "Plan Paris trip",
        "steps": [
            {"action": "search_flights", "mcp": "travel-mcp", "params": {...}},
            {"action": "book_flight",    "mcp": "travel-mcp", "params": {...}},
            {"action": "search_hotels",  "mcp": "travel-mcp", "params": {...}},
            {"action": "book_hotel",     "mcp": "travel-mcp", "params": {...}},
        ],
        "metadata": {"trip_id": "paris-2026-05"},
    },
)

print(plan.plan_hash)        # canonical hash
print(plan.merkle_root)      # Merkle tree root
print(plan.ordered_paths)    # per-step proofs

Your plan dict must contain steps. The SDK rejects empty or malformed plans - this is what prevents an attacker from substituting a different plan behind a valid token.

`get_intent_token(plan, validity_seconds=300)`

Mint a short-lived signed token bound to the plan:

token = client.get_intent_token(plan, validity_seconds=600)  # 10 min

print(token.token_id)
print(token.expires_at)
print(token.is_expired)
print(token.time_until_expiry)  # seconds

Tokens are JWT-style. Where the proxy performs verification it checks the signature (verification is currently skipped when the token has no key id, kid, so don't treat this as guaranteed on every path), checks the hash against the registered plan, and confirms the token hasn't expired. See the Security Model page for current signature-verification coverage.

`invoke(mcp, action, intent_token, params, user_email=None)`

Actually execute the tool call:

result = client.invoke(
    mcp="travel-mcp",
    action="book_flight",
    intent_token=token,
    params={"from": "SFO", "to": "CDG", "date": "2026-05-10"},
    user_email="alice@customer.com",
)

print(result.status)      # "success" | ...
print(result.data)        # the MCP's response body
print(result.duration_ms)

The call goes to the proxy, which verifies the token, checks policy, forwards to the MCP, and returns the response. params must match the steps in the plan - mismatch raises IntentMismatchException.

`invoke_with_policy()` - with hold / delegation

The richer variant. If the policy returns a hold (requires approval), invoke_with_policy can wait on the delegation and auto-retry once approved:

Options go in an InvokeOptions object, not as direct keyword arguments. The signature is invoke_with_policy(mcp, action, intent_token, params, options):

from armoriq_sdk import InvokeOptions

result = client.invoke_with_policy(
    mcp="travel-mcp",
    action="book_flight",
    intent_token=token,
    params={"amount": 5000},
    options=InvokeOptions(
        user_email="alice@customer.com",
        wait_for_approval=True,      # poll until approved or timeout
        delegation_timeout_ms=300_000,
    ),
)

InvokeOptions fields:

Field	Type	Description
`user_email`	str	End user the call is attributed to
`wait_for_approval`	bool	Poll the delegation until approved, rejected, or timeout
`delegation_timeout_ms`	int	How long to wait before failing closed
`on_hold`	callable	Called with a `HoldInfo` when the call is held
`requester_role`	str	Role asserted for policy evaluation
`requester_limit`	float	Amount limit asserted for policy evaluation

For high-stakes tool calls. See DelegationException and PolicyHoldException below.

Exception handling

Every exception inherits from ArmorIQException. Catch broadly for "something went wrong in ArmorIQ," or narrowly for specific conditions.

Exception	You'll see it when…	What to do
`ConfigurationException`	No API key resolved, or key is malformed	Run `armoriq login` or set `ARMORIQ_API_KEY`
`InvalidTokenException`	Token signature doesn't verify, or token doesn't match the plan	Capture a fresh plan and token
`TokenExpiredException`	Token was valid but `validity_seconds` elapsed	Get a new token with `get_intent_token`
`IntentMismatchException`	Your `invoke()` args don't match the captured plan's steps	Align `params` with the plan, or capture a new plan
`PolicyBlockedException`	The tool isn't in the caller's allow-list	Update policy in the dashboard
`PolicyHoldException`	The tool requires human approval	Use `invoke_with_policy(..., options=InvokeOptions(wait_for_approval=True))`
`DelegationException`	Delegation request failed or was denied	Inspect `exception.delegation_id` and retry
`MCPInvocationException`	The MCP server itself returned an error	Check the MCP server logs; not an ArmorIQ-layer issue

Example:

from armoriq_sdk import (
    ArmorIQClient,
    PolicyBlockedException,
    PolicyHoldException,
    TokenExpiredException,
)

try:
    result = client.invoke(mcp="travel-mcp", action="book_flight",
                           intent_token=token, params={...},
                           user_email="alice@customer.com")
except TokenExpiredException:
    token = client.get_intent_token(plan)
    result = client.invoke(mcp="travel-mcp", action="book_flight",
                           intent_token=token, params={...},
                           user_email="alice@customer.com")
except PolicyBlockedException as e:
    print(f"Blocked: {e.reason}")
except PolicyHoldException as e:
    print(f"Approval required: delegation_id={e.delegation_id}")

Plan helpers

Build plans from tool-call lists instead of writing the dict manually:

from armoriq_sdk import build_plan_from_tool_calls

# Your framework (LangChain, CrewAI, raw OpenAI) produced these tool calls:
tool_calls = [
    {"name": "travel-mcp.book_flight", "arguments": {...}},
    {"name": "travel-mcp.search_hotels", "arguments": {...}},
]

plan = build_plan_from_tool_calls(
    tool_calls=tool_calls,
    goal="Plan Paris trip",
)

default_tool_name_parser and hash_tool_calls are lower-level helpers for custom integrations - see source for details.

Full worked example

from armoriq_sdk import ArmorIQClient, TokenExpiredException

client = ArmorIQClient()

plan = client.capture_plan(
    llm="gpt-4o",
    prompt="What's the weather in Paris?",
    plan={
        "goal": "Weather lookup",
        "steps": [
            {
                "action": "get_weather",
                "mcp": "weather-mcp",
                "params": {"city": "Paris", "units": "celsius"},
            },
        ],
    },
)

token = client.get_intent_token(plan, validity_seconds=300)

try:
    result = client.invoke(
        mcp="weather-mcp",
        action="get_weather",
        intent_token=token,
        params={"city": "Paris", "units": "celsius"},
        user_email="alice@customer.com",
    )
    print(result.data)
finally:
    client.close()