Skip to content

symfonic.core

core

symfonic.core -- Agent Graph Framework core primitives (LOW-LEVEL).

Public API re-exports only. Internal modules should not be imported directly by consumer code.

Which layer do I want? * symfonic.core (this package) is the low-level engine: AgentGraph / AgentRuntime, tools, providers, and config primitives. Use it only when you are building your own orchestration. * symfonic.agent is the HIGH-LEVEL, batteries-included layer most users want: SymfonicAgent (memory hydration + consolidation, sub-agents, plugins, streaming, the FastAPI bridge). Start there: from symfonic.agent import SymfonicAgent. * Run symfonic guide (CLI) for the full decision tree and quickstarts.

AgentConfig dataclass 

AgentConfig(
    model: ModelConfig = ModelConfig(),
    compaction: CompactionConfig = CompactionConfig(),
    max_conversation_messages: int = 200,
    recursion_limit: int = 50,
    max_agent_depth: int = 2,
    ask_user_enabled: bool = False,
    role_models: dict[str, ModelConfig] = dict(),
    experimental_interrupt: bool = False,
    procedural_enforce_preconditions: bool = False,
)

Top-level agent configuration combining model and compaction settings.

with_anthropic classmethod 

with_anthropic(
    model_name: str = "claude-sonnet-4-5", **kwargs: object
) -> tuple[AgentConfig, AnthropicProvider]

Convenience constructor. Returns (config, provider) tuple.

Source code in src/symfonic/core/config.py 
205
206
207
208
209
210
211
212
213
214
@classmethod
def with_anthropic(
    cls,
    model_name: str = "claude-sonnet-4-5",
    **kwargs: object,
) -> tuple[AgentConfig, AnthropicProvider]:
    """Convenience constructor. Returns (config, provider) tuple."""
    from .providers import AnthropicProvider

    return cls(model=ModelConfig(model_name=model_name, **kwargs)), AnthropicProvider()

AgentEndEvent dataclass 

AgentEndEvent(
    run_id: str,
    final_response: str | None,
    duration_ms: float,
    node_count: int | None,
)

Emitted when agent execution completes.

In the streaming path, final_response and node_count are None because stream() does not accumulate final state.

AgentGraph 

AgentGraph(
    topology: str = "react_loop",
    state_class: type = BaseAgentState,
)

Graph definition factory. Accepts topology and compiles into CompiledGraph.

Not responsible for execution or configuration — that's AgentRuntime. Config is passed at compile time, not at construction time.

Source code in src/symfonic/core/graph.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    topology: str = "react_loop",
    state_class: type = BaseAgentState,
) -> None:
    if topology not in self._PRESET_REGISTRY:
        available = ", ".join(self._PRESET_REGISTRY.keys())
        raise ValueError(
            f"Unknown preset '{topology}'. Available presets: {available}"
        )
    self._topology = topology
    self._state_class = state_class
    self.registry = ToolRegistry()
    self._context_loaders: list[ContextLoaderRegistration] = []
    # v7.7.4: optional callable the engine registers to feed the
    # precondition_gate node the active procedural skills for the
    # current turn.  ``None`` (default) keeps the gate as a passive
    # passthrough even when the flag is on -- adopters who want
    # full enforcement either configure via the engine OR register
    # their own getter via ``set_active_skills_getter``.
    self._active_skills_getter: Any = None

add_context_loader 

add_context_loader(
    fn: Callable[..., Any],
    required: bool = False,
    timeout_seconds: float = 5.0,
    allow_collision_keys: set[str] | None = None,
) -> AgentGraph

Register a context loader. Returns self for chaining.

Source code in src/symfonic/core/graph.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def add_context_loader(
    self,
    fn: Callable[..., Any],
    required: bool = False,
    timeout_seconds: float = 5.0,
    allow_collision_keys: set[str] | None = None,
) -> AgentGraph:
    """Register a context loader. Returns self for chaining."""
    self._context_loaders.append(
        ContextLoaderRegistration(
            loader=fn,
            required=required,
            timeout_seconds=timeout_seconds,
            allow_collision_keys=allow_collision_keys or set(),
        )
    )
    return self

add_tool 

add_tool(tool: Any, **metadata: Any) -> AgentGraph

Register a single tool with metadata. Returns self for chaining.

Source code in src/symfonic/core/graph.py 
92
93
94
95
def add_tool(self, tool: Any, **metadata: Any) -> AgentGraph:
    """Register a single tool with metadata. Returns self for chaining."""
    self.registry.register(tool, **metadata)
    return self

add_tools 

add_tools(tools: list[Any]) -> AgentGraph

Register multiple tools. Returns self for chaining.

Source code in src/symfonic/core/graph.py 
87
88
89
90
def add_tools(self, tools: list[Any]) -> AgentGraph:
    """Register multiple tools. Returns self for chaining."""
    self.registry.register_many(tools)
    return self

compile 

compile(
    deps: BaseAgentDeps,
    config: AgentConfig | None = None,
    checkpointer: Any | None = None,
) -> Any

Validate capabilities, freeze registry, wire topology.

Parameters:

Name Type Description Default
deps BaseAgentDeps

Capability registry with registered providers.

 required
config AgentConfig | None

Agent configuration. Defaults to AgentConfig() if not provided.

 None
checkpointer Any | None

Optional LangGraph checkpointer for persistence/resume.

 None

Raises:

Type Description
ValueError

unknown preset name
ConfigurationError

capability flags violated
MissingCapabilityError

required capability absent from deps
Source code in src/symfonic/core/graph.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def compile(
    self,
    deps: BaseAgentDeps,
    config: AgentConfig | None = None,
    checkpointer: Any | None = None,
) -> Any:
    """Validate capabilities, freeze registry, wire topology.

    Args:
        deps: Capability registry with registered providers.
        config: Agent configuration. Defaults to AgentConfig() if not provided.
        checkpointer: Optional LangGraph checkpointer for persistence/resume.

    Raises:
        ValueError: unknown preset name
        ConfigurationError: capability flags violated
        MissingCapabilityError: required capability absent from deps
    """
    config = config or AgentConfig()
    preset = self._PRESET_REGISTRY[self._topology]

    # Validate required capabilities
    deps.validate_required(preset.required_capabilities, context=preset.name)

    # Validate capability flags
    if self.registry.all_tools() and not preset.supports_tools:
        raise ConfigurationError(
            f"Preset '{preset.name}' does not support tools, "
            f"but {len(self.registry)} tools are registered."
        )

    # Freeze registry
    self.registry.freeze()

    # Build state graph
    workflow = StateGraph(self._state_class)

    # Wire topology via preset
    preset(workflow, self, config)

    return workflow.compile(checkpointer=checkpointer)

register_preset classmethod 

register_preset(preset: GraphPreset) -> None

Register a custom preset globally.

Source code in src/symfonic/core/graph.py 
75
76
77
78
@classmethod
def register_preset(cls, preset: GraphPreset) -> None:
    """Register a custom preset globally."""
    cls._PRESET_REGISTRY[preset.name] = preset

set_active_skills_getter 

set_active_skills_getter(
    getter: Callable[[dict[str, Any]], Any],
) -> AgentGraph

Register the async callable the v7.7.4 :class:PreconditionGateNode uses to query active procedural skills.

Signature: async def getter(state: dict) -> list[skill_entry]. Typical implementation queries ProceduralLayer.query_skills(scope, "", include_drafts=False). The engine wires this automatically; callers building a graph by hand (no SymfonicAgent) register their own getter when opting into procedural_enforce_preconditions.

Returns self for chaining.

Source code in src/symfonic/core/graph.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def set_active_skills_getter(
    self, getter: Callable[[dict[str, Any]], Any],
) -> AgentGraph:
    """Register the async callable the v7.7.4
    :class:`PreconditionGateNode` uses to query active procedural
    skills.

    Signature: ``async def getter(state: dict) -> list[skill_entry]``.
    Typical implementation queries
    ``ProceduralLayer.query_skills(scope, "", include_drafts=False)``.
    The engine wires this automatically; callers building a graph
    by hand (no ``SymfonicAgent``) register their own getter when
    opting into ``procedural_enforce_preconditions``.

    Returns self for chaining.
    """
    self._active_skills_getter = getter
    return self

unregister_preset classmethod 

unregister_preset(name: str) -> None

Remove a custom preset. Raises KeyError if not found.

Source code in src/symfonic/core/graph.py 
80
81
82
83
84
85
@classmethod
def unregister_preset(cls, name: str) -> None:
    """Remove a custom preset. Raises KeyError if not found."""
    if name in ("react_loop", "plan_execute", "retrieval_augmented"):
        raise ValueError(f"Cannot unregister built-in preset '{name}'")
    del cls._PRESET_REGISTRY[name]

with_preset classmethod 

with_preset(name: str, **kwargs: Any) -> AgentGraph

Create an AgentGraph with a specific preset.

Source code in src/symfonic/core/graph.py 
70
71
72
73
@classmethod
def with_preset(cls, name: str, **kwargs: Any) -> AgentGraph:
    """Create an AgentGraph with a specific preset."""
    return cls(topology=name, **kwargs)

AgentRuntime 

AgentRuntime(
    graph: AgentGraph,
    deps: BaseAgentDeps,
    config: AgentConfig | None = None,
    timeout_seconds: float | None = None,
    checkpointer: Any | None = None,
)

Execute a compiled graph for a single agent invocation.

Eagerly compiles the graph in init -- raises immediately on misconfiguration (missing capabilities, invalid preset, etc.).

Source code in src/symfonic/core/runtime.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    graph: AgentGraph,
    deps: BaseAgentDeps,
    config: AgentConfig | None = None,
    timeout_seconds: float | None = None,
    checkpointer: Any | None = None,
) -> None:
    self._deps = deps
    self._config = config or AgentConfig()
    self._recursion_limit = self._config.recursion_limit
    self._timeout_seconds = timeout_seconds
    self._checkpointer = checkpointer
    self._workflow = graph.compile(deps, self._config, checkpointer=checkpointer)

run async 

run(
    query: str,
    *,
    callbacks: Sequence[CallbackHandler] | None = None,
    **state_overrides: Any,
) -> dict[str, Any]

Execute the graph and return final state dict with final_response.

v7.18.0 (Option B emission-ordering): the caller may pass _suppress_agent_end=True in state_overrides to disable the success-path on_agent_end dispatch in this method. The engine layer (SymfonicAgent.run) sets this flag so it can emit on_response_render AFTER the full transformation chain (synthesize-on-empty-final salvage -> fabrication -> metacog -> extraction-stripping) and THEN emit on_agent_end with the rewritten content. Composability rule (locked):

  • salvage -> fabrication -> metacog -> extraction-stripping -> on_response_render -> on_agent_end.
  • The rewriter runs at most ONCE per turn.

The error-path on_agent_end STILL fires from runtime -- engine cannot intercept the BaseException, and adopters relying on a forced-failure terminator event must still receive it. _suppress_agent_end is a runtime-internal flag and is NOT part of the public state_overrides surface; adopter code that passes arbitrary keys to runtime.run() ignores it per the legacy default-False contract.

Source code in src/symfonic/core/runtime.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
async def run(
    self,
    query: str,
    *,
    callbacks: Sequence[CallbackHandler] | None = None,
    **state_overrides: Any,
) -> dict[str, Any]:
    """Execute the graph and return final state dict with final_response.

    v7.18.0 (Option B emission-ordering): the caller may pass
    ``_suppress_agent_end=True`` in ``state_overrides`` to disable
    the success-path ``on_agent_end`` dispatch in this method.  The
    engine layer (``SymfonicAgent.run``) sets this flag so it can
    emit ``on_response_render`` AFTER the full transformation chain
    (synthesize-on-empty-final salvage -> fabrication -> metacog ->
    extraction-stripping) and THEN emit ``on_agent_end`` with the
    rewritten content.  Composability rule (locked):

    * salvage -> fabrication -> metacog -> extraction-stripping ->
      on_response_render -> on_agent_end.
    * The rewriter runs at most ONCE per turn.

    The error-path ``on_agent_end`` STILL fires from runtime --
    engine cannot intercept the BaseException, and adopters relying
    on a forced-failure terminator event must still receive it.
    ``_suppress_agent_end`` is a runtime-internal flag and is NOT
    part of the public ``state_overrides`` surface; adopter code
    that passes arbitrary keys to ``runtime.run()`` ignores it
    per the legacy default-False contract.
    """
    suppress_agent_end = bool(
        state_overrides.pop("_suppress_agent_end", False),
    )
    # v8.6.4 Ask 2(b): recursion-exhaustion salvage opt-in.  The engine
    # sets this flag when ``synthesize_on_empty_final`` is True so the
    # runtime can retain the in-flight state and recover gathered
    # ToolMessages when LangGraph hits ``recursion_limit`` -- instead of
    # propagating GraphRecursionError (which bypassed the engine's
    # normal-return salvage).  Popped here so it never reaches the graph
    # input state.  Default False -> the legacy ``ainvoke`` path runs
    # unchanged (byte-identical).
    recursion_salvage = bool(
        state_overrides.pop("_recursion_salvage", False),
    )
    input_state = self._build_input_state(query, **state_overrides)
    config = self._build_runnable_config(state_overrides)
    run_id = input_state["run_id"]
    mgr = self._resolve_callback_manager(callbacks)

    # Propagate merged manager to nodes via state so per-invocation
    # callbacks reach @observed_node (Finding 1).
    input_state["_callback_manager"] = mgr

    await mgr.on_agent_start(AgentStartEvent(query=query, run_id=run_id))
    start = time.monotonic()

    try:
        if recursion_salvage:
            # Drive via values-streaming so the last accumulated state
            # (including gathered ToolMessages) is retained when
            # ``recursion_limit`` is hit.  Equivalent final result to
            # ``ainvoke`` on the happy path -- values-mode yields the
            # full state after each super-step.
            result = await self._run_with_recursion_salvage(
                input_state, config,
            )
        else:
            coro = self._workflow.ainvoke(input_state, config=config)
            if self._timeout_seconds is not None:
                result = await asyncio.wait_for(
                    coro, timeout=self._timeout_seconds,
                )
            else:
                result = await coro
        duration_ms = (time.monotonic() - start) * 1000
        final_response = result.get("final_response")
        node_log = result.get("node_execution_log", [])

        if not suppress_agent_end:
            await mgr.on_agent_end(
                AgentEndEvent(
                    run_id=run_id,
                    final_response=final_response,
                    duration_ms=duration_ms,
                    node_count=len(node_log),
                )
            )

        return {
            "final_response": final_response,
            "messages": result.get("messages", []),
            "node_execution_log": node_log,
            # v7.18.0: expose runtime-leg latency + node_count so
            # the engine caller (which suppresses the runtime
            # ``on_agent_end``) can stamp them onto the
            # AgentEndEvent it emits post-transformation-chain.
            "_runtime_duration_ms": duration_ms,
            "_runtime_node_count": len(node_log),
            # v8.6.4 Ask 2(b): propagate the recursion-exhaustion
            # marker so the engine runs its synthesis salvage over the
            # gathered ToolMessages and emits the recursion_exhausted
            # ReactLoopEndEvent.  Absent (falsy) on the normal path.
            "_recursion_exhausted": result.get(
                "_recursion_exhausted", False,
            ),
        }
    except BaseException:
        duration_ms = (time.monotonic() - start) * 1000
        # Error-path ``on_agent_end`` fires unconditionally.  Even
        # when the caller asked to suppress success-path emission,
        # a forced-failure terminator must reach observability
        # consumers.  Engine cannot intercept BaseException so this
        # is the only reliable dispatch site for failure events.
        await asyncio.shield(
            mgr.on_agent_end(
                AgentEndEvent(
                    run_id=run_id,
                    final_response=None,
                    duration_ms=duration_ms,
                    node_count=None,
                )
            )
        )
        raise

stream async 

stream(
    query: str,
    *,
    callbacks: Sequence[CallbackHandler] | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[Any, None]

Yields stream events. Emits AgentStartEvent/AgentEndEvent via callbacks.

Source code in src/symfonic/core/runtime.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
async def stream(
    self,
    query: str,
    *,
    callbacks: Sequence[CallbackHandler] | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[Any, None]:
    """Yields stream events. Emits AgentStartEvent/AgentEndEvent via callbacks."""
    input_state = self._build_input_state(query, **state_overrides)
    config = self._build_runnable_config(state_overrides)
    run_id = input_state["run_id"]
    mgr = self._resolve_callback_manager(callbacks)

    # Propagate merged manager to nodes via state (Finding 1).
    input_state["_callback_manager"] = mgr

    await mgr.on_agent_start(AgentStartEvent(query=query, run_id=run_id))
    start = time.monotonic()

    exhausted = False
    try:
        if self._timeout_seconds is not None:
            deadline = asyncio.get_running_loop().time() + self._timeout_seconds
        else:
            deadline = None

        async for event in self._workflow.astream(input_state, config=config):
            if deadline is not None and asyncio.get_running_loop().time() > deadline:
                raise TimeoutError(
                    f"stream() exceeded timeout of {self._timeout_seconds}s"
                )
            yield event
        exhausted = True
    finally:
        duration_ms = (time.monotonic() - start) * 1000
        end_event = AgentEndEvent(
            run_id=run_id,
            final_response=None,
            duration_ms=duration_ms,
            node_count=None,
        )
        if exhausted:
            # Normal completion -- safe to await directly.
            await mgr.on_agent_end(end_event)
        else:
            # Early break (GeneratorExit) or CancelledError --
            # cannot reliably await inside an async generator
            # finally block, so schedule as a background task.
            # Keep a strong reference to prevent GC before completion.
            task = asyncio.get_running_loop().create_task(
                mgr.on_agent_end(end_event)
            )
            _background_tasks.add(task)
            task.add_done_callback(_background_tasks.discard)

stream_events async 

stream_events(
    query: str,
    *,
    callbacks: Sequence[CallbackHandler] | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[StreamEvent, None]

Yield typed StreamEvent objects from astream_events(v2).

Calls _workflow.astream_events and pipes each raw LangGraph event through EventTranspiler, yielding typed StreamEvent objects (TextDeltaEvent, ToolCallStartEvent, ThinkingDeltaEvent, etc.). AgentStartEvent / AgentEndEvent are dispatched via the callback manager as in stream().

Source code in src/symfonic/core/runtime.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
async def stream_events(
    self,
    query: str,
    *,
    callbacks: Sequence[CallbackHandler] | None = None,
    **state_overrides: Any,
) -> AsyncGenerator[StreamEvent, None]:
    """Yield typed StreamEvent objects from astream_events(v2).

    Calls ``_workflow.astream_events`` and pipes each raw LangGraph
    event through ``EventTranspiler``, yielding typed StreamEvent
    objects (TextDeltaEvent, ToolCallStartEvent, ThinkingDeltaEvent,
    etc.).  AgentStartEvent / AgentEndEvent are dispatched via the
    callback manager as in ``stream()``.
    """
    # v8.7.1 C3: a resume MUST re-enter the graph via a LangGraph
    # ``Command(resume=...)`` so the paused ``interrupt()`` returns the
    # payload instead of re-firing. When a ``_resume_command`` is present
    # we feed the Command to the graph as its input (LangGraph resumes
    # from the checkpoint) rather than a fresh input-state dict, while
    # still refreshing the per-invocation callback manager on state.
    resume_command = state_overrides.pop("_resume_command", None)
    input_state = self._build_input_state(query, **state_overrides)
    config = self._build_runnable_config(state_overrides)
    run_id = input_state["run_id"]
    mgr = self._resolve_callback_manager(callbacks)

    input_state["_callback_manager"] = mgr

    await mgr.on_agent_start(AgentStartEvent(query=query, run_id=run_id))
    start = time.monotonic()

    transpiler = EventTranspiler()
    exhausted = False
    try:
        if self._timeout_seconds is not None:
            deadline = asyncio.get_running_loop().time() + self._timeout_seconds
        else:
            deadline = None

        graph_input = (
            resume_command if resume_command is not None else input_state
        )
        raw_stream = self._workflow.astream_events(
            graph_input, version="v2", config=config
        )
        async for timestamped in transpiler.transpile(raw_stream):
            if deadline is not None and asyncio.get_running_loop().time() > deadline:
                raise TimeoutError(
                    f"stream_events() exceeded timeout of {self._timeout_seconds}s"
                )
            yield timestamped.event
        exhausted = True
    finally:
        duration_ms = (time.monotonic() - start) * 1000
        end_event = AgentEndEvent(
            run_id=run_id,
            final_response=None,
            duration_ms=duration_ms,
            node_count=None,
        )
        if exhausted:
            await mgr.on_agent_end(end_event)
        else:
            task = asyncio.get_running_loop().create_task(
                mgr.on_agent_end(end_event)
            )
            _background_tasks.add(task)
            task.add_done_callback(_background_tasks.discard)

AgentStartEvent dataclass 

AgentStartEvent(query: str, run_id: str)

Emitted when agent execution begins.

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)

BaseAgentState

Bases: TypedDict

Typed state contract for all nodes in the graph.

Used as a TypedDict schema for LangGraph's StateGraph. total=False makes all fields optional at construction time.

CallbackHandler

Bases: Protocol

Async callback protocol for agent graph lifecycle events.

Implement this protocol to receive notifications about agent start/end, node start/end/error, LLM invocation events, and token composition breakdowns during graph execution.

CallbackManager 

CallbackManager(handlers: Sequence[Any] = ())

Fan-out dispatcher for multiple CallbackHandler instances.

Satisfies the CallbackHandler protocol (composite pattern). When the handler list is empty, all methods return immediately (zero-cost).

Source code in src/symfonic/core/callbacks/manager.py 
49
50
def __init__(self, handlers: Sequence[Any] = ()) -> None:
    self._handlers: list[Any] = list(handlers)

is_noop property 

is_noop: bool

True when no handlers are registered (zero-cost path).

add 

add(handler: Any) -> None

Add a handler to the dispatch list.

Source code in src/symfonic/core/callbacks/manager.py 
52
53
54
def add(self, handler: Any) -> None:
    """Add a handler to the dispatch list."""
    self._handlers.append(handler)

describe_hook_handlers 

describe_hook_handlers(
    hook_name: str,
) -> list[tuple[str, bool]]

Introspection helper for adopter triage (v7.18.1).

Returns [(handler_qualname, implements_hook), ...] for every registered handler. Adopters debugging "which of my handlers is responsible for has_hook returning True?" call this to verify the registration without diffing internal state.

Example::

>>> mgr.describe_hook_handlers("on_response_render")
[('myadopter.SlackRenderCallback', True),
 ('symfonic.telemetry.MetricsHandler', False)]

Zero-cost when called outside a hot path -- intended for diagnostic logs and ad-hoc REPL introspection, not steady-state dispatch. The handler descriptor includes module + qualname so a forked CallbackManager handler shows up immediately as a non-symfonic module path.

Source code in src/symfonic/core/callbacks/manager.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def describe_hook_handlers(
    self, hook_name: str,
) -> list[tuple[str, bool]]:
    """Introspection helper for adopter triage (v7.18.1).

    Returns ``[(handler_qualname, implements_hook), ...]`` for every
    registered handler.  Adopters debugging "which of my handlers is
    responsible for ``has_hook`` returning ``True``?" call this to
    verify the registration without diffing internal state.

    Example::

        >>> mgr.describe_hook_handlers("on_response_render")
        [('myadopter.SlackRenderCallback', True),
         ('symfonic.telemetry.MetricsHandler', False)]

    Zero-cost when called outside a hot path -- intended for
    diagnostic logs and ad-hoc REPL introspection, not steady-state
    dispatch.  The handler descriptor includes module + qualname
    so a forked CallbackManager handler shows up immediately as a
    non-symfonic module path.
    """
    return [
        (
            f"{type(h).__module__}.{type(h).__qualname__}",
            hasattr(h, hook_name),
        )
        for h in self._handlers
    ]

has_hook 

has_hook(hook_name: str) -> bool

True when at least one handler implements the given hook.

Used at emission sites to skip event construction entirely for OPTIONAL hooks (e.g. on_llm_pre_call, on_user_correction) when no registered handler implements them. Preserves the zero-cost guarantee for optional extensions.

Source code in src/symfonic/core/callbacks/manager.py 
65
66
67
68
69
70
71
72
73
74
75
def has_hook(self, hook_name: str) -> bool:
    """True when at least one handler implements the given hook.

    Used at emission sites to skip event construction entirely for
    OPTIONAL hooks (e.g. ``on_llm_pre_call``, ``on_user_correction``)
    when no registered handler implements them.  Preserves the
    zero-cost guarantee for optional extensions.
    """
    if not self._handlers:
        return False
    return any(hasattr(h, hook_name) for h in self._handlers)

merge 

merge(other: CallbackManager) -> CallbackManager

Return a new CallbackManager combining both handler lists.

Source code in src/symfonic/core/callbacks/manager.py 
56
57
58
def merge(self, other: CallbackManager) -> CallbackManager:
    """Return a new CallbackManager combining both handler lists."""
    return CallbackManager(self._handlers + other._handlers)

on_agent_end async 

on_agent_end(event: AgentEndEvent) -> None

Dispatch AgentEndEvent to all handlers.

Source code in src/symfonic/core/callbacks/manager.py 
111
112
113
async def on_agent_end(self, event: AgentEndEvent) -> None:
    """Dispatch AgentEndEvent to all handlers."""
    await self._dispatch("on_agent_end", event)

on_agent_start async 

on_agent_start(event: AgentStartEvent) -> None

Dispatch AgentStartEvent to all handlers.

Source code in src/symfonic/core/callbacks/manager.py 
107
108
109
async def on_agent_start(self, event: AgentStartEvent) -> None:
    """Dispatch AgentStartEvent to all handlers."""
    await self._dispatch("on_agent_start", event)

on_before_tool_call async 

on_before_tool_call(
    event: Any, state: dict[str, Any]
) -> dict[str, Any] | None

Consult guard handlers before a tool executes; return a verdict.

The steering / guard seam (companion to the on_tool_call_dispatch rewriter). Handlers are consulted in registration order and the FIRST handler that returns a skip verdict wins — a guard veto short-circuits the chain. Handlers that return None / {"action": "proceed"} defer to the next handler.

Return value delivered to :class:InstrumentedToolNode:

  • None -> no guard skipped the call; execute normally.
  • {"action": "skip", "content": str, "is_error": bool} -> the tool node synthesizes a ToolMessage from content instead of executing. is_error defaults to True.

Malformed verdicts (missing/empty content on a skip) are dropped with a WARNING and treated as proceed so a buggy guard cannot silently swallow a call. A handler that raises is logged and skipped (the chain proceeds) — error isolation matches the other optional hooks.

Source code in src/symfonic/core/callbacks/manager.py 
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
async def on_before_tool_call(
    self,
    event: Any,
    state: dict[str, Any],
) -> dict[str, Any] | None:
    """Consult guard handlers before a tool executes; return a verdict.

    The steering / guard seam (companion to the ``on_tool_call_dispatch``
    rewriter). Handlers are consulted in registration order and the FIRST
    handler that returns a ``skip`` verdict wins — a guard veto
    short-circuits the chain. Handlers that return ``None`` /
    ``{"action": "proceed"}`` defer to the next handler.

    Return value delivered to :class:`InstrumentedToolNode`:

    * ``None`` -> no guard skipped the call; execute normally.
    * ``{"action": "skip", "content": str, "is_error": bool}`` -> the
      tool node synthesizes a ``ToolMessage`` from ``content`` instead of
      executing. ``is_error`` defaults to ``True``.

    Malformed verdicts (missing/empty ``content`` on a ``skip``) are
    dropped with a WARNING and treated as ``proceed`` so a buggy guard
    cannot silently swallow a call. A handler that raises is logged and
    skipped (the chain proceeds) — error isolation matches the other
    optional hooks.
    """
    if not self._handlers:
        return None
    for handler in self._handlers:
        method = getattr(handler, "on_before_tool_call", None)
        if method is None:
            continue
        try:
            result = await method(event, state)
        except Exception:
            logger.exception(
                "Callback handler %r raised in on_before_tool_call",
                handler,
            )
            continue
        if result is None:
            continue
        if not isinstance(result, dict):
            logger.warning(
                "Callback handler %r returned non-dict from "
                "on_before_tool_call (got %s); treating as proceed",
                handler,
                type(result).__name__,
            )
            continue
        action = result.get("action", "proceed")
        if action == "proceed":
            continue
        if action != "skip":
            logger.warning(
                "Callback handler %r returned unknown on_before_tool_call "
                "action %r; treating as proceed",
                handler,
                action,
            )
            continue
        content = result.get("content")
        if not isinstance(content, str) or not content:
            logger.warning(
                "Callback handler %r returned a 'skip' verdict without a "
                "non-empty string 'content'; treating as proceed",
                handler,
            )
            continue
        # First skip wins — a guard veto short-circuits the chain.
        return {
            "action": "skip",
            "content": content,
            "is_error": bool(result.get("is_error", True)),
        }
    return None

on_fabrication_detected async 

on_fabrication_detected(
    event: FabricationDetectedEvent,
) -> None

Dispatch FabricationDetectedEvent to handlers that implement it.

on_fabrication_detected is an OPTIONAL hook (the dispatch was implicit in v7.0.6 via raw getattr; v7.0.7 promotes it to a typed event with proper manager-level fan-out). Handlers that do not implement the method are silently skipped. Preserves the zero-cost invariant: callers gate construction of :class:~symfonic.core.contracts.callbacks.FabricationDetectedEvent on :meth:has_hook ("on_fabrication_detected") so no event objects are built when no handler listens.

Source code in src/symfonic/core/callbacks/manager.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
async def on_fabrication_detected(
    self, event: FabricationDetectedEvent,
) -> None:
    """Dispatch FabricationDetectedEvent to handlers that implement it.

    ``on_fabrication_detected`` is an OPTIONAL hook (the dispatch
    was implicit in v7.0.6 via raw ``getattr``; v7.0.7 promotes it
    to a typed event with proper manager-level fan-out).  Handlers
    that do not implement the method are silently skipped.
    Preserves the zero-cost invariant: callers gate construction
    of :class:`~symfonic.core.contracts.callbacks.FabricationDetectedEvent`
    on :meth:`has_hook` ``("on_fabrication_detected")`` so no
    event objects are built when no handler listens.
    """
    await self._dispatch_optional("on_fabrication_detected", event)

on_llm_end async 

on_llm_end(event: LLMEndEvent) -> None

Dispatch LLMEndEvent to all handlers.

Source code in src/symfonic/core/callbacks/manager.py 
141
142
143
async def on_llm_end(self, event: LLMEndEvent) -> None:
    """Dispatch LLMEndEvent to all handlers."""
    await self._dispatch("on_llm_end", event)

on_llm_pre_call async 

on_llm_pre_call(event: LLMPreCallEvent) -> None

Dispatch LLMPreCallEvent to handlers that implement it.

on_llm_pre_call is an OPTIONAL hook (added in v6.1.5); not every handler implements it, so we fan out only to handlers that define the attribute. Preserves the zero-cost invariant: if no handler implements the hook the loop is a no-op.

Source code in src/symfonic/core/callbacks/manager.py 
131
132
133
134
135
136
137
138
139
async def on_llm_pre_call(self, event: LLMPreCallEvent) -> None:
    """Dispatch LLMPreCallEvent to handlers that implement it.

    ``on_llm_pre_call`` is an OPTIONAL hook (added in v6.1.5); not
    every handler implements it, so we fan out only to handlers that
    define the attribute.  Preserves the zero-cost invariant: if no
    handler implements the hook the loop is a no-op.
    """
    await self._dispatch_optional("on_llm_pre_call", event)

on_llm_start async 

on_llm_start(event: LLMStartEvent) -> None

Dispatch LLMStartEvent to all handlers.

Source code in src/symfonic/core/callbacks/manager.py 
127
128
129
async def on_llm_start(self, event: LLMStartEvent) -> None:
    """Dispatch LLMStartEvent to all handlers."""
    await self._dispatch("on_llm_start", event)

on_node_end async 

on_node_end(event: NodeEndEvent) -> None

Dispatch NodeEndEvent to all handlers.

Source code in src/symfonic/core/callbacks/manager.py 
119
120
121
async def on_node_end(self, event: NodeEndEvent) -> None:
    """Dispatch NodeEndEvent to all handlers."""
    await self._dispatch("on_node_end", event)

on_node_error async 

on_node_error(event: NodeErrorEvent) -> None

Dispatch NodeErrorEvent to all handlers.

Source code in src/symfonic/core/callbacks/manager.py 
123
124
125
async def on_node_error(self, event: NodeErrorEvent) -> None:
    """Dispatch NodeErrorEvent to all handlers."""
    await self._dispatch("on_node_error", event)

on_node_start async 

on_node_start(event: NodeStartEvent) -> None

Dispatch NodeStartEvent to all handlers.

Source code in src/symfonic/core/callbacks/manager.py 
115
116
117
async def on_node_start(self, event: NodeStartEvent) -> None:
    """Dispatch NodeStartEvent to all handlers."""
    await self._dispatch("on_node_start", event)

on_procedure_selection async 

on_procedure_selection(
    event: ProcedureSelectionEvent,
) -> None

Dispatch ProcedureSelectionEvent to handlers that implement it (v7.20.0 T-7.20.0.9).

on_procedure_selection is an OPTIONAL hook; fans out only to handlers that define the attribute. Preserves the zero- cost invariant: the engine's force-resolver gates event construction on has_hook("on_procedure_selection") so no event objects are built when no handler listens.

Adopters subscribe to capture per-procedure billing attribution, A/B-test bucket assignment, audit trail entries, and production routing telemetry. See :class:~symfonic.core.contracts.callbacks.ProcedureSelectionEvent.

Source code in src/symfonic/core/callbacks/manager.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
async def on_procedure_selection(
    self, event: ProcedureSelectionEvent,
) -> None:
    """Dispatch ProcedureSelectionEvent to handlers that implement
    it (v7.20.0 T-7.20.0.9).

    ``on_procedure_selection`` is an OPTIONAL hook; fans out only
    to handlers that define the attribute.  Preserves the zero-
    cost invariant: the engine's force-resolver gates event
    construction on ``has_hook("on_procedure_selection")`` so no
    event objects are built when no handler listens.

    Adopters subscribe to capture per-procedure billing
    attribution, A/B-test bucket assignment, audit trail entries,
    and production routing telemetry.  See
    :class:`~symfonic.core.contracts.callbacks.ProcedureSelectionEvent`.
    """
    await self._dispatch_optional("on_procedure_selection", event)

on_react_loop_end async 

on_react_loop_end(event: ReactLoopEndEvent) -> None

Dispatch ReactLoopEndEvent to handlers that implement it (v7.17.0).

on_react_loop_end is an OPTIONAL hook fired by the React node when the loop terminates with an empty final_response AFTER the v7.17.0 content-block salvage attempt. Handlers that do not implement it are silently skipped. Preserves the zero-cost invariant: the react node gates event construction on :meth:has_hook ("on_react_loop_end") so no event objects are built when no handler listens.

Source code in src/symfonic/core/callbacks/manager.py 
181
182
183
184
185
186
187
188
189
190
191
192
async def on_react_loop_end(self, event: ReactLoopEndEvent) -> None:
    """Dispatch ReactLoopEndEvent to handlers that implement it (v7.17.0).

    ``on_react_loop_end`` is an OPTIONAL hook fired by the React node
    when the loop terminates with an empty ``final_response`` AFTER
    the v7.17.0 content-block salvage attempt.  Handlers that do not
    implement it are silently skipped.  Preserves the zero-cost
    invariant: the react node gates event construction on
    :meth:`has_hook` ``("on_react_loop_end")`` so no event objects
    are built when no handler listens.
    """
    await self._dispatch_optional("on_react_loop_end", event)

on_response_render async 

on_response_render(
    content: str,
    state: dict[str, Any],
    metadata: ResponseRenderEvent,
) -> str

Dispatch ResponseRenderEvent to handlers that implement it.

Unique among optional hooks: on_response_render RETURNS the rewritten string. The runtime substitutes the returned value for the original final_response BEFORE on_agent_end dispatch.

Chaining semantics for multiple handlers (registration order):

  • First handler receives content=original_content.
  • Each subsequent handler receives the previous handler's return value.
  • The final return value is delivered to the runtime.

Error isolation: if a handler raises, the exception is logged and that handler's contribution is SKIPPED (the chain proceeds with the previous content). This matches the error-isolation contract of :meth:_dispatch_optional -- a broken rewriter must not stall the runtime.

Handlers that do not implement the method are silently skipped. Preserves the zero-cost invariant: emission sites gate construction of :class:~symfonic.core.contracts.callbacks. ResponseRenderEvent on :meth:has_hook ("on_response_render") so no event objects are built when no handler listens.

v7.18.0 (Option B; Jarvio Slack-formatting adopter ask).

Source code in src/symfonic/core/callbacks/manager.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
async def on_response_render(
    self,
    content: str,
    state: dict[str, Any],
    metadata: ResponseRenderEvent,
) -> str:
    """Dispatch ResponseRenderEvent to handlers that implement it.

    Unique among optional hooks: ``on_response_render`` RETURNS
    the rewritten string.  The runtime substitutes the returned
    value for the original ``final_response`` BEFORE ``on_agent_end``
    dispatch.

    Chaining semantics for multiple handlers (registration order):

    * First handler receives ``content=original_content``.
    * Each subsequent handler receives the previous handler's
      return value.
    * The final return value is delivered to the runtime.

    Error isolation: if a handler raises, the exception is logged
    and that handler's contribution is SKIPPED (the chain proceeds
    with the previous content).  This matches the error-isolation
    contract of :meth:`_dispatch_optional` -- a broken rewriter
    must not stall the runtime.

    Handlers that do not implement the method are silently skipped.
    Preserves the zero-cost invariant: emission sites gate
    construction of :class:`~symfonic.core.contracts.callbacks.
    ResponseRenderEvent` on :meth:`has_hook`
    ``("on_response_render")`` so no event objects are built when
    no handler listens.

    v7.18.0 (Option B; Jarvio Slack-formatting adopter ask).
    """
    if not self._handlers:
        return content
    current = content
    for handler in self._handlers:
        method = getattr(handler, "on_response_render", None)
        if method is None:
            continue
        try:
            result = await method(current, state, metadata)
        except Exception:
            logger.exception(
                "Callback handler %r raised in on_response_render",
                handler,
            )
            continue
        # Defensive type contract: handlers MUST return a string.
        # A non-string return drops the rewriter's contribution
        # silently and proceeds with the previous content -- the
        # runtime never delivers a non-string final_response to
        # adopters.
        if isinstance(result, str):
            current = result
        else:
            logger.warning(
                "Callback handler %r returned non-string from "
                "on_response_render (got %s); skipping rewrite",
                handler,
                type(result).__name__,
            )
    return current

on_token_composition async 

on_token_composition(event: TokenCompositionEvent) -> None

Dispatch TokenCompositionEvent to all handlers.

Source code in src/symfonic/core/callbacks/manager.py 
145
146
147
async def on_token_composition(self, event: TokenCompositionEvent) -> None:
    """Dispatch TokenCompositionEvent to all handlers."""
    await self._dispatch("on_token_composition", event)

on_tool_call_dispatch async 

on_tool_call_dispatch(
    event: ToolCallDispatchEvent, state: dict[str, Any]
) -> dict[str, Any] | None

Dispatch ToolCallDispatchEvent to handlers that implement it.

Unique among optional hooks (alongside on_response_render): on_tool_call_dispatch RETURNS the rewritten {"tool_name", "args"} dict. The React node mutates ai_message.tool_calls in place before the AIMessage returns to LangGraph for dispatch.

Return-value semantics:

  • None -> pass-through (no rewrite contribution).
  • dict[str, Any] with keys {"tool_name": str, "args": dict} -> rewrite. The dict MUST NOT contain call_id -- the engine owns it. tool_name MUST be a string; args MUST be a dict. Malformed returns are dropped with a structured WARNING (see the type-validation block below).

Chaining semantics for multiple handlers (registration order):

  • First handler receives the original event (original tool_name / args).
  • Each handler that returns a dict re-emits a SHALLOW-COPIED event with the rewritten tool_name / args so the next handler sees the chain's current state.
  • Handlers that return None are pass-throughs (no contribution; the next handler sees the previous state).
  • The final return value is delivered to the React node.

Error isolation: a handler that raises is logged and SKIPPED (the chain proceeds with the previous state). Matches the error-isolation contract of _dispatch_optional -- a broken rewriter must not stall the React loop.

Validation gate: the React node verifies the final tool_name against event.available_tools and DROPS rewrites that target unregistered tools. This method does NOT enforce the gate -- it lives at the call site so the dropped-rewrite WARNING carries the React node context.

Handlers that do not implement the method are silently skipped. Preserves the zero-cost invariant: emission sites gate construction of :class:~symfonic.core.contracts.callbacks. ToolCallDispatchEvent on :meth:has_hook ("on_tool_call_dispatch") so no event objects are built when no handler listens.

v7.19.0 (Jarvio SL-02 CSV-export -> slack_upload_file + SL-04 per-user memory title shape).

Source code in src/symfonic/core/callbacks/manager.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
async def on_tool_call_dispatch(
    self,
    event: ToolCallDispatchEvent,
    state: dict[str, Any],
) -> dict[str, Any] | None:
    """Dispatch ToolCallDispatchEvent to handlers that implement it.

    Unique among optional hooks (alongside ``on_response_render``):
    ``on_tool_call_dispatch`` RETURNS the rewritten ``{"tool_name",
    "args"}`` dict.  The React node mutates ``ai_message.tool_calls``
    in place before the AIMessage returns to LangGraph for dispatch.

    Return-value semantics:

    * ``None`` -> pass-through (no rewrite contribution).
    * ``dict[str, Any]`` with keys ``{"tool_name": str, "args": dict}``
      -> rewrite.  The dict MUST NOT contain ``call_id`` -- the
      engine owns it.  ``tool_name`` MUST be a string; ``args``
      MUST be a dict.  Malformed returns are dropped with a
      structured WARNING (see the type-validation block below).

    Chaining semantics for multiple handlers (registration order):

    * First handler receives the original event (original
      ``tool_name`` / ``args``).
    * Each handler that returns a dict re-emits a SHALLOW-COPIED
      event with the rewritten ``tool_name`` / ``args`` so the
      next handler sees the chain's current state.
    * Handlers that return ``None`` are pass-throughs (no
      contribution; the next handler sees the previous state).
    * The final return value is delivered to the React node.

    Error isolation: a handler that raises is logged and SKIPPED
    (the chain proceeds with the previous state).  Matches the
    error-isolation contract of ``_dispatch_optional`` -- a broken
    rewriter must not stall the React loop.

    Validation gate: the React node verifies the final
    ``tool_name`` against ``event.available_tools`` and DROPS
    rewrites that target unregistered tools.  This method does
    NOT enforce the gate -- it lives at the call site so the
    dropped-rewrite WARNING carries the React node context.

    Handlers that do not implement the method are silently
    skipped.  Preserves the zero-cost invariant: emission sites
    gate construction of :class:`~symfonic.core.contracts.callbacks.
    ToolCallDispatchEvent` on :meth:`has_hook`
    ``("on_tool_call_dispatch")`` so no event objects are built
    when no handler listens.

    v7.19.0 (Jarvio SL-02 CSV-export -> slack_upload_file +
    SL-04 per-user memory title shape).
    """
    if not self._handlers:
        return None
    from dataclasses import replace as _dc_replace

    current_event = event
    current_rewrite: dict[str, Any] | None = None
    for handler in self._handlers:
        method = getattr(handler, "on_tool_call_dispatch", None)
        if method is None:
            continue
        try:
            result = await method(current_event, state)
        except Exception:
            logger.exception(
                "Callback handler %r raised in on_tool_call_dispatch",
                handler,
            )
            continue
        if result is None:
            # Pass-through -- no rewrite contribution.  The next
            # handler sees the previous state (which may itself
            # be the original event if no prior handler rewrote).
            continue
        # Defensive type validation -- a malformed return drops
        # the contribution silently so the chain can proceed.
        if not isinstance(result, dict):
            logger.warning(
                "Callback handler %r returned non-dict from "
                "on_tool_call_dispatch (got %s); skipping rewrite",
                handler,
                type(result).__name__,
            )
            continue
        _new_name = result.get("tool_name")
        _new_args = result.get("args")
        if not isinstance(_new_name, str) or not isinstance(
            _new_args, dict
        ):
            logger.warning(
                "Callback handler %r returned malformed "
                "on_tool_call_dispatch rewrite (tool_name=%r, "
                "args=%r); skipping",
                handler,
                type(_new_name).__name__,
                type(_new_args).__name__,
            )
            continue
        if "call_id" in result:
            logger.warning(
                "Callback handler %r included 'call_id' in "
                "on_tool_call_dispatch rewrite -- the engine owns "
                "call_id; dropping the key (rewrite proceeds with "
                "the original call_id)",
                handler,
            )
        current_rewrite = {
            "tool_name": _new_name,
            "args": _new_args,
        }
        # Re-emit the event with the rewritten fields so chained
        # handlers see the in-progress state, not the stale
        # original.  ``replace`` clones the frozen dataclass.
        current_event = _dc_replace(
            current_event,
            tool_name=_new_name,
            args=_new_args,
        )
    return current_rewrite

on_tool_result async 

on_tool_result(event: ToolResultCallbackEvent) -> None

Dispatch ToolResultCallbackEvent to handlers that implement it (v8.6.4).

on_tool_result is an OPTIONAL hook fired by the tool node (InstrumentedToolNode) AFTER a tool returns, carrying the produced result, call_id, and a deferred flag. Handlers that do not implement it are silently skipped via :meth:_dispatch_optional. Preserves the zero-cost invariant: the emit site gates event construction on :meth:has_hook ("on_tool_result") so no event objects are built when no handler listens.

Source code in src/symfonic/core/callbacks/manager.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
async def on_tool_result(self, event: ToolResultCallbackEvent) -> None:
    """Dispatch ToolResultCallbackEvent to handlers that implement it (v8.6.4).

    ``on_tool_result`` is an OPTIONAL hook fired by the tool node
    (``InstrumentedToolNode``) AFTER a tool returns, carrying the
    produced result, call_id, and a ``deferred`` flag.  Handlers that
    do not implement it are silently skipped via
    :meth:`_dispatch_optional`.  Preserves the zero-cost invariant:
    the emit site gates event construction on
    :meth:`has_hook` ``("on_tool_result")`` so no event objects are
    built when no handler listens.
    """
    await self._dispatch_optional("on_tool_result", event)

on_tool_routing_decision async 

on_tool_routing_decision(
    event: ToolRoutingDecisionEvent,
) -> None

Dispatch ToolRoutingDecisionEvent to handlers that implement it.

on_tool_routing_decision is an OPTIONAL hook (added in v7.0.1); fans out only to handlers that define the attribute. Preserves the zero-cost invariant: when tool_routing_mode is off the engine never constructs the event OR calls this method.

Source code in src/symfonic/core/callbacks/manager.py 
149
150
151
152
153
154
155
156
157
158
159
160
async def on_tool_routing_decision(
    self, event: ToolRoutingDecisionEvent,
) -> None:
    """Dispatch ToolRoutingDecisionEvent to handlers that implement it.

    ``on_tool_routing_decision`` is an OPTIONAL hook (added in
    v7.0.1); fans out only to handlers that define the attribute.
    Preserves the zero-cost invariant: when ``tool_routing_mode``
    is ``off`` the engine never constructs the event OR calls
    this method.
    """
    await self._dispatch_optional("on_tool_routing_decision", event)

CompactionConfig dataclass 

CompactionConfig(
    context_window_tokens: int | None = None,
    max_history_share: float = 0.5,
    chars_per_token: int = 4,
    keep_recent_messages: int = 10,
    max_chunk_tokens: int = 30000,
    compaction_model: str = "claude-haiku-4-5",
    soft_threshold_tokens: int = 4000,
    enable_memory_flush: bool = True,
)

Configuration for conversation compaction/summarization.

context_window_tokens is now optional (v7.1.2). When None (the new default), ContextWindowNode falls back to ModelConfig.context_window if set, then to the 200_000 safety net (Claude-sized) -- this gives non-Claude callers a way to pin the right budget without overriding CompactionConfig. Set explicitly (e.g. CompactionConfig(context_window_tokens=50_000)) to force a tighter or wider ceiling than the model declares.

ContextCollisionError

Bases: ValueError

Raised when two context loaders write the same state key.

GraphPreset

Bases: Protocol

Protocol for graph topology presets.

LLMLifecycleHook

Bases: Protocol

Async instrumentation for LLM call start/end lifecycle.

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)"
        )

ModelConfig dataclass 

ModelConfig(
    model_name: str = "claude-sonnet-4-5",
    temperature: float = 1.0,
    max_tokens: int = 16384,
    max_retries: int = 5,
    top_p: float | None = None,
    top_k: int | None = None,
    thinking: dict[str, object] | None = None,
    extra_headers: dict[str, str] = dict(),
    context_window: int | None = None,
    timeout_seconds: float | None = None,
    http_client: Any | None = None,
)

LLM model configuration. Provider-agnostic.

context_window (added in v7.1.2) declares the model's native context-window budget in tokens. When set, ContextWindowNode uses it as the hard ceiling unless CompactionConfig.context_window_tokens explicitly overrides it. None (default) preserves pre-v7.1.2 behaviour: the node falls through to the 200_000 safety net. Typical values: Claude 3/4 = 200_000, GPT-4 Turbo = 128_000, GPT-4 = 8_192, Llama-3 32k = 32_000.

http_client class-attribute instance-attribute 

http_client: Any | None = None

Underlying HTTP client (typically httpx.AsyncClient) for true per-chunk idle bound. Threaded to providers whose LangChain class accepts the kwarg -- http_async_client on OpenAI / DeepSeek / Kimi. Providers whose class does NOT accept it (Anthropic, Google, Ollama at the time of v7.25.0) emit a one-shot :class:HttpClientUnsupportedWarning on first encounter and ignore the field; timeout_seconds still threads. Typed Any so symfonic-core does not import httpx at module load -- adopters who don't pin a client pay no import cost. None (default) preserves the provider's own default client lifecycle.

timeout_seconds class-attribute instance-attribute 

timeout_seconds: float | None = None

Total-request timeout in seconds. Threaded to the provider's LangChain-class kwarg whose name varies by provider: default_request_timeout on Anthropic, timeout on OpenAI / DeepSeek / Kimi / Google. Coarse-grained; bounds the entire HTTP round-trip but does NOT enforce per-chunk idle. None (default) leaves the provider's own default in place (typically 600 for the OpenAI family; httpx-driven for Anthropic).

top_k class-attribute instance-attribute 

top_k: int | None = None

Top-k sampling: sample only from the k most likely tokens. Honoured by Anthropic, Google, and Ollama (their LangChain classes expose top_k). The OpenAI chat API has NO top_k parameter, so OpenAI / DeepSeek / Kimi ignore it and emit a one-shot warning. None leaves the provider default.

top_p class-attribute instance-attribute 

top_p: float | None = None

Nucleus sampling: keep the smallest token set whose cumulative probability >= top_p (0-1). Threaded to every provider's LangChain class (all accept top_p). None leaves the provider default. Note: most providers advise tuning EITHER temperature OR top_p, not both.

ModelProvider

Bases: Protocol

Protocol for LLM model construction.

supports_forced_tool_choice 

supports_forced_tool_choice(config: ModelConfig) -> bool

v7.8.5: provider-level introspection for the v7.8.3 procedural_force_first_action_tool lever.

Returns True when the provider can honour llm.bind_tools(tools, tool_choice=<name>) under the given ModelConfig. Returns False when the provider would reject forced tool_choice (Anthropic's API returns 400 when thinking is enabled alongside tool_choice={type:tool,...}; see Anthropic's published extended-thinking constraint).

Default implementation returns True -- providers that have a no-force constraint override. The engine's _maybe_resolve_forced_tool_choice consults this BEFORE stamping forced_tool_choice into the LangGraph state, so adopters who configure thinking see a clean WARN log and a forced no-op rather than a runtime 400 from the API boundary.

Implementations MUST be pure functions of the passed ModelConfig -- no I/O, no provider state mutation -- so the engine can call this on every turn without cost.

v7.8.3 architectural-line preservation: the framework constrains the choice set when it can; abstains cleanly when it cannot. Auto-disabling config.thinking would cross the line (reasoning is model-owned).

Source code in src/symfonic/core/providers.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def supports_forced_tool_choice(self, config: ModelConfig) -> bool:
    """v7.8.5: provider-level introspection for the v7.8.3
    ``procedural_force_first_action_tool`` lever.

    Returns ``True`` when the provider can honour
    ``llm.bind_tools(tools, tool_choice=<name>)`` under the given
    ``ModelConfig``.  Returns ``False`` when the provider would
    reject forced tool_choice (Anthropic's API returns ``400`` when
    ``thinking`` is enabled alongside ``tool_choice={type:tool,...}``;
    see Anthropic's published extended-thinking constraint).

    Default implementation returns ``True`` -- providers that have
    a no-force constraint override.  The engine's
    ``_maybe_resolve_forced_tool_choice`` consults this BEFORE
    stamping ``forced_tool_choice`` into the LangGraph state, so
    adopters who configure ``thinking`` see a clean WARN log and
    a forced no-op rather than a runtime 400 from the API
    boundary.

    Implementations MUST be pure functions of the passed
    ``ModelConfig`` -- no I/O, no provider state mutation -- so
    the engine can call this on every turn without cost.

    v7.8.3 architectural-line preservation: the framework
    constrains the choice set when it can; abstains cleanly when
    it cannot.  Auto-disabling ``config.thinking`` would cross the
    line (reasoning is model-owned).
    """
    return True

NodeEndEvent dataclass 

NodeEndEvent(
    node_name: str, run_id: str, duration_ms: float
)

Emitted after a graph node completes successfully.

NodeErrorEvent dataclass 

NodeErrorEvent(
    node_name: str, run_id: str, error: str, error_type: str
)

Emitted when a graph node raises an exception.

NodeLifecycleHook

Bases: Protocol

Async instrumentation for graph-node start/end/error lifecycle.

NodeStartEvent dataclass 

NodeStartEvent(node_name: str, run_id: str)

Emitted before a graph node executes.

ObservabilityHook

Bases: NodeLifecycleHook, LLMLifecycleHook, ToolLifecycleHook, Protocol

Composed protocol -- union of all lifecycle hooks.

Backward-compatible: any class implementing all 7 methods satisfies this protocol. New consumers should prefer the granular protocols when they only need a subset of methods.

TenantScope

Bases: BaseModel

Immutable hierarchical N-level multi-tenant request scope.

Canonical representation is the root-first path (path[0] broadest, path[-1] most-specific). The legacy tenant_id / sub_tenant_id fields are RETAINED for backward compatibility and, when no explicit path override is supplied, ARE the 1-/2-level path. When an explicit N-level path is supplied via the factories, tenant_id / sub_tenant_id read back off the path.

Isolation (design §5.d): a query at path P sees a memory at path Q iff Q is a prefix of P. Enforced via :meth:ancestor_prefix_paths exact-IN at the backend, NEVER a string prefix scan.

path property 

path: tuple[ScopeLevel, ...]

Root-first ordered path (canonical hierarchical representation).

Derived from the flat fields when no explicit override is set
  • tenant_id only → 1-level (tenant:tenant_id,)
    • sub_tenant_id → 2-level (tenant:tenant_id, sub_tenant:sub)

scope_depth property 

scope_depth: int

Number of levels in the path (1-based; root-only path is depth 1).

scope_path property 

scope_path: str

Materialised US-delimited, root-first path string for the backend.

org\x1facme\x1fbrand\x1fmyhalos\x1fconversation\x1fc-123. This is the stored column/field value (design §5.b).

ancestor_prefix_paths 

ancestor_prefix_paths() -> list[str]

The exact-IN list for the isolation filter (design §5.b/§5.d).

Returns the materialised scope_path string of EVERY prefix of this scope's path (root → … → self), inclusive. A query at path P uses this list as WHERE scope_path = ANY(list) so it matches a stored memory at path Q iff Q is a prefix of P — exact set membership, never a LIKE 'prefix%' scan (which would leak acme-corp to acme).

Source code in src/symfonic/core/scope.py 
340
341
342
343
344
345
346
347
348
349
350
351
352
def ancestor_prefix_paths(self) -> list[str]:
    """The exact-IN list for the isolation filter (design §5.b/§5.d).

    Returns the materialised ``scope_path`` string of EVERY prefix of this
    scope's path (root → … → self), inclusive.  A query at path P uses
    this list as ``WHERE scope_path = ANY(list)`` so it matches a stored
    memory at path Q iff Q is a prefix of P — exact set membership, never
    a ``LIKE 'prefix%'`` scan (which would leak ``acme-corp`` to ``acme``).
    """
    levels = self.path
    return [
        scope_path_for(levels[: i + 1]) for i in range(len(levels))
    ]

child 

child(kind: str, id: str) -> TenantScope

Return a new scope with one more (deeper) level appended.

org_scope.child("brand", "myhalos").child("conversation", "c-123") builds the full hierarchy. Namespace and inherits_from are preserved.

The result preserves the receiver's concrete type (type(self)) so chaining .child() off a subclass such as :class:~symfonic.agent.types.FrameworkTenantScope keeps the subclass (and its converter methods) instead of silently downcasting to the base TenantScope (P1 — subclass-drop on chaining).

Source code in src/symfonic/core/scope.py 
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def child(self, kind: str, id: str) -> TenantScope:
    """Return a new scope with one more (deeper) level appended.

    ``org_scope.child("brand", "myhalos").child("conversation", "c-123")``
    builds the full hierarchy.  Namespace and inherits_from are preserved.

    The result preserves the receiver's concrete type (``type(self)``) so
    chaining ``.child()`` off a subclass such as
    :class:`~symfonic.agent.types.FrameworkTenantScope` keeps the subclass
    (and its converter methods) instead of silently downcasting to the
    base ``TenantScope`` (P1 — subclass-drop on chaining).
    """
    new_path = (*self.path, ScopeLevel(kind=kind, id=id))
    return type(self)(
        tenant_id=new_path[0].id,
        sub_tenant_id=new_path[1].id if len(new_path) > 1 else None,
        namespace=self.namespace,
        inherits_from=self.inherits_from,
        path_override=new_path,
    )

from_path classmethod 

from_path(
    path: list[ScopeLevel] | tuple[ScopeLevel, ...],
    *,
    namespace: str | None = None,
) -> TenantScope

Construct an N-level scope from an explicit root-first path.

Source code in src/symfonic/core/scope.py 
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
@classmethod
def from_path(
    cls,
    path: list[ScopeLevel] | tuple[ScopeLevel, ...],
    *,
    namespace: str | None = None,
) -> TenantScope:
    """Construct an N-level scope from an explicit root-first path."""
    levels = tuple(path)
    if not levels:
        raise ValueError("TenantScope.from_path requires a non-empty path")
    tenant_id = levels[0].id
    sub_tenant_id = levels[1].id if len(levels) > 1 else None
    return cls(
        tenant_id=tenant_id,
        sub_tenant_id=sub_tenant_id,
        namespace=namespace,
        path_override=levels,
    )

from_request classmethod 

from_request(
    tenant_id: str,
    *,
    sub_tenant_id: str | None = None,
    namespace: str | None = None,
    path: list[ScopeLevel]
    | tuple[ScopeLevel, ...]
    | None = None,
) -> TenantScope

Construct a scope from request-edge fields (blessed adopter entry).

Supply path for N-level hierarchies, or the flat fields for the legacy 1-/2-level shape.

Source code in src/symfonic/core/scope.py 
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
@classmethod
def from_request(
    cls,
    tenant_id: str,
    *,
    sub_tenant_id: str | None = None,
    namespace: str | None = None,
    path: list[ScopeLevel] | tuple[ScopeLevel, ...] | None = None,
) -> TenantScope:
    """Construct a scope from request-edge fields (blessed adopter entry).

    Supply ``path`` for N-level hierarchies, or the flat fields for the
    legacy 1-/2-level shape.
    """
    if path is not None:
        return cls.from_path(path, namespace=namespace)
    return cls(
        tenant_id=tenant_id,
        sub_tenant_id=sub_tenant_id,
        namespace=namespace,
    )

from_state_dict classmethod 

from_state_dict(state: Mapping[str, Any]) -> TenantScope

Reconstruct a :class:TenantScope from a state-dict mapping.

Accepts UPPER_SNAKE and lower_snake casings (UPPER wins). Reads an explicit path / PATH key (list of {kind, id} dicts) when present — the N-level form; otherwise derives from the legacy flat fields. inherits_from is read recursively (v7.22 compat).

Source code in src/symfonic/core/scope.py 
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
@classmethod
def from_state_dict(cls, state: Mapping[str, Any]) -> TenantScope:
    """Reconstruct a :class:`TenantScope` from a state-dict mapping.

    Accepts UPPER_SNAKE and lower_snake casings (UPPER wins).  Reads an
    explicit ``path`` / ``PATH`` key (list of ``{kind, id}`` dicts) when
    present — the N-level form; otherwise derives from the legacy flat
    fields.  ``inherits_from`` is read recursively (v7.22 compat).
    """
    def _read(*keys: str) -> Any:
        for k in keys:
            if k in state and state[k] is not None:
                return state[k]
        return None

    raw_path = _read("PATH", "path")
    if isinstance(raw_path, (list, tuple)) and raw_path:
        # Fail loud on malformed levels rather than silently dropping
        # them (silent scope data loss). Round-2 fix (triage 2026-07-02).
        from symfonic.agent.types import ScopeValidationError

        built: list[ScopeLevel] = []
        for lvl in raw_path:
            if not (isinstance(lvl, Mapping) and "kind" in lvl and "id" in lvl):
                raise ScopeValidationError(
                    "Malformed scope path level in from_state_dict: "
                    f"{lvl!r} (each level requires 'kind' and 'id').",
                )
            built.append(ScopeLevel(kind=str(lvl["kind"]), id=str(lvl["id"])))
        levels = tuple(built)
        if levels:
            namespace = _read("NAMESPACE", "namespace")
            return cls.from_path(levels, namespace=namespace)

    tenant_id = _read("TENANT_ID", "tenant_id") or ""
    sub_tenant_id = _read("SUB_TENANT_ID", "sub_tenant_id")
    namespace = _read("NAMESPACE", "namespace")

    raw_inherits = _read("INHERITS_FROM", "inherits_from")
    inherits_from: tuple[TenantScope, ...] = ()
    if isinstance(raw_inherits, (list, tuple)) and raw_inherits:
        inherits_from = tuple(
            cls.from_state_dict(ancestor)
            for ancestor in raw_inherits
            if isinstance(ancestor, Mapping)
        )

    kwargs: dict[str, Any] = {"tenant_id": tenant_id}
    if sub_tenant_id is not None:
        kwargs["sub_tenant_id"] = sub_tenant_id
    if namespace is not None:
        kwargs["namespace"] = namespace
    if inherits_from:
        kwargs["inherits_from"] = inherits_from
    return cls(**kwargs)

narrow 

narrow(sub_tenant_id: str) -> TenantScope

Create a new scope narrowed to a sub-tenant (legacy 2-level shim).

Retained byte-identically from v7.22: sets both sub_tenant_id and namespace to the supplied value and preserves inherits_from. For N-level hierarchies prefer :meth:child.

Source code in src/symfonic/core/scope.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
def narrow(self, sub_tenant_id: str) -> TenantScope:
    """Create a new scope narrowed to a sub-tenant (legacy 2-level shim).

    Retained byte-identically from v7.22: sets both ``sub_tenant_id`` and
    ``namespace`` to the supplied value and preserves ``inherits_from``.
    For N-level hierarchies prefer :meth:`child`.
    """
    return TenantScope(
        tenant_id=self.tenant_id,
        sub_tenant_id=sub_tenant_id,
        namespace=sub_tenant_id,
        inherits_from=self.inherits_from,
    )

root classmethod 

root(
    kind: str, id: str, *, namespace: str | None = None
) -> TenantScope

Construct a 1-level root scope (kind:id,).

Source code in src/symfonic/core/scope.py 
376
377
378
379
@classmethod
def root(cls, kind: str, id: str, *, namespace: str | None = None) -> TenantScope:
    """Construct a 1-level root scope ``(kind:id,)``."""
    return cls.from_path([ScopeLevel(kind=kind, id=id)], namespace=namespace)

to_state_dict 

to_state_dict() -> dict[str, Any]

Convert to state-dict fields for BaseAgentState injection.

Backward-compatible: a legacy 1-/2-level scope (no explicit path override) emits EXACTLY the pre-v8.0 keys (tenant_id + sub_tenant_id [+ namespace] [+ inherits_from]) — byte-identical. An explicit N-level path additionally emits a path key (list of {kind, id} dicts) so the hierarchy round-trips.

Source code in src/symfonic/core/scope.py 
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
def to_state_dict(self) -> dict[str, Any]:
    """Convert to state-dict fields for BaseAgentState injection.

    Backward-compatible: a legacy 1-/2-level scope (no explicit path
    override) emits EXACTLY the pre-v8.0 keys
    (``tenant_id`` + ``sub_tenant_id`` [+ ``namespace``] [+
    ``inherits_from``]) — byte-identical.  An explicit N-level path
    additionally emits a ``path`` key (list of ``{kind, id}`` dicts) so
    the hierarchy round-trips.
    """
    d: dict[str, Any] = {
        "tenant_id": self.tenant_id,
        "sub_tenant_id": self.sub_tenant_id,
    }
    if self.namespace is not None:
        d["namespace"] = self.namespace
    if self.inherits_from:
        d["inherits_from"] = [
            ancestor.to_state_dict() for ancestor in self.inherits_from
        ]
    if self.path_override is not None:
        d["path"] = [
            {"kind": lvl.kind, "id": lvl.id} for lvl in self.path_override
        ]
    return d

ToolLifecycleHook

Bases: Protocol

Async instrumentation for tool invocation start/end lifecycle.

ToolRegistry 

ToolRegistry()

Tool registration, filtering, cost estimation, and freeze.

Frozen automatically by AgentGraph.compile(). After freeze, register() raises RuntimeError.

Source code in src/symfonic/core/tools/registry.py 
71
72
73
def __init__(self) -> None:
    self._registrations: dict[str, ToolRegistration] = {}
    self._frozen: bool = False

all_tools 

all_tools() -> list[BaseTool]

Return all registered tools.

Source code in src/symfonic/core/tools/registry.py 
146
147
148
def all_tools(self) -> list[BaseTool]:
    """Return all registered tools."""
    return [r.tool for r in self._registrations.values()]

cost_usd_for_invocation 

cost_usd_for_invocation(
    tool_names: list[str],
) -> float | None

Calculate total cost for a set of tool invocations.

Returns None if any tool has unknown cost.

Source code in src/symfonic/core/tools/registry.py 
215
216
217
218
219
220
221
222
223
224
225
226
def cost_usd_for_invocation(self, tool_names: list[str]) -> float | None:
    """Calculate total cost for a set of tool invocations.

    Returns None if any tool has unknown cost.
    """
    total = 0.0
    for name in tool_names:
        reg = self._registrations.get(name)
        if reg is None or reg.cost_usd is None:
            return None
        total += reg.cost_usd
    return total

filter 

filter(
    allowed_names: list[str] | None = None,
    categories: list[ToolCategory] | None = None,
    max_safety_level: SafetyLevel = SafetyLevel.DANGEROUS,
    agent_visible_only: bool = False,
    available_capabilities: set[type] | None = None,
    debug: bool = False,
) -> list[BaseTool]

Filter tools by criteria.

When debug=True, each excluded tool emits a logger.debug() message.

Source code in src/symfonic/core/tools/registry.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def filter(
    self,
    allowed_names: list[str] | None = None,
    categories: list[ToolCategory] | None = None,
    max_safety_level: SafetyLevel = SafetyLevel.DANGEROUS,
    agent_visible_only: bool = False,
    available_capabilities: set[type] | None = None,
    debug: bool = False,
) -> list[BaseTool]:
    """Filter tools by criteria.

    When debug=True, each excluded tool emits a logger.debug() message.
    """
    result: list[BaseTool] = []
    max_order = _SAFETY_ORDER[max_safety_level]

    for name, reg in self._registrations.items():
        # Name filter
        if allowed_names is not None and name not in allowed_names:
            if debug:
                logger.debug("Tool '%s' excluded — not in allowed_names", name)
            continue

        # Category filter
        if categories is not None and reg.category not in categories:
            if debug:
                logger.debug(
                    "Tool '%s' excluded — category %s not in filter",
                    name, reg.category,
                )
            continue

        # Safety filter
        if _SAFETY_ORDER[reg.safety_level] > max_order:
            if debug:
                logger.debug(
                    "Tool '%s' excluded — safety level %s exceeds max",
                    name, reg.safety_level,
                )
            continue

        # Agent visibility filter
        if agent_visible_only and not reg.visible_to_agents:
            if debug:
                logger.debug("Tool '%s' excluded — not visible to agents", name)
            continue

        # Capability filter
        if (
            reg.requires is not None
            and available_capabilities is not None
            and reg.requires not in available_capabilities
        ):
            if debug:
                logger.debug(
                    "Tool '%s' excluded — missing capability %s",
                    name,
                    reg.requires.__name__,
                )
            continue

        result.append(reg.tool)

    return result

freeze 

freeze() -> None

Freeze the registry. Called automatically by AgentGraph.compile().

Source code in src/symfonic/core/tools/registry.py 
138
139
140
def freeze(self) -> None:
    """Freeze the registry. Called automatically by AgentGraph.compile()."""
    self._frozen = True

get_registration 

get_registration(tool_name: str) -> ToolRegistration | None

Get registration metadata for a tool by name.

Source code in src/symfonic/core/tools/registry.py 
244
245
246
def get_registration(self, tool_name: str) -> ToolRegistration | None:
    """Get registration metadata for a tool by name."""
    return self._registrations.get(tool_name)

register 

register(tool: BaseTool, **metadata: Any) -> ToolRegistry

Register a single tool with optional metadata. Returns self for chaining.

Raises SymfonicAgentError(code='bad_request') if tool was produced by @symfonic_tool and the experimental flag is off.

Source code in src/symfonic/core/tools/registry.py 
120
121
122
123
124
125
126
127
128
129
130
def register(self, tool: BaseTool, **metadata: Any) -> ToolRegistry:
    """Register a single tool with optional metadata. Returns self for chaining.

    Raises ``SymfonicAgentError(code='bad_request')`` if ``tool`` was
    produced by ``@symfonic_tool`` and the experimental flag is off.
    """
    self._assert_mutable()
    lifted = self._apply_symfonic_metadata(tool, metadata)
    reg = ToolRegistration(tool=tool, **lifted)
    self._registrations[tool.name] = reg
    return self

register_many 

register_many(tools: list[BaseTool]) -> ToolRegistry

Register multiple tools with default metadata. Returns self for chaining.

Source code in src/symfonic/core/tools/registry.py 
132
133
134
135
136
def register_many(self, tools: list[BaseTool]) -> ToolRegistry:
    """Register multiple tools with default metadata. Returns self for chaining."""
    for tool in tools:
        self.register(tool)
    return self

set_experimental_tool_decorator 

set_experimental_tool_decorator(enabled: bool) -> None

Deprecated no-op. @symfonic_tool is GA and needs no gate.

Retained so existing engine wiring and adopter code that toggled the former experimental gate keep working. The enabled value is ignored — decorator-produced tools always register.

Source code in src/symfonic/core/tools/registry.py 
75
76
77
78
79
80
81
82
def set_experimental_tool_decorator(self, enabled: bool) -> None:
    """Deprecated no-op. ``@symfonic_tool`` is GA and needs no gate.

    Retained so existing engine wiring and adopter code that toggled
    the former experimental gate keep working. The ``enabled`` value
    is ignored — decorator-produced tools always register.
    """
    return None

tools_within_budget 

tools_within_budget(
    budget_usd: float, include_unknown_cost: bool = True
) -> list[BaseTool]

Return tools whose cost is within budget.

If include_unknown_cost is True, tools with no cost info are included.

Source code in src/symfonic/core/tools/registry.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def tools_within_budget(
    self, budget_usd: float, include_unknown_cost: bool = True
) -> list[BaseTool]:
    """Return tools whose cost is within budget.

    If include_unknown_cost is True, tools with no cost info are included.
    """
    result: list[BaseTool] = []
    for reg in self._registrations.values():
        if reg.cost_usd is None:
            if include_unknown_cost:
                result.append(reg.tool)
        elif reg.cost_usd <= budget_usd:
            result.append(reg.tool)
    return result

symfonic_tool 

symfonic_tool(
    *,
    scope: ScopeLiteral = "public",
    routing_mode: RoutingMode = "observe",
    visibility: Visibility = "visible",
    name: str | None = None,
    description: str | None = None,
) -> Callable[[Callable[..., Any]], StructuredTool]

Decorate a Python callable into a symfonic-compatible tool.

The resulting object is a :class:~langchain_core.tools.StructuredTool accepted by :meth:~symfonic.core.tools.registry.ToolRegistry.register without any further plumbing.

Parameters

scope: Logical visibility scope. "public" (default), "internal", or "admin". Symfonic does not yet enforce scope at the invocation tier; the field is captured today so policy code can consume it without a future signature change. routing_mode: Per-tool override for the framework's dynamic tool-routing verdict. Defaults to "observe"; "enforce" causes the router to actually narrow the manifest, "off" opts out of routing entirely. visibility: "visible" (default) keeps the tool in the agent-visible manifest; "hidden" flips the registry's visible_to_agents field to False so the tool is bound to the LangGraph runtime but never offered to the LLM in the manifest. Useful for system-level tools (telemetry emitters, feature-flag readers) that should be callable by orchestration code but never selected by the model. name: Override the tool name. Defaults to fn.__name__. description: Override the description. Defaults to the function's docstring.

Returns

Callable A decorator that, applied to a callable, returns a :class:StructuredTool with the symfonic metadata attached.

Notes
  • The decorated callable's async / sync / async-generator semantics are preserved on the underlying tool's func / coroutine slots. inspect.isasyncgenfunction continues to return True for generator tools so Phase 3's :func:~symfonic.core.tools.streaming.wrap_async_gen_tool works unchanged.
  • Registration requires no flag (GA). The resulting tool registers through ToolRegistry.register like any LangChain tool; its scope / routing_mode / visibility metadata is lifted into the ToolRegistration automatically.
Source code in src/symfonic/core/tools/decorator.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def symfonic_tool(
    *,
    scope: ScopeLiteral = "public",
    routing_mode: RoutingMode = "observe",
    visibility: Visibility = "visible",
    name: str | None = None,
    description: str | None = None,
) -> Callable[[Callable[..., Any]], StructuredTool]:
    """Decorate a Python callable into a symfonic-compatible tool.

    The resulting object is a :class:`~langchain_core.tools.StructuredTool`
    accepted by :meth:`~symfonic.core.tools.registry.ToolRegistry.register`
    without any further plumbing.

    Parameters
    ----------
    scope:
        Logical visibility scope.  ``"public"`` (default), ``"internal"``,
        or ``"admin"``.  Symfonic does not yet enforce ``scope`` at the
        invocation tier; the field is captured today so policy code can
        consume it without a future signature change.
    routing_mode:
        Per-tool override for the framework's dynamic tool-routing
        verdict.  Defaults to ``"observe"``; ``"enforce"`` causes the
        router to actually narrow the manifest, ``"off"`` opts out of
        routing entirely.
    visibility:
        ``"visible"`` (default) keeps the tool in the agent-visible
        manifest; ``"hidden"`` flips the registry's
        ``visible_to_agents`` field to ``False`` so the tool is bound
        to the LangGraph runtime but never offered to the LLM in the
        manifest.  Useful for system-level tools (telemetry emitters,
        feature-flag readers) that should be callable by orchestration
        code but never selected by the model.
    name:
        Override the tool name.  Defaults to ``fn.__name__``.
    description:
        Override the description.  Defaults to the function's docstring.

    Returns
    -------
    Callable
        A decorator that, applied to a callable, returns a
        :class:`StructuredTool` with the symfonic metadata attached.

    Notes
    -----
    * The decorated callable's async / sync / async-generator
      semantics are preserved on the underlying tool's ``func`` /
      ``coroutine`` slots.  ``inspect.isasyncgenfunction`` continues
      to return ``True`` for generator tools so Phase 3's
      :func:`~symfonic.core.tools.streaming.wrap_async_gen_tool` works
      unchanged.
    * Registration requires no flag (GA). The resulting tool registers
      through ``ToolRegistry.register`` like any LangChain tool; its
      ``scope`` / ``routing_mode`` / ``visibility`` metadata is lifted
      into the ``ToolRegistration`` automatically.
    """

    metadata = SymfonicToolMetadata(
        scope=scope,
        routing_mode=routing_mode,
        visibility=visibility,
    )

    def decorator(fn: Callable[..., Any]) -> StructuredTool:
        if not callable(fn):
            raise TypeError(
                "@symfonic_tool can only decorate callables, got "
                f"{type(fn).__name__!r}"
            )

        tool_name = name or fn.__name__
        tool_desc = description if description is not None else (
            inspect.getdoc(fn) or f"Symfonic tool '{tool_name}'."
        )

        # parse_docstring=True so ``Args:`` blocks surface as field
        # descriptions; error_on_invalid_docstring=False keeps the
        # decorator forgiving for callers who do not follow Google-style
        # docstrings.
        builder_kwargs: dict[str, Any] = {
            "name": tool_name,
            "description": tool_desc,
            "parse_docstring": True,
            "error_on_invalid_docstring": False,
        }

        if inspect.iscoroutinefunction(fn) or inspect.isasyncgenfunction(fn):
            # StructuredTool stores async impls on .coroutine.  Pass
            # ``func=None`` so the builder does not try to use the
            # async function as a sync one.
            built = StructuredTool.from_function(
                func=None,
                coroutine=fn,
                **builder_kwargs,
            )
        else:
            built = StructuredTool.from_function(func=fn, **builder_kwargs)

        # Attach metadata.  StructuredTool is a Pydantic model with
        # extra="forbid" on its model_config, so we use object.__setattr__
        # to bypass the validation and stash the attribute on the
        # instance (mirrors how LangChain itself attaches private state).
        object.__setattr__(built, _METADATA_ATTR, metadata)

        return built

    return decorator