symfonic.core.graph

graph

AgentGraph — graph definition factory.

Per ADR-PFX-002/008 and TDD SS2.8: Accepts graph topology and compiles it into a CompiledGraph. Config is owned by AgentRuntime, passed at compile time.

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

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

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

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

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

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

@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

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

@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

@classmethod
def with_preset(cls, name: str, **kwargs: Any) -> AgentGraph:
    """Create an AgentGraph with a specific preset."""
    return cls(topology=name, **kwargs)

ConfigurationError

Bases: Exception

Raised when capability flags are violated at compile time.