Extension seams

visvoai-core is deliberately not a framework you configure — it's a small loop you subclass. AgentRuntime (visvoai/core/runtime.py) exposes eight hook methods; build_graph calls each of them (when a runtime instance is passed) and falls back to a core default whenever a hook returns None. Every seam below is exercised by real consumers: the CLI overrides the agent node for per-turn context assembly, and a hosted platform adds HITL and background-task nodes, a Postgres persistence pair, and an auth-carrying context — same runtime, no forks, no private branches of this code.

_extend_graph(workflow, tool_configs)

Add extra nodes/edges to the StateGraph before it's compiled. It is called on every AgentRuntime.build_graph() — the base implementation is a no-op, so a bare runtime adds nothing.

class MyRuntime(AgentRuntime):
    def _extend_graph(self, workflow, tool_configs):
        workflow.add_node("hitl", my_hitl_node)
        workflow.add_edge("hitl", "agent")

This is the seam for an approval gate, a background-task node, or any other node your product's routing needs — combine with _get_interrupt_nodes to actually pause execution there.

_build_agent_node(ctx) / _build_tools_node(ctx)

Return a full replacement node body for "agent" / "tools", or None to keep the core default (call_model / ToolNode(all_tools)). Both receive a GraphBuildContext — the exact inputs the core builder used, so your replacement doesn't need to re-derive them:

@dataclass(frozen=True)
class GraphBuildContext:
    model: BaseChatModel
    core_tools: List[BaseTool]
    all_tools_map: Dict[str, BaseTool]
    all_tools: List[BaseTool]
    system_prompt: str
    tool_configs: Dict[str, Any] = field(default_factory=dict)
    per_round_retrieve: Optional[Any] = None
    lean_prompt: bool = False
    max_agent_steps: Optional[int] = None

Override _build_agent_node for per-turn system-prompt assembly (e.g. injecting a plan, trimming context by a token budget) without re-implementing tool binding, the step cap, or retrieval — build your replacement from ctx and it stays behaviorally consistent with core. Override _build_tools_node to gate execution behind an approval step before tools actually run.

_agent_routing(ctx) / _tools_routing(tool_configs)

_agent_routing returns (routing_fn, routing_map) for the "agent" → {...} conditional edge, or None for the core default (should_continue, {"tools": "tools", END: END}). Override to route the agent node through an intermediate node (e.g. a finalize-check) instead of straight to tools or END.

_tools_routing(tool_configs) is not optional to override in the same way — the base AgentRuntime implementation always routes "tools" back to "agent":

def _tools_routing(self, tool_configs: dict):
    def _route(state: AgentState) -> str:
        return "agent"
    return _route, {"agent": "agent"}

A subclass overrides this to add other destinations to the routing map (e.g. {"agent": "agent", "hitl": "hitl"}) once _extend_graph has added the corresponding node.

_get_checkpointer(checkpointer=None)

Return the checkpointer LangGraph should compile with. The base class passes through whatever you gave build_graph's checkpointer= argument unchanged; override to force your own (e.g. a Postgres-backed BaseCheckpointSaver) regardless of what the caller passed, or to layer setup around it.

graph = AgentRuntime().build_graph(
    model=model, core_tools=tools, system_prompt="...",
    checkpointer=MemorySaver(),   # in-memory; swap for Postgres/SQLite for durability
)

A checkpointer is what gives an agent memory across turns — the same thread_id passed to ask()/astream_events resumes the same conversation state.

_get_state_class()

Return the TypedDict LangGraph should use as its state schema. The public AgentState (visvoai/core/state.py) is intentionally lean:

class AgentState(TypedDict, total=False):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    active_mcp_tools: Annotated[List[str], _union_ordered]

messages uses LangGraph's add_messages reducer (append/merge by id). active_mcp_tools uses an order-preserving de-duplicating union reducer — it's the persistent half of dynamic tool binding (see Tool retrieval at scale). Core deliberately does not declare state it doesn't act on — plan-mode bookkeeping, approval flags, and similar surface-specific state belong on a subclass via TypedDict inheritance:

class MyState(AgentState, total=False):
    approval_request: Optional[dict]
    pending_background_tasks: List[str]

class MyRuntime(AgentRuntime):
    def _get_state_class(self):
        return MyState

_get_interrupt_nodes()

Return a list of node names LangGraph should interrupt_before — the human-in-the-loop pause points. Default None (no interrupts). Pair with a node added in _extend_graph:

class MyRuntime(AgentRuntime):
    def _get_interrupt_nodes(self):
        return ["approval_gate"]

RuntimeContext — your state carried to tools

RuntimeContext (visvoai/core/context.py) is the only context type visvoai-core tools receive by default:

@dataclass
class RuntimeContext:
    request_id: Optional[str] = None
    subagent_depth: int = 0
    parent_tool_call_id: Optional[str] = None

It's intentionally surface-agnostic — no auth, no datastore session, no HTTP concerns. A surface that needs more subclasses it and passes its own subclass to tools instead:

@dataclass
class MyContext(RuntimeContext):
    user_id: str = ""
    db_session: Any = None

subagent_depth and parent_tool_call_id exist specifically to support the subagent pattern — tracking nesting without any special-cased recursion machinery in core itself.

The summary table

Runnable version of the tool + persistence + runtime seams together: visvoai-core/examples/04_extend_the_runtime.py.

Next: Persistence.