nxtID Python SDK

Python SDK for integrating MetaKeep wallet, AuthID biometric IDV, and nxtlinq AIT on-chain permissions into AI agents, governed at the tool-call level by the nxtID Gateway.

---

Installation

pip install nxtid-sdk

Only httpx is a hard dependency. If your agent is built on FastAPI and you want the .mount(app) HTTP-route adapters (AITRouter, EntraAuth, SessionStore), install the fastapi extra:

pip install "nxtid-sdk[fastapi]"

Not on FastAPI? Skip the extra — every operation is a plain Python method (see Framework-agnostic core below), callable from Flask, Django, a Slack bot, a background worker, a plain script, whatever your agent is built on.

---

Framework-agnostic core

AITRouter, EntraAuth, and SessionStore are each split into two layers:

• Plain methods — wallet_connect, verify_start, verify_complete, permissions_request, permissions_admin_granted, permissions_refresh, session_restore, ait_status, guard_tool_call, login, config, create/get/get_or_create/delete — take a session object and plain Python arguments, return a plain dict, and raise NxtidHTTPError (never a web-framework exception) on failure. No FastAPI import happens anywhere in this layer.
• .mount(app) — an optional FastAPI adapter built on top of the plain methods: it wires them to FastAPI routes with request-body validation and converts NxtidHTTPError to fastapi.HTTPException. FastAPI/pydantic are imported lazily inside .mount()/._build_router() — never at module import time — so using the plain methods works even if FastAPI isn't installed at all.

If your agent uses FastAPI, call .mount(app) and never think about this again — see Typical Agent Integration Pattern below. If it doesn't, call the plain methods directly from whatever request handler your framework gives you:

from nxtid_sdk import NxtidSDK, AITRouter, EntraAuth, NxtidHTTPError

sdk = NxtidSDK.from_env()
ait_router = AITRouter(sdk, get_session=sessions.get, create_session=sessions.create)
entra = EntraAuth.from_env(get_session=get_or_create_session,
                            on_login=ait_router.on_login)

# e.g. inside a Flask view, a Slack slash-command handler, a plain script...
async def handle_wallet_connect(session_id: str, email: str) -> dict:
    session = sessions.get(session_id)
    if session is None:
        return {"error": "Session not found or expired."}, 404
    try:
        return await ait_router.wallet_connect(session, email=email)
    except NxtidHTTPError as exc:
        return {"error": exc.detail}, exc.status_code   # map to your framework's error shape

async def handle_entra_login(session_id: str, name: str, email: str, roles: list) -> dict:
    session = get_or_create_session(session_id)
    return await entra.login(session, name=name, email=email, roles=roles)

# Sessions: sessions.create()/get()/get_or_create()/delete() are plain dict
# operations, no HTTP involved — call them from anywhere.

---

Choosing how tool calls get governed

Once wallet + IDV + AIT permissions establish who the person behind the agent is, something still has to enforce that ceiling on every tool call. Two ways to do that:

• Gateway (recommended) — the agent's tool calls flow through an MCP server the nxtID Gateway fronts. The gateway checks each call against the person's AIT grant in real time, and a top-level agent can mint a narrower identity for a sub-agent (same person, same session, capped tools) — this is how nxtID governs production agents and the only way to get delegation. AITRouter enrolls with the gateway automatically the moment IDV completes; there's no separate setup step. See Gateway Integration.
• In-process (optional) — call sdk.guard_tool_call yourself right before executing each gated tool in your own dispatch loop, or import nxtid_sdk.auto to wire that same check into your agent framework's own tool-invocation hook with zero governance code at the call site. No MCP server or gateway required, so it's a reasonable starting point for a single simple agent or an early prototype — but there's no delegation, and you own the enforcement point instead of the gateway. See guard_tool_call below, In-Process Framework Governance, and Typical Agent Integration Pattern.

The two are not mutually exclusive: guard_tool_call is also the check run under the hood for tools that never reach the gateway (a handoff, a framework built-in), so agents that adopt the gateway still benefit from it at those edges — directly, or via nxtid_sdk.auto. Start with the gateway; reach for calling guard_tool_call directly only for paths it doesn't front.

---

Quick Start

from nxtid_sdk import NxtidSDK

sdk = NxtidSDK(
    api_key="your-api-key",
    api_secret="your-api-secret",
    env="production",          # "production" (default) or "staging"
    service_id="your-service-id",  # required for log_chat
)

# 1. Register your agent's tools on startup
await sdk.register_tools([
    {"name": "get_account_balance", "description": "Read-only finance access"},
    {"name": "transfer_funds",      "description": "Move money between accounts"},
])

# 2. Connect wallet by email
wallet = await sdk.connect_wallet(email="user@example.com")
# → WalletResult(wallet_address="0x...", email="user@example.com")

# 3. Start AuthID identity verification
verify = await sdk.start_verification(wallet.wallet_address)
# → VerifyStartResult(flow="login"|"onboarding"|"choose_document", widget_url="...", ...)

# 4. After user completes the widget
result = await sdk.complete_verification(
    wallet.wallet_address,
    operation_id=verify.operation_id,   # for onboarding flow
    transaction_id=verify.transaction_id,  # for login flow
)
# → VerifyCompleteResult(is_verified=True)

# 5. Check admin-granted permissions
perms = await sdk.get_permissions(wallet.wallet_address)
if perms.has_permission("transfer_funds"):
    ...

---

Environments

env	AIT Service	Dashboard
production	https://ait-service.nxtlinq.ai	https://dashboard.nxtlinq.ai
staging	https://staging-ait-service.nxtlinq.ai	https://dashboard-staging.nxtlinq.ai

---

API Reference

NxtidSDK

NxtidSDK(
    api_key: str,
    api_secret: str,
    env: str = "production",
    service_id: Optional[str] = None,
)

Credentials (api_key, api_secret, service_id) are obtained from nxtID Dashboard → Services → Register Third-party Agent.

NxtidSDK.from_env() → NxtidSDK | None

Builds an instance from env vars instead of wiring them by hand — returns None if not configured. Pass it straight to AITRouter either way; no if needed — AITRouter is None-safe and mounts unconditionally (see below):

sdk = NxtidSDK.from_env()
AITRouter(sdk, get_session=sessions.get, create_session=sessions.create).mount(app)

Reads NXTLINQ_SERVICE_API_KEY, NXTLINQ_SERVICE_API_SECRET, NXTLINQ_SERVICE_ID (optional), NXTLINQ_ENV ("production" default or "staging").

---

connect_wallet(email) → WalletResult

Gets or creates a MetaKeep non-custodial wallet for the given email address.

wallet = await sdk.connect_wallet(email="user@example.com")
wallet.wallet_address  # "0xABCDEF..."
wallet.email           # "user@example.com"

---

start_verification(wallet_address, id_type=None) → VerifyStartResult

Starts an AuthID identity verification session.

flow	Condition
"login"	Returning user (face scan only)
"onboarding"	New user with id_type provided
"choose_document"	New user without id_type (user selects type)

id_type is the AuthID document type code (e.g. "40" for passport).

result = await sdk.start_verification(wallet.wallet_address, id_type="40")
result.flow        # "onboarding"
result.widget_url  # URL to open in browser/iframe
result.operation_id   # pass to complete_verification (onboarding)
result.transaction_id # pass to complete_verification (login)

---

complete_verification(wallet_address, operation_id=None, transaction_id=None) → VerifyCompleteResult

Finalises the verification session after the user completes the widget. Polls AuthID for up to 60 s.

done = await sdk.complete_verification(
    wallet.wallet_address,
    operation_id=result.operation_id,
)
done.is_verified  # True | False

---

custom_verification(wallet_address) → VerifyCompleteResult

Marks a wallet verified by self-attestation — no third party (AuthID, Berify.me) involved. AuthID biometric IDV is an optional add-on tier on top of this, not the default this bypasses. Still registers the wallet and a HIT with the AIT Service (tagged method='custom', never 'authid'), so the wallet stays visible to the Dashboard's identity views and to any externalRef-based HIT lookup, unlike a bare local session.is_verified = True would.

Called through AITRouter.verify_custom / POST /verify/custom in normal usage — see allow_custom_verification under AITRouter below — rather than directly.

done = await sdk.custom_verification(wallet.wallet_address)
done.is_verified  # True | False

---

get_permissions(wallet_address) → PermissionsResult

Queries the AIT on-chain record for this wallet to determine which tools the admin has granted.

perms = await sdk.get_permissions(wallet.wallet_address)
perms.allowed        # True if any permission grant exists
perms.allowed_tools     # ["transfer_funds", "get_account_balance", ...] — the ceiling
perms.user_denied_tools # tools this wallet self-service-denied
perms.active_tools      # allowed_tools minus user_denied_tools, computed server-side
perms.reason            # optional explanation from nxtlinq

perms.has_permission("transfer_funds")  # True | False

---

get_admin_granted_tools(wallet_address) → list[str]

Returns the list of tools the admin has pre-approved for this user to self-service request. Empty list means the admin hasn't granted any tools yet.

tools = await sdk.get_admin_granted_tools(wallet.wallet_address)
# ["calculate", "get_company_info", "get_account_balance"]

---

set_permissions(wallet_address, permissions, granted_by=None, roles=None) → dict

Writes an AIT on-chain with the given tool list for wallet_address. Used for self-service permission requests (user selects a subset of admin-granted tools), and for role-based auto-grant (pass roles for audit purposes).

await sdk.set_permissions(
    wallet_address="0x...",
    permissions=["calculate", "get_company_info"],
    granted_by="user@example.com",
)

---

get_role_permissions(role) → list[str]

Returns the default tool list an admin has configured for a given Entra role — used to auto-grant AIT permissions right after IDV completes, without the user having to self-service pick anything.

tools = await sdk.get_role_permissions("Finance")
if tools:
    await sdk.set_permissions(wallet.wallet_address, tools, roles=["Finance"])

---

async guard_tool_call(tool_name, session) → None

Enforces identity and permission checks before a tool executes. Call this inside the agent's tool dispatch loop, before executing each gated tool (await it — it's async). Raises ToolCallBlockedError if the call should be blocked; returns None (no-op) if permitted. session needs wallet_address, is_verified, allowed_tools, admin_allowed_tools, permissions_configured — satisfied by NxtidSessionMixin. entra_roles is read too, for the role-based fallback (see below).

This makes a live API call every time — it re-fetches the current admin grant from the AIT service on every invocation rather than trusting a cached session snapshot, so a revoke takes effect on the very next tool call, not just the next page reload or widget reopen. A network failure here fails closed (ToolCallBlockedError), not open.

await sdk.guard_tool_call("transfer_funds", session)

Internally this is resolve_admin_grant + a membership check — the exact same call AITRouter's session_restore/permissions_refresh make, so real-time enforcement can never drift out of sync with what those endpoints report.

---

async resolve_admin_grant(session) → bool

Queries the current admin-granted AIT permission ceiling for session.wallet_address and sets session.permissions_configured / session.admin_allowed_tools / session.allowed_tools: personal grant if present, otherwise role-based group default from session.entra_roles (DB read only, no on-chain write), otherwise cleared. Returns True if some grant was found.

session.allowed_tools is set to the server-computed active set — the ceiling minus the wallet's own self-service denials (PermissionsResult.active_tools, a deny-list) — so when the admin widens the ceiling with a tool the user never explicitly turned off, it's active immediately. This is already the final, authoritative value: because it's a deny-list rather than a cached allow-list, "the user never touched this" and "the user explicitly chose to keep everything active" both simply mean "nothing denied" — there's no empty-list ambiguity for a caller to reconcile against a client-supplied selection.

async refresh_session_permissions(session) → bool

Convenience wrapper around resolve_admin_grant — refreshes session.allowed_tools/session.admin_allowed_tools from the current admin-granted ceiling in one call. No-ops (returns False) if wallet/IDV aren't established yet; never raises — keeps whatever was last cached on the session if the underlying call fails.

Call this unconditionally at the start of every turn, before building anything that reads session.allowed_tools (e.g. a system prompt's "here's what you can't use" hint). That value is otherwise only refreshed as a side effect of guard_tool_call, which only runs if a tool call is actually attempted — if your prompt tells the LLM a tool is blocked based on a stale allowed_tools, the LLM has no reason to try it, so guard_tool_call never runs, and the stale state never refreshes even after an admin grants new permissions:

await sdk.refresh_session_permissions(session)
allowed = [t for t in TOOL_DEFINITIONS if t["name"] in session.allowed_tools]
system_prompt = build_prompt(allowed)  # now reflects the current grant, not last turn's

---

register_tools(tools) → int

Registers the agent's tool catalogue with nxtID so admins can grant permissions per tool. Call once at agent startup.

count = await sdk.register_tools([
    {"name": "transfer_funds",      "description": "Move money between accounts"},
    {"name": "get_account_balance", "description": "Read-only finance access"},
])
# count = 2

---

log_chat(session, user_message, assistant_message, ...) → None

Sends one conversation turn to nxtID Dashboard for monitoring. Fails silently — logging is non-fatal. Requires service_id at SDK init time. The external user identity is derived from the session as wallet_address or verified_email / wallet_email or verified_email.

await sdk.log_chat(
    session,
    user_message="What's my balance?",
    assistant_message="Your balance is $500.",
    model="meta/llama-3.3-70b-instruct",
    has_tool_call=True,
    tool_name="get_account_balance",
    turn_index=0,
)

---

EntraAuth

Optional module for adding Microsoft Entra SSO login to a FastAPI agent. When enabled, mounts two routes onto the host app that the browser MSAL popup calls.

from nxtid_sdk import EntraAuth

entra = EntraAuth(
    client_id=os.getenv("AZURE_AD_CLIENT_ID"),
    tenant_id=os.getenv("AZURE_AD_TENANT_ID"),
    get_session=get_or_create_session,  # callable: session_id → Session object
)
entra.mount(app)  # app is a FastAPI instance

Or read the same two env vars in one call: EntraAuth.from_env(get_session=get_or_create_session, on_login=on_entra_login).

If client_id or tenant_id is empty/None, entra.enabled returns False and the routes respond accordingly — no error is raised.

Routes mounted (FastAPI) / plain methods (any framework):

Method	Path	Plain method	Description
GET	/auth/entra/config	entra.config() → dict	Returns { enabled, client_id, tenant_id } for MSAL initialisation
POST	/auth/entra	await entra.login(session, name, email, tenant_id=None, roles=None, id_token=None) → dict	Writes MSAL token claims onto the session, fires on_login

POST /auth/entra body:

{
  "session_id": "string",
  "name": "string",
  "email": "string",
  "tenant_id": "string (optional)",
  "roles": ["string"],
  "id_token": "string (optional) — the raw Entra ID token from MSAL. Omitting it is fine for role-based grants, but the gateway re-enrollment described in Gateway Integration below needs it to verify a returning user's sign-in, so pass it whenever you have it."
}

The get_session callable receives a session_id string and must return a session object with writable attributes: entra_authenticated, entra_name, entra_email, entra_tenant_id, entra_roles.

---

AITRouter

Bundles the wallet / AuthID IDV / AIT-permission HTTP endpoints onto a FastAPI app — the wallet/verify/permissions counterpart to EntraAuth.mount(app). Without this, every consuming app has to hand-write these ~9 routes itself.

from nxtid_sdk import NxtidSDK, AITRouter

sdk = NxtidSDK.from_env()  # may be None if credentials are missing
ait_router = AITRouter(sdk, get_session=sessions.get, create_session=sessions.create)
ait_router.mount(app)

sdk may be None — construct and mount AITRouter unconditionally either way, same pattern as EntraAuth. When sdk is None, /wallet/connect responds 503 and everything downstream (which all require a connected wallet first) degrades gracefully; check ait_router.enabled where your app needs to know (e.g. to decide whether to show the setup widget).

Registering tools at startup: if your app has no other startup/shutdown needs, skip writing your own lifespan function — ait_router.lifespan(tools) returns a ready-made one. Extra keys (e.g. LLM parameters schemas) are fine; only name/description are read server-side, so your tool-calling TOOL_DEFINITIONS list can usually be passed straight through.

app = FastAPI(lifespan=ait_router.lifespan(TOOL_DEFINITIONS))

Already have a lifespan for other startup work? Call register_tools inside it instead:

@asynccontextmanager
async def lifespan(app):
    await ait_router.register_tools(TOOL_DEFINITIONS)
    yield

Routes mounted (FastAPI) / plain methods (any framework):

Method	Path	Plain method	Description
POST	/wallet/connect	await ait_router.wallet_connect(session, email)	Connect a MetaKeep wallet by email
POST	/verify/start	await ait_router.verify_start(session, id_type=None)	Start AuthID IDV, returns a widget URL
POST	/verify/complete	await ait_router.verify_complete(session, operation_id=None, transaction_id=None)	Complete IDV; auto-loads permissions afterward
POST	/verify/custom	await ait_router.verify_custom(session)	Self-attested verification, no AuthID IDV — 403 unless allow_custom_verification=True
POST	/permissions/request	await ait_router.permissions_request(session, allowed_tools)	Self-service: user selects admin-approved tools
GET	/permissions/admin-granted	await ait_router.permissions_admin_granted(session)	Tools admin pre-approved for self-service selection
POST	/permissions/refresh	await ait_router.permissions_refresh(session)	Re-pull admin-granted permissions from the AIT service
POST	/session/restore	await ait_router.session_restore(session, data)	Restore identity state from browser localStorage (or any client-side cache)
GET	/ait/status/{session_id}	ait_router.ait_status(session) (sync)	Full identity snapshot for the given session
GET	/nxtlinq/widget.js	ait_router.widget_js() (sync)	Bundled NxtlinqWidget setup UI source
GET	/nxtlinq/session.js	ait_router.session_js() (sync)	Bundled NxtlinqSession localStorage helper source
GET	/nxtlinq/approvals	n/a — widget-only, see Gateway Integration	Pending step-up approval requests for this session's user
POST	/nxtlinq/approvals/{request_id}	n/a — widget-only, see Gateway Integration	Approve/deny one pending request as this session's user

All the async plain methods raise NxtidHTTPError(status_code, detail) on failure instead of any FastAPI type — catch that if you're calling them outside .mount().

Unlike EntraAuth, AITRouter takes two separate callables rather than one get-or-create function:

• get_session: (session_id: str) -> Session | None — a plain lookup (e.g. SessionStore.get). Every route except /session/restore calls this and responds 404 if it returns None, rather than silently operating on a brand-new, blank session. This matters in practice: a session that completed IDV can otherwise look "expired" as a confusing downstream error (e.g. permissions_refresh failing with "Complete IDV first") instead of a clear "session not found" — which happens whenever the in-memory session store loses state a browser tab's session_id still points to (a server restart, a process recycle, an expired TTL in a custom store).
• create_session: () -> Session — makes a brand-new session (e.g. SessionStore.create). Used only by /session/restore, the one route that's expected to run with no session yet.

The session object needs plain, writable attributes (no interface/protocol required):

• Written by AITRouter: wallet_address, wallet_email, is_verified, verified_email, verification_method, permissions_configured, allowed_tools, admin_allowed_tools
• Read by AITRouter (set these yourself, e.g. via EntraAuth's on_login): entra_roles, entra_email — used to auto-grant AIT permissions by role right after IDV completes

Use NxtidSessionMixin (below) instead of hand-declaring these fields.

Self-attested verification (allow_custom_verification): off by default. Pass AITRouter(sdk, ..., allow_custom_verification=True) to enable verify_custom / POST /verify/custom, which mark a session verified by self-attestation instead of running AuthID biometric IDV — a deliberate lower-assurance path for a host that wants one (e.g. a second demo page). The gateway trusts whatever verified claim the enrolling side asserts, so this genuinely grants the same downstream access as real IDV; keep it off unless a specific flow needs it, and never enable it for a host's only or primary login path. session.verification_method records which path was used ("authid" or "custom") — audit/observability only, every enforcement check still reads is_verified. On the frontend, dropping 'verify' from NxtlinqWidget.init's steps makes the widget call /verify/custom automatically after wallet connect instead of launching the AuthID widget — see Frontend Widget Integration below.

Role-based auto-grant with EntraAuth:

Wire ait_router.on_login — it auto-grants AIT permissions from the person's Entra role, and (assuming Gateway Integration, below) also re-enrolls a returning user with the gateway, which needs to happen on every login since the raw ID token it depends on is never persisted (see on_login there for why). It degrades to exactly the plain role-based grant when no gateway is configured, so it's always the right thing to wire even before you've set up the gateway:

entra = EntraAuth(..., get_session=get_or_create_session, on_login=ait_router.on_login)

Not using the gateway at all? auto_grant_by_role's signature — (session, roles=None) — matches on_login exactly and does the role-based grant alone, with no gateway re-enrollment attempt:

entra = EntraAuth(..., get_session=get_or_create_session, on_login=ait_router.auto_grant_by_role)

---

Gateway Integration

Recommended. The nxtID Gateway governs tool calls, not identity setup — it decides, per call, whether the person behind the agent is allowed to do that right now, and lets a top-level agent delegate a narrower identity to a sub-agent (same person, same session, capped tools). This is the default way to enforce the AIT permission ceiling once wallet/IDV/AIT permissions (above) have established who the person is — wire it in unless your agent has no MCP tool server for the gateway to front and never spawns sub-agents, in which case calling guard_tool_call yourself (see Choosing how tool calls get governed, above) is enough.

Enrollment is automatic — you do not write any of this yourself if you're using AITRouter:

1. AITRouter.verify_complete enrolls with the gateway right after IDV finishes, filling in session.login_token.
2. A returning user (wallet + IDV already done in an earlier page load) loses login_token on every reload, because the raw Entra ID token it depends on is deliberately never persisted client-side (see NxtidSessionMixin). Wire ait_router.on_login (above) as EntraAuth's on_login and this re-enrolls silently using the fresh ID token Entra just supplied — no popup, no visible action.

gateway_url defaults to the SDK's own — in production/staging this is nxtID's published gateway for that environment, derived the same way NxtidSDK's AIT-service/Dashboard URLs are, so an integration never has to carry our hostnames in its own configuration. Pass AITRouter(sdk, ..., gateway_url=...) explicitly only to point at a gateway we have not published yet (e.g. a local one during development).

await ait_router.register_config(*, mcp_url=None, entra_tenant_id=None, entra_audience=None) → None

Tells the gateway where your MCP tool server lives, so it can route a permitted call to it — and, if you use Entra, which tenant/audience to verify sign-ins against. You normally never call this yourself: ait_router.lifespan(TOOL_DEFINITIONS) runs it at startup, right after tool registration. Each value falls back to config you already have (mcp_url to NXTID_MCP_URL, the Entra pair to AZURE_AD_TENANT_ID/ AZURE_AD_CLIENT_ID), and only what's present is sent.

This is the one piece of gateway setup that fails silently. A missing mcp_url is not an error — startup still succeeds, register_config best-effort no-ops (an administrator can set the same fields from the Dashboard instead), and the gateway simply has nowhere to send a permitted tool call. If gateway-path tool calls are being refused for no apparent reason, confirm NXTID_MCP_URL is actually set:

# Gateway path only — where the gateway forwards a permitted tool call.
# Nothing warns you if this is missing; every tool call simply fails.
NXTID_MCP_URL=https://your-app.example.com/mcp

Or pass it directly at construction: AITRouter(sdk, ..., mcp_url=...).

Using a governed identity, once session.login_token is set:

me = await ait_router.identity_for(session, agent_id="support-bot")

# Claude Agent SDK — one call gets you the MCP server config + allowed_tools:
from claude_agent_sdk import ClaudeAgentOptions
options = ClaudeAgentOptions(**await me.agent_options(), max_turns=4)

# Any other framework: build your own MCP client with the same headers.
headers = me.mcp_headers()  # {"Authorization": "Bearer <token>"}

tools is optional and normally omitted from identity_for — leaving it out means the agent can use whatever the administrator has granted that person. Pass a list only to give this agent less than the person has.

Delegating to a sub-agent — same person, same session, capped further:

sub = await me.delegate(agent_id="researcher", tools=["get_order_status"])
ClaudeAgentOptions(**await sub.agent_options())

delegate can only ever remove access — the gateway refuses anything beyond what the parent identity itself can reach — and it validates the requested tool names up front against what the gateway will actually allow, so a typo fails immediately instead of silently producing a sub-agent that can't use anything.

Identity (nxtid_sdk.Identity) is the class both identity_for and delegate return, and the lower-level building block if you are not using AITRouter/sessions at all:

Method / property	Description
Identity.enroll(gateway_url, *, id_token, wallet_address, api_key, api_secret, verified=True, scope=None) → str	Mints a login token from a completed sign-in. Server-side only — never send these credentials to the browser. AITRouter calls this for you.
Identity.exchange(gateway_url, login_token, *, agent_id, session_id, framework="claude-agent-sdk", tools=None) → Identity	Turns a login token into a governed identity for one agent. AITRouter.identity_for calls this for you.
identity.delegate(*, agent_id, tools, framework="claude-agent-sdk") → Identity	Narrows to a sub-agent identity, as above.
identity.agent_options(gateway_url=None) → dict	Everything Claude Agent SDK needs (mcp_servers + allowed_tools), renewing the credential first if it's near expiry.
identity.mcp_headers() → dict	Just the Authorization header, for a non-Claude-Agent-SDK MCP client.
identity.subject / .agent / .session / .tools / .expires_in	Read-only claims off the credential (who, which agent, which session, delegation cap, seconds to expiry).

IdentityError is raised for anything gateway-side (refused enrolment/exchange, expired credential used directly instead of via agent_options, an unrecognised tool name passed to delegate).

Step-up approvals — when the gateway blocks a high-risk tool call, it opens a pending approval record and refuses the call until a human answers. AITRouter answers this with the identity the session already proved at sign-in (its entra_email/verified_email/wallet_email, in that order of trust) — no re-authentication, and the model never sees or touches the approval: the browser confirms with the session_id it already holds, which never enters a prompt.

This is fully automatic once AITRouter.mount(app) is wired — GET/POST /nxtlinq/approvals (above) are included on the same router as everything else, backed by an internal ApprovalBridge that talks to the gateway's /step-up/requests API with the SDK's Service credential. There is nothing to call directly unless you are not using .mount():

Method	Description
ait_router.set_gateway_url(url)	Point the router — and its approval bridge — at a gateway discovered after construction (e.g. one spawned on a dynamic port at startup). Keeps both in sync; do not set _gateway_url by hand.

Recording an approval always writes approved_by as session:<verified email> — never a bare name, never anything taken from the request body — so the gateway's hash-chained audit trail can tell a session-confirmed decision apart from one made through an unauthenticated link. A decision is rejected unless the request actually belongs to that session's user, checked server-side against the gateway's queue on every call.

See Frontend Widget Integration → NxtlinqWidget.approvals() below for the browser side.

---

In-Process Framework Governance (nxtid_sdk.auto)

The other enforcement path (see Choosing how tool calls get governed, above) — covers every tool call your agent framework makes in its own process: its function tools, built-ins (e.g. OpenAI Agents' Shell/ Computer tools), handoffs, and sub-agents running alongside it. nxtid_sdk.auto monkey-patches each installed framework's own tool-invocation entry point to route through the same permission + risk gate — no governance code at the call site.

# Zero-wiring: instruments every installed framework at import time.
import nxtid_sdk.auto as gov

# Per request/session, once wallet+IDV+AIT permissions are established:
gov.set_active(sdk, session)

# ...your normal agent code, unchanged — e.g. Claude Agent SDK, OpenAI Agents
# SDK, LangGraph, Google ADK, or AutoGen — every in-process tool call this
# framework makes is now governed automatically.

# Opt out of the import-time auto-enable (env var), then drive it by hand:
# NXTID_AUTO_INSTRUMENT=0
from nxtid_sdk import auto as gov
gov.enable(only=["claude-agent-sdk"])   # omit only= to enable every installed one
gov.disable()                            # restore the frameworks' original hooks

• Supported frameworks (auto-detected, patched independently — one failing to instrument never breaks the others): OpenAI Agents SDK, Claude Agent SDK, LangGraph, Google ADK, AutoGen.
• Risk grading mirrors the gateway: the nxtID Dashboard lets an admin classify each tool's risk tier (low/medium/high/critical) and a per-tier policy (allow/ask/block). The SDK fetches this from GET {gateway}/risk-config (same Service credential the SDK already carries, nothing new to configure) and caches it for NXTID_RISK_CONFIG_TTL seconds (default 60) — this is the fast in-process path, so it's cached instead of fetched per call. Any non-200 response, a timeout, or a malformed body all degrade to the pre-Dashboard local heuristic; a struggling gateway is never hammered, since a fetch failure just serves the stale cache for the rest of the TTL window.
• Local escalators still apply on top of the Dashboard level and can only raise it, never lower it: a shell-content overlay bumps risk up for command strings matching dangerous patterns (rm -rf, sudo, a pipe into sh, ...), and a per-tool min_level floor forces framework built-ins that can reach the host (e.g. OpenAI's Shell/Computer tools) to at least L4_HIGH regardless of what the Dashboard says.
• Permission revocation is unaffected by any of this caching — guard_tool_call's own check stays live on every call. Only a risk-level/policy edit made in the Dashboard takes up to one TTL to bind on this in-process path (vs. the next call on the gateway/MCP path).
• gov.set_active(sdk, session) is required, once per request, before any governed tool runs. It binds the identity through a contextvar, so concurrent requests never see each other's session — but a context with no binding has no identity to check a permission against, so the permission gate is skipped there and only risk grading and audit run. That's deliberate (an un-bound context degrades rather than hard-failing), and it means a code path that reaches a tool without set_active is not permission-enforced. Bind it at the start of every request, including background jobs and startup tasks that can reach a tool.

Need manual control over which framework hook gets which adapter, instead of the automatic monkey-patching above? nxtid_sdk.adapters.build_adapter(binding, framework=..., tools=...) returns a Governance bundle (run_hooks / can_use_tool / plugin / tools — whichever field your framework needs) built on the exact same guard_tool_call + risk-grade + audit core that nxtid_sdk.auto wires in for you automatically:

from nxtid_sdk.adapters import build_adapter
from nxtid_sdk.auto import current_binding

gov = build_adapter(current_binding, framework="claude-agent-sdk")
options = ClaudeAgentOptions(can_use_tool=gov.can_use_tool, ...)

When a tool call is blocked on this path, guard_tool_call raises ToolCallBlockedError — catch it in your tool dispatch loop and return exc.message as the tool result, so the LLM relays the real reason instead of guessing or fabricating one. Exactly three reasons are possible from guard_tool_call: it decides from the caller's own session and the live admin grant, and never sees an individual call's risk grade, so it can't conclude anything beyond these (see the ToolCallBlockedError table under Reference below).

from nxtid_sdk import ToolCallBlockedError

# In-process path — guard_tool_call raises exactly three reasons:
try:
    await sdk.guard_tool_call(tool_name, session)
except ToolCallBlockedError as exc:
    exc.reason        # "verification_required" | "awaiting_admin_grant" | "permission_denied"
    exc.message       # user-facing explanation — return this to the LLM
    exc.gateway_code  # None here: this did not come from the gateway

A call governed through nxtid_sdk.auto can additionally be stopped by the risk policy — a per-call judgement guard_tool_call never makes — and that does not arrive as a ToolCallBlockedError: depending on the framework it's either handed back to the model as a refusal or raised as the adapter's own NxtIDToolBlocked. Catch only ToolCallBlockedError and a risk-policy block slips past you.

---

NxtidSessionMixin

Dataclass mixin providing every field AITRouter and EntraAuth read/write. Inherit from it instead of hand-declaring the ~12 identity fields on your own Session every time you integrate nxtID into a new project.

from dataclasses import dataclass, field
from nxtid_sdk import NxtidSessionMixin

@dataclass
class Session(NxtidSessionMixin):
    messages: list = field(default_factory=list)   # your own app state

session = Session()  # session_id is auto-generated (UUID)

All mixin fields have defaults (session_id included, via a UUID factory), so — per the usual dataclass inheritance rule — any fields you add in your subclass need defaults too.

---

SessionStore

Generic in-memory session store keyed by session.session_id — the create/get/delete dict every nxtID integration otherwise hand-rolls, factored out once.

from nxtid_sdk import SessionStore

sessions = SessionStore(Session)   # Session is your NxtidSessionMixin subclass

session = sessions.create()
session = sessions.get(session_id)             # None if unknown
session = sessions.get_or_create(session_id)   # None/unknown id -> new session
sessions.delete(session_id)

Session lifecycle endpoints: most frontends need a way to create a fresh session and to reset one (e.g. on sign-out). sessions.mount(app) mounts both so you don't have to hand-write them:

sessions.mount(app)   # POST /session/new -> {"session_id": ...}, POST /session/reset {"session_id": ...} -> {"ok": true}

---

NxtidHTTPError

Framework-agnostic error raised by AITRouter/EntraAuth's plain methods (wallet_connect, verify_start, login, ...) — never a fastapi.HTTPException or any other web-framework type. .mount(app) catches this internally and converts it for you; call the plain methods directly and you need to catch it yourself.

from nxtid_sdk import NxtidHTTPError

try:
    result = await ait_router.wallet_connect(session, email=email)
except NxtidHTTPError as exc:
    exc.status_code  # e.g. 400, 502, 503
    exc.detail       # user-facing explanation

---

ToolCallBlockedError

Exception raised to block a tool execution before it reaches the LLM tool handler.

from nxtid_sdk import ToolCallBlockedError

try:
    await sdk.guard_tool_call(tool_name, session)
except ToolCallBlockedError as exc:
    exc.reason        # "verification_required" | "awaiting_admin_grant" | "permission_denied"
    exc.message       # user-facing explanation
    exc.gateway_code  # None here — see from_gateway_error below
    exc.status        # None here — see from_gateway_error below
    str(exc)          # == exc.message, so logging.exception prints something useful

Two producers, and they raise disjoint sets of reasons. guard_tool_call — the local pre-dispatch check — raises only the first three. It decides from the caller's session and the live admin grant and never sees a specific call's risk grade, so it cannot conclude anything about a step-up. step_up_required and step_up_denied come only from ToolCallBlockedError.from_gateway_error, for calls that went through the nxtID Gateway instead.

reason	Raised by	Meaning	Fixable via the setup widget?
verification_required	guard_tool_call, from_gateway_error	Wallet not connected, IDV not completed, or the admin-grant check itself failed transiently	Yes — handleChatEvent (modal mode) tries a silent credential refresh first; if that also fails, it signs the user out and opens the setup modal back on step 1 (see below)
awaiting_admin_grant	guard_tool_call, from_gateway_error	Wallet + IDV are done, but the admin hasn't granted this account any permission yet (no personal or role grant at all)	No — nothing for the end user to click through; only an admin can grant something, so the widget does not auto-open
permission_denied	guard_tool_call, from_gateway_error	The account has an admin grant, just not for this specific tool	No — the account is already fully set up, so the widget does not auto-open; show exc.message in chat instead
step_up_required	from_gateway_error only	The account IS authorized for this tool, but the gateway graded this call high-risk and is holding it for a human's yes/no. Nobody has answered yet	No — not a setup-state problem. Retry the call with identical arguments; pair with NxtlinqWidget.approvals() on the host side
step_up_denied	from_gateway_error only	Same hold, except a human reviewed and declined it. Terminal for this exact request	No — stop and tell the user. Retrying the unchanged call raises no new approval; only the user asking again does

Attributes:

Attribute	Type	Meaning
reason	str	One of the five above
message	str	The explanation to show or relay
gateway_code	str | None	The raw gateway deny code this was classified from (STEP_UP_REQUIRED, TOKEN_EXPIRED, AUTHORIZATION_WITHDRAWN, NOT_ENROLLED, RISK_POLICY_BLOCKED, NOT_AUTHORIZED). None whenever guard_tool_call raised it
status	str | None	The gateway payload's raw status for a step-up deny — "pending", "denied" or "unavailable". None for every non-step-up deny, for the step-up fallback branch (the gateway omits the field), when only the code matched as a substring, and whenever guard_tool_call raised it

str(exc) is exc.message — so logging.exception(...), a bare traceback and f"{exc}" print the explanation instead of an empty line.

status is what makes step_up_required actionable. One reason covers three gateway states that call for different handling, and the raw status is the only thing that tells them apart:

status	What actually happened	What to do
"pending"	An approval card exists and a human can still answer it	Retry with identical arguments. Safe by design — the gateway returns the same request rather than queueing a second card — and the call goes through the moment somebody approves. challenge_id in the raw text correlates it
"unavailable"	The approval service was unreachable, so no card was ever raised and nobody was asked anything	Treat as a transient fault, not as a wait. Retry shortly (the retry is what finally raises the card), or surface it — but a caller that treats this as pending waits forever for an answer to a question no human ever received. There is no challenge_id, because there is no request
None	No approval request was recorded at all; the caller simply lacks a step-up proof	Present a step-up proof and retry
"denied"	A human reviewed and declined	Already reflected in reason == "step_up_denied" — stop

reason deliberately collapses the first three, which is the right level for a host that only needs "blocked, nobody has answered yet". status is what keeps that collapse non-destructive, exactly as gateway_code does for the six deny codes.

if blocked.reason == "step_up_required":
    if blocked.status == "unavailable":
        ...   # nobody was asked — transient, not a wait
    else:
        ...   # a card is up (or none was recorded) — retry unchanged

ToolCallBlockedError.from_gateway_error(exc) → ToolCallBlockedError | None

For tool calls that go through the nxtID Gateway rather than through guard_tool_call. The gateway refuses by putting a JSON object in the tool error text (a failed MCP tools/call has exactly one channel — ToolError carries no structured data, and the MCP client libraries drop structuredContent on the error path). This classmethod turns that text into the same exception type the rest of the SDK raises, so your host has one thing to catch either way.

from nxtid_sdk import ToolCallBlockedError

try:
    await session.call_tool(tool_name, args)
except Exception as exc:
    blocked = ToolCallBlockedError.from_gateway_error(exc)
    if blocked is None:
        raise                          # not a gateway deny — not ours to handle

    if blocked.reason == "step_up_denied":
        ...                            # a human said no -> STOP, tell the user
    elif blocked.reason == "step_up_required":
        if blocked.status == "unavailable":
            ...                        # nobody was asked -> transient fault
        else:
            ...                        # a card is waiting -> retry, same arguments
    else:
        ...                            # show blocked.message

Returns None when the text carries no recognizable gateway deny code, so the caller can simply re-raise the original. Accepts an exception or a plain string — NeMo's MCPToolClient.acall does not raise on a failed tool call, it returns the error string.

Three things it absorbs that are easy to get wrong by hand:

• Declined is checked before pending. STEP_UP_REQUIRED covers both, distinguished by status; denied maps to step_up_denied and pending / unavailable / absent all map to step_up_required. It is a refinement, not an alternative — treating "a human said no" as "still waiting" turns a decision somebody made into a retry loop nobody will ever answer. The raw status is kept on the exception so the three unanswered states stay distinguishable (see the table above).
• The text is not assumed to be bare JSON. MCP clients prefix and wrap it, so the payload is located inside the text, and the code is matched as a plain substring when the JSON cannot be parsed at all.
• message prefers the gateway's human-safe sentence. The gateway sends both a guidance field (written for the model) and, on the five self-naming denies, a short user_message (safe to show a non-technical end user). message takes user_message when present, else guidance, else the raw text — because message is what gets shown or relayed. NOT_AUTHORIZED deliberately carries only code + guidance, so it yields the generic "Tool not found."

Retries must carry identical arguments. An approval is bound to the hash of the arguments the human actually saw; change any field and it becomes a new request nobody has answered. Retrying while still pending is safe — the gateway returns the same request rather than queueing a second card.

Factory method:

raise ToolCallBlockedError.permission_denied(
    tool_name="transfer_funds",
    admin_allowed_tools=session.admin_allowed_tools,  # optional
    tool_label=None,  # optional — display name shown instead of tool_name
)
# admin_allowed_tools does NOT contain "transfer_funds" (or "ALL"):
#   message: "You don't have permission to use transfer_funds. Please contact your administrator."
# admin_allowed_tools DOES contain it — the account is authorized, what's
# missing is the user's own opt-in (the self-service deny-list), so the
# sentence changes entirely rather than just dropping a suffix:
#   message: "Your assistant doesn't currently have your permission to use transfer_funds."
# tool_label lets a caller with its own name-to-label mapping (e.g. a setup
# widget's checkboxes) show that label instead of leaking the raw tool name.
# allowed_tools is accepted for backwards compatibility but is not currently
# used in either message.

---

Return Types

Class	Fields
WalletResult	wallet_address: str, email: str
VerifyStartResult	flow: str, widget_url: str|None, operation_id: str|None, transaction_id: str|None
VerifyCompleteResult	is_verified: bool
PermissionsResult	allowed: bool, allowed_tools: list[str], user_denied_tools: list[str], active_tools: list[str], reason: str|None, has_permission(tool_name) → bool

---

Frontend Widget Integration

AITRouter.mount(app) serves two browser-ready JS files — no build step, no npm package, just <script> tags:

• NxtlinqWidget (/nxtlinq/widget.js) — the setup UI: walks the user through Entra SSO → wallet connect → AuthID IDV → AIT permissions, calling AITRouter's HTTP routes under the hood.
• NxtlinqSession (/nxtlinq/session.js) — localStorage persistence + server-side session restore. Usable standalone even without the widget UI.
• NxtlinqWidget.approvals() (also shipped in widget.js) — the step-up confirmation card: a modal that pops up when the gateway is holding a high-risk tool call for this session's user to confirm.

NxtlinqWidget

<div id="nxtlinq-widget"></div>
<script src="/nxtlinq/session.js"></script>
<script src="/nxtlinq/widget.js"></script>
<script>
  let sessionId = null;

  NxtlinqWidget.init({
    container: '#nxtlinq-widget',
    apiBase: '',                 // '' = same origin as the page
    mode: 'inline',              // 'inline' (default) or 'modal' (gear-icon trigger + popup)
    steps: ['entra', 'wallet', 'verify', 'permissions'],
    entra: { clientId: '...', tenantId: '...', redirectUri: window.location.origin },
    tools: TOOL_DEFINITIONS,     // [{ name, label, desc }, ...] — same tools you passed to register_tools
    onStepComplete: (step, data) => {
      if (data.sessionId) sessionId = data.sessionId;
      // data is camelCase (walletAddress, allowedTools, ...) — map only the
      // fields you need into NxtlinqSession's snake_case shape.
      if (step === 'entra' && data.roles) NxtlinqSession.update(sessionId, { entra_roles: data.roles });
    },
    onComplete: (data) => {
      // Re-fetch a clean snake_case snapshot instead of hand-mapping the
      // camelCase callback payload — ait_status() already returns exactly
      // what NxtlinqSession.save() expects.
      fetch(`/ait/status/${data.sessionId}`).then(r => r.json())
        .then(snapshot => NxtlinqSession.save(data.sessionId, snapshot));
    },
    onError: (step, message) => console.error(step, message),
  });
</script>

Options:

Option	Default	Description
container	— (required)	CSS selector or element to mount into
mode	'inline'	'inline' renders the steps directly; 'modal' renders a trigger button that opens a popup panel
apiBase	''	Base URL of the agent API (same-origin if empty)
sessionId	null	Resume an existing server session. If omitted and NxtlinqSession has cached identity data (from a previous visit), the widget restores it via /session/restore before falling back to creating a brand-new session
steps	['entra','wallet','verify','permissions']	Which steps to show — drop 'entra' if you don't use EntraAuth; drop 'verify' to skip AuthID IDV and call /verify/custom instead (requires allow_custom_verification=True server-side, see AITRouter above)
entra	—	{ clientId, tenantId, redirectUri } — only needed if 'entra' is in steps
tools	[]	[{ name, label, desc }, ...] — tool catalogue rendered as checkboxes in the permissions step (same list you passed to register_tools). Without this, the permissions step has nothing to show. Auto-filtered to whatever the admin actually granted.
onStepComplete(step, data)	—	Fires after each step finishes. data uses camelCase (sessionId, walletAddress, allowedTools, permissionsConfigured, ...) — not the snake_case the REST API/NxtlinqSession use
onComplete(data)	—	Fires once every configured step is done. Same camelCase data shape as above
onError(step, message)	—	Fires on any step failure
onRestored(saved)	—	Fires if a saved session was restored from NxtlinqSession on load

mode: 'modal'-only options: triggerLabel (default '⚙️'), modalTitle (default '⚙️ Setup'), showSignOut (default true), onSignOut().

init() returns an instance with .destroy(), and in modal mode also .open() / .close().

handleChatEvent(event, ctx) (modal mode)

Wires a permission_required event from your chat/SSE stream straight into the widget — call it in your event handler and let it decide whether the gap is something the user can fix.

const widget = NxtlinqWidget.init({ container: '#nxtlinq-widget', mode: 'modal', ... });

// inside your SSE/chat event handler:
if (widget.handleChatEvent(event, { retry: () => sendMessage(lastText, n + 1), retryCount: n })) {
  return; // widget handled it (opened the modal, or is retrying silently)
}
// ... handle other event types yourself

Behavior by event.reason:

• permission_denied / awaiting_admin_grant — returns false without opening anything (nothing the user can click through — see the table above).
• verification_required, first occurrence (ctx.retryCount falsy), and local state shows a prior sign-in — does not open the modal immediately. This gap is very often just an expired gateway credential on an otherwise fully set-up session (e.g. the user was idle a while), so it first tries a silent credential refresh in the background (re-acquiring the browser's cached Entra token and re-enrolling with the gateway). If that succeeds, it calls ctx.retry() and the user never sees the modal at all.
• verification_required, first occurrence, but no local evidence of a prior sign-in (e.g. right after the user's own Sign out) — opens the modal directly, no silent attempt and no message. There is nothing to time out from, since a credential was never established this visit — calling that a "timeout" would be a false statement.
• verification_required, silent refresh failed — the browser's own signed-in session is gone too, not just the gateway credential. Treated as a real sign-out: clears the cached NxtlinqSession identity state, shows a "your session has timed out" notice inside the modal, resets it back to step 1, and opens it — so the user sees an honest logged-out state to sign back into, instead of a modal that still looks "already verified" over a credential that no longer works.
• verification_required, second+ occurrence (ctx.retryCount truthy) — treated the same as a failed refresh, not as a plain reopen: a "successful" silent refresh whose retry lands back on the exact same verification_required was never actually a fix (e.g. a silent Entra reauth only refreshes the browser's MSAL token — it never re-runs /enroll, so a separately-expired server-side login_token stays expired). Clears the cached identity, shows the timed-out notice, and reopens at step 1, rather than silently reopening a modal that still shows every step green with no explanation.

ctx is optional. The silent refresh still runs without it; if it succeeds the modal simply opens, and if it fails you get the sign-out above. A working credential never costs the user their session — only a refresh that actually failed does.

NxtlinqSession

<script src="/nxtlinq/session.js"></script>
<script>
  // On every page load — restores identity state via POST /session/restore
  NxtlinqSession.restore('').then(result => {
    if (result) console.log('restored session', result.session_id);
  });

  // Read the cached state without hitting the server
  const saved = NxtlinqSession.get();

  // identityData must use the same snake_case field names as ait_status()'s
  // response (wallet_address, allowed_tools, ...) — NOT NxtlinqWidget's
  // camelCase callback payloads, see the onComplete example above.
  NxtlinqSession.save(sessionId, identityData);
  NxtlinqSession.update(sessionId, { allowed_tools: [...] });  // merge a partial update

  // On sign-out
  NxtlinqSession.clear();
</script>

save/update/get/clear are synchronous localStorage operations. restore(apiBase) is the only one that hits the network — it posts the locally saved state to AITRouter's /session/restore and resolves with the server's up-to-date identity/permission snapshot, or null if nothing was saved or the request failed.

If you're using NxtlinqWidget, calling NxtlinqSession.restore() yourself before init() is optional — the widget does this internally when no sessionId is passed in. Call it yourself only if your app needs the restored state (e.g. to decide whether to mount the widget at all) before the widget initializes.

NxtlinqWidget.approvals()

The step-up confirmation card. It authenticates nobody itself — the session it polls with was already established by a Microsoft-signed sign-in, so "who is confirming" is a fact the backend already holds. The browser only ever carries session_id, which never enters a model prompt, so a language model cannot reach this even if injected content tells it to.

<script src="/nxtlinq/widget.js"></script>
<script>
  const watcher = NxtlinqWidget.approvals({
    apiBase: '',                       // '' = same origin as the page
    getSessionId: () => sessionId,     // however your app tracks the current session id
    interval: 3000,                    // poll interval in ms (default 3000)
    onDecision: (request, decision, ok) => {
      // decision: 'approve' | 'deny'; ok: whether it actually went through
    },
  });

  // Call eagerly right after an agent turn whose result says a call was held
  // for approval — safe and cheap even every turn, it never interrupts an
  // already-open card. The interval poll above is just the safety net for
  // approvals raised out-of-band.
  watcher.checkNow();

  // watcher.stop() to pause polling (e.g. on sign-out)
</script>

NxtlinqWidget.approvals(opts) starts polling GET /nxtlinq/approvals immediately and returns a watcher with .checkNow() and .stop(). When a request is pending it shows a modal with the tool name, risk tier, and a redacted view of the arguments (values that look like a password or API key are hidden), and lets the user Allow once or Don't allow — approving only ever authorizes that exact call, once; a repeat or a different call asks again.

When a decision doesn't go through, the card names why instead of showing one generic sentence for every failure: a request that expired and was cleaned up server-side (Mongo's TTL index removes the row outright, so it's simply absent) reads differently from one somebody else already answered, or one sent while signed out. A failure message stays on screen longer (4.5s) than a success (1.5s), since naming the reason is wasted if nobody has time to read it.

---

Typical Agent Integration Pattern

AITRouter + EntraAuth + SessionStore handle all the HTTP surface and session bookkeeping; your app only needs to mount them and, once a turn has a governed identity, build the MCP config the gateway fronts.

# session_store.py
from dataclasses import dataclass, field
from nxtid_sdk import NxtidSessionMixin, SessionStore

@dataclass
class Session(NxtidSessionMixin):
    messages: list = field(default_factory=list)   # your own app state

sessions = SessionStore(Session)
get_session, get_or_create_session = sessions.get, sessions.get_or_create

# main.py (FastAPI)
from claude_agent_sdk import ClaudeAgentOptions
from nxtid_sdk import NxtidSDK, AITRouter, EntraAuth

sdk = NxtidSDK.from_env()  # may be None if credentials are missing
ait_router = AITRouter(sdk, get_session=sessions.get, create_session=sessions.create)  # construct/mount unconditionally

# on_login role-grants AND (re-)enrolls with the gateway — see Gateway
# Integration above for why the re-enroll matters for returning users.
entra = EntraAuth.from_env(
    get_session=get_or_create_session,
    on_login=ait_router.on_login,
)

# No hand-written lifespan/session endpoints, and no `if ait_router`/`if sdk`
# guards anywhere — AITRouter and its methods are all None-safe internally.
app = FastAPI(lifespan=ait_router.lifespan(TOOL_DEFINITIONS))
ait_router.mount(app)
entra.mount(app)
sessions.mount(app)

# At the start of every turn, once wallet/IDV/AIT-permissions are done
# (session.login_token is set — see Gateway Integration above):
session = get_or_create_session(session_id)
me = await ait_router.identity_for(session, agent_id="support-bot")
options = ClaudeAgentOptions(**await me.agent_options(), max_turns=4)
# The gateway now enforces the AIT ceiling on every tool call this agent
# makes through `options.mcp_servers` — no per-call check to write yourself.

A call the gateway refuses comes back as a tool error whose text is the deny payload. Classify it with ToolCallBlockedError.from_gateway_error rather than string-matching it yourself — that is the only place step_up_required and step_up_denied come from.

from nxtid_sdk import ToolCallBlockedError

try:
    await session.call_tool(tool_name, args)
except Exception as exc:
    blocked = ToolCallBlockedError.from_gateway_error(exc)
    if blocked is None:
        raise
    ...   # blocked.reason / blocked.message / blocked.gateway_code / blocked.status

Without the gateway — no MCP server to front, no sub-agents: call guard_tool_call yourself before each gated tool instead of building options from a governed identity.

from nxtid_sdk import ToolCallBlockedError

# In your tool dispatch loop, before executing a gated tool:
try:
    await sdk.guard_tool_call(tool_name, session)
except ToolCallBlockedError as exc:
    # guard_tool_call raises only these three — the two step-up reasons come
    # only from ToolCallBlockedError.from_gateway_error (see above).
    exc.reason    # "verification_required" | "awaiting_admin_grant" | "permission_denied"
    exc.message   # user-facing explanation

# At the start of every turn, before building a system prompt that lists
# which tools are currently off-limits — see refresh_session_permissions above
# for why this can't just wait for guard_tool_call to run:
await sdk.refresh_session_permissions(session)

---

Requirements

• Python ≥ 3.10
• httpx >= 0.27
• fastapi >= 0.111, pydantic >= 2.7 (for AITRouter and EntraAuth)