Control flow API reference (Python)

These APIs control graph execution from inside nodes and tools. Use Command to update state and choose the next node in one return value, interrupt() to pause for human input and resume later, and handoff tools to let a model transfer control to another agent.

For concepts and walkthroughs, see Interrupts, Routing, Command and callbacks, Add human approval and Hand off between agents.

Import paths

Python

from tenxgraph.utils import Command, END
from tenxgraph.utils import interrupt, Interrupt, GraphInterrupt, pending_interrupt
from tenxgraph.prebuilt.tools import create_handoff_tool, is_handoff_tool

`Command`

A return value from a node function that combines a state update with an explicit routing decision. Returning Command(goto=...) sends execution to the named node, or to END to stop.

Python

from tenxgraph.core.state import AgentState
from tenxgraph.utils import Command, END

# Route on the last message; the matching nodes must exist in the graph
def router_node(state: AgentState, config: dict) -> Command:
    last = state.context[-1].text() if state.context else ""
    if "research" in last.lower():
        return Command(goto="RESEARCHER")
    elif "write" in last.lower():
        return Command(goto="WRITER")
    else:
        return Command(goto=END)

Signature

Python

Command(
    update=None,
    goto=None,
    graph=None,
    state=None,
)

Parameters

Parameter	Type	Default	Description
`update`	`StateT \| Message \| str \| BaseConverter \| None`	`None`	Applied before navigating. An `AgentState` replaces the current state (its new messages are recorded); a `Message`, `str`, converter or list of these is added as new message(s).
`goto`	`str \| None`	`None`	Name of the next node to execute, or `END` to terminate.
`graph`	`str \| None`	`None`	Target graph (`None` for the current one). Stored on the command; the current runtime does not use it for routing.
`state`	`StateT \| None`	`None`	Optional state to attach. When streaming, its new context messages are emitted.

Constants

Constant	Value	Description
`Command.PARENT`	`"PARENT"`	Value meant for `graph=` to name the parent graph. Not acted on by the current runtime.

Example

Add a message and navigate in one step:

Python

from tenxgraph.core.state import AgentState
from tenxgraph.utils import Command

# Record a note as an assistant message, then jump to BILLING_AGENT
def classify_node(state: AgentState, config: dict) -> Command:
    return Command(update="Routing to billing.", goto="BILLING_AGENT")

`interrupt()`

Pause the graph and wait for a resume value. Call interrupt() when a node or tool needs outside input: approval, a choice, or a correction. The first time it runs, it stops the graph. The graph saves its state and reports the interrupt. Resume by running the same thread again with a resume value.

Python

from tenxgraph.utils import interrupt

# Tool that pauses for approval before acting; issue_refund is your own function
async def refund(amount: float) -> str:
    decision = interrupt(
        {"amount": amount},
        message=f"Approve ${amount} refund?",
        response_schema={
            "type": "object",
            "properties": {"approved": {"type": "boolean"}}
        },
    )
    if not decision or not decision.get("approved"):
        return "Refund declined"
    return issue_refund(amount)

# First run: pauses at interrupt() (app and config come from your graph)
await app.ainvoke({"messages": [...]}, config)

# Resume on the same thread: interrupt() returns the value
await app.ainvoke({"resume": {"approved": True}}, config)

The interrupted node or tool runs again from the start on resume, so keep side effects after the interrupt() call. The thread needs a checkpointer and a thread_id in config to be resumable.

Signature

Python

interrupt(
    value=None,
    *,
    message=None,
    reason="input_required",
    response_schema=None,
)

Parameters

Parameter	Type	Default	Description
`value`	`Any`	`None`	Data for the client (what to approve, what to choose from).
`message`	`str \| None`	`None`	Human-readable prompt, shown by UIs such as CopilotKit.
`reason`	`str`	`"input_required"`	Machine-readable reason, e.g., `"tool_approval"` or `"human_verification"`.
`response_schema`	`dict[str, Any] \| None`	`None`	JSON Schema describing the expected resume value.

Returns

The resume value on the run that resumes this interrupt. None if the client cancelled.

Raises

Exception	When
`GraphInterrupt`	On the first run, to stop the graph. Do not catch it.
`RuntimeError`	When called outside a running graph node or tool.

Example

Multiple interrupts in one node, answered one per resume:

Python

from tenxgraph.core.state import AgentState
from tenxgraph.utils import interrupt

async def approval_step(state: AgentState, config: dict):
    step1 = interrupt({"step": 1}, message="Continue to step 1?")
    if not step1:
        return state
    step2 = interrupt({"step": 2}, message="Continue to step 2?")
    if not step2:
        return state
    return "All approved"

`Interrupt`

A pause requested by interrupt(), waiting for a resume value.

Signature

Python

class Interrupt(BaseModel):
    id: str
    key: str
    node: str
    value: Any
    message: str | None
    reason: str
    response_schema: dict[str, Any] | None
    tool_call_id: str | None

Fields

Field	Type	Description
`id`	`str`	Unique id of this pause; clients echo it when resuming.
`key`	`str`	Stable position of the call within its node or tool call.
`node`	`str`	Node that was running when the graph paused.
`value`	`Any`	Payload passed to `interrupt()`.
`message`	`str \| None`	Human-readable prompt.
`reason`	`str`	Why the graph paused (default `"input_required"`).
`response_schema`	`dict[str, Any] \| None`	JSON Schema the resume value should follow.
`tool_call_id`	`str \| None`	Tool call that paused (if `interrupt()` ran inside a tool).

Example

Inspect a pending interrupt after invoke. The state is only returned with ResponseGranularity.FULL:

Python

from tenxgraph.utils import ResponseGranularity, pending_interrupt

result = app.invoke(
    {"messages": [...]}, config, response_granularity=ResponseGranularity.FULL
)
paused = pending_interrupt(result["state"])
if paused:
    print(f"Waiting at {paused.node}: {paused.message}")
    result = app.invoke({"resume": {"approved": True}}, config)

`GraphInterrupt`

Exception raised by interrupt() to stop the graph. Derives from BaseException, so standard error handlers in tools and nodes do not catch it.

Signature

Python

class GraphInterrupt(BaseException):
    def __init__(self, interrupt: Interrupt) -> None: ...

    interrupt: Interrupt  # The pause details

The graph runtime catches GraphInterrupt. Do not catch it in nodes or tools, and note that a bare except BaseException would swallow it.

`pending_interrupt()`

Return the interrupt a paused thread is waiting on, or None.

Pass the AgentState from a run made with ResponseGranularity.FULL (result["state"]).

Signature

Python

pending_interrupt(state: AgentState) -> Interrupt | None

Parameters

Parameter	Type	Description
`state`	`AgentState`	The state from `result["state"]` after `invoke()` or `ainvoke()` with `ResponseGranularity.FULL`.

Returns

The pending Interrupt if the thread is paused at an interrupt() call, or None.

Example

Python

from tenxgraph.utils import ResponseGranularity, pending_interrupt

result = app.invoke(
    {"messages": [...]}, config, response_granularity=ResponseGranularity.FULL
)
pause = pending_interrupt(result["state"])
if pause:
    # Paused: resume with a value
    result = app.invoke({"resume": user_input}, config)
else:
    # Finished
    print(result["messages"][-1].text())

`create_handoff_tool()`

Factory that creates an LLM-callable tool. When the model calls it, the ToolNode detects the transfer_to_<agent> name and navigates directly to that node without executing the function body.

Python

from tenxgraph.core import ToolNode
from tenxgraph.prebuilt.tools import create_handoff_tool

transfer_to_researcher = create_handoff_tool(
    agent_name="RESEARCHER",
    description="Transfer to the research agent for detailed investigation.",
)

tools = ToolNode([transfer_to_researcher])

Signature

Python

create_handoff_tool(
    agent_name: str,
    description: str | None = None,
) -> Callable

Parameters

Parameter	Type	Default	Description
`agent_name`	`str`	required	Must match an existing node name in the graph exactly.
`description`	`str \| None`	`None`	Description shown to the LLM. Defaults to `"Transfer control to <agent_name> agent"`.

An agent_name containing an underscore logs a warning, because it can complicate name matching. Prefer names such as RESEARCHER.

Returns

A callable with special attributes:

__name__ = f"transfer_to_{agent_name}"
__doc__ = the description string
__handoff_tool__ = True
__target_agent__ = agent_name

Interception flow

When the last assistant message contains a tool call named transfer_to_<agent>, the ToolNode handler (invoke and stream) intercepts it before running any tool:

Text

Agent calls transfer_to_RESEARCHER
  -> ToolNode handler: is_handoff_tool("transfer_to_RESEARCHER") returns (True, "RESEARCHER")
  -> handler returns Command(goto="RESEARCHER")
  -> the tool function body is not called

Example

A coordinator that can hand off to a researcher. Set OPENAI_API_KEY first and install with pip install "10xgraph[openai]":

Python

from tenxgraph.core import Agent, StateGraph, ToolNode
from tenxgraph.core.state import AgentState
from tenxgraph.prebuilt.tools import create_handoff_tool
from tenxgraph.utils import END

handoff = create_handoff_tool("RESEARCH", "Transfer to the researcher for deep investigation.")

coordinator = Agent(model="gpt-4o", provider="openai", tool_node="COORD_TOOLS")
researcher = Agent(model="gpt-4o", provider="openai")

# Go to the tool node when the model asked for tools, otherwise finish
def after_coord(state: AgentState) -> str:
    last = state.context[-1]
    return "COORD_TOOLS" if last.role == "assistant" and last.tools_calls else END

graph = StateGraph()
graph.add_node("COORD", coordinator)
graph.add_node("COORD_TOOLS", ToolNode([handoff]))
graph.add_node("RESEARCH", researcher)
graph.set_entry_point("COORD")
graph.add_conditional_edges("COORD", after_coord, {"COORD_TOOLS": "COORD_TOOLS", END: END})
graph.add_edge("RESEARCH", END)

app = graph.compile()

For a fuller multi-agent walkthrough, see Hand off between agents.

`is_handoff_tool()`

Check whether a tool name follows the handoff convention.

Python

from tenxgraph.prebuilt.tools import is_handoff_tool

print(is_handoff_tool("transfer_to_researcher"))  # (True, 'researcher')
print(is_handoff_tool("get_weather"))             # (False, None)

Signature

Python

is_handoff_tool(tool_name: str) -> tuple[bool, str | None]

Parameters

Parameter	Type	Description
`tool_name`	`str`	The name to check.

Returns

A tuple (is_handoff, target_agent):

is_handoff (bool): True if the name matches "transfer_to_<agent>" pattern.
target_agent (str | None): The text after the transfer_to_ prefix, unchanged (case is preserved), or None if not a handoff tool or the prefix has no target.

Example

Python

is_hoff, target = is_handoff_tool("transfer_to_support_team")
assert is_hoff is True
assert target == "support_team"

Common errors

Error	Cause	Fix
`"should have been intercepted"` in logs	Handoff tool body executed.	Run the tool through a standard `ToolNode`, not a custom tool handler.
LLM doesn’t choose handoff tools	Description is unclear.	Make `description=` action-oriented and specific.
`Command.graph` has no effect	The runtime does not route on `graph`.	Route with `goto` within the current graph.
`pending_interrupt()` fails on a dict	Passed the `invoke()` result instead of the state.	Use `ResponseGranularity.FULL` and pass `result["state"]`.
`RuntimeError` from `interrupt()`	Called outside a graph node/tool.	Call `interrupt()` only during graph execution.
`ValueError` or `TypeError` from `create_handoff_tool`	`agent_name` is empty (`ValueError`) or not a string (`TypeError`).	Pass a non-empty string.