Aynigma Python SDK

Python SDK for Aynigma: IAM for AI agents and AI security guardrails.

Authorize agent tool calls, handle human approvals, enforce execution receipts, and scan prompts and responses with Protect.

python -m pip install aynigma
# Or, from a source checkout of the backend repo:
python -m pip install ./sdk/python
export AYNIGMA_API_KEY=psk_your_project_key

from aynigma import Aynigma

client = Aynigma()
result = client.guard("Hello", stage="prompt")
print(result.allowed)

The default API URL is https://api.aynigma.ai/api/v1. Supply base_url or set AYNIGMA_API_URL to use another deployment. Validation endpoints are /guardrails/validate/input/ and /guardrails/validate/output/; project keys are sent in the X-API-KEY header. These classic validation endpoints are deprecated. validate_input() and validate_output() continue to work but emit DeprecationWarning; migrate to guard(text, stage="prompt") and guard(text, stage="response") respectively.

Protect v2 stage-aware scans are also available from both sync and async clients. guard() accepts prompt, response, tool_call, tool_result, retrieval, or document; scope can be bound with company_id, project_id, system_id, environment_id, policy_id, and a per-generation conversation_id. When environment token pricing is enabled, direct guard() calls for prompt/response billing also require a fresh unique generation_id; the OpenAI wrappers create one automatically and reuse it across that model call's stages. Keep a stable conversation ID across turns, but never reuse a generation ID for a different provider call. A blocked scan raises the same BlockedError as v1. V1 remains available for existing integrations.

Pass end_user_id="customer-42" to guard() to associate detections with an end user for per-policy repeat-offender quarantine; IDs may be at most 128 characters and the server hashes them. SDK v2 guard calls automatically send X-Aynigma-Source: sdk, which is recorded on detection events. A ValidationResult exposes user_message and quarantined; when a blocked response includes user_message, it is also used as str(BlockedError) and is available on the exception's user_message attribute.

with Aynigma() as client:
	result = client.guard(
		"Retrieve customer account details",
		"prompt",
		system_id=4,
		environment_id=12,
		conversation_id="support-session-42",
		generation_id="support-session-42-generation-1",
	)

guard_openai_chat_v2() scans prompt, previous tool results, response, and proposed tool calls as separate stages. It rejects streaming rather than releasing unchecked chunks; use the TypeScript wrapper or bounded LiteLLM gateway for hold-back streaming.

Agent IAM quickstart

python -m pip install aynigma
# Or, from a source checkout of the backend repo:
python -m pip install ./sdk/python
export AYNIGMA_AGENT_ID=<agent uuid>
export AYNIGMA_AGENT_KEY=agt_...

With an agent ID and key the SDK trades the key for a 15 minute token, caches it, refreshes it at 80 % of its lifetime (or once after a 401), and sends Authorization: Bearer <token> on tool calls, receipt checks and MCP calls. The key is sent only to the token endpoint.

from aynigma import AgentAuthorizer, AgentTokenProvider, RemoteTool, acting_for_user

authorizer = AgentAuthorizer()      # reads AYNIGMA_AGENT_ID, AYNIGMA_AGENT_KEY, AYNIGMA_API_URL
create_ticket = RemoteTool(authorizer, "https://executor.internal.example", "ticket.create")

# Delegation: act for a signed-in person. Their JWT is exchanged for a delegated token
# (RFC 8693); the raw JWT is never forwarded and the backend records a verified chain.
with acting_for_user(current_user_jwt):
    create_ticket(title="Investigate alert")

# Narrow what the delegation may do, or hand it to a sub-agent.
scoped = authorizer.delegate(current_user_jwt, scope="db.read")
sub_authorizer = scoped.delegate_to(AgentTokenProvider(SUB_AGENT_ID, SUB_AGENT_KEY))

Receipts: every ALLOW returns a one-use receipt bound to the exact arguments and, with tokens, to the agent run. The executor calls ToolExecutor.execute (verify, run, report) with its own AgentAuthorizer. If an administrator kills the run, verify and tool calls fail with RunKilled (a ToolDenied), the SDK drops its cached tokens, and the next call starts a new run. A report for an already consumed receipt is still accepted after a kill and is recorded as after_kill. A run that only ended (idle) is replaced transparently and logged at INFO on the aynigma.agents logger.

Approvals: pause and resume

A rule with effect REVIEW raises ApprovalRequired until a person approves the call in Agent IAM > Activity. The approval belongs to the run it was raised in, so save approval_id and run_id when you catch it. If the work continues in another process (a queue worker, a restart), build the new authorizer with that run_id. Without it the new process starts a fresh run and the call is denied with run_mismatch.

from aynigma import AgentAuthorizer, ApprovalRequired

authorizer = AgentAuthorizer(agent_id=AGENT_ID, agent_key=AGENT_KEY, base_url=BASE_URL)
try:
    authorizer.run_tool(tool, params, handler)
except ApprovalRequired as pending:
    saved_approval_id, saved_run_id = pending.approval_id, pending.run_id   # store these with your task

# Later, after a reviewer approved it, possibly in a new process:
resumed = AgentAuthorizer(agent_id=AGENT_ID, agent_key=AGENT_KEY, base_url=BASE_URL, run_id=saved_run_id)
resumed.run_tool(tool, params, handler, approval_id=saved_approval_id)   # same tool, same params

run_id is None with the legacy X-AGENT-KEY header, which has no runs and needs no run_id to resume. If you pass your own token_provider, give that AgentTokenProvider the run_id instead; passing run_id to AgentAuthorizer as well raises ValueError. The arguments must match what was approved, or the call is denied with payload_mismatch.

Denials explain themselves: str(ToolDenied) starts with Agent tool call denied: <code> and adds a short hint for common codes (no_policy, run_mismatch, payload_mismatch, approval_already_used, approval_expired, environment, rate_limited). error.reason is the bare code and error.hint the hint, or None.

Debug a token from a shell: AYNIGMA_AGENT_KEY=agt_... aynigma-agents-token --agent-id <uuid> --claims. The key is never printed.

Verify an audit export offline

Auditors can check an audit export and its signed anchor without trusting the API. Standard library only.

aynigma-audit-verify --export audit.json --anchor anchor.json \
  --jwks https://api.aynigma.ai/api/v1/agents/.well-known/jwks.json \
  --chain-key-file chain.key --full-chain

audit.json is GET /api/v1/audit/export/?log_type=audit&export_format=json (an object with a chain header and records) and anchor.json is GET /api/v1/audit/anchors/<id>/. With the chain key (--chain-key-file, or AYNIGMA_AUDIT_CHAIN_KEY; --chain-key works but shows in the process list) every row is recomputed; without it only legacy rows are, and the report lists what was skipped. Legacy (v1) rows need no key, so they are trusted only up to the last_v1_entry_id signed inside a verified anchor (the export header's value is unsigned and never grants trust); a v1 row above it, or after a keyed row, is a forgery and fails. An export of only legacy rows with no verified anchor is INCONCLUSIVE. --full-chain means the export is the unfiltered chain, so each row must follow its predecessor.

Exit status: 0 pass, 1 a check failed, 2 inconclusive, 3 unusable input. PASS needs every row recomputed with the key, or an anchor whose head is in the export with an unbroken chain back from it (printed as "structure only", since v2 contents are not recomputed without the key). With neither the result is INCONCLUSIVE, never a pass. The hashing rules are in docs/AUDIT_CHAIN.md in the backend repo.

Migrating from X-AGENT-KEY: nothing breaks. A key with no agent ID keeps sending the header, and use_legacy_key=True (or AYNIGMA_USE_LEGACY_KEY=1) keeps it even when an agent ID is set. Since 0.3.0, AgentAuthorizer switches to tokens as soon as it has both an agent ID and a key. An explicit agent_key= with no agent_id= stays legacy, and AYNIGMA_AGENT_ID from the environment counts only together with AYNIGMA_AGENT_KEY from the environment. To move over, pass agent_id= or set both variables, and watch the agent.legacy_key_used audit event stop. acting_for_user and initiator_token_provider now exchange the user JWT instead of forwarding it; with the legacy header they forward it as before. Delegated tokens carry the delegation depth counted from the human (a direct human to agent hop is depth 1), so review max_delegation_depth conditions. See docs/AGENT_TOOL_EXECUTOR.md and docs/AGENT_IDENTITY.md.

Beyond the quickstart, use AgentAuthorizer and RemoteTool with a separately deployed receipt-enforcing executor. Optional adapters are as_langchain_tool, as_openai_agents_tool, as_crewai_tool, as_claude_agent_tool, as_google_adk_tool, and as_semantic_kernel_function; install their corresponding extras and pass a Pydantic argument model. OpenAIAgentsApprovalBridge uses the OpenAI Agents SDK's native approval interruption to preserve the exact model-selected call through backend REVIEW and receipt-backed resume. build_crewai_approval_flow uses CrewAI Flow human feedback and explicit persistence for the same handoff. ClaudeAgentApprovalBridge pairs the Claude Agent SDK's native PreToolUse defer hook with a receipt-backed MCP tool and caller-owned pending state. SemanticKernelApprovalBridge intercepts Semantic Kernel auto function invocation before a model-selected call and supports exact-call REVIEW resume through a caller-owned checkpoint. as_google_adk_approval_tool uses ADK's native confirmation event and persisted session state. Install aynigma[google-adk-postgres] for ADK's PostgreSQL session service; a local restart contract passed against a disposable PostgreSQL database. build_langgraph_approval_workflow adds a checkpointed pause/resume graph for REVIEW decisions and requires an explicit checkpointer. Install aynigma[langgraph-postgres] and use LangGraph's PostgresSaver for durable resume. See docs/AGENT_TOOL_EXECUTOR.md in the backend repository for the executor boundary and examples.

PostgresApprovalCheckpointStore is an optional encrypted store for serializable pending approval state shared across workers. Install aynigma[approval-postgres], call setup() once, then save() and one-use claim() with authenticated tenant and agent UUIDs. Schedule aynigma-approval-checkpoints-purge daily with AYNIGMA_APPROVAL_CHECKPOINT_DATABASE_URL supplied from the operator's secret store; it prints only the number purged. The application still serializes framework-specific session state and authorizes its resume endpoint. See the executor guide for retention and recovery rules.

from aynigma import Aynigma, guard_openai_chat_v2

with Aynigma() as client:
	response = guard_openai_chat_v2(
		client,
		openai_client,
		messages,
		model="gpt-4o-mini",
		guard_options={"system_id": 4, "environment_id": 12},
	)

Bedrock agent inventory

On a paid plan, run aynigma-agents-discover-cloud --company-id 42 --account-id 123456789012 --region us-east-1 with AYNIGMA_USER_JWT, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and optional AWS_SESSION_TOKEN set. The command sends the credentials once and prints only the bounded agent inventory. The backend checks the AWS account through STS before ListAgents and does not retain the credentials. See docs/AGENT_BEDROCK_DISCOVERY.md for scope and limits.

For Vertex Agent Engine, set AYNIGMA_USER_JWT and GOOGLE_ACCESS_TOKEN, then run aynigma-agents-discover-vertex --company-id 42 --project-id sample-project --region us-central1. The backend confirms the active Google Cloud project before listing reasoning engines. See docs/AGENT_VERTEX_DISCOVERY.md.

For a known Azure AI Foundry project, set AYNIGMA_USER_JWT and AZURE_ACCESS_TOKEN, then run aynigma-agents-discover-azure --company-id 42 --project-endpoint https://account.services.ai.azure.com/api/projects/project. See docs/AGENT_AZURE_DISCOVERY.md.

To find Foundry projects in a subscription first, set a separate AZURE_ARM_ACCESS_TOKEN and run aynigma-agents-discover-azure --company-id 42 --subscription-id 11111111-2222-4333-8444-555555555555. This lists bounded project metadata only; use a project token and endpoint to list agents.

For aggregate native Assess plugin review coverage, run aynigma-assess-plugin-performance --project-id 42 --days 90 with AYNIGMA_USER_JWT. It returns counts only; see docs/ASSESS_PLUGIN_PERFORMANCE.md for interpretation.

When registering an agent through the dashboard API, an administrator may provide a same-company project ID to attribute approvals and incidents to that project's compliance evidence. The project cannot be changed after registration. See docs/AGENT_COMPLIANCE_EVIDENCE.md.

Administrators can read registered-agent slot usage at GET /api/v1/agents/usage/?company=<id>. Free includes two slots; current paid caps are documented in docs/AGENT_PLAN_LIMITS.md.

For agent tools acting on behalf of an authenticated person, wrap the tool invocation in with acting_for_user(user_jwt): after importing acting_for_user from aynigma. The binding is scoped to the current Python task and carries through async framework adapters; see docs/AGENT_TOOL_EXECUTOR.md.

Receipts are required

Run in-process tools with auth.run_tool("ticket.create", {"title": "x"}, handler) or RemoteTool.local(auth, "ticket.create", handler). Both authorize, verify the one-use receipt, run the handler and report SUCCEEDED or FAILED. ToolExecutor refuses a call without a receipt unless you pass require_receipt=False, which is only for a backend running with AGENT_REQUIRE_EXECUTION_RECEIPT=False. See docs/AGENT_TOOL_EXECUTOR.md.