Contracts

zeroth.contracts.registry

Contract registry models.

Think of contracts like shared agreements about what data looks like. This package lets you store, version, and retrieve those agreements so that different parts of the system can communicate with a well-defined data shape.

ContractNotFoundError

Bases: ContractRegistryError

Raised when a requested contract version is missing.

This is raised when a lookup by contract name and version cannot find a matching registry entry.

ContractRegistryError

Bases: Exception

Base error for anything that goes wrong in the contract registry.

Catch this if you want to handle all contract-related errors in one place. More specific errors below inherit from this one.

ContractReference

Bases: BaseModel

A lightweight pointer to a contract by name and optional version.

Use this when you want to say "I need the contract called X" without loading the full contract record. If version is None, the latest version is assumed when the reference is resolved.

parse classmethod

parse(value: str) -> ContractReference

Parse an optional trailing @<positive-version> pin.

ContractRegistry

ContractRegistry(database: AsyncDatabase)

The main registry for storing and retrieving versioned contracts.

Backed by an async database, this class lets you register Pydantic models as named contracts with automatic version numbering, look them up later, and even resolve them back to the original Python class. It also provides helpers for registering governed tools and binding workflow steps to contracts.

Build the immutable legacy surface over the reserved default tenant.

Production tenant-aware paths use :meth:scoped; this constructor is retained as a compatibility wrapper for the published library surface.

scoped classmethod

scoped(
    database: AsyncDatabase,
    scope_context: ScopeContext
    | NullWorkspaceScopeContext
    | TenantWideScopeContext,
) -> ContractRegistry

Build a registry bound to one exact trusted tenant scope.

for_default_compatibility classmethod

for_default_compatibility(
    database: AsyncDatabase,
) -> ContractRegistry

Build the explicitly reserved legacy/default tenant registry.

for_scope

for_scope(
    scope_context: ScopeContext
    | NullWorkspaceScopeContext
    | TenantWideScopeContext,
) -> ContractRegistry

Return a new registry over the same database bound to scope_context.

register async

register(
    model_type: type[ModelT],
    *,
    name: str | None = None,
    metadata: Mapping[str, Any] | None = None,
    version: int | None = None,
) -> ContractVersion

Save a Pydantic model as a new contract version in the database.

If you don't provide a name, the model's class name is used. If you don't provide a version, it automatically picks the next one. Raises ContractVersionExistsError if that exact name+version already exists.

register_schema async

register_schema(
    name: str,
    json_schema: Mapping[str, Any],
    *,
    metadata: Mapping[str, Any] | None = None,
    version: int | None = None,
) -> ContractVersion

Save a schema-only contract (authored as raw JSON Schema, no Python class).

The stored model_path is empty — resolution synthesizes a Pydantic model that validates payloads against the schema exactly (see contracts.schema_model). Versioning matches register: same name re-registered gets the next version.

register_tool async

register_tool(
    tool: RegistrableTool,
    *,
    input_name: str | None = None,
    output_name: str | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> ToolContractBinding

Register the typed contracts backing a governed tool.

bind_step

bind_step(
    step: GovernedStepSpec,
    *,
    input_contract: ContractReference,
    output_contract: ContractReference,
    flow_name: str | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> StepContractBinding

Bind a governed step spec to registered contracts.

get async

get(
    name: str | ContractReference,
    version: int | None = None,
) -> ContractVersion

Look up a contract by name and optional version.

If no version is given, returns the latest version. Raises ContractNotFoundError if nothing matches.

resolve async

resolve(reference: ContractReference) -> ContractVersion

Look up a contract from a ContractReference. Shorthand for get().

resolve_model_type async

resolve_model_type(
    reference: ContractReference,
) -> type[BaseModel]

Get the actual Python class for a contract reference.

Looks up the contract, then imports and returns the original Pydantic model class. Results are cached so repeated calls are fast. Raises ContractTypeResolutionError if the class can't be imported.

list_versions async

list_versions(name: str) -> list[ContractVersion]

Return all registered versions of a contract, oldest first.

list_names async

list_names() -> list[str]

Return the names of all contracts in the registry, sorted alphabetically.

latest_version async

latest_version(name: str) -> int

Return the highest version number for a contract, or 0 if it doesn't exist yet.

delete async

delete(name: str, version: int | None = None) -> None

Remove a contract from the registry.

If version is given, only that specific version is deleted. If version is None, all versions of the named contract are deleted.

ContractVersion

Bases: BaseModel

The full record for a single version of a registered contract.

This is what you get back when you register or look up a contract. It contains the contract's name, version number, where to find the Python class, the JSON schema, any extra metadata, and when it was created.

StepContractBinding

Bases: BaseModel

Describes how a workflow step connects to its input and output contracts.

Similar to ToolContractBinding but for workflow steps. It records which contracts a step expects as input and output, along with which tool or agent (if any) the step uses.

ToolContractBinding

Bases: BaseModel

Describes how a governed tool connects to its input and output contracts.

When a tool is registered, this binding captures all the important details: which contracts define its input/output data shapes, how the tool should be executed, what capabilities it needs, and various behavioral flags like whether it has side effects or needs human approval.

contract_scope_context

contract_scope_context(
    tenant_id: str, workspace_id: str | None
) -> ScopeContext | NullWorkspaceScopeContext

Build the exact storage scope for a trusted contract owner.

validate_artifact_reference

validate_artifact_reference(data: dict[str, Any]) -> bool

Validate that a dict has valid ArtifactReference structure.

Performs structural validation only -- checks that all required fields (store, key, content_type, size) are present with correct types. Does NOT retrieve the actual payload from any store.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dict to validate as an ArtifactReference shape.	required

Returns:

Type	Description
bool	True if the dict has valid ArtifactReference structure, False otherwise.