API Reference

Protocols

`AIProviderProtocol`

Protocol for AI provider implementations.

chat

async def chat(
    messages: list[ChatMessageProtocol],
    tools: list[dict[str, Any]] | None = None,
    **kwargs: Any
) -> CompletionProtocol

source

Chat with optional tool calling.

Parameters

Parameter	Type	Description
`messages`	list[ChatMessageProtocol]	List of chat messages.
`tools`	list[dict[str, Any]] \| None	Optional list of JSON Schema tool definitions. Intentionally ``list[dict[str, Any]]`` — tool schemas are provider-specific JSON and their shape is not constrained here. **kwargs: Provider-specific options.

Returns

Type	Description
CompletionProtocol	Completion with optional tool calls.

`AIWorkflowNodeProtocol`

source

AI-domain graph node protocol for workflow execution.

A workflow node represents a single unit of work in a directed graph. Nodes receive the current shared state, perform their work, and return a state update dict that is merged into the global state.

name

property name() -> str

source

Unique node identifier within the workflow.

execute

async def execute(state: dict[str, Any]) -> dict[str, Any]

source

Execute this node with the current workflow state.

Parameters

Parameter	Type	Description
`state`	dict[str, Any]	Current shared workflow state.

Returns

Type	Description
dict[str, Any]	Dict of state updates to merge into the shared state.

`AgentExecutorProtocol`

source

Protocol for agent execution engines.

The executor runs an agent with governance checks, memory management, and observability integration.

run

async def run(
    agent: AgentProtocol,
    message: str,
    session_id: str | None = None,
    user_id: str | None = None,
    **kwargs: Any
) -> Any

source

Execute an agent and return the response.

Returns Result[AgentResponse, AgentError].

astream

async def astream(
    agent: AgentProtocol,
    message: str,
    session_id: str | None = None,
    user_id: str | None = None,
    **kwargs: Any
) -> Any

source

Stream agent execution events.

Yields AgentEvent objects as the agent executes, enabling real-time monitoring of thoughts, tool calls, and messages.

Parameters

Parameter	Type	Description
`agent`	AgentProtocol	The agent to execute.
`message`	str	User's input message.
`session_id`	str \| None	Session ID for multi-turn memory.
`user_id`	str \| None	User ID for governance tracking. **kwargs: Additional parameters passed to the strategy.

Yields

Type	Description
Any	AgentEvent objects with type, data, and run_id.

`AgentProtocol`

source

Protocol for AI agents.

An agent declares its identity, capabilities (tools), and persona (system prompt). The AgentExecutor uses this protocol to drive the reasoning loop.

name

property name() -> str

source

Unique agent identifier.

tools

property tools() -> list[ToolProtocol]

source

Tools available to this agent.

system_prompt

property system_prompt() -> str

source

System prompt defining the agent’s persona and constraints.

`AggregateRootProtocol`

source

Structural protocol for aggregate root objects.

Any class that implements event buffering and consistency boundaries satisfies this protocol. Type-check against it instead of the concrete AggregateRoot base class where possible.

add_event

def add_event(event: DomainEvent) -> None

source

collect_events

def collect_events() -> list[DomainEvent]

source

Return and clear all buffered domain events.

pull_events

def pull_events() -> list[DomainEvent]

source

Return buffered events without clearing them.

clear_events

def clear_events() -> None

source

Discard all buffered domain events.

has_uncommitted_events

property has_uncommitted_events() -> bool

source

Return True when there are buffered events not yet published.

`AsyncLockProtocol`

source

Protocol for a simple in-process async mutual-exclusion lock.

Mirrors the interface of asyncio.Lock. Implementations are single-process (i.e. not distributed); use DistributedLockProtocol or LockManagerProtocol when cross-process coordination is required.

Example

from quadkit.contracts.core.lock import AsyncLockProtocol

async def protected(lock: AsyncLockProtocol) -> None:
    async with lock:
        await do_critical_work()

from quadkit.contracts.core.lock import AsyncLockProtocol

async def protected(lock: AsyncLockProtocol) -> None:
    async with lock:
        await do_critical_work()

Container registration

container.singleton(AsyncLockProtocol, InMemoryAsyncLock)

container.singleton(AsyncLockProtocol, InMemoryAsyncLock)

acquire

async def acquire() -> bool

source

Acquire the lock, blocking until it becomes available.

Returns

Type	Description
bool	``True`` once the lock has been acquired.

release

def release() -> None

source

Release the lock.

Raises

Exception	Description
RuntimeError	If the lock is not currently held.

locked

def locked() -> bool

source

Return True if the lock is currently held by any coroutine.

`AsyncSecretStoreProtocol`

source

Async persistent store for sensitive named secret values.

Typical usage

store = await container.resolve(AsyncSecretStoreProtocol)
token = await store.get("stripe_api_key")

store = await container.resolve(AsyncSecretStoreProtocol)
token = await store.get("stripe_api_key")

get

async def get(name: str) -> str | None

source

Return the secret value for name, or None if absent.

Parameters

Parameter	Type	Description
`name`	str	Unique secret identifier.

get_bulk

async def get_bulk(*names: str) -> dict[str, str]

source

Return a mapping of name → value for all requested secrets.

Parameters

Parameter	Type	Description
`names`	str	One or more secret names.

Returns

Type	Description
dict[str, str]	Dict containing only the names that were found.

set

async def set(
    name: str,
    value: str
) -> None

source

Write or overwrite a secret value.

Parameters

Parameter	Type	Description
`name`	str	Unique secret identifier.
`value`	str	Plaintext secret value.

delete

async def delete(name: str) -> None

source

Remove a secret. No-op if absent.

Parameters

Parameter	Type	Description
`name`	str	Unique secret identifier.

`AsyncStringSerializerProtocol`

source

Protocol for value serialization.

Implementations handle conversion between Python objects and string representations suitable for storage or transmission. Used across extensions (cache, messaging, tasks, etc.) for consistent serialization.

Example

class JSONSerializer:
    async def serialize(self, value: Any) -> str:
        return json.dumps(value)

    async def deserialize(self, data: str) -> Any:
        return json.loads(data)

class JSONSerializer:
    async def serialize(self, value: Any) -> str:
        return json.dumps(value)

    async def deserialize(self, data: str) -> Any:
        return json.loads(data)

serialize

async def serialize(value: Any) -> str

source

Serialize a Python object to string.

Parameters

Parameter	Type	Description
`value`	Any	Python object to serialize.

Returns

Type	Description
str	String representation.

deserialize

async def deserialize(data: str) -> Any

source

Deserialize string back to Python object.

Parameters

Parameter	Type	Description
`data`	str	Serialized string data.

Returns

Type	Description
Any	Reconstructed Python object.

`AuditableProtocol`

source

Protocol for audit logging.

log_operation

async def log_operation(
    operation: str,
    record_id: Any,
    changes: dict[str, Any] | None = None,
    user_id: str | None = None
) -> None

source

get_audit_log

async def get_audit_log(
    record_id: Any | None = None,
    user_id: str | None = None,
    limit: int | None = None
) -> list[dict[str, Any]]

source

`AuthProviderProtocol`

source

Protocol for authentication and authorization providers.

Auth providers are responsible for user authentication, authorization, token management, and access control.

get_user

async def get_user(user_id: str) -> Any | None

source

Retrieve a user by their unique identifier.

Parameters

Parameter	Type	Description
`user_id`	str	Unique user identifier.

Returns

Type	Description
Any \| None	User object or None if not found.

verify_token

async def verify_token(token: str) -> Any

source

Verify an encoded auth token and return verification details.

Parameters

Parameter	Type	Description
`token`	str	Raw encoded auth token (JWT or similar).

Returns

Type	Description
Any	Verification result (typically ``Result[VerifiedToken, TokenError]``).

has_any_role

def has_any_role(
    user: Any,
    roles: list[str]
) -> bool

source

Return True if user holds at least one of the given roles.

Parameters

Parameter	Type	Description
`user`	Any	Authenticated user object.
`roles`	list[str]	Role names to check.

Returns

Type	Description
bool	True if the user has at least one of the supplied roles.

has_any_permission

def has_any_permission(
    user: Any,
    permissions: list[str]
) -> bool

source

Return True if user has at least one of the given permissions.

Parameters

Parameter	Type	Description
`user`	Any	Authenticated user object.
`permissions`	list[str]	Permission names to check.

Returns

Type	Description
bool	True if the user has at least one of the supplied permissions.

`AuthenticatedUserProtocol`

source

Protocol for authenticated user attached to requests.

This protocol defines the contract that any authenticated user implementation must follow for type-safe user handling.

Example

def get_user_profile(user: AuthenticatedUserProtocol) -> dict:
    return {"user_id": user.user_id, "roles": user.roles}

user_id

property user_id() -> str

source

Unique identifier for the user.

name

property name() -> str

source

Name of the authenticated user.

property email() -> str

source

Email address of the user.

is_active

property is_active() -> bool

source

Whether the user account is active.

is_verified

property is_verified() -> bool

source

Whether the user’s email is verified.

roles

property roles() -> list[str]

source

List of roles assigned to the user.

permissions

property permissions() -> list[str]

source

List of permissions granted to the user.

has_role

def has_role(role: str) -> bool

source

Check if user has a specific role.

Parameters

Parameter	Type	Description
`role`	str	The role to check for.

Returns

Type	Description
bool	True if the user has the role.

has_permission

def has_permission(permission: str) -> bool

source

Check if user has a specific permission.

Parameters

Parameter	Type	Description
`permission`	str	The permission to check for.

Returns

Type	Description
bool	True if the user has the permission.

`AuthorizerProtocol`

source

Protocol for authorization decisions.

Authorizers determine if an authenticated user can perform a specific action on a resource.

authorize

async def authorize(
    user: Any,
    action: str,
    resource: Any
) -> Result[bool, AuthorizationError]

source

Evaluate authorization; a successful evaluation may still deny.

Parameters

Parameter	Type	Description
`user`	Any	Authenticated user.
`action`	str	Action to perform (e.g., "read", "write", "delete").
`resource`	Any	Target resource.

Returns

Type	Description
Result[bool, AuthorizationError]	Ok(True) grants; Ok(False) denies; Err denies with a domain error. Consumers must inspect the boolean payload, not Result truthiness or is_ok() alone. Use can() for an exact boolean permission check. Infrastructure exceptions propagate; this is not row filtering.

check_access

async def check_access(
    user: Any,
    allowed_roles: set[str],
    resource: str | None = None,
    action: str | None = None
) -> bool

source

Check if user has access based on roles and permissions.

Parameters

Parameter	Type	Description
`user`	Any	Authenticated user.
`allowed_roles`	set[str]	Set of roles that grant access.
`resource`	str \| None	Target resource (optional).
`action`	str \| None	Action to perform (optional).

Returns

Type	Description
bool	True if access is granted.

can

async def can(
    user: Any,
    action: str,
    resource: str
) -> bool

source

Check if user can perform action on resource.

Parameters

Parameter	Type	Description
`user`	Any	Authenticated user.
`action`	str	Action to perform.
`resource`	str	Target resource.

Returns

Type	Description
bool	True if authorized.

can_view

async def can_view(
    user: Any,
    resource: str,
    record: Any = None
) -> bool

source

can_create

async def can_create(
    user: Any,
    resource: str
) -> bool

source

can_update

async def can_update(
    user: Any,
    resource: str,
    record: Any = None
) -> bool

source

can_delete

async def can_delete(
    user: Any,
    resource: str,
    record: Any = None
) -> bool

source

can_execute_action

async def can_execute_action(
    user: Any,
    resource: str,
    action: str,
    record: Any | None = None
) -> bool

source

set_roles

def set_roles(roles: dict[str, Any]) -> None

source

register_role

def register_role(
    name: str,
    role: Any
) -> None

source

remove_role

def remove_role(name: str) -> None

source

sync_from_db

async def sync_from_db(container: Any) -> None

source

`BlobStoreProtocol`

source

Protocol for blob storage operations.

This interface defines the contract for file storage backends (S3, GCS, Azure Blob, local filesystem, etc.).

Example

class S3BlobStore:
    async def upload(self, path: str, data: bytes, **options) -> FileInfo:
        await self._client.put_object(Bucket=self._bucket, Key=path, Body=data)
        return FileInfo(path=path, size=len(data), ...)

upload

async def upload(
    path: str,
    data: Uploadable,
    content_type: UploadOptions | str | None = None,
    **options: Any
) -> FileInfo

source

Upload data to the storage backend.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.
`data`	Uploadable	File content as bytes, text, file-like object, or async iterator.
`content_type`	UploadOptions \| str \| None	MIME type or UploadOptions for metadata and other upload settings. **options: Additional upload options.

Returns

Type	Description
FileInfo	FileInfo with path, size, and metadata.

download

async def download(path: str) -> bytes

source

Download file content into memory.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

Returns

Type	Description
bytes	File content as bytes.

stream

def stream(
    path: str,
    chunk_size: int = 8192
) -> AsyncIterator[bytes]

source

Stream file content (memory efficient).

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.
`chunk_size`	int	Size of each chunk in bytes.

Yields

Type	Description
AsyncIterator[bytes]	File content in chunks.

delete

async def delete(path: str) -> None

source

Delete a file.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

exists

async def exists(path: str) -> bool

source

Check if file exists.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

Returns

Type	Description
bool	True if file exists.

info

async def info(path: str) -> FileInfo

source

Get file metadata.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

Returns

Type	Description
FileInfo	FileInfo with size, content_type, etc.

list

def list(prefix: str = '') -> AsyncIterator[FileInfo]

source

List files with a given prefix.

Parameters

Parameter	Type	Description
`prefix`	str	Path prefix to filter by.

Yields

Type	Description
AsyncIterator[FileInfo]	FileInfo for each matching file.

get_url

async def get_url(path: str) -> str

source

Get public URL (if applicable).

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.

Returns

Type	Description
str	Public URL string.

get_presigned_url

async def get_presigned_url(
    path: str,
    expires_in: timedelta = timedelta(hours=1),
    method: str = 'GET'
) -> str

source

Get a temporary secure URL.

Parameters

Parameter	Type	Description
`path`	str	Storage path/key.
`expires_in`	timedelta	URL validity window as a timedelta (default one hour). Pass ``timedelta(minutes=5)` `for secure short-lived downloads or` `timedelta(hours=24)`` for bulk exports.
`method`	str	HTTP method (GET or PUT).

Returns

Type	Description
str	Presigned URL string.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check.

Returns

Type	Description
HealthCheckResult	Structured health check result.

`CORSPolicyProtocol`

source

Protocol for CORS (Cross-Origin Resource Sharing) policy configuration.

Defines how to evaluate CORS requests and provide the necessary headers and configuration for browser-based clients.

Example

class CORSPolicy:
    def is_origin_allowed(self, origin: str) -> bool:
        return origin in ("https://app.example.com", "https://admin.example.com")

    def get_allowed_headers(self) -> list[str]:
        return ["content-type", "authorization", "x-request-id"]

    def get_allowed_methods(self) -> list[str]:
        return ["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"]

    def get_max_age(self) -> int:
        return 3600

is_origin_allowed

def is_origin_allowed(origin: str) -> bool

source

Check if an origin is permitted for CORS requests.

Parameters

Parameter	Type	Description
`origin`	str	The value of the Origin header from a CORS preflight request.

Returns

Type	Description
bool	True if the origin is allowed, False otherwise.

get_allowed_headers

def get_allowed_headers() -> list[str]

source

Return the list of allowed request headers.

Returns

Type	Description
list[str]	List of header names that clients are allowed to send.
Common values	["content-type", "authorization", "x-request-id"].

get_allowed_methods

def get_allowed_methods() -> list[str]

source

Return the list of allowed HTTP methods.

Returns

Type	Description
list[str]	List of HTTP methods clients are allowed to use.
Typical	["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"].

get_max_age

def get_max_age() -> int

source

Return the preflight cache duration in seconds.

Returns

Type	Description
int	Maximum time (in seconds) browsers should cache preflight responses. Typical values: 3600 (1 hour) to 86400 (24 hours).

`CacheAdminProtocol`

source

Optional operational contract exposed by inspectable cache backends.

CacheBackendProtocol intentionally stays focused on cache data-plane operations. Operator consoles also need a safe, backend-neutral way to inspect and reset counters without reaching into private implementation fields. Backends that can expose local counters implement this optional protocol; distributed backends may return None for unavailable statistics.

get_stats

def get_stats() -> dict[str, int | float | str] | None

source

Return backend counters and capacity information, when available.

reset_stats

def reset_stats() -> None

source

Reset operator-facing counters without deleting cached values.

`CacheAwareProtocol`

source

Protocol for cache-aware operations.

invalidate_cache

async def invalidate_cache(
    keys: list[str] | None = None,
    patterns: list[str] | None = None
) -> None

source

get_cache_info

async def get_cache_info() -> dict[str, Any]

source

`CacheBackendProtocol`

source

Protocol for cache backend implementations.

This interface defines the contract that all cache backend implementations must follow. It provides a unified API for different storage mechanisms (memory, Redis, Memcached, etc.).

Example

class RedisBackend:
    async def get(self, key: str) -> Result[Any | None, CacheError]:
        try:
            data = await self._redis.get(key)
            return Ok(self._serializer.deserialize(data) if data else None)
        except Exception as exc:
            return Err(CacheError(str(exc)))

    async def set(
        self, key: str, value: Any, ttl: int | None = None
    ) -> Result[None, CacheError]:
        try:
            data = self._serializer.serialize(value)
            await self._redis.set(key, data, ex=ttl)
            return Ok(None)
        except Exception as exc:
            return Err(CacheError(str(exc)))

get

async def get(key: str) -> Result[Any | None, CacheError]

source

Get a value from the cache.

Parameters

Parameter	Type	Description
`key`	str	The cache key to retrieve.

Returns

Type	Description
Result[Any \| None, CacheError]	Ok(value) if found, Ok(None) if not found, Err(CacheError) on failure.

Note

The return type is Any by design: the cache is a heterogeneous store and the caller knows the expected type. Use cast(T, result.unwrap()) at call sites for type safety.

set

async def set(
    key: str,
    value: Any,
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set a value in the cache with optional TTL.

Parameters

Parameter	Type	Description
`key`	str	The cache key to set.
`value`	Any	The value to cache.
`ttl`	int \| None	Time to live in seconds (optional).

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

delete

async def delete(key: str) -> Result[bool, CacheError]

source

Delete a value from the cache.

Parameters

Parameter	Type	Description
`key`	str	The cache key to delete.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if deleted, Ok(False) if not found, Err(CacheError) on failure.

delete_many

async def delete_many(keys: list[str]) -> Result[int, CacheError]

source

Delete multiple values from the cache.

Parameters

Parameter	Type	Description
`keys`	list[str]	List of cache keys to delete.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

delete_pattern

async def delete_pattern(pattern: str) -> Result[int, CacheError]

source

Delete all keys matching a glob-style pattern.

Parameters

Parameter	Type	Description
`pattern`	str	Glob pattern (e.g. ``"pet:list:"). Supports` ``` as a wildcard matching any sequence of characters.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

exists

async def exists(key: str) -> Result[bool, CacheError]

source

Check if a key exists in the cache.

Parameters

Parameter	Type	Description
`key`	str	The cache key to check.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if exists, Ok(False) otherwise, Err(CacheError) on failure.

clear

async def clear() -> Result[None, CacheError]

source

Clear all values from the cache.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

get_many

async def get_many(keys: list[str]) -> Result[dict[str, Any], CacheError]

source

Get multiple values from the cache.

Parameters

Parameter	Type	Description
`keys`	list[str]	List of cache keys to retrieve.

Returns

Type	Description
Result[dict[str, Any], CacheError]	Ok(dict) mapping found keys to values, Err(CacheError) on failure.

set_many

async def set_many(
    items: dict[str, Any],
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set multiple values in the cache.

Parameters

Parameter	Type	Description
`items`	dict[str, Any]	Dictionary of key-value pairs to cache.
`ttl`	int \| None	Time to live in seconds for all items.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if all items set successfully, Err(CacheError) on failure.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a health check on the cache backend.

Returns

Type	Description
HealthCheckResult	Structured HealthCheckResult.

`CacheProviderProtocol`

source

Protocol for cache providers.

Cache providers are responsible for setting up caching backends, cache warming, and cache management.

`CliContributorProtocol`

source

Protocol for CLI contributors that register generators, commands, health checks, doctor checks, shell context, and hooks into the CLI.

Each contributor is discovered via the quadkit.cli.contributors entry point group. Install the package to make contributions available.

contributor_id

property contributor_id() -> str

source

Unique identifier for this contributor (e.g. ‘core’, ‘web’, ‘sql’).

get_generators

def get_generators() -> list[GeneratorDefinition]

source

Return all generator definitions this contributor provides.

Returns

Type	Description
list[GeneratorDefinition]	A list of GeneratorDefinition instances. May be empty.

get_commands

def get_commands() -> list[CommandContribution]

source

Return CLI command groups contributed by this package.

Returns

Type	Description
list[CommandContribution]	A list of CommandContribution instances. May be empty.

get_health_checks

def get_health_checks() -> list[HealthCheckContribution]

source

Return runtime health checks contributed by this package.

Health checks require a booted DI container.

Returns

Type	Description
list[HealthCheckContribution]	A list of HealthCheckContribution instances. May be empty.

get_doctor_checks

def get_doctor_checks() -> list[DoctorCheckContribution]

source

Return static environment/config diagnostic checks.

Doctor checks are sync and require no container.

Returns

Type	Description
list[DoctorCheckContribution]	A list of DoctorCheckContribution instances. May be empty.

get_shell_context

def get_shell_context() -> list[ShellContextContribution]

source

Return objects to inject into the interactive shell namespace.

Returns

Type	Description
list[ShellContextContribution]	A list of ShellContextContribution instances. May be empty.

get_hooks

def get_hooks() -> list[HookContribution]

source

Return CLI lifecycle hooks contributed by this package.

Returns

Type	Description
list[HookContribution]	A list of HookContribution instances. May be empty.

`ClockProtocol`

source

Injectable time source for the framework.

now

def now() -> datetime

source

Return the current timezone-aware UTC time.

monotonic

def monotonic() -> float

source

Return a monotonic elapsed-time counter.

timestamp

def timestamp() -> float

source

Return the current Unix timestamp in UTC seconds.

time

def time() -> float

source

Backward-compatible alias for timestamp.

`CommandBusProtocol`

source

Protocol for command bus implementations.

Example

class CommandBusProtocol:
    async def dispatch(self, command: Command) -> Any:
        handler = self._handlers[type(command)]
        return await handler.handle(command)

dispatch

async def dispatch(command: Any) -> Any

source

Dispatch a command to its handler.

Parameters

Parameter	Type	Description
`command`	Any	Command to dispatch.

Returns

Type	Description
Any	Result from the command handler.

`ConfigProtocol`

source

Protocol for configuration access across the framework.

Any configuration object (BaseConfig, QuadkitConfig, or custom implementations) can satisfy this protocol by implementing get(), get_section(), and has_section().

Example

config = container.resolve(ConfigProtocol)
db_url = config.get("database.url", "sqlite:///default.db")
logging = config.get_section("logging", LoggingConfig)

config = container.resolve(ConfigProtocol)
db_url = config.get("database.url", "sqlite:///default.db")
logging = config.get_section("logging", LoggingConfig)

environment

property environment() -> Environment

source

The active deployment environment.

is_production

property is_production() -> bool

source

Return True when the active environment is production.

is_development

property is_development() -> bool

source

Return True when the active environment is development.

is_testing

property is_testing() -> bool

source

Return True when the active environment is testing.

is_staging

property is_staging() -> bool

source

Return True when the active environment is staging.

is_debug

property is_debug() -> bool

source

Return True when debug mode is enabled.

get

def get(
    key: str,
    default: Any = None
) -> Any

source

Get a configuration value by dot-notation key.

Parameters

Parameter	Type	Description
`key`	str	Configuration key (e.g. ``"app.name"` `or` `"database.url"``).
`default`	Any	Value returned when the key is not found.

Returns

Type	Description
Any	The configuration value, or default if not found.

get_section

def get_section(
    name: str,
    model_cls: None = None
) -> dict[str, Any] | None

source

get_section

def get_section(
    name: str,
    model_cls: type[T]
) -> T

source

get_section

def get_section(
    name: str,
    model_cls: type[T] | None = None
) -> T | dict[str, Any] | None

source

Get a typed configuration section.

Parameters

Parameter	Type	Description
`name`	str	Section name (e.g. ``"logging"``).
`model_cls`	type[T] \| None	Optional model class to coerce the section into.

Returns

Type	Description
T \| dict[str, Any] \| None	A model instance when model_cls is provided, otherwise a raw dict or the attribute value; ``None`` when the section is absent and no model class was supplied.

has_section

def has_section(name: str) -> bool

source

Check whether a configuration section exists.

Parameters

Parameter	Type	Description
`name`	str	Section name to check.

Returns

Type	Description
bool	True if the section is present.

`ConnectionPoolProtocol`

source

Protocol for connection pools.

max_connections

property max_connections() -> int

source

connection_timeout

property connection_timeout() -> float

source

initialize

async def initialize() -> None

source

Initialize the connection pool.

shutdown

async def shutdown() -> None

source

Shutdown the connection pool.

get_connection

def get_connection() -> AbstractAsyncContextManager[Any]

source

Get a connection from the pool.

get_pool_stats

async def get_pool_stats() -> dict[str, Any]

source

Get pool statistics.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check pool health.

get_query_stats

async def get_query_stats(time_range_seconds: int = 3600) -> dict[str, Any]

source

Get query statistics.

warm

async def warm(count: int | None = None) -> None

source

Pre-create count connections to avoid cold-start latency.

Parameters

Parameter	Type	Description
`count`	int \| None	Number of connections to open. Defaults to ``min_connections`` (or the pool minimum) if not specified.

validate_connections

async def validate_connections() -> int

source

Validate all idle connections in the pool, evicting dead ones.

Returns

Type	Description
int	Number of valid connections remaining after validation.

`ConnectionProtocol`

source

Protocol for database connections.

Represents an active database connection that can execute queries.

execute

async def execute(
    query: str,
    *args: Any,
    timeout: float | None = None
) -> QueryResult

source

Execute a SQL query.

Parameters

Parameter	Type	Description
`query`	str	SQL query string with positional parameters ($1, $2, ...). *args: Positional arguments for the query.
`timeout`	float \| None	Optional timeout in seconds.

Returns

Type	Description
QueryResult	QueryResult with rows and metadata.

fetchrow

async def fetchrow(
    query: str,
    *args: Any,
    timeout: float | None = None
) -> dict[str, Any] | None

source

Fetch a single row from the database.

Parameters

Parameter	Type	Description
`query`	str	SQL query string. *args: Positional arguments for the query.
`timeout`	float \| None	Optional timeout in seconds.

Returns

Type	Description
dict[str, Any] \| None	Row dictionary or None if not found.

async def close() -> None

source

Close the connection.

fetch

async def fetch(
    query: str,
    *args: Any,
    **kwargs: Any
) -> list[dict[str, Any]]

source

Fetch rows from the database.

Parameters

Parameter	Type	Description
`query`	str	SQL query string. args: Positional query parameters. *kwargs: Named query parameters.

Returns

Type	Description
list[dict[str, Any]]	List of row dictionaries.

`ContainerProtocol`

source

Combined DI container protocol covering registration, resolution, and validation.

Use this instead of Any when a component needs to both register services (via singleton/transient) and resolve them (via resolve) within the same method. This is a structural Protocol — any object implementing both ContainerRegistrarProtocol and ContainerResolverProtocol satisfies this type.

`ContainerRegistrarProtocol`

source

Protocol for registering dependencies in the container.

Used during the registration phase where resolution is not yet permitted.

transient

def transient(
    service_type: type[T],
    factory: Any,
    validate: bool = True
) -> None

source

singleton

def singleton(
    service_type: type[T],
    instance: T | None = None,
    *,
    name: str | None = None,
    factory: Any | None = None,
    validate: bool = True
) -> None

source

singleton

def singleton(
    service_type: Any,
    instance: Any = None,
    *,
    name: str | None = None,
    factory: Any | None = None,
    validate: bool = True
) -> None

source

singleton

def singleton(
    service_type: Any,
    instance: Any = None,
    *,
    name: str | None = None,
    factory: Any | None = None,
    validate: bool = True
) -> None

source

Parameters

Parameter	Type	Description
`service_type`	Any	The abstract type (protocol/class) being registered. Accepts both concrete classes (``type[T]` `with full type inference) and Protocol types (via` `Any`` fallback).
`instance`	Any	Pre-built singleton instance.
`name`	str \| None	Optional string key. When provided, the binding is stored under this name instead of ``service_type. Resolve via` `Annotated[T, Named(name)]``.
`factory`	Any \| None	Factory callable for lazy creation.
`validate`	bool	Whether to validate protocol conformance.

scoped

def scoped(
    service_type: type[T],
    factory: Any,
    validate: bool = True,
    *,
    name: str | None = None
) -> None

source

Parameters

Parameter	Type	Description
`service_type`	type[T]	The abstract type being registered.
`factory`	Any	Factory callable, called once per scope.
`validate`	bool	Whether to validate protocol conformance.
`name`	str \| None	Optional string key for named scoped registration. Resolve via ``Annotated[T, Named(name)]``.

has

def has(service_type: Any) -> bool

source

Check if a service is registered.

resolve_sync

def resolve_sync(service_type: type[T]) -> T

source

Synchronously resolve an already-instantiated singleton.

Intended for register-phase configuration reads where pre-seeded singletons (such as QuadkitConfig) are already instantiated.

bind

def bind(
    service_type: type[T],
    instance: T
) -> None

source

Bind a pre-built singleton instance, overwriting any existing binding.

Unlike singleton(), this works on frozen containers. Designed for updating singleton instances during the boot phase (e.g. wrapping a store with a tenancy decorator).

The service must already be registered as a singleton.

Parameters

Parameter	Type	Description
`service_type`	type[T]	The registered service type to rebind.
`instance`	T	The replacement singleton instance.

`ContainerResolverProtocol`

source

Protocol for resolving dependencies from the container.

Async-first code should depend on this protocol. It provides only an asynchronous resolve() method and makes no guarantees about sync behaviour. All synchronous helpers have been removed to keep the API simple and forward‑looking.

Used during the boot phase and runtime after registration is complete.

resolve

async def resolve(
    service_type: type[T],
    *,
    bypass_visibility: bool = False
) -> T

source

resolve

async def resolve(
    service_type: Any,
    *,
    bypass_visibility: bool = False
) -> Any

source

resolve

async def resolve(
    service_type: Any,
    *,
    bypass_visibility: bool = False
) -> Any

source

Asynchronously resolve a service by its type.

The framework is async-first and does not support string or other loose keys. Consumers should only register and resolve using types so that the DI container remains fully type-safe.

Accepts both concrete classes (type[T] with full return-type inference) and Protocol types (via Any fallback).

Parameters

Parameter	Type	Description
`service_type`	Any	The service type to resolve.
`bypass_visibility`	bool	If True, skip module visibility enforcement. Use only in framework-internal resolution paths.

resolve_sync

def resolve_sync(service_type: type[T]) -> T

source

Synchronously resolve an already-instantiated singleton.

Intended for decorators and other synchronous contexts where starting a coroutine is impossible, such as register-phase configuration reads and sync message handlers. Only singletons already instantiated during boot are resolvable through this method.

Parameters

Parameter	Type	Description
`service_type`	type[T]	The service type to resolve.

Returns

Type	Description
T	The resolved instance.

Raises

Exception	Description
UnresolvableDependencyError	If the service is not a singleton or has not been instantiated yet.

call

async def call(
    func: Callable[Ellipsis, Awaitable[T] | T],
    *args: Any,
    **kwargs: Any
) -> T

source

Call a function with dependency injection.

create_scope

def create_scope() -> Any

source

Create a request-scoped resolution context.

has

def has(service_type: Any) -> bool

source

Check if a service is registered.

bind

def bind(
    service_type: type[T],
    instance: T
) -> None

source

Bind a pre-built singleton instance, overwriting any existing binding.

Works on frozen containers and is designed for updating singleton instances during the boot phase (e.g. wrapping a store with a tenancy decorator). The service must already be registered as a singleton.

Parameters

Parameter	Type	Description
`service_type`	type[T]	The registered service type to rebind.
`instance`	T	The replacement singleton instance.

resolve_optional

async def resolve_optional(service_type: type[T]) -> T | None

source

resolve_optional

async def resolve_optional(service_type: Any) -> Any | None

source

resolve_optional

async def resolve_optional(service_type: Any) -> Any | None

source

Resolve a service, returning None if not registered.

This provides a graceful way to handle optional dependencies without catching exceptions.

Parameters

Parameter	Type	Description
`service_type`	Any	The type to resolve.

Returns

Type	Description
Any \| None	The resolved instance or None if the service is not registered.

resolve_all

async def resolve_all(service_type: type[T]) -> list[T]

source

resolve_all

async def resolve_all(service_type: Any) -> list[Any]

source

resolve_all

async def resolve_all(service_type: Any) -> list[Any]

source

Resolve all registered implementations that are subtypes of a service type.

Useful for collecting multiple implementations of a protocol or abstract base class (e.g. all registered EventHandlerProtocol instances).

Parameters

Parameter	Type	Description
`service_type`	Any	The base type whose implementations to resolve.

Returns

Type	Description
list[Any]	A list of resolved instances.

`ContextPrunerProtocol`

source

Relevance-based conversation history pruning.

Unlike sliding-window truncation, implementations score each turn for relevance to the current query and preserve high-value turns regardless of their position in the history.

prune

async def prune(
    history: Sequence[ChatMessageProtocol],
    current_query: str,
    max_turns: int
) -> list[ChatMessageProtocol]

source

Prune history to at most max_turns, preserving relevant turns.

Parameters

Parameter	Type	Description
`history`	Sequence[ChatMessageProtocol]	Full conversation history as ChatMessageProtocol instances.
`current_query`	str	The current user query (used for relevance scoring).
`max_turns`	int	Maximum number of turns to retain.

Returns

Type	Description
list[ChatMessageProtocol]	Pruned history in chronological order.

`CursorPageProtocol`

source

Protocol for cursor-based paginated result pages.

Implemented by CursorPage and Relay-compliant GraphQL page types. Use this for code that should work with any cursor-pagination result.

items

property items() -> list[T]

source

Items in this page.

next_cursor

property next_cursor() -> str | None

source

Opaque cursor for the next page, or None if no more.

has_more

property has_more() -> bool

source

Whether there are more results after this page.

`DatabaseProviderProtocol`

source

Protocol for database providers.

This defines the interface that all database providers must implement, regardless of the underlying database technology.

Example

class PostgresProvider:
    async def connect(self) -> None:
        self._pool = await asyncpg.create_pool(self._dsn)

    async def execute_query(
        self,
        sql: str,
        params: list[Any] | None = None,
    ) -> QueryResult:
        async with self._pool.acquire() as conn:
            rows = await conn.fetch(sql, *params or [])
            return QueryResult(rows=list(map(dict, rows)), ...)

connect

async def connect() -> None

source

Establish connection to the database.

disconnect

async def disconnect() -> None

source

Close connection to the database.

is_connected

async def is_connected() -> bool

source

Check if database is connected.

get_primary_pool

async def get_primary_pool() -> ConnectionPoolProtocol

source

Return the primary connection pool.

For multi-backend setups, returns the backend marked primary: true. Raises NoPrimaryBackendError if no primary is marked or zero backends.

Returns

Type	Description
ConnectionPoolProtocol	The primary connection pool.

Raises

Exception	Description
NoPrimaryBackendError	If no primary backend is configured.

execute_query

async def execute_query(
    sql: str,
    params: list[Any] | None = None,
    **kwargs: Any
) -> QueryResult

source

Execute a SELECT query.

A statement that returns no rows (DDL, PRAGMA, a data-modifying statement without RETURNING) is not an error: it succeeds with an empty row list. The framework’s stores issue CREATE TABLE IF NOT EXISTS through this method when they provision their own table, and an implementation that treats “no result set” as failure makes that provisioning unusable.

Parameters

Parameter	Type	Description
`sql`	str	SQL query string.
`params`	list[Any] \| None	Query parameters. JSON-shaped values (``dict/list``) are encoded for binding — see encode_bind_value. **kwargs: Additional options.

Returns

Type	Description
QueryResult	QueryResult with rows and execution metadata.

execute_insert

async def execute_insert(
    table: str,
    data: dict[str, Any],
    **kwargs: Any
) -> InsertResult

source

Execute an INSERT operation.

JSON-shaped values in data (dict/list) are encoded to JSON text before binding, matching the JSONB/TEXT columns the framework’s stores declare. Relational drivers cannot bind a Python container, so an implementation that passes one through fails at the driver — silently, when the caller does not inspect success.

Parameters

Parameter	Type	Description
`table`	str	Table name.
`data`	dict[str, Any]	Column-value mapping. **kwargs: Additional options.

Returns

Type	Description
InsertResult	InsertResult with inserted ID and influenced rows.

Raises

Exception	Description
IntegrityError	Constraint violations (foreign key, unique, NOT NULL, CHECK) propagate as the typed ``quadkit.sql.exceptions.IntegrityError`` family rather than folding into a failed result, so callers can answer for them by type.

execute_update

async def execute_update(
    table: str,
    data: dict[str, Any],
    where_clause: str,
    where_params: list[Any] | None = None,
    **kwargs: Any
) -> UpdateResult

source

Execute an UPDATE operation.

JSON-shaped values in data and where_params are encoded for binding, as in execute_insert.

A value may be a Delta instead of a plain value. It renders col = col + ? — a relative move, so a counter changes inside the statement rather than through a read the caller then writes back. An implementation that binds it as if it were a value writes the Delta object itself into the column, so this is part of the contract, not an optional extension.

Parameters

Parameter	Type	Description
`table`	str	Table name.
`data`	dict[str, Any]	Column-value updates, or ``Delta`` moves.
`where_clause`	str	WHERE condition.
`where_params`	list[Any] \| None	Parameters for WHERE clause. **kwargs: Additional options.

Returns

Type	Description
UpdateResult	UpdateResult with affected rows.

Raises

Exception	Description
IntegrityError	As in execute_insert.

execute_delete

async def execute_delete(
    table: str,
    where_clause: str,
    where_params: list[Any] | None = None,
    **kwargs: Any
) -> DeleteResult

source

Execute a DELETE operation.

JSON-shaped values in where_params are encoded for binding, as in execute_insert.

Parameters

Parameter	Type	Description
`table`	str	Table name.
`where_clause`	str	WHERE condition.
`where_params`	list[Any] \| None	Parameters for WHERE clause. **kwargs: Additional options.

Returns

Type	Description
DeleteResult	DeleteResult with affected rows.

Raises

Exception	Description
IntegrityError	As in execute_insert — a delete restricted by referencing rows raises the typed foreign-key error.

execute

async def execute(
    sql: str,
    params: Any = None
) -> QueryResult

source

Execute a raw SQL query with parameters.

Parameters

Parameter	Type	Description
`sql`	str	SQL query string.
`params`	Any	Query parameters.

Returns

Type	Description
QueryResult	QueryResult with execution results.

transaction

def transaction(isolation_level: IsolationLevel | None = None) -> AbstractAsyncContextManager[Any]

source

Context manager for transactions.

Parameters

Parameter	Type	Description
`isolation_level`	IsolationLevel \| None	Optional ANSI SQL isolation level. When ``None`` the driver's default isolation level is used.

Example

async with db.transaction(isolation_level=IsolationLevel.SERIALIZABLE):
    await db.execute("INSERT INTO ...")
    await db.execute("UPDATE ...")

begin_transaction

async def begin_transaction() -> None

source

Begin a transaction.

commit_transaction

async def commit_transaction() -> None

source

Commit current transaction.

rollback_transaction

async def rollback_transaction() -> None

source

Rollback current transaction.

table_exists

async def table_exists(table_name: str) -> bool

source

Check if a table exists.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check on database connection.

scoped_context

def scoped_context() -> AbstractAsyncContextManager[Any]

source

Return an async context manager that establishes a scoped session.

The scoped context binds a database session to the current async context so that get_scoped_connection can retrieve it without passing the connection around explicitly.

Returns

Type	Description
AbstractAsyncContextManager[Any]	Async context manager that yields no value (or the session).

get_scoped_connection

async def get_scoped_connection() -> ConnectionProtocol

source

Return the connection bound to the current scoped context.

Must be called within an active scoped_context block.

Returns

Type	Description
ConnectionProtocol	Active ConnectionProtocol for the current scope.

acquire

async def acquire() -> ConnectionProtocol

source

Acquire a connection from the pool for manual management.

Use this when you need fine-grained control over connection lifecycle, but prefer scoped_context when possible for automatic cleanup.

Returns

Type	Description
ConnectionProtocol	An acquired connection that must be released via release.

Example

conn = await db.acquire()
try:
    result = await conn.execute("SELECT * FROM users")
finally:
    await db.release(conn)

release

async def release(connection: ConnectionProtocol) -> None

source

Release a connection back to the pool.

Parameters

Parameter	Type	Description
`connection`	ConnectionProtocol	Connection acquired via acquire.

`DistributedLockProtocol`

source

Protocol for distributed locking.

Distributed locks coordinate access to shared resources across multiple processes, threads, or services. They provide:

Mutual exclusion: Only one holder at a time
Safety: Locks expire after a TTL to prevent deadlocks
Non-blocking acquire: Returns immediately if lock is held

Implementations must be safe for use in async contexts and handle network partitions gracefully.

Example

lock = RedisLock(redis, "my-resource", ttl=30)

# Non-blocking acquire
if await lock.acquire():
    try:
        await process()
    finally:
        await lock.release()

# Blocking with timeout
if await lock.acquire_blocking(timeout=5.0):
    try:
        await process()
    finally:
        await lock.release()

# Context manager (preferred)
async with lock:
    await process()

lock = RedisLock(redis, "my-resource", ttl=30)

# Non-blocking acquire
if await lock.acquire():
    try:
        await process()
    finally:
        await lock.release()

# Blocking with timeout
if await lock.acquire_blocking(timeout=5.0):
    try:
        await process()
    finally:
        await lock.release()

# Context manager (preferred)
async with lock:
    await process()

acquire

async def acquire() -> bool

source

Attempt to acquire the lock non-blocking.

Returns immediately with True if the lock was acquired, False if the lock is already held by another process.

Returns

Type	Description
bool	True if lock was acquired, False otherwise

acquire_blocking

async def acquire_blocking(timeout: float | None = None) -> bool

source

Attempt to acquire the lock with optional timeout.

Blocks until the lock is acquired or the timeout expires.

Parameters

Parameter	Type	Description
`timeout`	float \| None	Maximum seconds to wait. None means wait forever.

Returns

Type	Description
bool	True if lock was acquired, False if timeout expired

release

async def release() -> bool

source

Release the lock.

Returns

Type	Description
bool	True if lock was released by this call, False if lock was not held or already expired

is_held

async def is_held() -> bool

source

Check if this instance currently holds the lock.

Returns

Type	Description
bool	True if this instance holds the lock

extend

async def extend(additional_time: float) -> bool

source

Extend the lock TTL.

Useful for long-running operations that need to prevent the lock from expiring mid-operation.

Parameters

Parameter	Type	Description
`additional_time`	float	Seconds to add to the TTL

Returns

Type	Description
bool	True if TTL was extended, False if lock not held

`DocumentVectorStoreProtocol`

source

Document-centric vector store API for RAG and AI packages.

This is the AI-layer contract for add/search/delete against embedded documents. For connection lifecycle and collection management, use VectorStoreProtocol (infrastructure layer).

All vector database drivers (Qdrant, ChromaDB, PGVector, etc.) are adapted to this shape so callers can swap backends.

add

async def add(documents: list[DocumentProtocol]) -> Result[list[str], VectorError]

source

Add documents to the vector store.

Parameters

Parameter	Type	Description
`documents`	list[DocumentProtocol]	List of documents with text and metadata.

Returns

Type	Description
Result[list[str], VectorError]	``Ok(list[str])` `with document IDs on success, or` `Err(VectorStoreError)`` on failure.

batch_upsert

async def batch_upsert(
    documents: list[DocumentProtocol],
    batch_size: int = 100
) -> Result[int, VectorError]

source

Upsert documents in batches.

Parameters

Parameter	Type	Description
`documents`	list[DocumentProtocol]	List of documents to upsert.
`batch_size`	int	Number of documents per batch. defaults to 100.

Returns

Type	Description
Result[int, VectorError]	``Ok(int)` `with the total count of documents upserted, or` `Err(VectorStoreError)`` on failure.

async def search(
    query: list[float],
    *,
    top_k: int = 10,
    filters: MetadataFilter | None = None,
    score_threshold: float | None = None
) -> Result[list[SearchResultProtocol], VectorError]

source

Search for similar documents using a vector.

Parameters

Parameter	Type	Description
`query`	list[float]	Query embedding vector.
`top_k`	int	Number of results to return.
`filters`	MetadataFilter \| None	Metadata filters.
`score_threshold`	float \| None	Minimum relevance score.

Returns

Type	Description
Result[list[SearchResultProtocol], VectorError]	``Ok(list of search results)` `on success, or` `Err(VectorStoreError)`` on failure.

search_text

async def search_text(
    query: str,
    *,
    top_k: int = 10,
    filters: MetadataFilter | None = None
) -> Result[list[SearchResultProtocol], VectorError]

source

Search for similar documents using text.

Parameters

Parameter	Type	Description
`query`	str	Query text.
`top_k`	int	Number of results to return.
`filters`	MetadataFilter \| None	Metadata filters.

Returns

Type	Description
Result[list[SearchResultProtocol], VectorError]	``Ok(list of search results)` `on success, or` `Err(VectorStoreError)`` on failure.

delete

async def delete(ids: list[str]) -> Result[int, VectorError]

source

Delete documents by ID.

Parameters

Parameter	Type	Description
`ids`	list[str]	List of document IDs to delete.

Returns

Type	Description
Result[int, VectorError]	``Ok(int)` `with the count of deleted documents on success, or` `Err(VectorStoreError)`` on failure.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight connectivity check.

Returns

Type	Description
HealthCheckResult	Structured health check result.

`DomainEventPublisherProtocol`

source

Protocol for publishing domain events.

publish

async def publish(event: Any) -> None

source

Publish a domain event.

Parameters

Parameter	Type	Description
`event`	Any	DomainEvent instance.

`EmbeddingClientProtocol`

source

Protocol for embedding client implementations.

embed

async def embed(texts: list[str]) -> list[list[float]]

source

Generate embeddings for texts.

Parameters

Parameter	Type	Description
`texts`	list[str]	List of strings to embed.

Returns

Type	Description
list[list[float]]	List of embedding vectors.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight connectivity check.

Returns

Type	Description
HealthCheckResult	Structured health check result.

async def close() -> None

source

Close the client and release resources.

`EpisodicMemoryProtocol`

source

Protocol for episodic memory (conversation history with temporal grounding).

Episodic memory stores timestamped events and conversations, supporting hybrid retrieval based on recency and relevance.

record

async def record(entry: MemoryEntry) -> None

source

Record a new episode/conversation turn.

Parameters

Parameter	Type	Description
`entry`	MemoryEntry	The episode to record

recall

async def recall(query: MemoryQuery) -> list[MemorySearchResult]

source

Recall episodes based on relevance and recency.

Parameters

Parameter	Type	Description
`query`	MemoryQuery	Query parameters with weighting preferences

Returns

Type	Description
list[MemorySearchResult]	Ranked list of matching episodes

forget

async def forget(
    entry_id: str,
    owner_id: str
) -> None

source

Forget/delete a specific episode, scoped to owner_id.

Parameters

Parameter	Type	Description
`entry_id`	str	ID of the episode to forget
`owner_id`	str	Owner scope; only this owner's episode may be forgotten

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check if the episodic memory backend is reachable and healthy.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for a response.

Returns

Type	Description
HealthCheckResult	Health check result with status and details.

`EventBusProtocol`

source

Protocol for event bus implementations.

The event bus manages event publication and subscription.

Example

```python
class InMemoryEventBus:
    async def publish(self, event: DomainEvent) -> "Result[None, EventError]":
        for handler in self._handlers.get(type(event), []):
            await handler.handle(event)
        return Ok(None)

    def subscribe(self, event_type, handler):
        self._handlers.setdefault(event_type, []).append(handler)

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>publish</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>publish</span><span class='qk-c'>(</span><span class='qk-n'>event</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'><a class='qk-typelink' href='/quadkit/packages/foundation/quadkit-contracts/api/#result'>Result</a><span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> EventError<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L87' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Publish an event to all subscribers.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`event`</td><td class='qk-tt'>Any</td><td class='qk-td'>DomainEvent to publish.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/foundation/quadkit-contracts/api/#result'>Result</a><span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> EventError<span class='qk-c'>]</span></td><td class='qk-td'>Ok(None) when the event is successfully enqueued for dispatch. Err(EventError) when the event cannot be accepted (e.g.\ no handlers registered and the bus requires at least one).</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>subscribe</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>subscribe</span><span class='qk-c'>(</span>
    <span class='qk-n'>event_type</span><span class='qk-c'>: </span><span class='qk-t'>type</span><span class='qk-c'>,</span>
    <span class='qk-n'>handler</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/foundation/quadkit-contracts/api/#eventhandlerprotocol'>EventHandlerProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> Callable<span class='qk-c'>[</span>Ellipsis<span class='qk-c'>,</span> Awaitable<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>]</span></span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L100' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Subscribe a handler to an event type.

Implementations accept either an ``EventHandlerProtocol`` object or a
plain async callable — the latter is what ``@event_handler`` produces
and what the flagship
EventBus dispatches directly.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`event_type`</td><td class='qk-tt'>type</td><td class='qk-td'>Type of event to subscribe to.</td></tr><tr><td class='qk-tn'>`handler`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/foundation/quadkit-contracts/api/#eventhandlerprotocol'>EventHandlerProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> Callable<span class='qk-c'>[</span>Ellipsis<span class='qk-c'>,</span> Awaitable<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>]</span></td><td class='qk-td'>Handler to call when event is published.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>flush</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>flush</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L118' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Wait until all pending events have been dispatched.

Useful in tests and during shutdown to drain in-flight handler
dispatches before tearing the bus down.  Implemented by the flagship
EventBus; engines that dispatch
synchronously may implement it as a no-op.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>unsubscribe</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>unsubscribe</span><span class='qk-c'>(</span>
    <span class='qk-n'>event_type</span><span class='qk-c'>: </span><span class='qk-t'>type</span><span class='qk-c'>,</span>
    <span class='qk-n'>handler</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/foundation/quadkit-contracts/api/#eventhandlerprotocol'>EventHandlerProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> Callable<span class='qk-c'>[</span>Ellipsis<span class='qk-c'>,</span> Awaitable<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>]</span></span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L128' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Remove a handler subscription for an event type.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`event_type`</td><td class='qk-tt'>type</td><td class='qk-td'>Type of event to unsubscribe from.</td></tr><tr><td class='qk-tn'>`handler`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/foundation/quadkit-contracts/api/#eventhandlerprotocol'>EventHandlerProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> Callable<span class='qk-c'>[</span>Ellipsis<span class='qk-c'>,</span> Awaitable<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>]</span></td><td class='qk-td'>Handler to remove.</td></tr></tbody></table></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `EventHandlerProtocol`

</div>

<span class='qk-meta' data-api-type='Protocols'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L27' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Protocol for event handlers.

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>handle</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>handle</span><span class='qk-c'>(</span><span class='qk-n'>event</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'><a class='qk-typelink' href='/quadkit/packages/foundation/quadkit-contracts/api/#result'>Result</a><span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> EventError<span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L30' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Handle an event.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`event`</td><td class='qk-tt'>Any</td><td class='qk-td'>DomainEvent to handle.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/foundation/quadkit-contracts/api/#result'>Result</a><span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> EventError<span class='qk-c'>]</span></td><td class='qk-td'>``Ok(None)`` on success, ``Err(EventError)`` if handling fails in an expected, recoverable way.</td></tr></tbody></table></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `EventStoreProtocol`

</div>

<span class='qk-meta' data-api-type='Protocols'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/store.py#L14' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Protocol for event store implementations.

The event store persists and retrieves domain events.


**Example**

```python
class PostgresEventStore:
    async def append(self, stream_id, events, expected_version=None):
        async with self.db.transaction():
            for event in events:
                await self.db.insert("events", event.to_dict())

append

async def append(
    stream_id: str,
    events: list[Any],
    expected_version: int | None = None
) -> int

source

Append events to a stream.

Parameters

Parameter	Type	Description
`stream_id`	str	Unique stream identifier.
`events`	list[Any]	List of events to append.
`expected_version`	int \| None	Expected stream version for optimistic concurrency.

Returns

Type	Description
int	New stream version.

Raises

Exception	Description
ConcurrencyError	If expected version doesn't match.

read

async def read(
    stream_id: str,
    from_version: int = 0,
    to_version: int | None = None,
    limit: int | None = None
) -> list[Any]

source

Read events from a stream.

Every shipped store — abstract, in-memory, SQLite, Postgres, MongoDB, Redis, outbox and streaming — declares this exact signature. The protocol previously said (stream_id, start, count); because the container only checks member presence, that drift bound cleanly and surfaced only as type: ignore[call-arg] at call sites (S23). Widening here is backward compatible: the added parameter is optional, so positional and keyword callers both keep working.

Parameters

Parameter	Type	Description
`stream_id`	str	Unique stream identifier.
`from_version`	int	First version to return; 0 reads from the start.
`to_version`	int \| None	Last version to return, inclusive; ``None`` is unbounded.
`limit`	int \| None	Maximum number of events to return.

Returns

Type	Description
list[Any]	List of events.

read_all

async def read_all(
    position: int = 0,
    count: int | None = None
) -> list[Any]

source

Read events from all streams.

Parameters

Parameter	Type	Description
`position`	int	Starting global position.
`count`	int \| None	Maximum events to read.

Returns

Type	Description
list[Any]	List of events.

`ExportableProtocol`

source

Protocol for data export.

export_data

async def export_data(
    file_format: str,
    filters: dict[str, Any] | None = None,
    fields: list[str] | None = None
) -> bytes

source

get_supported_formats

def get_supported_formats() -> list[str]

source

`GracefulShutdownProtocol`

source

Standard protocol for services that support graceful shutdown.

Any service that holds background tasks, open connections, or in-flight work SHOULD implement this protocol. The framework will call shutdown() on all registered services during teardown.

Behavioral Contract: - shutdown() MUST be idempotent — calling it twice is safe. - shutdown() SHOULD complete within a reasonable timeout. - shutdown() MUST NOT raise exceptions. - shutdown() SHOULD release all held resources (tasks, connections, file handles) before returning.

shutdown

async def shutdown() -> None

source

Request graceful shutdown and wait for completion.

`HasherProtocol`

source

General-purpose hashing protocol (non-password-specific).

algorithm

property algorithm() -> str

source

Hash algorithm name (for example sha256 or blake2b).

digest

def digest(data: str | bytes) -> str

source

Hash input data and return an encoded digest.

Parameters

Parameter	Type	Description
`data`	str \| bytes	Input string or bytes to hash.

Returns

Type	Description
str	Encoded hash string.

verify_digest

def verify_digest(
    data: str | bytes,
    expected: str
) -> bool

source

Constant-time verification against an expected digest.

Parameters

Parameter	Type	Description
`data`	str \| bytes	Input string or bytes to hash.
`expected`	str	Expected encoded digest.

Returns

Type	Description
bool	True when the digest matches, False otherwise.

hash

async def hash(value: str) -> str

source

Backward-compatible async alias for digest.

verify

async def verify(
    value: str,
    hashed_value: str
) -> bool

source

Backward-compatible async alias for verify_digest.

`HealthCheckAggregatorProtocol`

source

Protocol for services that aggregate health checks from multiple providers.

Implementations collect HealthCheckProtocol instances, run them concurrently, and return a single AggregateHealthResult.

def register(
    name: str,
    check: HealthCheckProtocol,
    *,
    category: HealthCheckCategory = HealthCheckCategory.READINESS
) -> None

source

Parameters

Parameter	Type	Description
`name`	str	Unique component name for this check.
`check`	HealthCheckProtocol	Object implementing HealthCheckProtocol.
`category`	HealthCheckCategory	Probe category for this check. Defaults to HealthCheckCategory.READINESS.

run_all

async def run_all(
    timeout: float = 5.0,
    *,
    category: HealthCheckCategory | None = None
) -> AggregateHealthResult

source

Run all registered checks concurrently and aggregate results.

Parameters

Parameter	Type	Description
`timeout`	float	Per-check timeout in seconds.
`category`	HealthCheckCategory \| None	Optional probe category filter.

Returns

Type	Description
AggregateHealthResult	AggregateHealthResult containing all component results.

run_liveness

async def run_liveness(timeout: float = 5.0) -> AggregateHealthResult

source

Run only liveness checks and aggregate results.

run_readiness

async def run_readiness(timeout: float = 5.0) -> AggregateHealthResult

source

Run only readiness checks and aggregate results.

run_startup

async def run_startup(timeout: float = 5.0) -> AggregateHealthResult

source

Run only startup checks and aggregate results.

`HealthCheckProtocol`

source

Protocol for health check capability.

Expected Behavior:

health_check: MUST return a result within a reasonable timeout (~5s).
health_check: SHOULD NOT raise exceptions; wrap errors in HealthCheckResult.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

`HealthCheckRegistryProtocol`

source

Protocol for a categorised health check registry.

Implementations (e.g. HealthChecker) store checks tagged with a HealthCheckCategory so that callers can query subsets independently:

run_liveness — is the process alive and not deadlocked?
run_readiness — is the process ready to accept traffic?
run_startup — has initial startup completed?

This maps directly to the three Kubernetes probe types.

add

def add(
    name: str,
    check: Callable[[], Any],
    *,
    timeout: float | None = None,
    critical: bool = True,
    category: HealthCheckCategory = HealthCheckCategory.READINESS
) -> None

source

Parameters

Parameter	Type	Description
`name`	str	Unique identifier for the check.
`check`	Callable[[], Any]	Callable that performs the check.
`timeout`	float \| None	Optional per-check timeout in seconds.
`critical`	bool	Whether a non-healthy result should make the aggregate readiness status ``UNHEALTHY. Defaults to` `True``.
`category`	HealthCheckCategory	HealthCheckCategory value. Defaults to ``READINESS``.

run_all

async def run_all() -> tuple[Any, dict[str, Any]]

source

Run all registered checks regardless of category.

Returns

Type	Description
tuple[Any, dict[str, Any]]	``(aggregate_status, per_check_results)`` tuple.

run_liveness

async def run_liveness() -> tuple[Any, dict[str, Any]]

source

Run only LIVENESS checks.

Returns

Type	Description
tuple[Any, dict[str, Any]]	``(aggregate_status, per_check_results)`` tuple.

run_readiness

async def run_readiness() -> tuple[Any, dict[str, Any]]

source

Run only READINESS checks.

Returns

Type	Description
tuple[Any, dict[str, Any]]	``(aggregate_status, per_check_results)`` tuple.

run_startup

async def run_startup() -> tuple[Any, dict[str, Any]]

source

Run only STARTUP checks.

Returns

Type	Description
tuple[Any, dict[str, Any]]	``(aggregate_status, per_check_results)`` tuple.

`HttpRequestLoggerProtocol`

source

Protocol for HTTP request/response logging middleware.

Defines how to log completed HTTP requests with duration, status code, and optional metadata for monitoring and audit purposes.

Example

class RequestLogger:
    async def log_request(
        self,
        method: str,
        path: str,
        status_code: int,
        duration_ms: float,
        request_id: str | None = None,
        **metadata: Any,
    ) -> None:
        logger.info(
            "request_completed",
            method=method,
            path=path,
            status=status_code,
            duration_ms=duration_ms,
            request_id=request_id,
            **metadata,
        )

log_request

async def log_request(
    method: str,
    path: str,
    status_code: int,
    duration_ms: float,
    request_id: str | None = None,
    **metadata: Any
) -> None

source

Log a completed HTTP request.

Parameters

Parameter	Type	Description
`method`	str	HTTP method (GET, POST, etc.).
`path`	str	Request path URI.
`status_code`	int	HTTP response status code.
`duration_ms`	float	Request processing duration in milliseconds.
`request_id`	str \| None	Optional request identifier for tracing. **metadata: Additional context-specific data (client_id, user_id, etc.).

`IdGeneratorProtocol`

source

Injectable identity generator.

strategy

property strategy() -> IdStrategy

source

Return the active ID generation strategy.

generate

def generate() -> str

source

Generate a new unique identifier.

generate_for

def generate_for(entity_type: str) -> str

source

Generate a new identifier for a specific entity type.

`IdempotencyMiddlewareProtocol`

source

Protocol for HTTP idempotency deduplication middleware.

process

async def process(
    headers: dict[str, str],
    handler: Any,
    *args: Any,
    **kwargs: Any
) -> Any

source

Run the handler with idempotency deduplication.

ttl

property ttl() -> float

source

Default TTL (seconds) for cached idempotency results.

`IdempotencyStoreProtocol`

source

Protocol for storing and checking idempotency keys.

get

async def get(key: str) -> Any | None

source

Retrieve a stored result by idempotency key.

Parameters

Parameter	Type	Description
`key`	str	The idempotency key.

Returns

Type	Description
Any \| None	The stored result, or ``None`` when the key is not present or expired. Store failures raise.

Note

Implementations return the unwrapped value directly — this is the shape implemented by every concrete store (InMemoryIdempotencyStore, RedisIdempotencyStore, DatabaseIdempotencyStore, and the multimedia in-memory store) and consumed by IdempotencyMiddleware. The previous revision declared a Result wrapper here that no implementation ever returned.

get_record

async def get_record(key: str) -> Result[IdempotencyRecord | None, IdempotencyError]

source

Retrieve the full idempotency record, including metadata.

set

async def set(
    key: str,
    value: Any,
    ttl: float | None = None
) -> None

source

Store a result with an optional time-to-live.

Parameters

Parameter	Type	Description
`key`	str	The idempotency key.
`value`	Any	The result to store.
`ttl`	float \| None	Time-to-live in seconds, or ``None`` for no expiry.

delete

async def delete(key: str) -> None

source

Remove an idempotency record by key.

acquire

async def acquire(
    key: str,
    ttl: int
) -> bool

source

Atomically claim an idempotency key if it is not already held.

Parameters

Parameter	Type	Description
`key`	str	The idempotency key to acquire.
`ttl`	int	Time-to-live in seconds for the claimed key.

Returns

Type	Description
bool	``True` `if this caller should proceed.` `False`` if the key is already claimed. Store failures raise.

`JobProtocol`

source

Protocol for JobProtocol objects.

`KeyDerivationProtocol`

source

Protocol for key derivation services.

derive

async def derive(
    secret: str,
    *,
    salt: bytes | None = None
) -> str

source

Derive an encoded key from a secret.

Parameters

Parameter	Type	Description
`secret`	str	Input secret to derive from.
`salt`	bytes \| None	Optional salt. When omitted, implementations generate one.

Returns

Type	Description
str	Stable encoded key derivation string.

verify

async def verify(
    secret: str,
    encoded: str
) -> bool

source

Verify a secret against a derived key payload.

Parameters

Parameter	Type	Description
`secret`	str	Input secret to verify.
`encoded`	str	Stable encoded key derivation string.

Returns

Type	Description
bool	True when the secret matches the encoded payload, False otherwise.

hash

async def hash(
    secret: str,
    *,
    salt: bytes | None = None
) -> str

source

Backward-compatible async alias for derive.

`LLMClientProtocol`

source

Protocol for LLM client implementations.

All LLM providers (OpenAI, Anthropic, Google, etc.) should implement this interface.

complete

async def complete(
    messages: Sequence[ChatMessageProtocol],
    *,
    model: str | None = None,
    temperature: float | None = None,
    max_tokens: int | None = None,
    tools: Sequence[Any] | None = None,
    stop_sequences: Sequence[str] | None = None,
    **kwargs: Any
) -> Result[CompletionProtocol, LLMError]

source

Generate a completion from messages.

Parameters

Parameter	Type	Description
`messages`	Sequence[ChatMessageProtocol]	List of chat messages.
`model`	str \| None	Optional model override.
`temperature`	float \| None	Optional sampling temperature.
`max_tokens`	int \| None	Optional max output tokens.
`tools`	Sequence[Any] \| None	Optional tool definitions.
`stop_sequences`	Sequence[str] \| None	Optional list of stop sequences for early termination. **kwargs: Provider-specific options.

Returns

Type	Description
Result[CompletionProtocol, LLMError]	``Ok(Completion)` `on success,` `Err(LLMError)`` on recoverable failure.

stream_chat

def stream_chat(
    messages: list[ChatMessageProtocol],
    *,
    model: str | None = None,
    temperature: float | None = None,
    max_tokens: int | None = None,
    tools: list[Any] | None = None,
    stop_sequences: list[str] | None = None,
    **kwargs: Any
) -> AsyncStream[StreamChunk, LLMError]

source

Start a streaming completion.

This method returns an AsyncStream immediately (not wrapped in Result). The stream is established lazily when iteration begins. Setup failures and mid-stream failures are both surfaced through the stream’s typed error channel.

Parameters

Parameter	Type	Description
`messages`	list[ChatMessageProtocol]	List of chat messages.
`model`	str \| None	Optional model override.
`temperature`	float \| None	Optional sampling temperature.
`max_tokens`	int \| None	Optional max output tokens.
`tools`	list[Any] \| None	Optional tool definitions.
`stop_sequences`	list[str] \| None	Optional list of stop sequences for early termination. **kwargs: Provider-specific options.

Returns

Type	Description
AsyncStream[StreamChunk, LLMError]	``AsyncStream[StreamChunk, LLMError]` `that yields chunks or surfaces typed errors through terminal operations (collect(),` `first(),` `drain()``).

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight connectivity check.

Returns

Type	Description
HealthCheckResult	Structured health check result.

async def close() -> None

source

Close the client and release resources.

`LockManagerProtocol`

source

Protocol for a process-local or distributed lock manager.

A lock manager creates and tracks named locks for coordinating mutual exclusion of concurrent operations. Callers use acquire to obtain a per-key async context manager.

Implementations range from purely in-memory (single-process deduplication) to Redis-backed or advisory-lock-backed distributed variants.

Example

from quadkit.contracts.core.lock import LockManagerProtocol

async def process(manager: LockManagerProtocol) -> None:
    async with manager.acquire("resource:42", timeout=30):
        await do_work()

from quadkit.contracts.core.lock import LockManagerProtocol

async def process(manager: LockManagerProtocol) -> None:
    async with manager.acquire("resource:42", timeout=30):
        await do_work()

Container registration

container.singleton(LockManagerProtocol, InMemoryLockManager)

container.singleton(LockManagerProtocol, InMemoryLockManager)

acquire

def acquire(
    key: str,
    timeout: float = 60.0
) -> AbstractAsyncContextManager[Any]

source

Return an async context manager that holds the named lock.

The caller must use the returned value as an async context manager. The lock is acquired on __aenter__ and released on __aexit__.

Parameters

Parameter	Type	Description
`key`	str	Unique, stable identifier for the resource to protect.
`timeout`	float	Informational TTL in seconds; implementations may use this to automatically expire stale locks.

Returns

Type	Description
AbstractAsyncContextManager[Any]	An ``AbstractAsyncContextManager`` for the named lock.

`LockStoreProtocol`

source

Persistent distributed lock store.

Individual lock handles are not returned; all operations are done via the lock name and owner identifier.

acquire

async def acquire(
    lock_name: str,
    owner: str,
    ttl: int
) -> bool

source

Attempt to acquire the named lock.

Parameters

Parameter	Type	Description
`lock_name`	str	Globally unique lock identifier.
`owner`	str	Identifier of the requesting owner (e.g. host + PID).
`ttl`	int	Lock time-to-live in seconds. The lock is automatically released after this duration to prevent deadlocks.

Returns

Type	Description
bool	``True` `if the lock was acquired;` `False`` if it is already held.

release

async def release(
    lock_name: str,
    owner: str
) -> bool

source

Release the named lock only if it is held by owner.

Parameters

Parameter	Type	Description
`lock_name`	str	Globally unique lock identifier.
`owner`	str	Must match the owner that acquired the lock.

Returns

Type	Description
bool	``True` `if the lock was released;` `False`` if it was not held by owner or did not exist.

extend

async def extend(
    lock_name: str,
    owner: str,
    ttl: int
) -> bool

source

Extend the TTL of a currently-held lock.

Parameters

Parameter	Type	Description
`lock_name`	str	Globally unique lock identifier.
`owner`	str	Must match the owner that currently holds the lock.
`ttl`	int	New time-to-live in seconds from now.

Returns

Type	Description
bool	``True` `if the extension was applied;` `False`` if the lock was not held by owner.

`MCPServerProtocol`

source

Protocol for the MCP server.

The server routes JSON-RPC messages to handlers and returns JSON-RPC responses.

handle_message

async def handle_message(message: dict[str, Any]) -> dict[str, Any] | None

source

Handle a JSON-RPC message and return the response.

Returns None for notifications (no response expected).

`MCPToolProviderProtocol`

source

Protocol for providing tools to the MCP server.

Satisfied by:

ToolRegistryAdapter (bridges quadkit-agents ToolRegistry)
Any custom tool provider

list_tools

async def list_tools() -> list[dict[str, Any]]

source

List all available tools in MCP format.

Returns list of dicts with name, description, inputSchema keys.

call_tool

async def call_tool(
    name: str,
    arguments: dict[str, Any]
) -> dict[str, Any]

source

Execute a tool by name with given arguments.

`MCPTransportProtocol`

source

Protocol for MCP transport implementations.

A transport handles the I/O layer — reading requests and writing responses. The MCP server is transport-agnostic.

start

async def start() -> None

source

Start the transport.

stop

async def stop() -> None

source

Stop the transport.

`MemoryConsolidatorProtocol`

source

Protocol for consolidating and optimizing memory.

Consolidation compresses old entries, deduplicates, extracts entities, and manages memory size to prevent unbounded growth.

consolidate

async def consolidate(entries: list[MemoryEntry]) -> ConsolidationResult

source

Consolidate a batch of memory entries.

Applies compression, deduplication, entity extraction, and pruning strategies to optimize memory storage and retrieval.

Parameters

Parameter	Type	Description
`entries`	list[MemoryEntry]	Entries to consolidate

Returns

Type	Description
ConsolidationResult	Consolidation result with statistics

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check if the memory consolidator backend is reachable and healthy.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for a response.

Returns

Type	Description
HealthCheckResult	Health check result with status and details.

`MemoryProtocol`

source

Protocol for conversation memory.

add_message

async def add_message(message: Any) -> None

source

Add a message to memory.

Parameters

Parameter	Type	Description
`message`	Any	Message to add.

get_messages

async def get_messages() -> list[Any]

source

Get all messages from memory.

Returns

Type	Description
list[Any]	List of messages.

clear

async def clear() -> None

source

Clear all messages from memory.

`MemoryStoreProtocol`

source

Protocol for storing and retrieving memory entries.

Implementations should provide persistent or semi-persistent storage for memory entries, supporting both sequential and search-based access.

store

async def store(entry: MemoryEntry) -> None

source

Store a single memory entry.

Parameters

Parameter	Type	Description
`entry`	MemoryEntry	The memory entry to store

retrieve

async def retrieve(query: MemoryQuery) -> list[MemorySearchResult]

source

Search for memory entries based on a query.

Parameters

Parameter	Type	Description
`query`	MemoryQuery	Query parameters

Returns

Type	Description
list[MemorySearchResult]	List of matching entries with scores, ordered by relevance

get_recent

async def get_recent(
    n: int,
    owner_id: str
) -> list[MemoryEntry]

source

Get the N most recent entries owned by owner_id.

Parameters

Parameter	Type	Description
`n`	int	Number of entries to return
`owner_id`	str	Owner scope; only this owner's entries are returned

Returns

Type	Description
list[MemoryEntry]	List of recent entries in descending temporal order

delete

async def delete(
    entry_id: str,
    owner_id: str
) -> None

source

Delete an entry by ID, scoped to owner_id.

Parameters

Parameter	Type	Description
`entry_id`	str	ID of the entry to delete
`owner_id`	str	Owner scope; only this owner's entry may be deleted

clear

async def clear(owner_id: str) -> None

source

Clear all entries owned by owner_id from the store.

Parameters

Parameter	Type	Description
`owner_id`	str	Owner scope; only this owner's entries are cleared

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a lightweight connectivity check.

`MetricsExpositionProtocol`

source

Protocol for exporting text/binary metrics expositions (e.g. Prometheus).

content_type

property content_type() -> str

source

MIME content type for the rendered exposition.

export

def export() -> str | bytes

source

Render the current metrics exposition payload.

`MigrationManagerProtocol`

source

Protocol for migration management.

initialize_migration_table

async def initialize_migration_table() -> None

source

Initialize the migration tracking table.

get_applied_migrations

async def get_applied_migrations() -> list[MigrationRecord]

source

Get list of applied migrations.

apply_migration

async def apply_migration(
    version: str,
    name: str,
    sql: str
) -> bool

source

Apply a migration.

rollback_migration

async def rollback_migration(version: str) -> bool

source

Rollback a migration.

get_pending_migrations

async def get_pending_migrations(available_migrations: list[str]) -> list[str]

source

Get migrations that haven’t been applied yet.

`MigrationRunnerProtocol`

source

Executes database migrations in the correct order.

Implementations must be idempotent: running run_migrations() on an already up-to-date schema must be a no-op. Each migration is identified by a unique string version identifier (e.g. "20240101_001" or the Alembic revision hash).

Typical usage

runner = await container.resolve(MigrationRunnerProtocol)
result = await runner.run_migrations()
if result:
    logger.info("migrations_applied", count=len(result))
else:
    logger.info("schema_up_to_date")

runner = await container.resolve(MigrationRunnerProtocol)
result = await runner.run_migrations()
if result:
    logger.info("migrations_applied", count=len(result))
else:
    logger.info("schema_up_to_date")

run_migrations

async def run_migrations() -> list[str]

source

Apply all pending migrations.

Returns

Type	Description
list[str]	List of migration version identifiers that were applied. Empty list means the schema was already up-to-date.

rollback

async def rollback(target: str | None = None) -> list[str]

source

Roll back migrations to target.

Parameters

Parameter	Type	Description
`target`	str \| None	Version identifier to roll back to. ``None`` rolls back the most recent migration only.

Returns

Type	Description
list[str]	List of migration version identifiers that were rolled back.

get_current_version

async def get_current_version() -> str | None

source

Return the version identifier of the most recently applied migration.

Returns

Type	Description
str \| None	Version string, or ``None`` if no migrations have been applied.

get_pending_migrations

async def get_pending_migrations() -> list[str]

source

Return version identifiers of migrations not yet applied.

Returns

Type	Description
list[str]	Ordered list of pending migration version identifiers.

`ObjectMapperProtocol`

source

Protocol for a registry-based object-to-object mapper.

Provides a clean map() API for converting between registered type pairs without ad-hoc conversion logic scattered through the codebase.

map

def map(
    source: Any,
    dest_type: type[Any],
    *,
    validate: bool = False,
    validator: Any | None = None
) -> Any

source

Map a source object to the destination type.

Uses a registered mapping function to transform the source instance into an instance of dest_type.

Parameters

Parameter	Type	Description
`source`	Any	The source object to transform.
`dest_type`	type[Any]	The target type to produce.
`validate`	bool	Whether to validate the result after mapping.
`validator`	Any \| None	Optional validator instance to use.

Returns

Type	Description
Any	A new instance of dest_type.

Raises

Exception	Description
MappingNotFoundError	If no mapping is registered for this type pair.

def register(
    source_type: type[Any],
    dest_type: type[Any],
    mapper_func: Any
) -> None

source

Parameters

Parameter	Type	Description
`source_type`	type[Any]	The type to map from.
`dest_type`	type[Any]	The type to map to.
`mapper_func`	Any	A callable that accepts a source instance and returns a dest instance.

`OffsetPageProtocol`

source

Protocol for offset-based page request parameters.

Implemented by offset/limit style pagination inputs. Use this for code that should work with any offset-pagination type.

offset

property offset() -> int

source

Zero-based record offset.

limit

property limit() -> int

source

Maximum number of records to return.

`OnApplicationBootstrapProtocol`

source

Provider implements this to run after application bootstrap.

Called after ALL providers have been booted successfully. At this point every service is fully resolved and available.

Behavioral Contract: - Implementers MUST NOT throw exceptions — doing so indicates a fatal application startup failure. - Implementers CAN resolve any service from the container. - Implementers SHOULD start background tasks, schedulers, or listeners here rather than in on_module_init().

on_application_bootstrap

async def on_application_bootstrap() -> None

source

Run after all providers are booted.

`OnApplicationShutdownProtocol`

source

Provider implements this to run after shutdown completes.

Called after ALL providers have been shut down. Use this for final cleanup like closing log files or flushing metrics.

Behavioral Contract: - Implementers MUST NOT resolve services — the container is already disposed at this point. - Implementers MUST NOT throw exceptions. - Implementers SHOULD release any remaining OS resources.

on_application_shutdown

async def on_application_shutdown(signal: str | None = None) -> None

source

Run after all providers are shut down.

`OnBeforeShutdownProtocol`

source

Provider implements this to run before shutdown begins.

Called before any provider’s shutdown() method is invoked. Use this to drain queues, finish in-flight requests, or save state.

Behavioral Contract: - Implementers MUST complete within a reasonable timeout. - Implementers SHOULD NOT start new work — only finish existing work. - Implementers MUST NOT throw exceptions.

on_before_shutdown

async def on_before_shutdown(signal: str | None = None) -> None

source

Run before shutdown begins.

`OnModuleInitProtocol`

source

Provider implements this to run initialization logic.

Called once when the provider is first booted, BEFORE other providers that depend on it.

Behavioral Contract: - Implementers MUST be idempotent — calling on_module_init() twice MUST NOT produce side effects or raise. - Implementers SHOULD complete quickly — blocking here delays the entire boot sequence. - Implementers MUST NOT resolve services from other providers that have not yet been booted.

on_module_init

async def on_module_init() -> None

source

Initialize provider after registration but before other providers boot.

`PasswordHasherProtocol`

source

Protocol for password hashing services.

Responsible for hashing passwords and verifying them against hashes.

hash

async def hash(password: str) -> str

source

Hash a plain text password.

Parameters

Parameter	Type	Description
`password`	str	Plain text password.

Returns

Type	Description
str	Hashed password string.

verify

async def verify(
    password: str,
    hashed_password: str
) -> bool

source

Verify a password against a hash.

Parameters

Parameter	Type	Description
`password`	str	Plain text password to check.
`hashed_password`	str	Stored hash to compare against.

Returns

Type	Description
bool	True if password matches hash, False otherwise.

needs_rehash

def needs_rehash(hashed_password: str) -> bool

source

Return True when the stored hash should be re-computed.

Implementations parse the cost parameters from the self-describing stored hash and compare them against their configured target. Unparseable or unknown formats return True (fail-closed) — safe because rehashing only ever runs after a successful verify().

Parameters

Parameter	Type	Description
`hashed_password`	str	Stored hash string.

Returns

Type	Description
bool	True when the hash is below the configured cost target.

rehash_if_needed

async def rehash_if_needed(
    password: str,
    hashed_password: str | None
) -> str | None

source

Rehash password when hashed_password is below the cost target.

Django check_password(setter=...) pattern: run this after a successful verify() and persist the returned hash to upgrade stored credentials in place. Returns None when no upgrade is needed (or no stored hash exists).

Parameters

Parameter	Type	Description
`password`	str	Plain text password (already verified against the hash).
`hashed_password`	str \| None	Stored hash string, or None.

Returns

Type	Description
str \| None	A freshly computed hash when an upgrade is needed, else None.

`PoolManagerProtocol`

source

Manager that tracks pools by name.

get_stats

def get_stats() -> dict[str, PoolStatsProtocol]

source

get_pool

async def get_pool(name: str) -> PoolProtocol

source

`PoolProtocol`

source

Protocol for an individual connection pool.

get_stats

def get_stats() -> PoolStatsProtocol

source

async def close() -> None

source

`PoolStatsProtocol`

source

Snapshot of pool statistics.

`PromptAssemblerProtocol`

source

Assembles prompt layers in cache-friendly static-to-dynamic order.

The static-first ordering maximizes provider-side KV cache reuse. The assembler also injects provider-specific cache annotations (Anthropic’s cache_control breakpoints, DeepSeek’s 64-token padding, etc.).

assemble

def assemble(
    system: str,
    tools: list[Any] | None,
    reference_docs: list[str] | None,
    few_shot: list[ChatMessage] | None,
    history: list[ChatMessage],
    query: str,
    provider: str,
    dynamic_metadata: str | None = None
) -> list[ChatMessage]

source

Assemble a complete prompt message list.

Parameters

Parameter	Type	Description
`system`	str	System instructions (static, cached).
`tools`	list[Any] \| None	Tool/function definitions.
`reference_docs`	list[str] \| None	Reference documents as text (semi-static).
`few_shot`	list[ChatMessage] \| None	Few-shot example messages (static, cached).
`history`	list[ChatMessage]	Chat history messages (dynamic).
`query`	str	Current user query (dynamic).
`provider`	str	Provider name for cache annotation strategy.
`dynamic_metadata`	str \| None	Timestamps, user IDs, etc. (dynamic).

Returns

Type	Description
list[ChatMessage]	Ordered list of ChatMessage instances ready for the LLM client.

`PromptCompressorProtocol`

source

Compresses text to fit within a token budget.

Implementations range from learned compression (LLMLingua-2) to heuristic truncation. The protocol guarantees that the returned text fits within target_token_count.

Placement note: Lives in ai/rag.py because its primary consumer is the RAG context compression stage.

compress

async def compress(
    text: str,
    target_token_count: int,
    force_tokens: list[str] | None = None
) -> str

source

Compress text to fit within target_token_count.

Parameters

Parameter	Type	Description
`text`	str	The raw text to compress.
`target_token_count`	int	Maximum tokens in the result.
`force_tokens`	list[str] \| None	Tokens that must never be removed.

Returns

Type	Description
str	Compressed text fitting within the budget.

`ProviderProtocol`

source

Contract for framework providers.

Providers are responsible for:

Registering bindings into the container (register phase)
Initializing external resources (boot phase)
Cleaning up resources on shutdown (shutdown phase)

Expected Behavior:

register: MUST be declarative. No external side effects or resolution.
boot: CAN resolve services. MUST be idempotent if called twice.
shutdown: MUST gracefully close resources. SHOULD NOT raise.

name

property name() -> str

source

Unique provider identifier (e.g. ‘auth’, ‘database’).

priority

property priority() -> ProviderPriority

source

Initialization order hint.

dependencies

property dependencies() -> Sequence[str]

source

Names of providers that must be booted before this one.

async def register(container: ContainerRegistrarProtocol) -> None

source

Bind services into the container. declarative phase.

boot

async def boot(container: BootContainerProtocol) -> None

source

Initialize and wire services. Resolution and registration allowed here.

shutdown

async def shutdown() -> None

source

Tear down resources on application exit.

on_error

async def on_error(
    error: Exception,
    phase: str
) -> None

source

Called when boot() or shutdown() raises an exception.

Override to perform cleanup on startup/shutdown failure.

Parameters

Parameter	Type	Description
`error`	Exception	The exception that was raised.
`phase`	str	Either 'boot' or 'shutdown'.

`QueryBusProtocol`

source

Protocol for query bus implementations.

Example

class QueryBusProtocol:
    async def execute(self, query: Query) -> Any:
        handler = self._handlers[type(query)]
        return await handler.handle(query)

execute

async def execute(query: Any) -> Any

source

Execute a query through its handler.

Parameters

Parameter	Type	Description
`query`	Any	Query to execute.

Returns

Type	Description
Any	Query result.

`QueryLoggerProtocol`

source

Protocol for logging database queries.

Implement this protocol to enable query logging in database providers. The concrete quadkit.sql.logging.BaseQueryLogger implements this interface and adds helper methods for querying the stored entries.

log_query

async def log_query(entry: QueryLogEntry) -> None

source

Log a query execution.

get_recent_queries

async def get_recent_queries(limit: int = 100) -> list[QueryLogEntry]

source

Return the most recent limit entries.

get_slow_queries

async def get_slow_queries(
    threshold_seconds: float,
    limit: int = 50
) -> list[QueryLogEntry]

source

Return entries whose execution time exceeds threshold_seconds.

get_query_stats

async def get_query_stats(time_range_seconds: int = 3600) -> dict[str, Any]

source

Return aggregated statistics over a time window.

`ReadOnlyRepositoryProtocol`

source

Protocol for read-only repository operations.

Use this for query-only access patterns (CQRS query side).

Example

class UserQueryRepository:
    async def get(self, id: str) -> User | None:
        return await self.db.query("users").where(id=id).first()

    async def list(self, skip: int = 0, limit: int = 100) -> list[User]:
        return await self.db.query("users").offset(skip).limit(limit).all()

get

async def get(item_id: str) -> T | None

source

Get entity by ID.

Parameters

Parameter	Type	Description
`item_id`	str	Entity identifier.

Returns

Type	Description
T \| None	Entity if found, None otherwise.

list

async def list(
    skip: int = 0,
    limit: int = 100,
    **filters: Any
) -> list[T]

source

List entities with pagination.

Parameters

Parameter	Type	Description
`skip`	int	Number of records to skip.
`limit`	int	Maximum records to return. **filters: Optional filter criteria.

Returns

Type	Description
list[T]	List of entities.

find_by_spec

async def find_by_spec(spec: SpecificationProtocol[T]) -> list[T]

source

Find entities matching a complex specification.

Parameters

Parameter	Type	Description
`spec`	SpecificationProtocol[T]	The DDD specification to evaluate.

Returns

Type	Description
list[T]	List of matching entities.

count

async def count(**filters: Any) -> int

source

Count entities matching filters.

Parameters

Parameter	Type	Description

Returns

Type	Description
int	Total count of matching entities.

`RedisClientProtocol`

source

The commands Quadkit issues against a raw Redis client.

Structural by design: redis.asyncio.Redis, fakeredis and purpose built doubles satisfy it without inheriting from it. Both the synchronous command methods below and the connection lifecycle stay the client’s own; only what the framework calls is named here.

The return values stay Any because every Redis command answers with the decoded payload of its own shape — callers narrow what they read.

eval

async def eval(
    script: str,
    numkeys: int,
    *keys_and_args: Any
) -> Any

source

Run script server-side.

Parameters

Parameter	Type	Description
`script`	str	Lua source for the command.
`numkeys`	int	How many of the following arguments are key names. *keys_and_args: Key names followed by the script arguments.

Returns

Type	Description
Any	Whatever the script returns.

get

async def get(name: str) -> Any

source

Return the value stored at name.

Parameters

Parameter	Type	Description
`name`	str	Key to read.

Returns

Type	Description
Any	The stored value, or ``None`` when the key is absent.

set

async def set(
    name: str,
    value: Any,
    *,
    ex: int | None = None,
    nx: bool = False,
    xx: bool = False
) -> Any

source

Store value at name.

Parameters

Parameter	Type	Description
`name`	str	Key to write.
`value`	Any	Payload to store.
`ex`	int \| None	Expiry in seconds.
`nx`	bool	Only write when the key does not exist.
`xx`	bool	Only write when the key already exists.

Returns

Type	Description
Any	The command result (``None, or a truthy value under` `nx/xx``).

setex

async def setex(
    name: str,
    time: int,
    value: Any
) -> Any

source

Store value at name with a lifetime.

Parameters

Parameter	Type	Description
`name`	str	Key to write.
`time`	int	Lifetime in seconds.
`value`	Any	Payload to store.

Returns

Type	Description
Any	The command result.

expire

async def expire(
    name: str,
    time: int
) -> Any

source

Set the lifetime of an existing key.

Parameters

Parameter	Type	Description
`name`	str	Key to expire.
`time`	int	Lifetime in seconds.

Returns

Type	Description
Any	``True`` when the timeout was applied.

zadd

async def zadd(
    name: str,
    mapping: Mapping[str, float]
) -> Any

source

Add members to the sorted set at name.

Parameters

Parameter	Type	Description
`name`	str	Sorted-set key.
`mapping`	Mapping[str, float]	Member name to score.

Returns

Type	Description
Any	How many members were newly added.

zcard

async def zcard(name: str) -> Any

source

Count the members of the sorted set at name.

Parameters

Parameter	Type	Description
`name`	str	Sorted-set key.

Returns

Type	Description
Any	The cardinality of the set.

zremrangebyscore

async def zremrangebyscore(
    name: str,
    min: float | str,
    max: float | str
) -> Any

source

Drop sorted-set members whose score falls inside a range.

Parameters

Parameter	Type	Description
`name`	str	Sorted-set key.
`min`	float \| str	Inclusive lower score bound, or ``"-inf"``.
`max`	float \| str	Inclusive upper score bound, or ``"+inf"``.

Returns

Type	Description
Any	How many members were removed.

`RepositoryProtocol`

source

Protocol for full repository operations.

Extends ReadOnlyRepositoryProtocol with write operations.

Example

class UserRepository:
    async def save(self, entity: User) -> User:
        if entity.id:
            await self.db.update("users", entity.dict()).where(id=entity.id)
        else:
            entity.id = await self.db.insert("users", entity.dict())
        return entity

    async def delete(self, id: str) -> bool:
        result = await self.db.delete("users").where(id=id)
        return result.affected_rows > 0

save

async def save(entity: T) -> T

source

Save (create or update) an entity.

Parameters

Parameter	Type	Description
`entity`	T	Entity to save.

Returns

Type	Description
T	Saved entity with any generated fields populated.

delete

async def delete(item_id: str) -> bool

source

Delete entity by ID.

Parameters

Parameter	Type	Description
`item_id`	str	Entity identifier.

Returns

Type	Description
bool	True if deleted, False if not found.

save_many

async def save_many(entities: list[T]) -> list[T]

source

Save (create or update) multiple entities in a single operation.

Implementations SHOULD execute this as a batch for efficiency.

Parameters

Parameter	Type	Description
`entities`	list[T]	Entities to save.

Returns

Type	Description
list[T]	Saved entities with any generated fields populated.

delete_many

async def delete_many(item_ids: list[str]) -> int

source

Delete multiple entities by ID.

Implementations SHOULD execute this as a batch for efficiency.

Parameters

Parameter	Type	Description
`item_ids`	list[str]	Entity identifiers to delete.

Returns

Type	Description
int	Number of entities actually deleted.

`SagaManagerProtocol`

source

Protocol for saga lifecycle management.

The manager routes incoming events to all registered sagas and coordinates their execution.

process

async def process(event: Any) -> None

source

Process an event through all relevant sagas.

Parameters

Parameter	Type	Description
`event`	Any	Domain event to route and process.

`SagaProtocol`

source

Protocol for saga / process-manager implementations.

Sagas coordinate long-running business processes that span multiple aggregates by reacting to domain events and dispatching commands.

handle

async def handle(event: Any) -> list[Any]

source

Handle a domain event and produce commands.

Parameters

Parameter	Type	Description
`event`	Any	Domain event to handle.

Returns

Type	Description
list[Any]	List of commands to dispatch to the command bus.

`SearchEngineProtocol`

source

Protocol for search engine backends.

All operations are index-scoped: the target index name is the first argument of every method. This matches the shape implemented by every concrete backend (Meilisearch, Elasticsearch, SQLite FTS, PostgreSQL FTS, MySQL FTS) and by the DefaultSearchEngine facade, which delegates to backends using these signatures.

(The previous revision declared a single-index shape with optional index_name kwargs; the concrete backends never matched it.)

async def search(
    index: str,
    query: str,
    filters: dict[str, Any] | None = None,
    limit: int = 20,
    offset: int = 0,
    sort: list[str] | None = None,
    **kwargs: Any
) -> Any

source

Execute a search query against index.

Parameters

Parameter	Type	Description
`index`	str	Target index name.
`query`	str	Search query string.
`filters`	dict[str, Any] \| None	Search filters (backend-specific structure).
`limit`	int	Maximum results to return.
`offset`	int	Results offset for pagination.
`sort`	list[str] \| None	Sort terms, ``"field:asc\|desc"` `strings. **kwargs: Backend-specific options (e.g.` `rule``).

Returns

Type	Description
Any	``Ok(SearchResponse)` `on success,` `Err(SearchError)`` on failure — the canonical Result type.

index_document

async def index_document(
    index: str,
    document_id: str,
    document: dict[str, Any],
    **kwargs: Any
) -> Any

source

Index a document.

Parameters

Parameter	Type	Description
`index`	str	Target index name.
`document_id`	str	Unique document identifier.
`document`	dict[str, Any]	Document data to index. **kwargs: Backend-specific options.

index_many

async def index_many(
    index: str,
    documents: list[tuple[str, dict[str, Any]]],
    **kwargs: Any
) -> Any

source

Index multiple documents in a single bulk operation.

Callers should prefer this over N repeated index_document calls when processing batches. Backends may use a native bulk API to avoid per-document round-trip overhead.

Parameters

Parameter	Type	Description
`index`	str	Target index name.
`documents`	list[tuple[str, dict[str, Any]]]	Sequence of ``(document_id, document)`` pairs. **kwargs: Backend-specific options.

get_document

async def get_document(
    index: str,
    document_id: str,
    **kwargs: Any
) -> dict[str, Any] | None

source

Retrieve a document by ID, or None when absent.

Parameters

Parameter	Type	Description
`index`	str	Target index name.
`document_id`	str	Document identifier to retrieve. **kwargs: Backend-specific options.

delete_document

async def delete_document(
    index: str,
    document_id: str,
    **kwargs: Any
) -> Any

source

Delete a document from the index.

Parameters

Parameter	Type	Description
`index`	str	Target index name.
`document_id`	str	Document identifier to delete. **kwargs: Backend-specific options.

create_index

async def create_index(
    index: str,
    **kwargs: Any
) -> Any

source

Create the index (and schema) if it does not exist.

delete_index

async def delete_index(
    index: str,
    **kwargs: Any
) -> Any

source

Delete the index and all its documents.

index_exists

async def index_exists(
    index: str,
    **kwargs: Any
) -> Any

source

Check whether the index exists.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform health check.

Returns

Type	Description
HealthCheckResult	Structured health check result.

`SecretStoreProtocol`

source

Protocol for retrieving, writing, and deleting named secrets.

Secret names may use any naming convention; a hierarchical path (e.g. "database/password") is recommended for readability and to align with most provider APIs.

All mutating operations (set_secret, delete_secret) are synchronous at the protocol level — implementations may perform async I/O internally but the public contract accepts simple calls from both sync and async contexts.

Example

store = EnvSecretStore()
val = store.get_secret("MY_API_KEY")

store = EnvSecretStore()
val = store.get_secret("MY_API_KEY")

Async-first usage via a wrapping coroutine

val = await asyncio.to_thread(store.get_secret, "MY_API_KEY")

val = await asyncio.to_thread(store.get_secret, "MY_API_KEY")

get_secret

def get_secret(name: str) -> str

source

Return the value of a secret by name.

Parameters

Parameter	Type	Description
`name`	str	Unique secret identifier (e.g. ``"stripe/api-key"``).

Returns

Type	Description
str	The plaintext secret value.

Raises

Exception	Description
SecretNotFoundError	If no secret with that name exists.
SecretAccessError	If the caller lacks permission.

set_secret

def set_secret(
    name: str,
    value: str
) -> None

source

Write or overwrite a secret.

Parameters

Parameter	Type	Description
`name`	str	Unique secret identifier.
`value`	str	Plaintext secret value to store.

Raises

Exception	Description
SecretAccessError	If the caller lacks permission to write.

delete_secret

def delete_secret(name: str) -> None

source

Delete a secret by name.

Non-existent secrets are silently ignored (idempotent delete).

Parameters

Parameter	Type	Description
`name`	str	Unique secret identifier.

Raises

Exception	Description
SecretAccessError	If the caller lacks permission to delete.

has_secret

def has_secret(name: str) -> bool

source

Return True if a secret with name exists, False otherwise.

Parameters

Parameter	Type	Description
`name`	str	Unique secret identifier.

`SemanticCacheProtocol`

source

Embedding-based semantic similarity cache for LLM responses.

Provides three-tier lookup: exact hash match (Tier 1), vector similarity match (Tier 2), and cache miss (caller invokes LLM).

Placement note: Lives in ai/llm.py (not infra/cache/) because it carries AI-domain semantics (model tracking in store(), LLM response caching).

lookup

async def lookup(query: str) -> str | None

source

Look up a query in the cache.

Checks Tier 1 (exact hash) then Tier 2 (vector similarity).

Parameters

Parameter	Type	Description
`query`	str	The user query string.

Returns

Type	Description
str \| None	Cached response string, or ``None`` on cache miss.

store

async def store(
    query: str,
    response: str,
    model: str
) -> None

source

Store a query-response pair in both tiers.

Parameters

Parameter	Type	Description
`query`	str	The user query string.
`response`	str	The LLM response to cache.
`model`	str	The model that produced the response.

invalidate

async def invalidate(query: str) -> bool

source

Invalidate a cached entry by query.

Parameters

Parameter	Type	Description
`query`	str	The user query string to invalidate.

Returns

Type	Description
bool	``True` `if the entry was found and removed,` `False`` otherwise.

`SemanticMemoryProtocol`

source

Protocol for semantic memory (extracted facts and entities).

Semantic memory stores generalized knowledge in the form of facts, entities, and their relationships for long-term retention.

store_fact

async def store_fact(
    subject: str,
    predicate: str,
    object_: str,
    confidence: float
) -> None

source

Store a fact as a subject-predicate-object triple.

Parameters

Parameter	Type	Description
`subject`	str	The subject entity
`predicate`	str	The relationship type
`object_`	str	The object entity
`confidence`	float	Confidence score (0-1) for this fact

query_facts

async def query_facts(subject: str) -> list[dict[str, Any]]

source

Query facts by subject.

Parameters

Parameter	Type	Description
`subject`	str	The subject entity to query

Returns

Type	Description
list[dict[str, Any]]	List of facts with this subject

get_entity_facts

async def get_entity_facts(entity: str) -> list[dict[str, Any]]

source

Get all facts mentioning an entity (subject or object).

Parameters

Parameter	Type	Description
`entity`	str	The entity name

Returns

Type	Description
list[dict[str, Any]]	List of facts involving this entity

update_fact

async def update_fact(
    fact_id: str,
    confidence: float
) -> None

source

Update the confidence score of a fact.

Parameters

Parameter	Type	Description
`fact_id`	str	ID of the fact to update
`confidence`	float	New confidence score (0-1)

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check if the semantic memory backend is reachable and healthy.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for a response.

Returns

Type	Description
HealthCheckResult	Health check result with status and details.

`SerializerProtocol`

source

General-purpose serializer/deserializer with typed round-trip support.

serialize

def serialize(obj: Any) -> bytes

source

Serialize obj to raw bytes.

deserialize

def deserialize(
    data: bytes,
    type_: type[T]
) -> T

source

Deserialize data into an instance of type_.

`StateStoreProtocol`

source

Persistent key-value store for arbitrary application state.

Supports bulk read/write operations on top of the basic get/set/delete primitives. Implementations may use any persistent backend (SQL, Redis, blob storage) that satisfies the interface.

Typical usage

state = await container.resolve(StateStoreProtocol)
await state.set("session:abc123", {"user_id": "u-1"}, ttl=3600)
data = await state.get("session:abc123")

state = await container.resolve(StateStoreProtocol)
await state.set("session:abc123", {"user_id": "u-1"}, ttl=3600)
data = await state.get("session:abc123")

get

async def get(key: str) -> Any | None

source

Get a value by key.

Parameters

Parameter	Type	Description
`key`	str	The key to retrieve.

Returns

Type	Description
Any \| None	The value if found, None otherwise.

set

async def set(
    key: str,
    value: Any,
    ttl: int | None = None
) -> None

source

Set a value with optional TTL.

Parameters

Parameter	Type	Description
`key`	str	The key to set.
`value`	Any	The value to store.
`ttl`	int \| None	Optional time-to-live in seconds.

delete

async def delete(key: str) -> bool

source

Delete a key.

Parameters

Parameter	Type	Description
`key`	str	The key to delete.

Returns

Type	Description
bool	True if deleted, False if not found.

exists

async def exists(key: str) -> bool

source

Check if a key exists.

Parameters

Parameter	Type	Description
`key`	str	The key to check.

Returns

Type	Description
bool	True if exists, False otherwise.

expire

async def expire(
    key: str,
    ttl: int
) -> bool

source

Set expiration on a key.

Parameters

Parameter	Type	Description
`key`	str	The key to expire.
`ttl`	int	Time-to-live in seconds.

Returns

Type	Description
bool	True if timeout was set, False if key doesn't exist.

ttl

async def ttl(key: str) -> int

source

Get remaining TTL for a key.

Parameters

Parameter	Type	Description
`key`	str	The key to check.

Returns

Type	Description
int	TTL in seconds, -1 if no expiry, -2 if key doesn't exist.

get_many

async def get_many(keys: list[str]) -> dict[str, Any]

source

Return a mapping of keys → values for all keys that exist.

Parameters

Parameter	Type	Description
`keys`	list[str]	List of storage keys to fetch.

Returns

Type	Description
dict[str, Any]	Dict containing only keys that were found; absent keys are omitted.

set_many

async def set_many(
    items: dict[str, Any],
    ttl: int | None = None
) -> None

source

Persist multiple key-value pairs in a single operation.

Parameters

Parameter	Type	Description
`items`	dict[str, Any]	Mapping of storage keys to JSON-serializable values.
`ttl`	int \| None	Optional time-to-live in seconds applied to all entries.

`StrategyProtocol`

source

Protocol for agent reasoning strategies.

A strategy implements the reasoning loop that drives an agent’s behavior. Built-in strategies: ReActStrategy (reason → act → observe) and PlanAndExecuteStrategy (plan → execute steps).

execute

async def execute(
    message: str,
    tools: list[ToolProtocol],
    history: list[dict[str, Any]],
    llm: Any,
    **kwargs: Any
) -> Any

source

Execute the reasoning strategy.

Parameters

Parameter	Type	Description
`message`	str	The user's input message.
`tools`	list[ToolProtocol]	Tools available to the agent.
`history`	list[dict[str, Any]]	Conversation history as list of message dicts.
`llm`	Any	LLM client for reasoning. **kwargs: Additional strategy-specific parameters (system_prompt, temperature, tool_registry, etc.)

Returns

Type	Description
Any	Result[AgentResponse, Exception]

`TaskExecutorProtocol`

source

Protocol for task executor implementations.

Executors process tasks from a queue by dispatching them to registered handlers.

Example

class TaskExecutorProtocol:
    async def execute(self, task: Task) -> Any:
        handler = self._handlers.get(task.name)
        if handler:
            return await handler(task.payload)
        raise UnknownTaskError(task.name)

execute

async def execute(task: Any) -> Any

source

Execute a task using the registered handler.

Parameters

Parameter	Type	Description
`task`	Any	Task to execute.

Returns

Type	Description
Any	Task execution result.

Raises

Exception	Description
UnknownTaskError	If no handler registered for task.

register_handler

def register_handler(
    task_name: str,
    handler: Any
) -> None

source

Parameters

Parameter	Type	Description
`task_name`	str	Name of the task type.
`handler`	Any	Async callable to handle the task.

get_handler

def get_handler(task_name: str) -> Any | None

source

Get the handler for a task type.

Parameters

Parameter	Type	Description
`task_name`	str	Name of the task type.

Returns

Type	Description
Any \| None	Handler if registered, None otherwise.

`TaskProviderProtocol`

source

Protocol for task provider implementations.

Task providers integrate task processing with the framework, providing dependency injection, lifecycle management, and health monitoring.

async def register(container: Any) -> None

source

Parameters

Parameter	Type	Description
`container`	Any	DI container to register services in.

boot

async def boot(container: Any | None = None) -> None

source

Start the task provider.

Called by the framework on application startup.

Parameters

Parameter	Type	Description
`container`	Any \| None	Optional DI container.

shutdown

async def shutdown(app: Any) -> None

source

Shutdown the task provider gracefully.

Called by the framework on application shutdown.

Parameters

Parameter	Type	Description
`app`	Any	Application instance.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check task provider health.

Returns

Type	Description
HealthCheckResult	HealthCheckResult with current status and metrics.

register_handler

def register_handler(
    task_name: str,
    handler: Any
) -> None

source

Parameters

Parameter	Type	Description
`task_name`	str	Name of the task type.
`handler`	Any	Async handler function.

register_scheduled_task

def register_scheduled_task(task_func: Any) -> None

source

Parameters

Parameter	Type	Description
`task_func`	Any	Task function decorated with @scheduled.

enqueue_job

async def enqueue_job(job: Any) -> str

source

Enqueue a job for processing.

Parameters

Parameter	Type	Description
`job`	Any	JobProtocol instance to enqueue.

Returns

Type	Description
str	JobProtocol ID.

build_idempotency_manager

def build_idempotency_manager(storage: IdempotencyStoreProtocol) -> IdempotencyManagerProtocol

source

Build an idempotency manager over storage.

Cross-package consumers build the idempotency machinery through this factory instead of importing quadkit-tasks internals directly.

Parameters

Parameter	Type	Description
`storage`	IdempotencyStoreProtocol	Storage backend for idempotency keys.

Returns

Type	Description
IdempotencyManagerProtocol	An IdempotencyManagerProtocol implementation bound to storage.

build_idempotent_task_manager

def build_idempotent_task_manager(
    queue_client: Any,
    idempotency_manager: IdempotencyManagerProtocol
) -> IdempotentTaskManagerProtocol

source

Build an idempotent task manager over queue_client.

Cross-package consumers obtain their duplicate-protected task submitter through this factory instead of importing quadkit-tasks internals directly.

Parameters

Parameter	Type	Description
`queue_client`	Any	Task queue client (``TaskQueueProtocol``) to submit jobs through.
`idempotency_manager`	IdempotencyManagerProtocol	Idempotency manager detecting duplicates.

Returns

Type	Description
IdempotentTaskManagerProtocol	An IdempotentTaskManagerProtocol implementation.

schedule_job

def schedule_job(
    job_template: Any,
    cron_expression: str,
    job_id: str | None = None
) -> str | None

source

Schedule a job with cron expression.

Parameters

Parameter	Type	Description
`job_template`	Any	JobProtocol or JobTemplateProtocol instance.
`cron_expression`	str	Cron expression for scheduling.
`job_id`	str \| None	Optional job ID.

Returns

Type	Description
str \| None	Scheduled job ID if successful, None otherwise.

unschedule_job

def unschedule_job(job_id: str) -> bool

source

Remove a scheduled job.

Parameters

Parameter	Type	Description
`job_id`	str	JobProtocol ID to unschedule.

Returns

Type	Description
bool	True if job was unscheduled, False otherwise.

get_worker_stats

def get_worker_stats() -> dict[str, Any] | None

source

Get worker pool statistics.

Returns

Type	Description
dict[str, Any] \| None	Dictionary with worker statistics or None.

get_scheduled_jobs

def get_scheduled_jobs() -> dict[str, Any] | None

source

Get scheduled jobs information.

Returns

Type	Description
dict[str, Any] \| None	Dictionary with scheduled jobs info or None.

`TaskQueueProtocol`

source

Protocol for task queue implementations.

This defines the contract for background task queues (Redis-based, PostgreSQL-based, in-memory, etc.).

Example

class RedisTaskQueue:
    async def enqueue(self, task: Task) -> str:
        task_id = str(uuid4())
        await self._redis.rpush("tasks", task.to_json())
        return task_id

    async def dequeue(self) -> Task | None:
        data = await self._redis.lpop("tasks")
        return Task.from_json(data) if data else None

enqueue

async def enqueue(task: Any) -> Result[str, TaskQueueError]

source

Add a task to the queue.

Parameters

Parameter	Type	Description
`task`	Any	Task instance to enqueue.

Returns

Type	Description
Result[str, TaskQueueError]	Ok(task_id) on success; Err(TaskQueueError) if the queue reports a recoverable failure (e.g. full queue, invalid payload). Only infrastructure failures (lost connection) are raised as exceptions.

dequeue

async def dequeue() -> Any | None

source

Remove and return the next task.

Returns

Type	Description
Any \| None	Next Task or None if queue is empty.

get_task_count

async def get_task_count() -> int

source

Get the number of tasks in the queue.

Returns

Type	Description
int	Number of pending tasks.

clear

async def clear() -> None

source

Clear all tasks from the queue.

get_job

async def get_job(job_id: str) -> Any | None

source

Get a job by ID.

Parameters

Parameter	Type	Description
`job_id`	str	Unique job identifier.

Returns

Type	Description
Any \| None	Job data or None if not found.

delete_job

async def delete_job(job_id: str) -> bool

source

Delete a job from the queue.

Parameters

Parameter	Type	Description
`job_id`	str	Unique job identifier.

Returns

Type	Description
bool	True if the job was deleted.

count

async def count(
    queue_name: str,
    status: Any | None = None
) -> int

source

Count queued jobs.

Parameters

Parameter	Type	Description
`queue_name`	str	Queue to inspect.
`status`	Any \| None	Optional status filter.

Returns

Type	Description
int	Number of matching jobs.

ack

async def ack(task_id: str) -> None

source

Acknowledge successful processing of a dequeued task.

Signals that the task has been processed successfully and can be permanently discarded from any in-flight tracking.

Parameters

Parameter	Type	Description
`task_id`	str	ID of the task to acknowledge.

nack

async def nack(
    task_id: str,
    requeue: bool = True
) -> None

source

Negative-acknowledge a dequeued task.

Signals that the task could not be processed. The task is optionally requeued for another processing attempt.

Parameters

Parameter	Type	Description
`task_id`	str	ID of the task to negative-acknowledge.
`requeue`	bool	If True, return the task to the queue for retry. If False, discard the task permanently.

async def close() -> None

source

Close the queue connection and cleanup resources.

`TenantConfigProviderProtocol`

source

Per-tenant configuration key-value store.

Provides a low-level get/set interface. The higher-level TenantConfigService adds default fallback and event emission on top of this protocol.

get_config

async def get_config(
    tenant_id: str,
    key: str
) -> Any | None

source

Retrieve a single configuration value for a tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose configuration is queried.
`key`	str	The configuration key.

Returns

Type	Description
Any \| None	The stored value, or ``None`` if the key is not set for this tenant.

get_all_config

async def get_all_config(tenant_id: str) -> dict[str, Any]

source

Retrieve all configuration entries for a tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose configuration is retrieved.

Returns

Type	Description
dict[str, Any]	A dictionary of all key-value pairs for the tenant. Returns an empty dict if no overrides are set.

set_config

async def set_config(
    tenant_id: str,
    key: str,
    value: Any
) -> None

source

Set a configuration value for a tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose configuration is updated.
`key`	str	The configuration key.
`value`	Any	The new value (must be JSON-serialisable).

`TenantIsolationStrategyProtocol`

source

Pluggable data isolation strategy.

Implementations provide the mechanics of isolating tenant data at the database layer (row-level, schema-per-tenant, or database-per-tenant).

Attributes: name: Strategy identifier used by the registry ("row_level", "schema", "database").

apply_isolation

async def apply_isolation(
    tenant_id: str,
    context: dict[str, Any]
) -> None

source

Apply tenant isolation to the given execution context.

For row-level isolation this is a no-op; for schema isolation this sets the search_path in context.

Parameters

Parameter	Type	Description
`tenant_id`	str	The active tenant.
`context`	dict[str, Any]	Mutable execution context dict to annotate.

remove_isolation

async def remove_isolation(tenant_id: str) -> None

source

Remove any active isolation for the tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant whose isolation context to tear down.

provision_isolation

async def provision_isolation(tenant_id: str) -> Result[None, TenantError]

source

Provision isolation resources for a newly created tenant.

For row-level isolation this is a no-op. For schema isolation this creates the schema.

Parameters

Parameter	Type	Description
`tenant_id`	str	The newly created tenant.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)` `on success,` `Err(TenantError)`` on failure.

deprovision_isolation

async def deprovision_isolation(tenant_id: str) -> Result[None, TenantError]

source

Tear down isolation resources for a deactivated tenant.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant being deactivated.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)` `on success,` `Err(TenantError)`` on failure.

`TenantProviderProtocol`

source

Storage-agnostic tenant CRUD operations.

Applications may supply their own implementation by binding a class to this protocol in the DI container. The default implementations are InMemoryTenantProvider (for testing/dev).

get_tenant

async def get_tenant(tenant_id: str) -> TenantInfo | None

source

Retrieve a tenant by its unique identifier.

Parameters

Parameter	Type	Description
`tenant_id`	str	The unique tenant identifier.

Returns

Type	Description
TenantInfo \| None	The TenantInfo record, or ``None`` if no tenant with that ID exists.

get_tenant_by_slug

async def get_tenant_by_slug(slug: str) -> TenantInfo | None

source

Retrieve a tenant by its URL-safe slug.

Parameters

Parameter	Type	Description
`slug`	str	The tenant slug (e.g. ``acme-corp``).

Returns

Type	Description
TenantInfo \| None	The matching TenantInfo, or ``None`` if not found.

list_tenants

async def list_tenants(
    *,
    active_only: bool = True
) -> list[TenantInfo]

source

List tenants, optionally filtering to active ones only.

Parameters

Parameter	Type	Description
`active_only`	bool	When ``True` `(default), return only tenants with` `status == ACTIVE``.

Returns

Type	Description
list[TenantInfo]	List of matching TenantInfo records.

create_tenant

async def create_tenant(command: CreateTenantCommand) -> Result[TenantInfo, TenantError]

source

Persist a new tenant record.

Parameters

Parameter	Type	Description
`command`	CreateTenantCommand	CreateTenantCommand with the new tenant's attributes.

Returns

Type	Description
Result[TenantInfo, TenantError]	``Ok(TenantInfo)` `on success,` `Err(TenantError)`` on failure.

update_tenant

async def update_tenant(
    tenant_id: str,
    command: UpdateTenantCommand
) -> Result[TenantInfo, TenantError]

source

Update mutable fields on an existing tenant record.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to update.
`command`	UpdateTenantCommand	UpdateTenantCommand with the fields to apply.

Returns

Type	Description
Result[TenantInfo, TenantError]	``Ok(TenantInfo)` `with the updated record, or` `Err(TenantError)``.

deactivate_tenant

async def deactivate_tenant(tenant_id: str) -> Result[None, TenantError]

source

Mark a tenant as inactive.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to deactivate.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)` `on success,` `Err(TenantError)`` on failure.

activate_tenant

async def activate_tenant(tenant_id: str) -> Result[None, TenantError]

source

Mark a tenant as active.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to activate.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)` `on success,` `Err(TenantError)`` on failure.

suspend_tenant

async def suspend_tenant(
    tenant_id: str,
    reason: str | None = None
) -> Result[None, TenantError]

source

Mark a tenant as suspended.

Parameters

Parameter	Type	Description
`tenant_id`	str	Identifier of the tenant to suspend.
`reason`	str \| None	Optional human-readable reason for the suspension.

Returns

Type	Description
Result[None, TenantError]	``Ok(None)` `on success,` `Err(TenantError)`` on failure.

`TenantResolverProtocol`

source

Resolves tenant identity from request context.

Implementations are tried in priority order by CompositeResolver. Lower priority value = tried first = higher trust level.

Attributes: name: Unique resolver name (e.g. "header", "jwt_claim"). priority: Ordering weight; lower = tried first.

resolve

async def resolve(context: TenantResolutionContext) -> str | None

source

Attempt to resolve the tenant identifier from the given context.

Parameters

Parameter	Type	Description
`context`	TenantResolutionContext	Immutable snapshot of request data.

Returns

Type	Description
str \| None	The resolved ``tenant_id` `string, or` `None`` if this resolver cannot determine the tenant from the provided context.

`TokenCounterProtocol`

source

Model-aware token counter.

Implementations use the exact tokenizer for the target model (tiktoken for OpenAI, AutoTokenizer for HuggingFace, mistral-common for Mistral). A character-estimate fallback is always available.

Note: Methods are synchronous because tokenization is CPU-bound with no I/O.

count

def count(text: str) -> int

source

Count tokens in a text string.

count_messages

def count_messages(messages: Sequence[ChatMessageProtocol]) -> int

source

Count tokens in a list of chat messages, including message overhead.

model

property model() -> str

source

The model this counter is calibrated for.

`TokenManagerProtocol`

source

Protocol for token management implementations.

Defines the contract for JWT token creation, verification, and refresh operations.

Example

class JWTTokenManager:
    def create_token(self, user: AuthenticatedUserProtocol) -> AuthToken:
        payload = {"sub": user.user_id, "roles": user.roles}
        access = jwt.encode(payload, self._secret, algorithm="HS256")
        return AuthToken(access_token=access, ...)

create_token

def create_token(user: AuthenticatedUserProtocol) -> Any

source

Create an authentication token for a user.

Parameters

Parameter	Type	Description
`user`	AuthenticatedUserProtocol	The authenticated user.

Returns

Type	Description
Any	AuthToken containing access and refresh tokens.

verify_token

def verify_token(token: str) -> Result[VerifiedToken, TokenError]

source

Verify and decode a token.

Returns a Result rather than raising domain exceptions so that the caller can pattern-match on success vs. failure explicitly. Infrastructure errors (cache unavailable, network failure) are still raised as exceptions and must not be wrapped in Result.

Parameters

Parameter	Type	Description
`token`	str	The JWT token string.

Returns

Type	Description
Result[VerifiedToken, TokenError]	``Ok(VerifiedToken)` `if the token is valid and not revoked, or` `Err(TokenError)`` for any expected domain failure.

refresh_token

def refresh_token(refresh_token: str) -> Result[Any, TokenError]

source

Refresh an access token using a refresh token.

Parameters

Parameter	Type	Description
`refresh_token`	str	The refresh token string.

Returns

Type	Description
Result[Any, TokenError]	``Ok(AuthToken)` `if the refresh token is valid, or` `Err(TokenError)`` for expected domain failures.

`ToolProtocol`

source

Protocol for agent tools.

Tools are the atomic capabilities an agent can invoke during reasoning. Each tool has a name, description, JSON parameter schema (for LLM function calling), and an async execute method.

Satisfied by:

@tool decorated functions (FunctionTool)
Classes extending Tool base class

name

property name() -> str

source

Unique tool identifier.

description

property description() -> str

source

Human-readable description for the LLM.

parameters_schema

property parameters_schema() -> dict[str, Any]

source

JSON Schema describing the tool’s parameters.

Auto-generated from type hints by the @tool decorator, or manually defined for class-based tools.

Format follows OpenAI function calling schema

{
    "type": "object",
    "properties": {
        "order_id": {"type": "string"},
        "reason": {"type": "string"}
    },
    "required": ["order_id"]
}

{
    "type": "object",
    "properties": {
        "order_id": {"type": "string"},
        "reason": {"type": "string"}
    },
    "required": ["order_id"]
}

execute

async def execute(**kwargs: Any) -> Any

source

Execute the tool with the given arguments.

Returns the tool’s result. Errors should be raised as exceptions — the executor wraps them in Result.

`ToolRegistryProtocol`

source

Protocol for tool registries.

A registry stores tools by name and provides execution with error handling. When module visibility is enabled, tool access is checked against the compiled module graph.

def register(
    tool: ToolProtocol,
    module_class: type | None = None
) -> None

source

get

def get(name: str) -> ToolProtocol | None

source

Get a tool by name.

list_tools

def list_tools() -> list[ToolProtocol]

source

List all registered tools.

list_visible_tools

def list_visible_tools() -> list[ToolProtocol]

source

List tools visible to the current caller module.

list_visible_tool_names

def list_visible_tool_names() -> list[str]

source

List names of tools visible to the current caller module.

set_module_graph

def set_module_graph(graph: CompiledModuleGraphProtocol | None) -> None

source

Set the compiled module graph for visibility enforcement.

set_caller_module

def set_caller_module(module_class: type | None) -> None

source

Set the calling module for visibility checks.

execute

async def execute(
    name: str,
    **kwargs: Any
) -> Result[Any, ToolError]

source

Execute a tool by name.

Returns Result[Any, ToolError].

`TransactionalProtocol`

source

Protocol for transactional operations.

begin_transaction

async def begin_transaction() -> Any

source

commit_transaction

async def commit_transaction(transaction: Any) -> None

source

rollback_transaction

async def rollback_transaction(transaction: Any) -> None

source

`UnitOfWorkProtocol`

source

Protocol for Unit of Work pattern.

Coordinates multiple repository operations within a single transaction boundary.

Example

async with uow:
    user = await uow.users.get(user_id)
    user.email = new_email
    await uow.users.save(user)
    await uow.audit_log.save(AuditEntry(...))
    await uow.commit()

commit

async def commit() -> None

source

Commit all changes in this unit of work.

rollback

async def rollback() -> None

source

Rollback all changes in this unit of work.

on_commit

def on_commit(callback: Callable[[], Any]) -> None

source

The callback may be a plain function or a coroutine function (or a sync function returning an awaitable); awaitable results are awaited.

on_rollback

def on_rollback(callback: Callable[[], Any]) -> None

source

The callback may be a plain function or a coroutine function (or a sync function returning an awaitable); awaitable results are awaited.

register_new

def register_new(entity: Any) -> None

source

register_event

def register_event(event: DomainEvent) -> None

source

Handlers or repositories may call this method when they detect that a domain event needs to be published once the transaction successfully commits. The unit of work is responsible for retaining the events until they are gathered by the application and dispatched via an event bus.

collect_events

def collect_events() -> list[DomainEvent]

source

Return and clear all domain events registered with this unit of work.

Calling this method yields the list of events that have been registered either explicitly via register_event or implicitly by the UoW when entities with collect_events semantics are registered. The returned list is cleared from the unit of work; subsequent calls will return an empty list until more events are registered.

register_dirty

def register_dirty(entity: Any) -> None

source

register_deleted

def register_deleted(entity: Any) -> None

source

`UserProtocol`

source

Protocol for user entity.

Defines the minimal contract for a user in the system.

user_id

property user_id() -> str

source

Unique user identifier.

property email() -> str

source

User email address.

is_active

property is_active() -> bool

source

Whether user account is active.

`ValidatableProtocol`

source

Protocol for data validation.

validate_record

async def validate_record(
    record: dict[str, Any],
    operation: str = 'create'
) -> dict[str, Any]

source

get_validation_rules

def get_validation_rules() -> dict[str, Any]

source

`VectorStoreProtocol`

source

Top-level vector store lifecycle and collection management.

Implementations manage connections, collection CRUD, and delegate vector operations to VectorCollectionProtocol instances.

connect

async def connect() -> None

source

Establish connection to the vector store.

disconnect

async def disconnect() -> None

source

Close all connections and release resources.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check store connectivity and readiness.

list_collections

async def list_collections() -> list[CollectionInfo]

source

List all collections with metadata.

create_collection

async def create_collection(config: CollectionConfig) -> None

source

Create a new vector collection.

Raises

Exception	Description
CollectionAlreadyExistsError	If collection name is taken.
VectorConfigError	If config is invalid for this backend.

delete_collection

async def delete_collection(name: str) -> None

source

Delete a collection and all its vectors.

Raises

Exception	Description
CollectionNotFoundError	If collection does not exist.

collection_exists

async def collection_exists(name: str) -> bool

source

Check whether a collection exists.

get_collection

async def get_collection(name: str) -> VectorCollectionProtocol

source

Get a handle to an existing collection.

Raises

Exception	Description
CollectionNotFoundError	If collection does not exist.

add_texts

async def add_texts(
    texts: list[str],
    embeddings: list[list[float]] | None = None,
    metadatas: list[dict[str, Any]] | None = None,
    collection_name: str | None = None
) -> UpsertResult

source

Add texts to a collection.

Convenience method that handles collection routing. If embeddings are not provided, the implementation may compute them internally or raise an error.

`WebhookSignatureVerifierProtocol`

source

Verifies the authenticity of inbound webhook payloads.

Implementations must provide both a verification method and a signature computation method so callers can independently validate or generate signatures without exposing the underlying algorithm.

The canonical implementation is HMAC-SHA256, but any MAC or asymmetric scheme that fulfils this protocol is acceptable.

Typical usage

verifier = HMACWebhookVerifier()
if not verifier.verify(payload=body, signature=sig_header, secret=secret):
    raise PermissionError("Invalid webhook signature")

verifier = HMACWebhookVerifier()
if not verifier.verify(payload=body, signature=sig_header, secret=secret):
    raise PermissionError("Invalid webhook signature")

verify

def verify(
    payload: bytes,
    signature: str,
    secret: str
) -> bool

source

Return True when the signature matches the payload.

Implementations must use a constant-time comparison to prevent timing side-channel attacks.

Parameters

Parameter	Type	Description
`payload`	bytes	Raw request body bytes.
`signature`	str	Signature string as received in the request header (may include algorithm prefix such as ``"sha256=..."``).
`secret`	str	Shared secret used to compute the expected signature.

Returns

Type	Description
bool	``True` `if the signature is valid,` `False`` otherwise.

compute_signature

def compute_signature(
    payload: bytes,
    secret: str
) -> str

source

Compute the expected signature for payload using secret.

Parameters

Parameter	Type	Description
`payload`	bytes	Raw request body bytes.
`secret`	str	Shared secret.

Returns

Type	Description
str	Hex-encoded signature string in the same format that verify expects as input.

`WorkflowNodeProtocol`

source

Protocol for individual workflow step nodes.

A workflow node represents a single unit of work in a directed graph. Nodes receive the current shared state, perform their work, and return a state update dict that is merged into the global state.

name

property name() -> str

source

Unique node identifier within the workflow.

execute

async def execute(state: dict[str, Any]) -> dict[str, Any]

source

Execute this node with the current workflow state.

Parameters

Parameter	Type	Description
`state`	dict[str, Any]	Current shared workflow state.

Returns

Type	Description
dict[str, Any]	Dict of state updates to merge into the shared state.

`WorkingMemoryProtocol`

source

Protocol for working memory that assembles context for the current request.

Working memory is the in-context window that gets passed to the LLM, assembled from episodic and semantic memory based on token budgets.

assemble

async def assemble(
    query: str,
    token_budget: int,
    *,
    owner_id: str,
    session_id: str | None = None
) -> list[MemoryEntry]

source

Assemble context window from available memory tiers.

Retrieves relevant entries from episodic and semantic memory and fits them within the token budget.

Parameters

Parameter	Type	Description
`query`	str	The current query/prompt
`token_budget`	int	Maximum tokens available for context
`owner_id`	str	Owner scope; only this owner's memory is assembled
`session_id`	str \| None	Optional session scope for session-specific memory

Returns

Type	Description
list[MemoryEntry]	Ordered list of memory entries for context

add

async def add(entry: MemoryEntry) -> None

source

Add a new entry to the working memory stream.

Parameters

Parameter	Type	Description
`entry`	MemoryEntry	The entry to add

get_context_entries

async def get_context_entries() -> list[MemoryEntry]

source

Get current assembled context entries.

Returns

Type	Description
list[MemoryEntry]	List of entries currently in context window

flush

async def flush() -> None

source

Clear the current context assembly.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check if the working memory backend is reachable and healthy.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for a response.

Returns

Type	Description
HealthCheckResult	Health check result with status and details.

Classes

`AgentResponse`

source

Complete response from an agent execution.

Contains the final message, the full reasoning trace (steps), all tool calls made, token usage, cost, and timing metadata.

Note: ToolCall and ReasoningStep are defined in quadkit-ai-agents and imported here for use in this type’s field annotations.

tool_call_count

property tool_call_count() -> int

source

Number of tool calls made.

step_count

property step_count() -> int

source

Number of reasoning steps taken.

successful_tool_calls

property successful_tool_calls() -> list[Any]

source

Tool calls that completed without error.

failed_tool_calls

property failed_tool_calls() -> list[Any]

source

Tool calls that failed.

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to a JSON-compatible dict.

`AggregateHealthResult`

source

Composite health result that collects checks from multiple components.

The aggregate status follows a worst-case rule:

Any UNHEALTHY check → overall UNHEALTHY
Any DEGRADED check (no UNHEALTHY) → overall DEGRADED
All HEALTHY → overall HEALTHY
No checks registered → UNKNOWN

status

property status() -> HealthStatus

source

Overall status, computed from component checks.

is_healthy

def is_healthy() -> bool

source

Return True only if all components are healthy.

to_dict

def to_dict() -> dict[str, Any]

source

Convert to a dictionary suitable for a /health HTTP response.

`AuditEntry`

source

Immutable record of a security-relevant or compliance-relevant operation.

Canonical audit type used by all packages. Supersedes the former AuditEvent (simpler, no old/new values) and the AuditEntry from contracts/observability/audit.

Attributes: action: Dot-notation action identifier (e.g. "user.update"). actor_id: ID of the user or service that performed the action. resource_type: Kind of affected resource (e.g. "User"). resource_id: ID of the affected resource. outcome: "success" or "failure". severity: Severity level (defaults to MEDIUM). occurred_at: UTC datetime of the event (defaults to now). metadata: Arbitrary additional context. old_values: Field values before the action (optional). new_values: Field values after the action (optional). source: Originating subsystem (e.g. "sql", "admin", "ai"). tenant_id: Multi-tenant scoping (optional). correlation_id: Request correlation ID for tracing (optional). causation_id: Causing event ID for event-driven traces (optional). command_payload_hash: SHA-256 hash of the command payload (optional). payload_size_bytes: Size of the payload in bytes (optional). checksum: HMAC-SHA256 checksum of the persisted row, populated by stores on read-back (None for pre-checksum entries).

`ChatMessage`

source

Concrete chat message data class shared across AI packages.

Satisfies ChatMessageProtocol. Suitable for construction in any package that needs to build messages without importing from quadkit-ai-llm.

`CircuitBreakerConfig`

source

Configuration for circuit breaker.

Attributes: failure_threshold: Number of failures before opening the circuit. recovery_timeout: Seconds in open state before attempting half-open. expected_exception: Exception types that count as failures. failure_if: Predicate narrowing expected_exception to the instances that mean an outage; the rest are decisions and leave the circuit’s health alone. success_threshold: Consecutive successes required to close from half-open. timeout: Per-call timeout in seconds. name: Human-readable identifier for this breaker instance. sliding_window_seconds: Window for computing the failure rate. failure_rate_threshold: Fraction of calls that must fail to trip. backend: State store backend for distributed circuit breaker coordination. "memory" (default) — in-process state, no coordination. "redis" — uses a CacheBackendProtocol resolved from the DI container. "consul" — uses Consul KV for cross-datacenter coordination.

`CursorPage`

source

Cursor-based pagination for large datasets.

Attributes: items: The entities in this page. next_cursor: Opaque cursor for the next page, or None if no more. prev_cursor: Opaque cursor for the previous page, or None. has_more: Whether there are more results after this page. has_previous: Whether there are results before this page. total_count: Optional total count (may be None for performance).

to_dict

def to_dict() -> dict[str, Any]

source

Serialize to API-friendly dictionary.

`DeleteResult`

source

Result of a delete operation.

`DomainEvent`

source

Base class for domain events.

__init__

def __init__(**kwargs: Any) -> None

source

Initialize the event setting base fields and any extra subclass fields.

Declared dataclass fields receive their default values when not supplied. Extra keyword arguments (e.g. fields declared on plain subclasses without @dataclass) are set directly on the instance via object.__setattr__, which bypasses the frozen=True guard safely during construction.

to_dict

def to_dict() -> dict[str, Any]

source

Return a JSON-serializable dictionary representation.

from_dict

def from_dict(
    cls,
    data: dict[str, Any]
) -> DomainEvent

source

Rebuild an event from to_dict output.

The inverse of to_dict for the base fields it stringified: a UUID that travelled as text comes back as a UUID, a timestamp as a datetime. Anything else is handed to the constructor unchanged, so a round trip through an outbox table or a JSON log yields an event that compares field-for-field with the one that was stored.

for_aggregate

def for_aggregate(
    aggregate_id: UUID,
    aggregate_type: str
) -> DomainEvent

source

`Duration`

source

Immutable duration value object.

The class stores total seconds internally and provides lightweight parsing, formatting, and arithmetic helpers for configuration and API boundaries.

seconds

def seconds(
    cls,
    value: float
) -> Duration

source

Create a duration from seconds.

minutes

def minutes(
    cls,
    value: float
) -> Duration

source

Create a duration from minutes.

hours

def hours(
    cls,
    value: float
) -> Duration

source

Create a duration from hours.

days

def days(
    cls,
    value: float
) -> Duration

source

Create a duration from days.

zero

def zero(cls) -> Duration

source

Create a zero duration.

parse

def parse(
    cls,
    value: str
) -> Duration

source

Parse a human-readable duration string.

Supported units are seconds (s), minutes (m), hours (h), and days (d). Chained forms like 1h30m are supported.

seconds_value

property seconds_value() -> float

source

Backward-compatible alias for the total seconds.

to_timedelta

def to_timedelta() -> timedelta

source

Convert the duration to datetime.timedelta.

`ErrorDetail`

source

Single error detail in an HTTP error response.

Attributes: code: Machine-readable error code (e.g., “INVALID_INPUT”). message: Human-readable error message. field: Optional field name that caused the error (for form validation).

`ErrorResponseDTO`

source

Standard HTTP error response envelope.

Provides a consistent structure for error responses across all endpoints. Should be used with appropriate HTTP status codes (4xx, 5xx).

Attributes: error: High-level error category. message: Human-readable summary of what went wrong. details: List of specific error details (empty for simple errors). request_id: Optional request identifier for tracing and support.

`FlagEvaluation`

source

The result of evaluating a single feature flag.

Attributes: key: The flag identifier. value: The evaluated value (type depends on FlagType). flag_type: The type of flag evaluated. reason: Machine-readable reason code (e.g. “DEFAULT”, “TARGETING”). variant: Variant key for VARIANT-type flags, None otherwise. metadata: Arbitrary evaluation metadata from the provider.

`FlagType`

source

The evaluation strategy of a feature flag.

`GeneratorDefinition`

source

Describes a single executable scaffolding generator contributed to the CLI.

Generators are discovered via CliContributorProtocol.get_generators() and assembled into the unified gen command by CommandAssembler. generator_path is required so every definition can be executed once it has been discovered and registered.

supports

def supports(capability: str) -> bool

source

Whether this generator declares capability.

Discovery must be declarative: bridges should not probe a generator by calling it with structured input and interpreting a TypeError after files may already have been staged.

make

def make(
    cls,
    name: str,
    *,
    description: str,
    generator_path: str,
    output_dir: str = 'src',
    contributor: str = '',
    category: str = 'general',
    options: tuple[GeneratorOption, Ellipsis] = (),
    title: str | None = None,
    capabilities: Iterable[str] = ()
) -> GeneratorDefinition

source

Build a definition with conventional title derivation.

The title defaults to "Generate <Title Case Of Name>" — e.g. auth_guard becomes Generate Auth Guard. Pass title to override for names where the derivation reads poorly.

Parameters

Parameter	Type	Description
`name`	str	Snake_case generator identifier (also the CLI command name).
`description`	str	One-line description shown in help text.
`generator_path`	str	Import path ``"pkg.module:ClassName"`` of the generator implementation.
`output_dir`	str	Default output directory relative to the project root.
`contributor`	str	``contributor_id`` of the owning contributor.
`category`	str	Grouping bucket used by interactive listings.
`options`	tuple[GeneratorOption, Ellipsis]	Typed CLI options accepted by the generator.
`title`	str \| None	Explicit human-readable title overriding the derivation.
`capabilities`	Iterable[str]	Declarative generator capabilities bridges may require before invoking the generator.

Returns

Type	Description
GeneratorDefinition	A frozen GeneratorDefinition.

`GeneratorOption`

source

A single CLI option/flag accepted by a generator command.

Used for auto-generated help text and input validation.

`HealthCheckResult`

source

Result of a health check.

to_dict

def to_dict() -> dict[str, Any]

source

Convert to dictionary format.

is_healthy

def is_healthy() -> bool

source

Check if status is healthy.

is_degraded

def is_degraded() -> bool

source

Check if status is degraded.

`HealthStatus`

source

Unified health status.

`IdStrategy`

source

Available ID generation strategies.

`InsertResult`

source

Result of an insert operation.

`JobStatus`

source

JobProtocol execution status.

`Lifecycle`

source

Provider lifecycle stages.

`LockInfo`

source

Immutable snapshot of a distributed lock's state.

Attributes: resource: The resource being locked. holder: Identifier of the current holder. acquired_at: Unix timestamp when the lock was acquired. expires_at: Unix timestamp when the lock expires (TTL). metadata: Additional implementation-specific data.

is_expired

property is_expired() -> bool

source

Check if the lock has expired.

ttl_remaining

property ttl_remaining() -> float | None

source

Get remaining TTL in seconds.

`MigrationRecord`

source

Record of a database migration.

`PaginatedResponseDTO`

source

Standard paginated HTTP response envelope.

Encapsulates a page of results with metadata for navigation. Supports both offset-based and cursor-based pagination.

Attributes: items: The entities in this page. total: Total count of items across all pages. page: Current page number (1-indexed, 0-indexed for cursor). page_size: Maximum items returned per page. has_next: Whether there are more pages after this one. has_prev: Whether there are pages before this one. next_cursor: Opaque cursor for the next page (for cursor-based pagination).

`ProjectionTier`

source

Alert routing tier for a projection or SLO.

Attributes: P0_PAGE: Page on-call immediately (PagerDuty, etc.). P1_BUSINESS_HOURS: Route to Slack during business hours; queue outside business hours. P2_DIGEST: Accumulate in a buffer and flush as a weekly digest.

`ProviderPriority`

source

Provider initialization priority.

Providers are booted in ascending order of priority. Lower values run first; the typical ordering is:

CRITICAL (0) – absolutely foundational services that others depend on (e.g. configuration, diagnostics).
INFRASTRUCTURE (10) – low-level plumbing such as database connections, messaging clients, and cache providers.
SECURITY (20) – authentication/authorization infrastructure.
NORMAL (30) – everyday domain services with no special ordering.
APPLICATION (40) – application-level tools (CLI, admin utilities) that depend on domain services but are not themselves domain logic.
DOMAIN (50) – business-logic providers that may depend on earlier layers.
PRESENTATION (80) – web/API layers and other entry points.
COMMS (90) – outbound communication providers (email, SMS, webhooks) often run late to avoid interfering with core initialization.
LOW (100) – lowest-priority, optional providers that can boot last.

`QueryLogEntry`

source

Represents a database query log entry.

This is a concrete dataclass rather than a Protocol so that callers can instantiate it directly.

`QueryResult`

source

Result of a database query.

Implements the iterator and sequence protocols so that code expecting a plain list[dict] from a query result continues to work after the return type is normalised to QueryResult.

Example

result = await provider.execute_query("SELECT * FROM users")
for row in result:          # iterate directly
    print(row["email"])
if not result:              # bool coercion
    raise LookupError("no rows")
first = result[0]           # index access
rows = list(result)         # convert to plain list

result = await provider.execute_query("SELECT * FROM users")
for row in result:          # iterate directly
    print(row["email"])
if not result:              # bool coercion
    raise LookupError("no rows")
first = result[0]           # index access
rows = list(result)         # convert to plain list

`Result`

source

Base Result type. Not abstract — Ok and Err are the only variants.

is_ok

def is_ok() -> bool

source

is_err

def is_err() -> bool

source

unwrap

def unwrap() -> T

source

unwrap_err

def unwrap_err() -> E

source

unwrap_or

def unwrap_or(default: T) -> T

source

unwrap_or_else

def unwrap_or_else(op: Callable[[E], T]) -> T

source

map_sync

def map_sync(op: Callable[[T], U]) -> Result[U, E]

source

map_err

def map_err(op: Callable[[E], F]) -> Result[T, F]

source

and_then_sync

def and_then_sync(op: Callable[[T], Result[U, E]]) -> Result[U, E]

source

or_else_sync

def or_else_sync(op: Callable[[E], Result[T, F]]) -> Result[T, F]

source

expect

def expect(message: str) -> T

source

match

def match(
    ok: Callable[[T], U],
    err: Callable[[E], U]
) -> U

source

map

async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

source

and_then

async def and_then(op: Callable[[T], Awaitable[Result[U, E]]]) -> Result[U, E]

source

or_else

async def or_else(op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]

source

flatten

def flatten() -> Result[Any, E]

source

filter

def filter(
    predicate: Callable[[T], bool],
    error: E
) -> Result[T, E]

source

ok_or

def ok_or(default: U) -> T | U

source

from_exception

def from_exception(
    cls,
    exc: Exception,
    ok_type: type[T] = type(None)
) -> Result[T, Exception]

source

Wrap a caught exception into an Err result.

to_optional

def to_optional() -> T | None

source

inspect

def inspect(op: Callable[[T], None]) -> Result[T, E]

source

inspect_err

def inspect_err(op: Callable[[E], None]) -> Result[T, E]

source

`RetryConfig`

source

Configuration for retry policy.

`ServiceScope`

source

Dependency injection scopes.

`SpecificationProtocol`

source

Protocol for the SpecificationProtocol pattern.

This class was previously defined in quadkit.contracts.specification before the 2026 reorg. It has been moved here alongside the other DDD primitives. A migration mapping ensures imports are rewritten.

is_satisfied_by

def is_satisfied_by(candidate: T) -> bool

source

`TenantInfo`

source

Core tenant identity record.

The config field here is the tenant’s static provisioning config (plan limits, feature flags set at creation time). Runtime per-tenant overrides are managed separately by TenantConfigProviderProtocol / TenantConfigService. The two are independent — TenantInfo.config is a snapshot stored alongside the tenant record; TenantConfigService is a live, cached, key-value overlay.

Attributes: tenant_id: Unique identifier for the tenant. slug: URL-safe, human-readable identifier (e.g. acme-corp). name: Display name of the tenant. status: Current lifecycle status. plan: Subscription plan name (optional). config: Static provisioning configuration snapshot. metadata: Arbitrary application-defined metadata. created_at: Timestamp when the tenant was created.

`TenantResolutionContext`

source

Immutable snapshot of request data available for tenant resolution.

Passed to every TenantResolverProtocol during the resolution chain so resolvers can inspect request metadata without depending on any web framework type.

Attributes: headers: HTTP request headers (lowercased keys). host: The Host header value (e.g. acme.app.com). path: The request path (e.g. /api/tenants/acme/users). claims: Decoded JWT claims from the current authentication context.

`TenantStatus`

source

Lifecycle status of a tenant.

Attributes: ACTIVE: Tenant is fully operational and accepts requests. INACTIVE: Tenant has been deactivated and no longer accepts requests. SUSPENDED: Tenant has been suspended (e.g. for non-payment) and is temporarily blocked from accessing the system. PROVISIONING: Tenant is being created; isolation resources are being set up.

`TokenBudget`

source

Immutable token budget that pipeline stages consult.

Every field is calculated once at the start of a request. Stages read remaining capacity via properties — they never mutate the budget. Builder methods return new instances (immutable value semantics).

total_used

property total_used() -> int

source

Total tokens consumed by all components.

remaining

property remaining() -> int

source

Tokens available for additional content.

remaining_for_history

property remaining_for_history() -> int

source

Tokens available specifically for chat history.

over_budget

property over_budget() -> bool

source

True if total consumption exceeds the model context limit.

with_rag_tokens

def with_rag_tokens(rag_tokens: int) -> TokenBudget

source

Return a new budget with updated RAG token count.

with_history_tokens

def with_history_tokens(history_tokens: int) -> TokenBudget

source

Return a new budget with updated history token count.

`UpdateResult`

source

Result of an update operation.

Exceptions

`AgentError`

source

Base exception for all agent errors.

__init__

def __init__(
    message: str = 'Agent error',
    **kwargs: Any
) -> None

source

`DuplicateHandlerError`

source

Duplicate handler registration.

Raised when attempting to register a handler for a type that already has one.

__init__

def __init__(
    message: str = 'Duplicate handler',
    message_type: str | None = None,
    **kwargs: Any
) -> None

source

`HandlerNotFoundError`

source

Event/command handler not found.

Raised when no handler is registered for a specific event or command type.

__init__

def __init__(
    message: str = 'Handler not found',
    handler_type: str | None = None,
    message_type: str | None = None,
    **kwargs: Any
) -> None

source

`MCPError`

source

Base exception for all MCP errors.

__init__

def __init__(
    message: str = 'MCP error',
    **kwargs: Any
) -> None

source

`MCPInitializationError`

source

MCP server initialization failed.

Raised when the MCP server cannot be initialized, usually due to missing required handlers or configuration errors.

__init__

def __init__(
    message: str = 'MCP initialization failed',
    *,
    reason: str | None = None,
    **kwargs: Any
) -> None

source

`MCPMethodNotFoundError`

source

Unknown MCP method requested by client.

Raised when the client requests an MCP method that the server does not support.

__init__

def __init__(
    message: str = 'MCP method not found',
    *,
    method: str | None = None,
    **kwargs: Any
) -> None

source

`MCPPromptError`

source

Prompt retrieval or list failed.

Raised when a prompt operation fails, such as when a prompt is not found or arguments are invalid.

__init__

def __init__(
    message: str = 'MCP prompt error',
    *,
    prompt_name: str | None = None,
    **kwargs: Any
) -> None

source

`MCPProtocolError`

source

Protocol violation (malformed message, invalid state).

Raised when an MCP message violates the protocol specification, such as missing required fields or invalid JSON-RPC structure.

__init__

def __init__(
    message: str = 'MCP protocol error',
    *,
    details: dict[str, Any] | None = None,
    **kwargs: Any
) -> None

source

`MCPResourceError`

source

Resource read or list failed.

Raised when a resource operation fails, such as when a resource is not found or cannot be read.

__init__

def __init__(
    message: str = 'MCP resource error',
    *,
    uri: str | None = None,
    **kwargs: Any
) -> None

source

`MCPToolCallError`

source

Tool call failed during MCP execution.

Raised when a tool invocation fails, either due to the tool not existing or raising an exception.

__init__

def __init__(
    message: str = 'MCP tool call failed',
    *,
    tool_name: str | None = None,
    **kwargs: Any
) -> None

source

`SecretAccessError`

source

Raised when the caller lacks permission to access or modify a secret.

Attributes: secret_name: Name of the secret that was denied access. operation: The operation that was denied (read, write, delete, etc.).

__init__

def __init__(
    secret_name: str,
    operation: str = 'read',
    **kwargs: Any
) -> None

source

`SecretNotFoundError`

source

Raised when a requested secret does not exist.

__init__

def __init__(
    secret_name: str,
    **kwargs: Any
) -> None

source

`StrategyError`

source

Reasoning strategy failed.

Raised when the agent’s strategy encounters an error during reasoning (LLM failure, invalid response, etc.).

__init__

def __init__(
    message: str = 'Strategy execution failed',
    *,
    strategy_name: str | None = None,
    **kwargs: Any
) -> None

source

`TenantConfigError`

source

Raised when per-tenant configuration access or mutation fails.

__init__

def __init__(
    message: str = 'Tenant config error',
    **kwargs: Any
) -> None

source

Initialise a TenantConfigError.

Parameters

Parameter	Type	Description
`message`	str	Human-readable description of the config error. **kwargs: Additional context forwarded to the base class.

`TenantError`

source

Base class for all tenancy-related domain errors.

__init__

def __init__(
    message: str = 'Tenant error',
    **kwargs: Any
) -> None

source

Initialise a TenantError.

Parameters

Parameter	Type	Description
`message`	str	Human-readable description of the error. **kwargs: Additional context forwarded to the base class.

`TenantInactiveError`

source

Raised when an operation is attempted on an inactive tenant.

__init__

def __init__(
    tenant_id: str = '',
    **kwargs: Any
) -> None

source

Initialise a TenantInactiveError.

Parameters

Parameter	Type	Description
`tenant_id`	str	The inactive tenant identifier. **kwargs: Additional context forwarded to the base class.

`TenantNotFoundError`

source

Raised when a requested tenant does not exist.

__init__

def __init__(
    tenant_id: str = '',
    **kwargs: Any
) -> None

source

Initialise a TenantNotFoundError.

Parameters

Parameter	Type	Description
`tenant_id`	str	The tenant identifier that was not found. **kwargs: Additional context forwarded to the base class.

`TenantProvisioningError`

source

Raised when tenant provisioning (isolation setup) fails.

__init__

def __init__(
    message: str = 'Tenant provisioning failed',
    **kwargs: Any
) -> None

source

Initialise a TenantProvisioningError.

Parameters

Parameter	Type	Description
`message`	str	Human-readable description of the provisioning failure. **kwargs: Additional context forwarded to the base class.

`TenantResolutionError`

source

Raised when tenant resolution fails unexpectedly.

__init__

def __init__(
    message: str = 'Tenant resolution failed',
    **kwargs: Any
) -> None

source

Initialise a TenantResolutionError.

Parameters

Parameter	Type	Description
`message`	str	Human-readable description of the resolution failure. **kwargs: Additional context forwarded to the base class.

`TenantSlugConflictError`

source

Raised when a tenant slug is already in use.

__init__

def __init__(
    slug: str = '',
    **kwargs: Any
) -> None

source

Initialise a TenantSlugConflictError.

Parameters

Parameter	Type	Description
`slug`	str	The conflicting tenant slug. **kwargs: Additional context forwarded to the base class.

`TenantSuspendedError`

source

Raised when an operation is attempted on a suspended tenant.

__init__

def __init__(
    tenant_id: str = '',
    **kwargs: Any
) -> None

source

Initialise a TenantSuspendedError.

Parameters

Parameter	Type	Description
`tenant_id`	str	The suspended tenant identifier. **kwargs: Additional context forwarded to the base class.

`ToolError`

source

Base exception for tool errors.

__init__

def __init__(
    message: str = 'Tool error',
    **kwargs: Any
) -> None

source

`ValidationError`

source

Data validation error.

__init__

def __init__(
    message: str = 'Validation failed',
    errors: list[FieldError] | None = None,
    **kwargs: Any
) -> None

source

errors

property errors() -> list[FieldError]

source

Return the list of field errors.

add_error

def add_error(
    field: str,
    message: str,
    code: str = 'invalid'
) -> ValidationError

source

Add a field error and return self for chaining.