prefactor_core.schema_registry module

prefactor_core.schema_registry module

Schema registry for span type definitions.

This module provides a SchemaRegistry that allows registration of span schemas from multiple packages before agent instances are created.

class prefactor_core.schema_registry.SchemaRegistry

Bases: object

Central registry for span schemas - allows pre-registration.

Multiple components can register their schemas independently before agent instance creation. The registry aggregates all into a single format suitable for the API.

The API supports three ways to define span schemas, in increasing order of expressiveness:

• span_schemas: flat map of span name → params JSON schema
• span_result_schemas: flat map of span name → result JSON schema
• span_type_schemas: structured list with params, result, title, description, and template per span type

Use register() for simple payload schemas, register_result() to add a result schema for an existing entry, or register_type() for the full structured form. All three approaches can be mixed; to_agent_schema_version() emits whichever fields are populated.

Example

registry = SchemaRegistry()

Simple params-only schema

registry.register(“langchain:agent”, {“type”: “object”})

Full structured schema with result and display metadata

registry.register_type(

name=”agent:llm”, params_schema={

“type”: “object”, “properties”: {

“model”: {“type”: “string”}, “prompt”: {“type”: “string”},

}, “required”: [“model”, “prompt”],

}, result_schema={

“type”: “object”, “properties”: {“response”: {“type”: “string”}},

}, title=”LLM Call”, description=”A call to a language model”, template=”{{model}}: {{prompt}} → {{response}}”,

)

Convert to API format

version = registry.to_agent_schema_version(“combined-1.0.0”)

get(schema_name: str) → dict[str, Any] | None

Get a params schema by name.

• Parameters: schema_name – The schema identifier to look up
• Returns: The schema dict if found, None otherwise

has_schema(schema_name: str) → bool

Check if a params schema is registered for a span type.

• Parameters: schema_name – The schema identifier to check
• Returns: True if the schema is registered, False otherwise

list_schemas() → list[str]

List all registered span schema names (params schemas only).

• Returns: List of registered schema names

merge(other: SchemaRegistry) → None

Merge schemas from another registry into this one.

• Parameters: other – Another SchemaRegistry to merge. Conflicting schemas from the other registry will be rejected.
• Raises: ValueError – If there are conflicting schema names in any category.

register(schema_name: str, schema: dict[str, Any]) → None

Register a params schema for a span type.

Adds to span_schemas (the flat params-schema map). Use register_type() if you also need a result schema, title, description, or template.

• Parameters:
  • schema_name – Unique identifier for this span type (e.g., “langchain:llm”)
  • schema – JSON Schema dict defining the span payload structure
• Raises: ValueError – If schema_name is already registered.

register_result(schema_name: str, result_schema: dict[str, Any]) → None

Register a result schema for a span type.

Adds to span_result_schemas (the flat result-schema map). The span type does not need to have a params schema registered first.

• Parameters:
  • schema_name – Span type identifier (e.g., “agent:llm”)
  • result_schema – JSON Schema dict defining the span result payload
• Raises: ValueError – If a result schema for schema_name is already registered.

register_type(name: str, params_schema: dict[str, Any], result_schema: dict[str, Any] | None = None, title: str | None = None, description: str | None = None, template: str | None = None) → None

Register a full structured span type schema.

Adds to span_type_schemas. This is the richest form and supports all API fields: params schema, result schema, human-readable title, description, and a display template.

• Parameters:
  • name – Span type name (e.g., “agent:llm”)
  • params_schema – JSON Schema for the span payload (params)
  • result_schema – Optional JSON Schema for the span result payload
  • title – Optional human-readable title (defaults to name on the API)
  • description – Optional description of the span type
  • template – Optional display template using {{field}} interpolation
• Raises: ValueError – If name is already registered as a span type schema.

register_unsafe(schema_name: str, schema: dict[str, Any]) → None

Register a params schema, overwriting if it already exists.

• Parameters:
  • schema_name – Unique identifier for this span type
  • schema – JSON Schema dict defining the span payload structure

to_agent_schema_version(external_id: str) → dict[str, Any]

Convert registry contents to API-compatible agent_schema_version format.

Emits span_schemas, span_result_schemas, and span_type_schemas for whichever have been populated.

• Parameters: external_id – External identifier for this combined schema version
• Returns: Dict with external_identifier and whichever schema fields are non-empty.