aria-agentkit

Version-pinned ARIA backend integration for the Microsoft Agent Governance Toolkit.

aria-agentkit is a first-class, contract-tested backend integration for the official AGT governance surfaces. It connects your AGT-governed agents to EmpowerID ARIA — the enterprise governance platform for AI agents — through rigorous adapters for policy evaluation, governance-grade audit export, and MCP remote configuration.

Built on authzen-policy-backend, which provides the standalone AuthZEN backend for any PolicyEvaluator use case. This package adds AGT-native extension point adapters on top.

What This Package Does

Integration	AGT Extension Point	ARIA Service
ARIAToolInterceptor	ToolCallInterceptor	AuthZEN PDP
ARIAPolicyProvider	PolicyProviderInterface	AuthZEN PDP
ARIAAuditBackend	AuditBackend	Receipt Vault
claude_desktop_config	—	MCP Gateway
mcp_session_params	—	MCP Gateway
mcp_session_params_factory	—	MCP Gateway

What This Package Does NOT Do

• Reimplement AGT governance primitives
• Proxy LLM traffic (see aria-shield-sdk for that)
• Provide budget management (deferred until transactional semantics are designed)
• Provide approval workflows (deferred until persisted workflow model is built)

Version Compatibility

aria-agentkit	agent-os-kernel	Python
0.1.x	3.0.0 – 3.0.x	>= 3.10

The compat module detects the installed AGT version at import time and raises RuntimeError on unsupported versions.

Installation

pip install aria-agentkit

Quick Start

1. Tool Call Interception via AuthZEN PDP

Every tool call is evaluated against the ARIA PDP before execution. The interceptor posts a rigorous AuthZEN 1.0 request directly — no re-mapping, no abstraction leaks.

from agent_os.integrations.base import CompositeInterceptor
from aria_agentkit import ARIAToolInterceptor

with ARIAToolInterceptor(
    pdp_url="https://pdp.example.com/access/v1/evaluation",
    pdp_application="my-agent-platform",
    token="my-bearer-token",
) as interceptor:
    # Register with AGT's composite interceptor
    composite = CompositeInterceptor()
    composite.add(interceptor)

    # Every tool call now gets policy-checked:
    #   tool_name  →  action.name = "tool.invoke"
    #   agent_id   →  subject.id  = "auth:agent:agentmesh:{agent_id}"
    #   arguments  →  context.original_args_hash (SHA-256)

2. Governance-Grade Audit Export

Audit entries are redacted, hash-chained into tamper-evident receipts, and exported to Receipt Vault with retry and dead-letter routing.

from agent_os.audit_logger import GovernanceAuditLogger
from aria_agentkit import ARIAAuditBackend

dead_letters = []

with ARIAAuditBackend(
    receipt_vault_url="https://receipts.example.com",
    dead_letter_callback=lambda batch, reason: dead_letters.extend(batch),
) as audit:
    logger = GovernanceAuditLogger()
    logger.add_backend(audit)

    # Entries are automatically:
    # 1. Redacted  (password, secret, token, api_key, credential)
    # 2. Hash-chained  (each receipt links to the previous via prev_hash)
    # 3. Queued with idempotency keys
    # 4. Exported with exponential backoff (capped at 5s)

3. Policy Provider with Agent Filtering

from aria_agentkit import ARIAPolicyProvider

with ARIAPolicyProvider(
    pdp_base_url="https://pdp.example.com",
    pdp_application="my-agent-platform",
    token="my-bearer-token",
    cache_ttl=300.0,
) as provider:
    # Fetch policies, filtered by agent identity
    policies = provider.get_policies(agent_id="hr-onboarding-agent")

    # Policies with metadata.agent_id matching "hr-onboarding-agent"
    # are returned, plus all untagged (global) policies.

4. Claude Desktop MCP Configuration

from aria_agentkit.mcp.claude_desktop import claude_desktop_config

config = claude_desktop_config(
    server_name="aria-gateway",
    gateway_url="https://mcp.example.com/v1/mcp",
    token="my-bearer-token",
)
# Returns a dict ready to merge into Claude Desktop's mcpServers config
# with Streamable HTTP transport, Bearer auth, and MCP protocol version

5. MCP SDK with Refreshable Tokens

For production MCP connections where tokens expire:

from authzen_backend.auth import ClientCredentialsAuth
from aria_agentkit.mcp.sdk_session import mcp_session_params_factory
import httpx

creds = ClientCredentialsAuth(
    token_endpoint="https://idp.example.com/oauth/token",
    client_id="my-mcp-client",
    client_secret="...",
)
client = httpx.Client()

# Factory produces fresh params on each call — token is never stale
factory = mcp_session_params_factory(
    gateway_url="https://mcp.example.com/v1/mcp",
    token_factory=lambda: creds.get_token(client),
)

params = factory()  # Fresh token, proper Origin header, streamable-http

For static tokens, use the simpler mcp_session_params() instead.

Architecture

graph LR
    subgraph agt [AGT Runtime]
        PE[PolicyEvaluator]
        TCI[ToolCallInterceptor]
        GAL[GovernanceAuditLogger]
    end

    subgraph kit [aria-agentkit]
        APP[ARIAPolicyProvider]
        ATI[ARIAToolInterceptor]
        AAB["ARIAAuditBackend<br/>(outbox + hash chain)"]
        MCP[MCP config emitters]
    end

    subgraph aria [ARIA Services]
        PDP[AuthZEN PDP]
        RV[Receipt Vault]
        GW[MCP Gateway]
    end

    PE --> APP
    TCI --> ATI
    GAL --> AAB

    APP --> PDP
    ATI --> PDP
    AAB --> RV
    MCP --> GW

AuthZEN Request Mapping

The interceptor maps AGT ToolCallRequest fields into OpenID AuthZEN 1.0 evaluation requests:

AuthZEN Field	Value
action.name	"tool.invoke" (constant)
resource.type	"mcp_tool"
resource.id	"mcp://{server_name}/{tool_name}"
subject.type	"agent"
subject.id	"auth:agent:agentmesh:{agent_id}"
context.call_id	From ToolCallRequest.call_id
context.original_args_hash	SHA-256 of canonical arguments
context.pdp_application	Constructor parameter

PDP constraints in the response are mapped to ToolCallResult.modified_arguments for parameter clamping.

graph LR
    TCR["ToolCallRequest<br/>tool_name, agent_id,<br/>arguments, metadata"] --> MAP[map_tool_call]
    MAP --> AZR["AuthZEN Request<br/>subject, action,<br/>resource, context"]
    AZR --> PDP[POST to PDP]
    PDP --> RESP["{ decision, context }"]
    RESP --> TCRes["ToolCallResult<br/>allowed, reason,<br/>modified_arguments,<br/>audit_entry"]

Audit Pipeline

flowchart TD
    E[AuditEntry] --> R[Redact sensitive keys]
    R --> H[Hash-chain with prev_hash]
    H --> Q[Queue in outbox]
    Q --> F{Flush to Receipt Vault}
    F -->|Success| D[Done]
    F -->|Failure| RT[Retry with backoff]
    RT -->|"Max retries exceeded"| DL[Dead-letter callback]
    RT -->|Retry succeeds| D

Properties:

• Redaction — password, secret, token, api_key, credential are recursively redacted before persistence (including inside nested lists)
• Hash Chaining — Each receipt includes a prev_hash linking to the prior entry, producing a tamper-evident chain
• Idempotency — Each receipt gets a unique idempotency key for deduplication
• Retry with Backoff — Failed exports are retried with exponential backoff (capped at max_backoff, default 5.0s)
• Dead Letter — Persistently failing batches are routed to a configurable callback

Note: The outbox is in-memory. Unflushed entries are lost on process termination. Call flush() or close() (or use the context manager) before shutdown.

MCP Configuration

Two helpers are provided for connecting to the ARIA MCP Gateway:

Helper	Use Case	Token
mcp_session_params()	Static token, simple setup	str
mcp_session_params_factory()	Refreshable tokens, production	Callable[[], str]

Both produce dicts with url, headers (Bearer auth, Mcp-Protocol-Version, Origin), and transport: "streamable-http" — ready for the MCP Python SDK or Claude Desktop.

Security

• Fail-closed by default — PDP communication errors produce denial, not silent pass-through. Set fail_closed=False to propagate exceptions instead.
• Recursive redaction — Sensitive keys are scrubbed from audit payloads before they enter the hash chain, including values nested inside lists.
• No credentials in repr — __repr__ on all classes shows configuration (URL, application) but never tokens or secrets.
• HTTPS enforced — The underlying authzen-policy-backend enforces HTTPS for PDP URLs in production.

See SECURITY.md for reporting vulnerabilities.

Testing

87 tests | 97% coverage | mypy strict | ruff | contract tests against AGT 3.0.x

pip install -e ".[dev]"
pytest tests/ -v
ruff check src/ tests/
mypy src/

Test categories:

• Unit tests — mapping, interceptor, provider, audit, MCP config
• Contract tests — ToolCallInterceptor, AuditBackend, PolicyProviderInterface, PolicyEvaluator against real AGT types
• Golden tests — snapshot assertions on AuthZEN request mapping
• Concurrency tests — 50 concurrent interceptor evaluations, 100 concurrent audit writers

Ecosystem

┌─────────────────────────────────────────────────┐
│         aria-agentkit (this package)             │
│  AGT-native adapters for ARIA services           │
│  Depends on: authzen-policy-backend              │
├─────────────────────────────────────────────────┤
│         authzen-policy-backend                   │
│  Standalone AuthZEN ExternalPolicyBackend         │
│  Zero dependency on agent-os-kernel              │
└─────────────────────────────────────────────────┘

• authzen-policy-backend — Use this if you only need PolicyEvaluator.add_backend() without AGT extension points. Zero dependency on agent-os-kernel.
• aria-agentkit (this package) — Use this if you're building on the full AGT stack and want ToolCallInterceptor, AuditBackend, PolicyProviderInterface, and MCP config helpers.
• aria-shield-sdk (coming soon) — LLM traffic proxying through ARIA Shield. Separate from governance.

Development

git clone https://github.com/EmpowerID/aria-agentkit.git
cd aria-agentkit
pip install -e ".[dev]"
pytest tests/ -v
ruff check src/ tests/
mypy src/

Links

• PyPI
• authzen-policy-backend
• Microsoft Agent Governance Toolkit
• EmpowerNow ARIA
• OpenID AuthZEN Specification

License

MIT — see LICENSE.