Agent Enrollment Protocol for Python

Official Python software development kit for the Agent Enrollment Protocol, the open protocol for Agent enrollment, Service-issued credentials, and authenticated Agent access.

Installation

python -m pip install agent-enrollment-protocol

Python 3.11 or newer is required. The distribution provides one typed package with modules for each integration role:

Core is synchronous and transport-independent. Agent, Service, Platform, and their integration interfaces are asynchronous. Applications provide durable stores and security policy through typed protocols.

Framework integrations are optional and remain separate from Core and role behavior.

Runnable examples

The repository includes self-contained examples that exercise real signed protocol flows without external infrastructure:

uv run python examples/agent_service.py
uv run python examples/hosted_platform.py

The first example composes an Agent, Service, JWT authentication, every built-in credential profile, a protected resource, and the framework-neutral ASGI adapter. The second demonstrates Platform discovery, Service DID resolution, provisioning, DID publication, verified delegated signing, identity listing, and suspension that prevents further signing. See examples/README.md for the integration boundaries and production replacements.

The Service module includes stored credential profiles for each built-in Grant Type:

Each factory accepts application-owned credential issuance and storage implementations. The included memory store is intended for examples and local development.

Concrete Grant Types and extensions can define additional Grant and Revoke request members. Pass those members through GrantOptions.parameters or RevokeOptions.parameters; the Agent retains control of the standard Grant Type and Revoke selector fields.

Protected HTTP requests

After explicitly enrolling an Agent with a Service, use fetch_protected_resource() to request a protected URL. It returns the SDK's transport-independent HttpResponse:

from agent_enrollment_protocol.agent import (
    FetchProtectedResourceOptions,
    fetch_protected_resource,
)
from agent_enrollment_protocol.core import HttpRequest

response = await fetch_protected_resource(
    FetchProtectedResourceOptions(
        agent=agent,
        request=HttpRequest(method="GET", url="https://service.example/private"),
        carrier="AEP-Authorization",
    )
)

The helper uses advertised OpenAPI security when it can resolve a complete supported requirement; otherwise it waits for a valid 401 AEP challenge. It reuses stored credentials or signs a resource-bound assertion, and can request a supported Grant for an already-enrolled identity. It does not enroll automatically. Set grant_type to select a credential type, or credential_id to select one stored credential. A missing credential_id match fails instead of requesting a replacement Grant.

Requests use replayable byte bodies. Redirects are handled manually: each same-origin JWT retry gets a fresh assertion bound to its target, and an origin change drops caller headers and requires a new anonymous AEP challenge. The defaults are five redirects and thirty seconds for the entire operation, including discovery and authentication. Task cancellation propagates to the active operation. Responses obey the Agent's maximum_response_bytes bound.

Put payment credentials and custom secret headers in additional_authentication_headers, not request.headers. They are sent only on authenticated requests to the original origin. Select carrier="AEP-Authorization" when payment authentication occupies Authorization. Additional headers cannot override the selected AEP credential.

An optional transport implements AsyncHttpTransport; it must return redirects without following them. The helper closes its default HttpxTransport after the operation but does not close a caller-provided transport. Existing ServiceSession.authentication_headers() remains available for applications that manage their own HTTP requests.

Service ASGI integration

agent_enrollment_protocol.adapters provides a framework-neutral ASGI integration with no additional dependency. AepAsgiApplication serves Inspect and every command advertised by the Service. It enforces the command methods, media type, request-body limit, and idempotency header boundary, and supplies cache metadata and conditional requests for Inspect.

AepAuthenticationMiddleware protects a downstream ASGI application and exposes the authenticated Agent principal through principal_from_scope(). Place the protocol application outside the authentication middleware so AEP's own command routes remain directly accessible:

from agent_enrollment_protocol.adapters import (
    AepAsgiApplication,
    AepAuthenticationMiddleware,
    principal_from_scope,
)

protected_application = AepAuthenticationMiddleware(
    application,
    service,
    resource_origin="https://service.example",
)
asgi_application = AepAsgiApplication(service, protected_application)

The downstream application can obtain its immutable principal from the ASGI scope:

principal = principal_from_scope(scope)
if principal is None:
    raise RuntimeError("The route requires AEP authentication")

The application then applies its own account permissions and scope checks. Authentication alone does not authorize every operation. The complete Service construction, credential issuers and example DID verifier are in examples/agent_service.py.

Use a separate unprotected application branch for public resources. For local development, allow_insecure_loopback=True permits an HTTP localhost or loopback resource origin; production origins require HTTPS.

Agent with a hosted identity Platform

PlatformIdentityProvider lets an Agent use a remote AEP Platform for Service-scoped identity custody and delegated assertion signing. It discovers the Platform, recovers an existing active identity before provisioning one, caches discovery metadata according to HTTP cache directives, and supplies the resulting signer directly to Agent.

import os

from agent_enrollment_protocol.agent import (
    Agent,
    AgentOptions,
    PlatformIdentityProvider,
    PlatformIdentityProviderOptions,
)


async def authentication_headers() -> dict[str, str]:
    return {"Authorization": f"Bearer {os.environ['AEP_PLATFORM_ACCESS_TOKEN']}"}


async with PlatformIdentityProvider(
    PlatformIdentityProviderOptions(
        authentication_headers=authentication_headers,
        platform_url="https://platform.example",
    )
) as identities:
    async with Agent(AgentOptions(identity_provider=identities)) as agent:
        result = await agent.service("https://service.example").enroll()

The Platform authentication callback is evaluated for each private request so applications can refresh short-lived credentials. Supply pending_sign_resolver when the Platform can return 202 Accepted during delegated signing. The resolver receives the immutable retry interval and opaque Platform context; returning updated context starts the next signing stage with a distinct idempotency key. Without a resolver, pending signing raises PlatformSignPendingError for the application to continue explicitly.

Grant recovers an existing identity through find_identity_by_service_did when the identity provider implements ExistingIdentityProvider and the Agent's local identity store is empty. Recovery does not provision or enroll an identity. The Agent checks Service Status before requesting credentials when the Service advertises Status.

The default HTTP transport requires TLS 1.3 or later and verifies certificates using the system trust configuration. Applications supplying an HTTP client or transport control its TLS settings and must enforce the same protocol minimum. The SDK does not change a supplied client's settings.

Hosted identity Platform

agent_enrollment_protocol.platform implements discovery, Service-scoped Agent identity provisioning, DID document publication, identity listing and lifecycle, delegated signing, and optional hosted verification.

Applications supply caller authorization, Service DID resolution, key custody, and durable stores. The included memory stores are suitable for local development, not production key custody or durable idempotency.

from datetime import timedelta

from agent_enrollment_protocol.core import SigningAlgorithm
from agent_enrollment_protocol.platform import DiscoveryOptions, Platform, PlatformOptions

platform = Platform(
    PlatformOptions(
        authorizer=authorizer,
        did_host="platform.example",
        did_url_template="https://platform.example/agents/{agent_did_id}/did.json",
        discovery=DiscoveryOptions(
            endpoint_base="/v1/aep",
            lifecycle_endpoint="/v1/aep/agent-identities/{agent_identity_id}",
            list_endpoint="/v1/aep/agent-identities",
            platform_name="Example Platform",
            provision_endpoint="/v1/aep/agent-identities",
            sign_endpoint="/v1/aep/agent-identities/{agent_identity_id}/sign",
        ),
        key_store=key_store,
        maximum_lifetime=timedelta(minutes=5),
        service_did_resolver=service_did_resolver,
        signing_algorithms=(SigningAlgorithm.ES256,),
    )
)

Map platform.discovery() to /.well-known/aep-platform and the remaining methods to the paths advertised by DiscoveryOptions. Authenticate private Platform routes before constructing their RequestContext; the Platform also invokes the supplied authorizer for every private operation. Enable hosted verification only with a replay store.

For hosted verification, the authorizer decides whether the authenticated caller may verify the supplied Agent assertion. A trusted Service can call with its own credentials; it does not need the identity owner's credentials. Authorize that relationship explicitly rather than permitting every authenticated caller to verify every identity. Signing and identity management retain their owner checks.

Application-owned storage

Default memory stores retain state only for the lifetime of one process. To retain an Agent's identity and session credentials across restarts, supply AgentOptions.identity_store and credential_store. A Service supplies durable enrollment, credential, replay and idempotency stores; a Platform supplies identity, key, idempotency and, when using hosted verification, replay storage. Inspect and discovery caches can remain transient.

Custom stores implement the asynchronous interfaces exported by their role module. In particular, Service EnrollmentStore.find_or_create() must atomically reuse or create one enrollment for an Agent DID. Service IdempotencyStore.execute() and Platform PlatformIdempotencyStore.execute() must atomically run or replay an operation and reject conflicting reuse of the key. Retain command results for at least one hour. Replay-store consume() must atomically refuse a second use of an assertion identifier. Independent reads followed by writes do not provide these guarantees when requests run concurrently or the application has multiple instances.

Upgrading persisted state from 0.1.1

Custom Service stores continue receiving ServiceCredentialRecord with a BuiltInCredential. There is no conversion to a verifier-only format in that public storage contract. Salted verifier changes belong to the private MemoryServiceCredentialStore representation, which exists only for the lifetime of its process. Do not delete or reissue credentials to upgrade a durable store. Keep identities, enrollments, revocation state, and completed command-idempotency responses too; deleting completed responses can repeat operations.

Applications that persist InspectCacheEntry must serialize its age, date, and expires fields along with cache_control, validators, document, final URL, and cached_at. The additional fields default to empty strings, so old constructors remain callable, but they cannot recover freshness information that was not stored. Invalidate only old Inspect-cache entries and let the Agent fetch them again. Do not clear credential or command-idempotency storage with that cache. For mixed-version processes, use separate Inspect-cache namespaces so an older writer cannot discard the metadata required by an upgraded reader.

Development

Install the locked development environment and run the complete merge gate:

uv sync --all-groups --locked
make verify

Run the shared Agent, Service, and Platform conformance suites against the public Python APIs:

make conformance

The command reads the adjacent ../aep-specs checkout by default and writes role reports to .conformance/reports/. Set AEP_SPECS_DIR when the specifications are checked out elsewhere.

Run bidirectional Agent, Service, and Platform interoperability against the Node.js SDK:

make interoperability

The command reads the adjacent ../aep-node checkout by default and writes a four-row evidence report to .interop/reports/aep-python-node-interoperability.json. Set AEP_NODE_DIR when the Node.js SDK is checked out elsewhere.

See aep-specs for the normative drafts, schemas, registries, examples, and test vectors.

Releases

Maintainers run the Release workflow from main. It verifies the package and a clean consumer, runs shared conformance and Node.js interoperability, publishes through PyPI Trusted Publishing, attests the distributions, and creates the matching tag and GitHub release with the verification reports.

Security

See SECURITY.md for vulnerability reporting.

License

MIT.