Contracts

NodeContract

Bases: BaseModel

Schema contract binding a node to its expected input and output types.

Maps a node_id to its input_schema and output_schema Pydantic model classes that state must conform to at each boundary.

Example

class Query(BaseModel):
    text: str

class Result(BaseModel):
    urls: list[str]

contract = NodeContract(node_id="search", input_schema=Query, output_schema=Result)

ContractViolation

Bases: Exception

Raised when node input or output fails schema validation.

Carries the node_id, direction ("input" or "output"), the raw dict that failed, and a list of Pydantic validation errors.

BoundaryValidator

Callable that validates a raw dict against a contract's input or output schema.

Returns the validated Pydantic model on success, raises :class:ContractViolation on failure.

__call__(contract, direction, raw)

Validate raw against the appropriate schema of contract.

Selects the input or output schema based on direction and returns a validated Pydantic model instance. Raises ContractViolation on validation failure.

ContractRegistry

Registry mapping node IDs to their contracts with built-in validation.

Example

registry = ContractRegistry()
registry.register(NodeContract(node_id="search", input_schema=Query, output_schema=Result))

validated_input = registry.validate_input("search", {"text": "python"})
validated_output = registry.validate_output("search", {"urls": ["https://..."]})

register(contract)

Register a contract for a node, replacing any existing contract for that node ID.

get(node_id)

Return the contract for node_id, or raise KeyError if not registered.

validate_input(node_id, raw)

Validate raw against the input schema of the contract for node_id.

validate_output(node_id, raw)

Validate raw against the output schema of the contract for node_id.