acapy_agent.core package

Submodules

acapy_agent.core.conductor module

acapy_agent.core.dispatcher module

acapy_agent.core.error module

Common exception classes.

exception acapy_agent.core.error.BaseError(*args, error_code: str | None = None, **kwargs)

Bases: Exception

Generic exception class which other exceptions should inherit from.

property message: str

Accessor for the error message.

property roll_up: str

Accessor for nested error messages rolled into one line.

For display: aiohttp.web errors truncate after newline.

exception acapy_agent.core.error.ProfileDuplicateError(*args, error_code: str | None = None, **kwargs)

Bases: ProfileError

Profile with the given name already exists.

exception acapy_agent.core.error.ProfileError(*args, error_code: str | None = None, **kwargs)

Bases: BaseError

Base error for profile operations.

exception acapy_agent.core.error.ProfileNotFoundError(*args, error_code: str | None = None, **kwargs)

Bases: ProfileError

Requested profile was not found.

exception acapy_agent.core.error.ProfileSessionInactiveError(*args, error_code: str | None = None, **kwargs)

Bases: ProfileError

Error raised when a profile session is not currently active.

exception acapy_agent.core.error.ProtocolDefinitionValidationError(*args, error_code: str | None = None, **kwargs)

Bases: BaseError

Error raised when there is a problem validating a protocol definition.

exception acapy_agent.core.error.ProtocolMinorVersionNotSupported(*args, error_code: str | None = None, **kwargs)

Bases: BaseError

Minimum minor version protocol error.

Error raised when protocol support exists but minimum minor version is higher than in @type parameter.

exception acapy_agent.core.error.StartupError(*args, error_code: str | None = None, **kwargs)

Bases: BaseError

Error raised when there is a problem starting the conductor.

acapy_agent.core.event_bus module

A simple event bus.

class acapy_agent.core.event_bus.Event(topic: str, payload: Any | None = None)

Bases: object

A simple event object.

property payload

Return this event’s payload.

property topic: str

Return this event’s topic.

with_metadata(metadata: EventMetadata) → EventWithMetadata

Annotate event with metadata and return EventWithMetadata object.

class acapy_agent.core.event_bus.EventBus

Bases: object

A simple event bus implementation.

async notify(profile: Profile, event: Event)

Notify subscribers of event.

Parameters:

• profile (Profile) – context of the event

• event (Event) – event to emit

async shutdown()

Shutdown the event bus and clean up background tasks.

subscribe(pattern: Pattern, processor: Callable)

Subscribe to an event.

Parameters:

• pattern (Pattern) – compiled regular expression for matching topics

• processor (Callable) – async callable accepting profile and event

unsubscribe(pattern: Pattern, processor: Callable)

Unsubscribe from an event.

This method is idempotent. Repeated calls to unsubscribe will not result in errors.

Parameters:

• pattern (Pattern) – regular expression used to subscribe the processor

• processor (Callable) – processor to unsubscribe

wait_for_event(waiting_profile: Profile, pattern: Pattern, cond: Callable[[Event], bool] | None = None) → Iterator[Awaitable[Event]]

Capture an event and retrieve its value.

class acapy_agent.core.event_bus.EventMetadata(pattern: Pattern, match: Match[str])

Bases: NamedTuple

Metadata passed alongside events to add context.

match: Match[str]

Alias for field number 1

pattern: Pattern

Alias for field number 0

class acapy_agent.core.event_bus.EventWithMetadata(topic: str, payload: Any, metadata: EventMetadata)

Bases: Event

Event with metadata passed alongside events to add context.

property metadata: EventMetadata

Return metadata.

class acapy_agent.core.event_bus.MockEventBus

Bases: EventBus

A mock EventBus for testing.

async notify(profile: Profile, event: Event)

Append the event to MockEventBus.events.

async shutdown()

Mock shutdown method for testing.

acapy_agent.core.goal_code_registry module

Handle registration and publication of supported goal codes.

class acapy_agent.core.goal_code_registry.GoalCodeRegistry

Bases: object

Goal code registry.

goal_codes_matching_query(query: str) → Sequence[str]

Return a list of goal codes matching a query string.

register_controllers(*controller_sets)

Add new controllers.

Parameters:

controller_sets – Mappings of controller to coroutines

acapy_agent.core.oob_processor module

acapy_agent.core.plugin_registry module

Handle registration of plugin modules for extending functionality.

class acapy_agent.core.plugin_registry.PluginRegistry(blocklist: Set[str] | None = None)

Bases: object

Plugin registry for indexing application plugins.

async init_context(context: InjectionContext) → None

Call plugin setup methods on the current context.

async load_protocol_version(context: InjectionContext, mod: ModuleType, version_definition: dict | None = None) → None

Load a particular protocol version.

async load_protocols(context: InjectionContext, plugin: ModuleType) → None

For modules that don’t implement setup, register protocols manually.

property plugin_names: Sequence[str]

Accessor for a list of all plugin modules.

property plugins: Sequence[ModuleType]

Accessor for a list of all plugin modules.

post_process_routes(app) → None

Call route binary file response OpenAPI fixups if applicable.

async register_admin_routes(app) → None

Call route registration methods on the current context.

register_package(package_name: str) → Sequence[ModuleType]

Register all modules (sub-packages) under a given package name.

register_plugin(module_name: str) → ModuleType | None

Register a plugin module.

register_protocol_events(context: InjectionContext) → None

Call route register_events methods on the current context.

validate_version(version_list, module_name)

Validate version dict format.

acapy_agent.core.profile module

Classes for managing profile information within a request context.

class acapy_agent.core.profile.Profile(*, context: InjectionContext | None = None, name: str | None = None, created: bool = False)

Bases: ABC

Base abstraction for handling identity-related state.

BACKEND_NAME: str | None = None

DEFAULT_NAME: str = 'default'

property backend: str

Accessor for the backend implementation name.

async close()

Close the profile instance.

property context: InjectionContext

Accessor for the injection context.

property created: bool

Accessor for the created flag indicating a new profile.

inject(base_cls: Type[InjectType], settings: Mapping[str, object] = None) → InjectType

Get the provided instance of a given class identifier.

Parameters:

• cls – The base class to retrieve an instance of

• base_cls – The base class to retrieve

• settings – An optional mapping providing configuration to the provider

Returns:

An instance of the base class, or None

inject_or(base_cls: Type[InjectType], settings: Mapping[str, object] = None, default: InjectType | None = None) → InjectType | None

Get the provided instance of a given class identifier or default if not found.

Parameters:

• base_cls – The base class to retrieve an instance of

• settings – An optional dict providing configuration to the provider

• default – default return value if no instance is found

Returns:

An instance of the base class, or None

property is_anoncreds: bool

Check if this profile uses an AnonCreds-compatible backend.

property name: str

Accessor for the profile name.

async notify(topic: str, payload: Any)

Signal an event.

async remove()

Remove the profile.

abstractmethod session(context: InjectionContext | None = None) → ProfileSession

Start a new interactive session with no transaction support requested.

property settings: BaseSettings

Accessor for scope-specific settings.

abstractmethod transaction(context: InjectionContext | None = None) → ProfileSession

Start a new interactive session with commit and rollback support.

If the current backend does not support transactions, then commit and rollback operations of the session will not have any effect.

class acapy_agent.core.profile.ProfileManager

Bases: ABC

Handle provision and open for profile instances.

abstractmethod async open(context: InjectionContext, config: Mapping[str, Any] = None) → Profile

Open an instance of an existing profile.

abstractmethod async provision(context: InjectionContext, config: Mapping[str, Any] = None) → Profile

Provision a new instance of a profile.

class acapy_agent.core.profile.ProfileManagerProvider

Bases: BaseProvider

The standard profile manager provider which keys off the selected wallet type.

MANAGER_TYPES = {'askar': 'acapy_agent.askar.profile.AskarProfileManager', 'askar-anoncreds': 'acapy_agent.askar.profile_anon.AskarAnonProfileManager', 'kanon-anoncreds': 'acapy_agent.kanon.profile_anon_kanon.KanonAnonProfileManager'}

provide(settings: BaseSettings, injector: BaseInjector)

Create the profile manager instance.

class acapy_agent.core.profile.ProfileSession(profile: Profile, *, context: InjectionContext | None = None, settings: Mapping[str, Any] = None)

Bases: ABC

An active connection to the profile management backend.

property active: bool

Accessor for the session active state.

async commit()

Commit any updates performed within the transaction.

If the current session is not a transaction, then nothing is performed.

property context: InjectionContext

Accessor for the associated injection context.

async emit_event(topic: str, payload: Any, force_emit: bool = False)

Emit an event.

If we are in an active transaction, just queue the event, otherwise emit it.

Parameters:

• topic (str) – The topic of the event.

• payload (Any) – The payload of the event.

• force_emit (bool, optional) – If True, force the event to be emitted even if there is an active transaction. Defaults to False.

property handle: ProfileSessionHandle

Accessor for the session handle.

inject(base_cls: Type[InjectType], settings: Mapping[str, object] = None) → InjectType

Get the provided instance of a given class identifier.

Parameters:

• base_cls (Type[InjectType]) – The base class to retrieve an instance of.

• settings (Mapping[str, object], optional) – An optional mapping providing configuration to the provider.

Returns:

An instance of the base class, or None.

Return type:

InjectType

Raises:

ProfileSessionInactiveError – If the profile session is inactive.

inject_or(base_cls: Type[InjectType], settings: Mapping[str, object] = None, default: InjectType | None = None) → InjectType | None

Get the provided instance of a given class identifier or default if not found.

Parameters:

• base_cls – The base class to retrieve an instance of

• settings – An optional dict providing configuration to the provider

• default – default return value if no instance is found

Returns:

An instance of the base class, or None

property is_transaction: bool

Check if the session supports commit and rollback operations.

property profile: Profile

Accessor for the associated profile instance.

async rollback()

Roll back any updates performed within the transaction.

If the current session is not a transaction, then nothing is performed.

property settings: BaseSettings

Accessor for scope-specific settings.

class acapy_agent.core.profile.ProfileSessionHandle

Bases: ABC

Abstract interface for profile session handles.

This interface defines the common methods that are available across different session handle implementations (aries_askar.Session and DBStoreSession).

abstractmethod async close() → None

Close the session.

abstractmethod async commit() → None

Commit any updates performed within the transaction.

abstractmethod async count(category: str, tag_filter: str | dict = None) → int

Count the records matching a category and tag filter.

abstractmethod async fetch(category: str, name: str, *, for_update: bool = False) → Entry | None

Fetch a record from the store by category and name.

abstractmethod async fetch_all(category: str = None, tag_filter: str | dict = None, limit: int = None, *, order_by: str | None = None, descending: bool = False, for_update: bool = False) → EntryList

Fetch all records matching a category and tag filter.

abstractmethod async insert(category: str, name: str, value: str | bytes = None, tags: dict = None, expiry_ms: int = None, value_json=None) → None

Insert a new record into the store.

abstract property is_transaction: bool

Check if the session supports commit and rollback operations.

abstractmethod async remove(category: str, name: str) → None

Remove a record by category and name.

abstractmethod async remove_all(category: str = None, tag_filter: str | dict = None) → int

Remove all records matching a category and tag filter.

abstractmethod async replace(category: str, name: str, value: str | bytes = None, tags: dict = None, expiry_ms: int = None, value_json=None) → None

Replace a record in the store matching a category and name.

abstractmethod async rollback() → None

Roll back any updates performed within the transaction.

acapy_agent.core.protocol_registry module

Handle registration and publication of supported protocols.

class acapy_agent.core.protocol_registry.ProtocolDefinition(ident: ProtocolIdentifier, min: MessageVersion, current: MessageVersion, controller: str | None = None)

Bases: object

Protocol metadata used to register and resolve message types.

controller: str | None = None

current: MessageVersion

ident: ProtocolIdentifier

min: MessageVersion

property minor_versions_supported: bool

Accessor for whether minor versions are supported.

class acapy_agent.core.protocol_registry.ProtocolRegistry

Bases: object

Protocol registry for indexing message families.

property message_types: Sequence[str]

Accessor for a list of all message types.

async prepare_disclosed(context: InjectionContext, protocols: Sequence[str])

Call controllers and return publicly supported message families and roles.

property protocols: Sequence[str]

Accessor for a list of all message protocols.

protocols_matching_query(query: str) → Sequence[str]

Return a list of message protocols matching a query string.

register_controllers(*controller_sets)

Add new controllers.

Parameters:

controller_sets – Mappings of message families to coroutines

register_message_types(typeset: Mapping[str, str | type], version_definition: dict[str, Any] | VersionDefinition | None = None)

Add new supported message types.

Parameters:

• typeset (Mapping[str, Union[str, type]]) – Mappings of message types to register.

• version_definition (Optional[Union[dict[str, Any], VersionDefinition]]) – Optional version definition dict.

resolve_message_class(message_type: str) → DeferLoad | type | None

Resolve a message_type to a message class.

Given a message type identifier, this method returns the corresponding registered message class.

Parameters:

message_type – Message type to resolve

Returns:

The resolved message class

class acapy_agent.core.protocol_registry.VersionDefinition(min: MessageVersion, current: MessageVersion)

Bases: object

Version definition.

current: MessageVersion

classmethod from_dict(data: dict) → VersionDefinition

Create a version definition from a dict.

min: MessageVersion

acapy_agent.core.util module

Core utilities and constants.