Registry System#

Centralized component registry with configuration-driven application loading, convention-based module discovery, and explicit component registration.

Core Registry Classes#

RegistryManager#

class framework.registry.RegistryManager(application_names=None)[source]#

Bases: object

Centralized registry for all ALS Expert Agent components.

This class provides the single point of access for capabilities, nodes, context classes, and data sources throughout the framework. It replaces the fragmented registry system with a unified approach that eliminates circular imports through lazy loading and provides dependency-ordered initialization.

The registry system follows a strict initialization order to handle dependencies: 1. Context classes (required by capabilities) 2. Data sources (required by capabilities) 3. Core nodes (infrastructure components) 4. Capabilities (domain-specific functionality) 5. Framework prompt providers (application-specific prompts) 6. Workflow templates (predefined execution patterns)

All components are loaded lazily using module path and class name metadata, preventing circular import issues while maintaining full introspection capabilities.

Note

The registry is typically accessed through the global functions get_registry() and initialize_registry() rather than instantiated directly.

Warning

Registry initialization must complete successfully before any components can be accessed. Failed initialization will raise RegistryError.

Initialize registry manager with convention-based application loading.

Creates a new registry manager instance that will load the specified application registries using naming conventions. The manager builds a merged configuration by combining the framework registry with all specified application registries.

The initialization process follows these steps: 1. Load framework registry from framework.registry.registry 2. Load application registries from applications.{app_name}.registry 3. Merge configurations with application overrides taking precedence 4. Prepare component registries for lazy loading

Applications are loaded using the convention-based pattern where each application name corresponds to a registry module at: applications.{app_name}.registry

Each application registry module must contain exactly one class implementing the RegistryConfigProvider interface.

Parameters:

application_names (list[str], optional) – List of application names to load using conventions. If None or empty, only the framework registry will be loaded. Application names should match directory names under src/applications/

Raises:

Note

The registry manager is not initialized after construction. Call initialize() to perform component loading and make components available for access.

Examples

Create registry with specific applications:

>>> manager = RegistryManager(["als_expert", "wind_turbine"])
>>> manager.initialize()  # Load components
>>> capability = manager.get_capability("pv_address_finding")

Create framework-only registry:

>>> manager = RegistryManager([])  # No applications
>>> manager.initialize()
>>> memory_cap = manager.get_capability("memory")  # Framework capability

Typical usage through global functions:

>>> # Preferred approach - use global functions
>>> from framework.registry import initialize_registry, get_registry
>>> initialize_registry()  # Uses config to determine applications
>>> registry = get_registry()

See also

initialize_registry() : Preferred way to create and initialize registry get_registry() : Access the global singleton registry instance initialize() : Initialize components after construction RegistryConfigProvider : Interface that applications must implement

Component Access Methods

Capabilities:

get_capability(name)[source]#

Retrieve registered capability instance by name.

Parameters:

name (str) – Unique capability name from registration

Returns:

Capability instance if registered, None otherwise

Return type:

framework.base.BaseCapability, optional

get_all_capabilities()[source]#

Retrieve all registered capability instances.

Returns:

List of all registered capability instances

Return type:

list[framework.base.BaseCapability]

Context Classes:

get_context_class(context_type)[source]#

Retrieve context class by type identifier.

Parameters:

context_type (str) – Context type identifier (e.g., ‘PV_ADDRESSES’)

Returns:

Context class if registered, None otherwise

Return type:

Type[framework.base.CapabilityContext], optional

get_all_context_classes()[source]#

Get dictionary of all registered context classes by context type.

This method provides access to all registered context classes indexed by their context type identifiers. It enables introspection of the complete context system and supports dynamic context handling patterns.

Returns:

Dictionary mapping context types to their corresponding context classes

Return type:

Dict[str, Type[CapabilityContext]]

Examples

Access all context classes:

>>> registry = get_registry()
>>> context_classes = registry.get_all_context_classes()
>>> pv_class = context_classes.get("PV_ADDRESSES")
>>> if pv_class:
...     instance = pv_class(pvs=["test:pv"])

Infrastructure Nodes:

get_node(name)[source]#

Retrieve registered node instance by name.

Parameters:

name (str) – Unique node name from registration

Returns:

Node instance if registered, None otherwise

Return type:

framework.base.BaseCapabilityNode, optional

get_all_nodes()[source]#

Retrieve all registered nodes as (name, callable) pairs.

Returns:

Dictionary mapping node names to their callable instances

Return type:

Dict[str, Any]

Data Sources:

get_data_source(name)[source]#

Retrieve data source provider instance by name.

Parameters:

name (str) – Unique data source name from registration

Returns:

Data source provider instance if registered, None otherwise

Return type:

Any, optional

get_all_data_sources()[source]#

Retrieve all registered data source provider instances.

Returns:

List of all registered data source provider instances

Return type:

list[Any]

Services:

get_service(name)[source]#

Retrieve registered service graph by name.

Parameters:

name (str) – Unique service name from registration

Returns:

Compiled LangGraph service instance if registered, None otherwise

Return type:

Any, optional

get_all_services()[source]#

Retrieve all registered service graph instances.

Returns:

List of all registered service graph instances

Return type:

list[Any]

Registry Management

initialize()[source]#

Initialize all component registries in dependency order.

Performs complete initialization of the registry system by loading all registered components in the proper dependency order. This method handles the transition from configuration metadata to actual component instances, importing modules and instantiating classes as needed.

The initialization process follows strict dependency order to ensure components are available when needed by dependent components:

  1. Context Classes: Data structures used by capabilities

  2. Data Sources: External data providers used by capabilities

  3. Core Nodes: Infrastructure components (router, orchestrator, etc.)

  4. Services: Internal LangGraph service graphs

  5. Capabilities: Domain-specific functionality

  6. Framework Prompt Providers: Application-specific prompt customizations

During initialization, the registry: - Dynamically imports all component modules using lazy loading - Instantiates classes and functions as specified in registrations - Validates that all components can be loaded successfully - Creates proper cross-references between related components - Sets up prompt provider overrides and customizations

Component Loading Details:

  • Context Classes: Imported and registered by type identifier

  • Data Sources: Instantiated and optionally health-checked

  • Nodes: Decorator-created functions registered for LangGraph

  • Capabilities: Instantiated with decorator-created node functions

  • Services: Service graphs compiled and made available

  • Analyzers: Policy and domain analyzers prepared for use

Raises:

  • RegistryError – If any component fails to load, import, or initialize. This includes missing modules, invalid class names, or instantiation failures

  • ConfigurationError – If configuration is invalid, has circular dependencies, or contains inconsistent component definitions

  • ImportError – If any component module cannot be imported

  • AttributeError – If any component class or function is not found in its module

Note

This method is idempotent - multiple calls have no effect if the registry is already initialized. The initialization state is tracked internally.

Warning

All component imports occur during this call. Import failures, missing dependencies, or circular imports will prevent framework operation. Ensure all component modules and dependencies are available.

Examples

Basic initialization:

>>> registry = RegistryManager(["als_expert"])
>>> registry.initialize()  # Load all components
>>> print(registry.get_stats())  # Check what was loaded

Handle initialization errors:

>>> try:
...     registry.initialize()
... except RegistryError as e:
...     print(f"Registry initialization failed: {e}")
...     # Check configuration or fix missing components

Check initialization status:

>>> if not registry._initialized:
...     registry.initialize()
>>> capability = registry.get_capability("my_capability")

See also

validate_configuration() : Validate configuration before initialization get_stats() : Check initialization results and component counts clear() : Reset registry to uninitialized state initialize_registry() : Global function that calls this method

validate_configuration()[source]#

Validate registry configuration for consistency and completeness.

Performs comprehensive validation of the registry configuration including duplicate name checking, dependency validation, and structural consistency. This method should be called before initialization to catch configuration errors early.

Returns:

List of validation error messages, empty if valid

Return type:

list[str]

Examples

Validate before initialization:

>>> registry = get_registry()
>>> errors = registry.validate_configuration()
>>> if errors:
...     print(f"Configuration errors: {errors}")
... else:
...     registry.initialize()
export_registry_to_json(output_dir=None)[source]#

Export registry metadata for external tools and plan editors.

Creates comprehensive JSON export of all registered components including capabilities, context types, and workflow templates. This data is used by execution plan editors and other external tools to understand system capabilities.

Parameters:

output_dir (str, optional) – Directory path for saving JSON files, if None returns data only

Returns:

Complete registry metadata with capabilities, context types, and templates

Return type:

dict[str, Any]

Raises:

Exception – If file writing fails when output_dir is specified

Examples

Export to directory:

>>> registry = get_registry()
>>> data = registry.export_registry_to_json("/tmp/registry")
>>> print(f"Exported {data['metadata']['total_capabilities']} capabilities")

Get data without saving:

>>> data = registry.export_registry_to_json()
>>> capabilities = data['capabilities']

Registry Statistics and Debugging

get_stats()[source]#

Retrieve comprehensive registry statistics for debugging.

Returns:

Dictionary containing counts and lists of registered components

Return type:

dict[str, Any]

Examples

Check registry status:

>>> registry = get_registry()
>>> stats = registry.get_stats()
>>> print(f"Loaded {stats['capabilities']} capabilities")
>>> print(f"Available: {stats['capability_names']}")
__init__(application_names=None)[source]#

Initialize registry manager with convention-based application loading.

Creates a new registry manager instance that will load the specified application registries using naming conventions. The manager builds a merged configuration by combining the framework registry with all specified application registries.

The initialization process follows these steps: 1. Load framework registry from framework.registry.registry 2. Load application registries from applications.{app_name}.registry 3. Merge configurations with application overrides taking precedence 4. Prepare component registries for lazy loading

Applications are loaded using the convention-based pattern where each application name corresponds to a registry module at: applications.{app_name}.registry

Each application registry module must contain exactly one class implementing the RegistryConfigProvider interface.

Parameters:

application_names (list[str], optional) – List of application names to load using conventions. If None or empty, only the framework registry will be loaded. Application names should match directory names under src/applications/

Raises:

Note

The registry manager is not initialized after construction. Call initialize() to perform component loading and make components available for access.

Examples

Create registry with specific applications:

>>> manager = RegistryManager(["als_expert", "wind_turbine"])
>>> manager.initialize()  # Load components
>>> capability = manager.get_capability("pv_address_finding")

Create framework-only registry:

>>> manager = RegistryManager([])  # No applications
>>> manager.initialize()
>>> memory_cap = manager.get_capability("memory")  # Framework capability

Typical usage through global functions:

>>> # Preferred approach - use global functions
>>> from framework.registry import initialize_registry, get_registry
>>> initialize_registry()  # Uses config to determine applications
>>> registry = get_registry()

See also

initialize_registry() : Preferred way to create and initialize registry get_registry() : Access the global singleton registry instance initialize() : Initialize components after construction RegistryConfigProvider : Interface that applications must implement

initialize()[source]#

Initialize all component registries in dependency order.

Performs complete initialization of the registry system by loading all registered components in the proper dependency order. This method handles the transition from configuration metadata to actual component instances, importing modules and instantiating classes as needed.

The initialization process follows strict dependency order to ensure components are available when needed by dependent components:

  1. Context Classes: Data structures used by capabilities

  2. Data Sources: External data providers used by capabilities

  3. Core Nodes: Infrastructure components (router, orchestrator, etc.)

  4. Services: Internal LangGraph service graphs

  5. Capabilities: Domain-specific functionality

  6. Framework Prompt Providers: Application-specific prompt customizations

During initialization, the registry: - Dynamically imports all component modules using lazy loading - Instantiates classes and functions as specified in registrations - Validates that all components can be loaded successfully - Creates proper cross-references between related components - Sets up prompt provider overrides and customizations

Component Loading Details:

  • Context Classes: Imported and registered by type identifier

  • Data Sources: Instantiated and optionally health-checked

  • Nodes: Decorator-created functions registered for LangGraph

  • Capabilities: Instantiated with decorator-created node functions

  • Services: Service graphs compiled and made available

  • Analyzers: Policy and domain analyzers prepared for use

Raises:

  • RegistryError – If any component fails to load, import, or initialize. This includes missing modules, invalid class names, or instantiation failures

  • ConfigurationError – If configuration is invalid, has circular dependencies, or contains inconsistent component definitions

  • ImportError – If any component module cannot be imported

  • AttributeError – If any component class or function is not found in its module

Note

This method is idempotent - multiple calls have no effect if the registry is already initialized. The initialization state is tracked internally.

Warning

All component imports occur during this call. Import failures, missing dependencies, or circular imports will prevent framework operation. Ensure all component modules and dependencies are available.

Examples

Basic initialization:

>>> registry = RegistryManager(["als_expert"])
>>> registry.initialize()  # Load all components
>>> print(registry.get_stats())  # Check what was loaded

Handle initialization errors:

>>> try:
...     registry.initialize()
... except RegistryError as e:
...     print(f"Registry initialization failed: {e}")
...     # Check configuration or fix missing components

Check initialization status:

>>> if not registry._initialized:
...     registry.initialize()
>>> capability = registry.get_capability("my_capability")

See also

validate_configuration() : Validate configuration before initialization get_stats() : Check initialization results and component counts clear() : Reset registry to uninitialized state initialize_registry() : Global function that calls this method

export_registry_to_json(output_dir=None)[source]#

Export registry metadata for external tools and plan editors.

Creates comprehensive JSON export of all registered components including capabilities, context types, and workflow templates. This data is used by execution plan editors and other external tools to understand system capabilities.

Parameters:

output_dir (str, optional) – Directory path for saving JSON files, if None returns data only

Returns:

Complete registry metadata with capabilities, context types, and templates

Return type:

dict[str, Any]

Raises:

Exception – If file writing fails when output_dir is specified

Examples

Export to directory:

>>> registry = get_registry()
>>> data = registry.export_registry_to_json("/tmp/registry")
>>> print(f"Exported {data['metadata']['total_capabilities']} capabilities")

Get data without saving:

>>> data = registry.export_registry_to_json()
>>> capabilities = data['capabilities']
get_capability(name)[source]#

Retrieve registered capability instance by name.

Parameters:

name (str) – Unique capability name from registration

Returns:

Capability instance if registered, None otherwise

Return type:

framework.base.BaseCapability, optional

get_always_active_capability_names()[source]#

Get names of capabilities marked as always active.

Returns:

List of capability names that are always active

Return type:

List[str]

get_all_capabilities()[source]#

Retrieve all registered capability instances.

Returns:

List of all registered capability instances

Return type:

list[framework.base.BaseCapability]

get_capabilities_overview()[source]#

Generate a text overview of all registered capabilities.

Returns:

Human-readable overview of capabilities and their descriptions

Return type:

str

get_node(name)[source]#

Retrieve registered node instance by name.

Parameters:

name (str) – Unique node name from registration

Returns:

Node instance if registered, None otherwise

Return type:

framework.base.BaseCapabilityNode, optional

get_all_nodes()[source]#

Retrieve all registered nodes as (name, callable) pairs.

Returns:

Dictionary mapping node names to their callable instances

Return type:

Dict[str, Any]

get_context_class(context_type)[source]#

Retrieve context class by type identifier.

Parameters:

context_type (str) – Context type identifier (e.g., ‘PV_ADDRESSES’)

Returns:

Context class if registered, None otherwise

Return type:

Type[framework.base.CapabilityContext], optional

get_context_class_by_name(class_name)[source]#

Retrieve context class by class name identifier.

Parameters:

class_name (str) – Python class name (e.g., ‘PVAddresses’)

Returns:

Context class if found, None otherwise

Return type:

Type[framework.base.CapabilityContext], optional

is_valid_context_type(context_type)[source]#

Check if a context type is registered in the registry.

Parameters:

context_type (str) – Context type identifier (e.g., ‘PV_ADDRESSES’)

Returns:

True if context type is registered, False otherwise

Return type:

bool

get_all_context_types()[source]#

Get list of all registered context types.

Returns:

List of all registered context type identifiers

Return type:

list[str]

get_all_context_classes()[source]#

Get dictionary of all registered context classes by context type.

This method provides access to all registered context classes indexed by their context type identifiers. It enables introspection of the complete context system and supports dynamic context handling patterns.

Returns:

Dictionary mapping context types to their corresponding context classes

Return type:

Dict[str, Type[CapabilityContext]]

Examples

Access all context classes:

>>> registry = get_registry()
>>> context_classes = registry.get_all_context_classes()
>>> pv_class = context_classes.get("PV_ADDRESSES")
>>> if pv_class:
...     instance = pv_class(pvs=["test:pv"])
get_data_source(name)[source]#

Retrieve data source provider instance by name.

Parameters:

name (str) – Unique data source name from registration

Returns:

Data source provider instance if registered, None otherwise

Return type:

Any, optional

get_all_data_sources()[source]#

Retrieve all registered data source provider instances.

Returns:

List of all registered data source provider instances

Return type:

list[Any]

get_service(name)[source]#

Retrieve registered service graph by name.

Parameters:

name (str) – Unique service name from registration

Returns:

Compiled LangGraph service instance if registered, None otherwise

Return type:

Any, optional

get_all_services()[source]#

Retrieve all registered service graph instances.

Returns:

List of all registered service graph instances

Return type:

list[Any]

get_execution_policy_analyzers()[source]#

Retrieve all registered execution policy analyzer instances.

Creates instances of execution policy analyzers with empty configurable. The actual configurable will be provided when the analyzers are used.

Returns:

List of execution policy analyzer instances

Return type:

list[Any]

get_domain_analyzers()[source]#

Retrieve all registered domain analyzer instances.

Creates instances of domain analyzers with empty configurable. The actual configurable will be provided when the analyzers are used.

Returns:

List of domain analyzer instances

Return type:

list[Any]

get_available_data_sources(state)[source]#

Retrieve available data sources for current execution context.

Filters all registered data sources based on their availability for the current agent state and returns them in registration order (framework providers first, then applications).

Parameters:

state (framework.state.AgentState) – Current agent state for availability checking

Returns:

Available data source providers in registration order

Return type:

list[Any]

Note

Providers without is_available() method are assumed to be available.

property context_types#

Dynamic object providing context type constants as attributes.

Creates a dynamic object where each registered context type is accessible as an attribute with its string value.

Returns:

Object with context types as attributes

Return type:

object

Examples

Access context types:

>>> registry = get_registry()
>>> pv_type = registry.context_types.PV_ADDRESSES
>>> print(pv_type)  # "PV_ADDRESSES"
property capability_names#

Dynamic object providing capability names as constants with debug fallback.

Creates a dynamic object where each registered capability name is accessible as an uppercase constant attribute. If a capability name is not registered, returns the expected string and logs a warning (useful for development).

Returns:

Object with capability names as constant attributes

Return type:

object

Examples

Access capability names as constants:

>>> registry = get_registry()
>>> pv_finding = registry.capability_names.PV_ADDRESS_FINDING
>>> print(pv_finding)  # "pv_address_finding"

Graceful fallback for missing capabilities:

>>> viz_name = registry.capability_names.DATA_VISUALIZATION  # Not registered
>>> print(viz_name)  # "data_visualization" (with warning logged)
validate_configuration()[source]#

Validate registry configuration for consistency and completeness.

Performs comprehensive validation of the registry configuration including duplicate name checking, dependency validation, and structural consistency. This method should be called before initialization to catch configuration errors early.

Returns:

List of validation error messages, empty if valid

Return type:

list[str]

Examples

Validate before initialization:

>>> registry = get_registry()
>>> errors = registry.validate_configuration()
>>> if errors:
...     print(f"Configuration errors: {errors}")
... else:
...     registry.initialize()
get_stats()[source]#

Retrieve comprehensive registry statistics for debugging.

Returns:

Dictionary containing counts and lists of registered components

Return type:

dict[str, Any]

Examples

Check registry status:

>>> registry = get_registry()
>>> stats = registry.get_stats()
>>> print(f"Loaded {stats['capabilities']} capabilities")
>>> print(f"Available: {stats['capability_names']}")
clear()[source]#

Clear all registry data and reset initialization state.

Removes all registered components and marks the registry as uninitialized. This method is primarily used for testing to ensure clean state between tests.

Warning

This method clears all registered components. Only use for testing or complete registry reset scenarios.

Global Registry Functions#

framework.registry.get_registry()[source]#

Retrieve the global registry manager singleton instance.

Returns the global registry manager instance, creating it automatically if it doesn’t exist. This function provides the primary access point for the registry system throughout the framework, ensuring consistent access to the same registry instance across all framework components.

The registry instance is created from the global configuration. Applications are loaded from the ‘applications’ configuration key and their registries are loaded using the pattern: applications.{app_name}.registry

Singleton Behavior:

  • First call creates the registry instance from configuration

  • Subsequent calls return the same instance

  • Registry persists for the lifetime of the application

  • Thread-safe access to the global instance

Registry Creation Process:

  1. Read ‘applications’ list from global configuration

  2. Create RegistryManager with configured application names

  3. Build merged configuration (framework + applications)

  4. Return uninitialized registry instance

Returns:

The global registry manager singleton instance. The instance may not be initialized - call initialize_registry() or registry.initialize() to load components before accessing them

Return type:

RegistryManager

Raises:

  • ConfigurationError – If global configuration is invalid or applications configuration is malformed

  • RuntimeError – If registry creation fails due to configuration issues

Note

The returned registry instance may not be initialized. Components are not available until initialize_registry() or registry.initialize() is called.

Warning

This function creates the registry from global configuration on first access. Ensure configuration is properly set before calling this function.

Examples

Basic registry access:

>>> from framework.registry import get_registry, initialize_registry
>>>
>>> # Initialize first, then access
>>> initialize_registry()
>>> registry = get_registry()
>>> capability = registry.get_capability("pv_address_finding")

Check if registry is initialized:

>>> registry = get_registry()
>>> if not registry._initialized:
...     registry.initialize()
>>> stats = registry.get_stats()

Access registry from different modules:

>>> # Same instance returned everywhere
>>> from framework.registry import get_registry
>>> registry1 = get_registry()
>>> registry2 = get_registry()
>>> assert registry1 is registry2  # Same instance

See also

initialize_registry() : Initialize the registry system with components reset_registry() : Reset the global registry instance RegistryManager : The registry manager class returned by this function Registry and Discovery : Registry access patterns

framework.registry.initialize_registry(auto_export=True)[source]#

Initialize the global registry system with all components.

Performs complete initialization of the global registry system, including loading of application registries, component loading in dependency order, and optional export of registry metadata for external tools. This function should be called once during application startup before accessing any framework components.

The initialization process: 1. Gets or creates the global registry singleton instance 2. Loads all components in proper dependency order 3. Validates that all components loaded successfully 4. Optionally exports registry metadata to JSON files 5. Sets up the registry for component access throughout the application

Component Loading Order:

  • Context classes (data structures)

  • Data sources (external data providers)

  • Core nodes (infrastructure components)

  • Services (internal LangGraph service graphs)

  • Capabilities (domain-specific functionality)

  • Framework prompt providers (application customizations)

Auto-Export Functionality:

When auto_export is True, the function automatically exports complete registry metadata to JSON files for use by external tools, execution plan editors, and documentation systems. Export includes capability definitions, context types, and component relationships.

Parameters:

auto_export (bool) – Whether to automatically export registry metadata to JSON files after successful initialization. Export files are saved to the configured registry_exports_dir location

Raises:

  • RegistryError – If registry initialization fails due to component loading errors, missing modules, or invalid component definitions

  • ConfigurationError – If global configuration is invalid, application configurations are malformed, or dependencies are inconsistent

  • ImportError – If any component module cannot be imported

  • AttributeError – If any component class or function is not found

Note

This function is idempotent - multiple calls have no effect if the registry is already initialized. The initialization state persists for the application lifetime.

Warning

This function must be called before accessing any framework components. Attempting to access components before initialization will result in empty or incomplete results.

Examples

Basic application startup:

>>> from framework.registry import initialize_registry, get_registry
>>>
>>> # Initialize during application startup
>>> initialize_registry()
>>>
>>> # Now components are available
>>> registry = get_registry()
>>> capability = registry.get_capability("pv_address_finding")

Initialize without metadata export:

>>> # Skip JSON export for faster initialization
>>> initialize_registry(auto_export=False)

Handle initialization errors:

>>> try:
...     initialize_registry()
...     print("Registry initialized successfully")
... except RegistryError as e:
...     print(f"Registry initialization failed: {e}")
...     # Handle missing components or configuration issues

Check initialization results:

>>> initialize_registry()
>>> registry = get_registry()
>>> stats = registry.get_stats()
>>> print(f"Loaded {stats['capabilities']} capabilities")

See also

get_registry() : Access the initialized registry instance reset_registry() : Reset registry for testing or reconfiguration RegistryManager.initialize() : Core initialization method called by this function RegistryManager.export_registry_to_json() : Export functionality used when auto_export=True

Registration Configuration#

Configuration Interface#

class framework.registry.RegistryConfigProvider[source]#

Bases: ABC

Abstract interface for application registry configuration.

All applications must implement this interface to provide their registry configuration in a standardized, type-safe manner. This interface eliminates naming ambiguity and ensures consistent registry patterns across all applications in the framework ecosystem.

The framework loads classes implementing this interface from each configured application’s registry module using the convention-based path: applications.{app_name}.registry

This design provides several key benefits:

  • Type Safety: All registry configurations are strongly typed

  • Convention-Based Loading: Standardized module path patterns

  • Validation: Automatic validation of registry structure

  • Extensibility: Easy to add new component types without breaking changes

  • Documentation: Self-documenting registry configurations

Implementation Requirements:

  1. Exactly one class per application registry module must implement this interface

  2. The get_registry_config() method must return a complete RegistryConfig

  3. All component registrations must use valid module paths and class names

  4. Context type dependencies (provides/requires) must be consistent

The registry system will call get_registry_config() once during initialization and merge the returned configuration with the framework’s base registry. Applications can override framework components by using the same names.

Raises:

  • RegistryError – If multiple implementations found in one module

  • RegistryError – If no implementation found in application registry module

  • NotImplementedError – If get_registry_config() is not implemented

Examples

Basic application registry:

>>> class MyAppRegistryProvider(RegistryConfigProvider):
...     def get_registry_config(self) -> RegistryConfig:
...         return RegistryConfig(
...             capabilities=[
...                 CapabilityRegistration(
...                     name="my_capability",
...                     module_path="applications.myapp.capabilities.my_capability",
...                     class_name="MyCapability",
...                     description="Application-specific functionality",
...                     provides=["MY_RESULTS"],
...                     requires=["INPUT_DATA"]
...                 )
...             ],
...             context_classes=[
...                 ContextClassRegistration(
...                     context_type="MY_RESULTS",
...                     module_path="applications.myapp.context_classes",
...                     class_name="MyResults"
...                 )
...             ]
...         )

Advanced registry with services and data sources:

>>> class AdvancedAppRegistryProvider(RegistryConfigProvider):
...     def get_registry_config(self) -> RegistryConfig:
...         return RegistryConfig(
...             capabilities=[...],
...             context_classes=[...],
...             data_sources=[
...                 DataSourceRegistration(
...                     name="app_database",
...                     module_path="applications.myapp.data_sources.database",
...                     class_name="AppDatabaseProvider",
...                     description="Application-specific database access",
...                     health_check_required=True
...                 )
...             ],
...             services=[
...                 ServiceRegistration(
...                     name="data_processor",
...                     module_path="applications.myapp.services.processor",
...                     class_name="DataProcessorService",
...                     description="Multi-step data processing workflow",
...                     provides=["PROCESSED_DATA"],
...                     requires=["RAW_DATA"],
...                     internal_nodes=["validator", "transformer", "aggregator"]
...                 )
...             ]
...         )

Note

The framework will merge application registries with the base framework registry, allowing applications to override framework components by using the same names.

Warning

Each application registry module must contain exactly one class implementing this interface. Multiple implementations or missing implementations will cause RegistryError during framework initialization.

See also

RegistryConfig : Configuration structure returned by implementations RegistryManager : Manager that discovers and uses these providers Building Your First Capability : Complete application development guide

Abstract Methods

abstractmethod get_registry_config()[source]#

Get complete application registry configuration.

This method is called once during registry initialization to retrieve the complete component configuration for the application. The returned RegistryConfig will be merged with the framework’s base registry, allowing applications to extend or override framework functionality.

The configuration should include all components that the application provides: capabilities, context classes, data sources, services, and any other registered components. Components are loaded lazily using the module_path and class_name metadata provided in the registrations.

Implementation Guidelines:

  • Return a complete RegistryConfig with all application components

  • Use descriptive names and clear documentation for all components

  • Ensure provides/requires relationships are consistent across components

  • Validate that all module paths and class names are correct

  • Consider component initialization order when defining dependencies

Returns:

Complete registry configuration for this application including all capabilities, context classes, data sources, services, and other components that should be available in the framework

Return type:

RegistryConfig

Raises:

  • NotImplementedError – If not implemented by subclass (required by ABC)

  • ImportError – If any component module paths are invalid

  • AttributeError – If any component class names are not found

Note

This method is called exactly once during registry initialization. The framework caches the returned configuration, so dynamic changes after initialization are not supported.

Warning

All module paths must be importable and all class names must exist in their respective modules. Invalid paths will cause registry initialization to fail.

Examples

Minimal application configuration:

>>> def get_registry_config(self) -> RegistryConfig:
...     return RegistryConfig(
...         capabilities=[
...             CapabilityRegistration(
...                 name="hello_world",
...                 module_path="applications.myapp.capabilities.hello",
...                 class_name="HelloWorldCapability",
...                 description="Simple greeting capability",
...                 provides=["GREETING"],
...                 requires=[]
...             )
...         ],
...         context_classes=[
...             ContextClassRegistration(
...                 context_type="GREETING",
...                 module_path="applications.myapp.context_classes",
...                 class_name="GreetingContext"
...             )
...         ]
...     )

Complex application with all component types:

>>> def get_registry_config(self) -> RegistryConfig:
...     return RegistryConfig(
...         capabilities=[...],  # Domain-specific capabilities
...         context_classes=[...],  # Data structures
...         data_sources=[...],  # External data providers
...         services=[...],  # Internal service graphs
...         framework_prompt_providers=[...]  # Custom prompts
...     )

See also

RegistryConfig : Structure of the returned configuration CapabilityRegistration : Capability registration metadata ContextClassRegistration : Context class registration metadata Registry and Discovery : Component registration guide

abstractmethod get_registry_config()[source]#

Get complete application registry configuration.

This method is called once during registry initialization to retrieve the complete component configuration for the application. The returned RegistryConfig will be merged with the framework’s base registry, allowing applications to extend or override framework functionality.

The configuration should include all components that the application provides: capabilities, context classes, data sources, services, and any other registered components. Components are loaded lazily using the module_path and class_name metadata provided in the registrations.

Implementation Guidelines:

  • Return a complete RegistryConfig with all application components

  • Use descriptive names and clear documentation for all components

  • Ensure provides/requires relationships are consistent across components

  • Validate that all module paths and class names are correct

  • Consider component initialization order when defining dependencies

Returns:

Complete registry configuration for this application including all capabilities, context classes, data sources, services, and other components that should be available in the framework

Return type:

RegistryConfig

Raises:

  • NotImplementedError – If not implemented by subclass (required by ABC)

  • ImportError – If any component module paths are invalid

  • AttributeError – If any component class names are not found

Note

This method is called exactly once during registry initialization. The framework caches the returned configuration, so dynamic changes after initialization are not supported.

Warning

All module paths must be importable and all class names must exist in their respective modules. Invalid paths will cause registry initialization to fail.

Examples

Minimal application configuration:

>>> def get_registry_config(self) -> RegistryConfig:
...     return RegistryConfig(
...         capabilities=[
...             CapabilityRegistration(
...                 name="hello_world",
...                 module_path="applications.myapp.capabilities.hello",
...                 class_name="HelloWorldCapability",
...                 description="Simple greeting capability",
...                 provides=["GREETING"],
...                 requires=[]
...             )
...         ],
...         context_classes=[
...             ContextClassRegistration(
...                 context_type="GREETING",
...                 module_path="applications.myapp.context_classes",
...                 class_name="GreetingContext"
...             )
...         ]
...     )

Complex application with all component types:

>>> def get_registry_config(self) -> RegistryConfig:
...     return RegistryConfig(
...         capabilities=[...],  # Domain-specific capabilities
...         context_classes=[...],  # Data structures
...         data_sources=[...],  # External data providers
...         services=[...],  # Internal service graphs
...         framework_prompt_providers=[...]  # Custom prompts
...     )

See also

RegistryConfig : Structure of the returned configuration CapabilityRegistration : Capability registration metadata ContextClassRegistration : Context class registration metadata Registry and Discovery : Component registration guide

RegistryConfig#

class framework.registry.RegistryConfig(capabilities, context_classes, core_nodes=<factory>, data_sources=<factory>, services=<factory>, domain_analyzers=<factory>, execution_policy_analyzers=<factory>, framework_prompt_providers=<factory>, framework_exclusions=<factory>, initialization_order=<factory>)[source]#

Bases: object

Complete registry configuration with all component metadata.

Contains the complete configuration for the framework registry including all component registrations and initialization ordering. Enhanced for LangGraph migration with support for decorators and advanced features.

Most fields are optional with sensible defaults to improve UX for applications. Applications typically only need to define capabilities, context_classes, and optionally data_sources and framework_prompt_providers.

Parameters:

  • capabilities (list[CapabilityRegistration]) – Registration entries for domain capabilities

  • context_classes (list[ContextClassRegistration]) – Registration entries for context data classes

  • core_nodes (list[NodeRegistration]) – Registration entries for infrastructure nodes (optional)

  • data_sources (list[DataSourceRegistration]) – Registration entries for external data sources (optional)

  • services (list[ServiceRegistration]) – Registration entries for internal service graphs (optional)

  • domain_analyzers (list[DomainAnalyzerRegistration]) – Registration entries for domain analyzers (optional)

  • execution_policy_analyzers (list[ExecutionPolicyAnalyzerRegistration]) – Registration entries for execution policy analyzers (optional)

  • framework_prompt_providers (list[FrameworkPromptProviderRegistration]) – Registration entries for prompt providers (optional)

  • framework_exclusions (dict[str, list[str]]) – Framework component names to exclude by type (optional)

  • initialization_order (list[str]) – Component type initialization sequence (optional)

Component Lists

capabilities: List[CapabilityRegistration]#

Registration entries for domain capabilities (required).

context_classes: List[ContextClassRegistration]#

Registration entries for context data classes (required).

core_nodes: List[NodeRegistration]#

Registration entries for infrastructure nodes (optional).

data_sources: List[DataSourceRegistration]#

Registration entries for external data sources (optional).

services: List[ServiceRegistration]#

Registration entries for internal service graphs (optional).

capabilities: List[CapabilityRegistration]#
context_classes: List[ContextClassRegistration]#
core_nodes: List[NodeRegistration]#
data_sources: List[DataSourceRegistration]#
services: List[ServiceRegistration]#
domain_analyzers: List[DomainAnalyzerRegistration]#
execution_policy_analyzers: List[ExecutionPolicyAnalyzerRegistration]#
framework_prompt_providers: List[FrameworkPromptProviderRegistration]#
framework_exclusions: Dict[str, List[str]]#
initialization_order: List[str]#
__init__(capabilities, context_classes, core_nodes=<factory>, data_sources=<factory>, services=<factory>, domain_analyzers=<factory>, execution_policy_analyzers=<factory>, framework_prompt_providers=<factory>, framework_exclusions=<factory>, initialization_order=<factory>)#

Registration Classes#

Component Registration#

class framework.registry.CapabilityRegistration(name, module_path, class_name, description, provides, requires, always_active=False, functional_node=None, example_usage='')[source]#

Bases: object

Registration metadata for capabilities.

Defines the metadata required for lazy loading of capability classes that implement specific functionality for agent systems. Enhanced for LangGraph migration with support for convention-based decorators and advanced features.

Parameters:

  • name (str) – Unique capability name for registration

  • module_path (str) – Python module path for lazy import

  • class_name (str) – Class name within the module

  • description (str) – Human-readable description of capability

  • provides (list[str]) – List of context types this capability produces

  • requires (list[str]) – List of context types this capability needs

  • always_active (bool) – Whether capability is always active (no classification needed), defaults to False

  • functional_node (str) – Name of the functional node for execution (from capability.node attribute)

  • example_usage (str) – Example of how this capability is used

Key Fields

name: str#

Unique capability name for registration.

module_path: str#

Python module path for lazy import.

class_name: str#

Class name within the module.

provides: List[str]#

Context types this capability produces.

requires: List[str]#

Context types this capability needs as input.

name: str#
module_path: str#
class_name: str#
description: str#
provides: List[str]#
requires: List[str]#
always_active: bool = False#
functional_node: str = None#
example_usage: str = ''#
__init__(name, module_path, class_name, description, provides, requires, always_active=False, functional_node=None, example_usage='')#
class framework.registry.ContextClassRegistration(context_type, module_path, class_name)[source]#

Bases: object

Registration metadata for context data classes.

Defines the metadata required for lazy loading of context classes that represent structured data passed between capabilities.

Parameters:

  • context_type (str) – String identifier for the context type (e.g., ‘PV_ADDRESSES’)

  • module_path (str) – Python module path for lazy import

  • class_name (str) – Class name within the module

Key Fields

context_type: str#

String identifier for the context type (e.g., ‘PV_ADDRESSES’).

module_path: str#

Python module path for lazy import.

class_name: str#

Class name within the module.

context_type: str#
module_path: str#
class_name: str#
__init__(context_type, module_path, class_name)#
class framework.registry.NodeRegistration(name, module_path, function_name, description)[source]#

Bases: object

Registration metadata for infrastructure node functions.

Defines the metadata required for lazy loading of functional nodes.

Parameters:

  • name (str) – Unique identifier for the node in the registry

  • module_path (str) – Python module path for lazy import

  • function_name (str) – Function name within the module (decorated with @infrastructure_node)

  • description (str) – Human-readable description of node functionality

Key Fields

name: str#

Unique identifier for the node in the registry.

module_path: str#

Python module path for lazy import.

function_name: str#

Function name within the module (decorated with @infrastructure_node).

name: str#
module_path: str#
function_name: str#
description: str#
__init__(name, module_path, function_name, description)#
class framework.registry.DataSourceRegistration(name, module_path, class_name, description, health_check_required=True)[source]#

Bases: object

Registration metadata for external data source providers.

Defines the metadata required for lazy loading of data source provider classes that provide access to external systems and databases.

Parameters:

  • name (str) – Unique identifier for the data source in the registry

  • module_path (str) – Python module path for lazy import

  • class_name (str) – Class name within the module

  • description (str) – Human-readable description of data source

  • health_check_required (bool) – Whether provider requires health checking

Key Fields

name: str#

Unique identifier for the data source in the registry.

health_check_required: bool#

Whether provider requires health checking.

name: str#
module_path: str#
class_name: str#
description: str#
health_check_required: bool = True#
__init__(name, module_path, class_name, description, health_check_required=True)#
class framework.registry.ServiceRegistration(name, module_path, class_name, description, provides=<factory>, requires=<factory>, internal_nodes=<factory>)[source]#

Bases: object

Registration metadata for internal service graphs.

Services are separate LangGraph graphs that can be called by capabilities without interfering with the main graph routing. Each service manages its own internal node flow and returns control to the calling capability.

Parameters:

  • name (str) – Unique identifier for the service in the registry

  • module_path (str) – Python module path for lazy import

  • class_name (str) – Service class name within the module

  • description (str) – Human-readable description of service functionality

  • provides (list[str]) – List of context types this service produces

  • requires (list[str]) – List of context types this service needs

  • internal_nodes (list[str]) – List of node names internal to this service

Key Fields

internal_nodes: List[str]#

List of node names internal to this service.

name: str#
module_path: str#
class_name: str#
description: str#
provides: List[str]#
requires: List[str]#
internal_nodes: List[str]#
__init__(name, module_path, class_name, description, provides=<factory>, requires=<factory>, internal_nodes=<factory>)#

Specialized Registration#

class framework.registry.FrameworkPromptProviderRegistration(application_name, module_path, description, prompt_builders=<factory>)[source]#

Bases: object

Registration metadata for application-specific prompt providers.

Defines the metadata required for dependency injection of application-specific prompt builders. Uses the professional pattern of “start with defaults, override specific components” - applications only declare what they want to customize, everything else uses framework defaults.

Parameters:

  • application_name (str) – Application identifier (e.g., ‘als_expert’)

  • module_path (str) – Python module path for lazy import

  • description (str) – Human-readable description of prompt provider

  • prompt_builders (dict[str, str]) – Mapping of prompt types to override with custom builder classes

Examples

Basic application override:

FrameworkPromptProviderRegistration(
    application_name="als_expert",
    module_path="applications.als_expert.framework_prompts",
    description="ALS-specific prompt customizations",
    prompt_builders={
        "orchestrator": "ALSOrchestratorPromptBuilder",
        "memory_extraction": "ALSMemoryExtractionPromptBuilder"
        # time_range_parsing not listed = uses framework default
    }
)

Key Fields

application_name: str#

Application identifier (e.g., ‘als_expert’).

prompt_builders: Dict[str, str]#

Mapping of prompt types to override with custom builder classes.

application_name: str#
module_path: str#
description: str#
prompt_builders: Dict[str, str]#
__init__(application_name, module_path, description, prompt_builders=<factory>)#
class framework.registry.ExecutionPolicyAnalyzerRegistration(name, module_path, class_name, description, priority=50)[source]#

Bases: object

Registration metadata for configurable execution policy analyzers.

Defines the metadata required for lazy loading of execution policy analyzer classes that make execution mode and approval decisions based on code analysis.

Parameters:

  • name (str) – Unique identifier for the policy analyzer in the registry

  • module_path (str) – Python module path for lazy import

  • class_name (str) – Class name within the module

  • description (str) – Human-readable description of policy analyzer

  • priority (int) – Analysis priority (lower numbers = higher priority)

name: str#
module_path: str#
class_name: str#
description: str#
priority: int = 50#
__init__(name, module_path, class_name, description, priority=50)#
class framework.registry.DomainAnalyzerRegistration(name, module_path, class_name, description, priority=50)[source]

Bases: object

Registration metadata for configurable domain analyzers.

Defines the metadata required for lazy loading of domain analyzer classes that analyze generated code for domain-specific patterns and operations.

Parameters:

  • name (str) – Unique identifier for the domain analyzer in the registry

  • module_path (str) – Python module path for lazy import

  • class_name (str) – Class name within the module

  • description (str) – Human-readable description of domain analyzer

  • priority (int) – Analysis priority (lower numbers = higher priority)

name: str
module_path: str
class_name: str
description: str
priority: int = 50
__init__(name, module_path, class_name, description, priority=50)

Registry Architecture#

The registry system uses a two-tier architecture with configuration-driven application loading:

Framework Registry: Core infrastructure loaded from framework.registry.registry

Application Registries: Domain-specific components from applications.{app}.registry (applications must be listed in global configuration)

Initialization Order:

Components are initialized in strict dependency order:

  1. Context classes (required by capabilities)

  2. Data sources (required by capabilities)

  3. Core nodes (infrastructure components)

  4. Services (internal LangGraph service graphs)

  5. Capabilities (domain-specific functionality)

  6. Framework prompt providers (application-specific prompts)

Lazy Loading:

All components use lazy loading to prevent circular import issues. Components are imported and instantiated only during the initialization phase, not at module load time.

Registry Export System#

The registry provides comprehensive export functionality for external tool integration and debugging:

Automatic Export: Registry metadata is automatically exported during initialize_registry() when auto_export=True (default).

Manual Export: Use RegistryManager.export_registry_to_json() for on-demand export of registry state.

Export Configuration: Default export directory is configured via file_paths.registry_exports_dir in config.yml (defaults to _agent_data/registry_exports/).

Export Structure: Exports create standardized JSON files containing capability definitions, context types, and metadata suitable for consumption by external tools, execution plan editors, and debugging utilities.

Integration Pattern: The export system enables air-gapped integration where external tools need component metadata but cannot execute live Python code.

See also

Base Components

Base component classes and decorators for registered components

State and Context Management

State and context management for component data

Configuration System

Configuration system used by registry initialization

Registry and Discovery

Complete guide to registry system and component discovery

On this page
Edit on GitHub