Registry

registry

Classes

RegistryError

Bases: RuntimeError

Raised when a registry operation fails.

TypeRegistry

TypeRegistry()

In-memory registry for blackboard artifact types.

Source code in src/flock/registry.py

def __init__(self) -> None:
    self._by_name: dict[str, type[BaseModel]] = {}
    self._by_cls: dict[type[BaseModel], str] = {}

Functions

resolve_name

resolve_name(type_name: str) -> str

Resolve a type name (simple or qualified) to its canonical form.

Parameters:

Name	Type	Description	Default
type_name	str	Simple name ("Document") or qualified ("main.Document")	required

Returns:

Type	Description
str	Canonical type name from registry

Raises:

Type	Description
RegistryError	Type not found or ambiguous

Source code in src/flock/registry.py

def resolve_name(self, type_name: str) -> str:
    """
    Resolve a type name (simple or qualified) to its canonical form.

    Args:
        type_name: Simple name ("Document") or qualified ("__main__.Document")

    Returns:
        Canonical type name from registry

    Raises:
        RegistryError: Type not found or ambiguous
    """
    # If already canonical, return as-is (O(1) lookup)
    if type_name in self._by_name:
        return type_name

    # Search for models with matching simple name (O(n) scan)
    matches = []
    for canonical_name, model_cls in self._by_name.items():
        if model_cls.__name__ == type_name:
            matches.append(canonical_name)

    if len(matches) == 0:
        raise RegistryError(f"Unknown artifact type '{type_name}'.")
    if len(matches) == 1:
        return matches[0]
    raise RegistryError(
        f"Ambiguous type name '{type_name}'. Matches: {', '.join(matches)}. Use qualified name."
    )

FunctionRegistry

FunctionRegistry()

Registry for deterministic callable helpers (flock_tool).

Source code in src/flock/registry.py

def __init__(self) -> None:
    self._callables: dict[str, Callable[..., Any]] = {}

Functions

flock_type

flock_type(model: type[BaseModel] | None = None, *, name: str | None = None) -> Any

Decorator to register a Pydantic model as a blackboard artifact type.

Source code in src/flock/registry.py

def flock_type(model: type[BaseModel] | None = None, *, name: str | None = None) -> Any:
    """Decorator to register a Pydantic model as a blackboard artifact type."""

    def _wrap(cls: type[BaseModel]) -> type[BaseModel]:
        type_registry.register(cls, name=name)
        return cls

    if model is None:
        return _wrap
    return _wrap(model)

flock_tool

flock_tool(func: Callable[..., Any] | None = None, *, name: str | None = None) -> Any

Decorator to register a deterministic helper function for agents.

Source code in src/flock/registry.py

def flock_tool(func: Callable[..., Any] | None = None, *, name: str | None = None) -> Any:
    """Decorator to register a deterministic helper function for agents."""

    def _wrap(callable_: Callable[..., Any]) -> Callable[..., Any]:
        function_registry.register(callable_, name=name)
        return callable_

    if func is None:
        return _wrap
    return _wrap(func)