Skip to content

symfonic.core.deps

deps

Capability Registry — register and retrieve capabilities by type.

Per ADR-PFX-012: BaseAgentDeps is a generic capability registry. Nodes access capabilities via deps.get(Type), deps.require(Type), deps.has(Type).

BaseAgentDeps 

BaseAgentDeps(**capabilities: Any)

Capability Registry for agent dependencies.

Register capabilities by type and retrieve them via get/require/has. ObservabilityHook is always registered with a NoOp default.

Supports constructor kwargs for convenience::

deps = BaseAgentDeps(ModelProvider=my_provider, DocumentStore=my_store)

Kwarg keys are resolved by matching against known capability types (Protocol classes) defined in this package.

Source code in src/symfonic/core/deps.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def __init__(self, **capabilities: Any) -> None:
    self._capabilities: dict[type, Any] = {}

    # Resolve kwargs like ModelProvider=impl to actual types
    if capabilities:
        type_registry = self._get_capability_type_registry()
        for key, impl in capabilities.items():
            cap_type = type_registry.get(key)
            if cap_type is None:
                raise TypeError(
                    f"Unknown capability name '{key}'. "
                    f"Known capabilities: {', '.join(sorted(type_registry))}"
                )
            self._capabilities[cap_type] = impl

    # Auto-register NoOpObservabilityHook if no ObservabilityHook provided
    self._ensure_default_observability()
    # Auto-register empty CallbackManager if none provided
    self._ensure_default_callbacks()

get 

get(capability_type: type) -> Any | None

Optional: returns impl or None. Never raises.

Source code in src/symfonic/core/deps.py 
124
125
126
def get(self, capability_type: type) -> Any | None:
    """Optional: returns impl or None. Never raises."""
    return self._capabilities.get(capability_type)

has 

has(capability_type: type) -> bool

Conditional: returns bool. Never raises.

Source code in src/symfonic/core/deps.py 
135
136
137
def has(self, capability_type: type) -> bool:
    """Conditional: returns bool. Never raises."""
    return capability_type in self._capabilities

list_capabilities 

list_capabilities() -> set[type]

Alias for registered_types(). Debug-friendly name.

Source code in src/symfonic/core/deps.py 
143
144
145
def list_capabilities(self) -> set[type]:
    """Alias for registered_types(). Debug-friendly name."""
    return self.registered_types()

register 

register(capability_type: type, impl: Any) -> BaseAgentDeps

Register a capability by its type. Returns self for chaining.

Source code in src/symfonic/core/deps.py 
119
120
121
122
def register(self, capability_type: type, impl: Any) -> BaseAgentDeps:
    """Register a capability by its type. Returns self for chaining."""
    self._capabilities[capability_type] = impl
    return self

registered_types 

registered_types() -> set[type]

Return all registered capability types.

Source code in src/symfonic/core/deps.py 
139
140
141
def registered_types(self) -> set[type]:
    """Return all registered capability types."""
    return set(self._capabilities.keys())

require 

require(capability_type: type) -> Any

Required: returns impl or raises MissingCapabilityError.

Source code in src/symfonic/core/deps.py 
128
129
130
131
132
133
def require(self, capability_type: type) -> Any:
    """Required: returns impl or raises MissingCapabilityError."""
    impl = self._capabilities.get(capability_type)
    if impl is None:
        raise MissingCapabilityError(capability_type)
    return impl

validate_required 

validate_required(
    required: list[type], context: str = ""
) -> None

Raises MissingCapabilityError listing all absent types.

Source code in src/symfonic/core/deps.py 
147
148
149
150
151
def validate_required(self, required: list[type], context: str = "") -> None:
    """Raises MissingCapabilityError listing all absent types."""
    missing = [t for t in required if not self.has(t)]
    if missing:
        raise MissingCapabilityError(missing)

MissingCapabilityError 

MissingCapabilityError(capabilities: type | list[type])

Bases: Exception

Raised when a required capability is not registered in BaseAgentDeps.

Accepts either a single type or a list of types (for bulk validation). Always produces a self-explanatory message with the registration hint.

Source code in src/symfonic/core/deps.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(self, capabilities: type | list[type]) -> None:
    if isinstance(capabilities, list):
        names = ", ".join(c.__name__ for c in capabilities)
        super().__init__(
            f"Missing capabilities: {names}. "
            f"Register via BaseAgentDeps({names}=impl) "
            f"or deps.register(CapabilityType, impl)"
        )
    else:
        cap = capabilities
        super().__init__(
            f"Missing capability: {cap.__name__}. "
            f"Register via BaseAgentDeps({cap.__name__}=impl) "
            f"or deps.register({cap.__name__}, your_impl)"
        )