Agent Langgraph¶

Agent LangGraph — mvp.agent_langgraph

Cluster: Agents & LLM | Type: component | MCP Tools: 28

Overview¶

Stateful graph workflow agent backed by real LangGraph StateGraph. The MCP runtime uses a long-lived compiled graph cache and durable local checkpointing by default. Defines workflows as lists of typed nodes and edges; supports pass, append, set, increment, llm, tool, and branch node operations, enabling conditional routing and component invocation without boilerplate. MCP execution responses surface completion_state and structured node-level degradation evidence; the read-only capabilities op reports supported operations, transforms, condition syntax, checkpoint modes, and optional backend availability without reading secrets.

Production checkpointing caveat

The default MCP checkpointer persists LangGraph checkpoints to a local file and is suitable for a single MCP process on a simple VPS-style deployment. For multi-worker, multi-container, or horizontally scaled deployments, use a shared durable checkpointer such as Postgres rather than sharing the local checkpoint file.

When to use:

Building multi-step stateful pipelines where graph topology is data-driven rather than hard-coded
Implementing conditional branching workflows that checkpoint state between steps
Wiring G6 components together as named graph nodes with explicit data-flow edges

Example:

from mvp.agent_langgraph import AgentLangGraphBlock, GraphInput, GraphNode, GraphEdge

block = AgentLangGraphBlock(name="langgraph")
result = block.infer(GraphInput(
    nodes=[
        GraphNode(name="init", operation="set", field="status", value="started"),
        GraphNode(name="count", operation="increment", field="steps"),
    ],
    edges=[GraphEdge(source="init", target="count"), GraphEdge(source="count", target="END")],
    initial_state={"status": "", "steps": 0},
))
# result.ok → True; result.value → GraphOutput with final_state, execution_path

Works well with: agent_langchain, navigator, autonomous_orchestrator

Public API¶

`AgentLangGraphBlock(LifecycleMixin, AIBlock[GraphInput, GraphOutput, None])`¶

Stateful graph workflow agent using real LangGraph StateGraph.

Field	Type	Default
`name`	`str`	'agent_langgraph'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: GraphInput) -> Result[GraphOutput]`¶

`GraphNode(BaseModel)`¶

Field	Type	Default
`name`	`str`	required
`operation`	`NodeOp`	'pass'
`field`	`str`	'messages'
`value`	`str`	''
`condition`	`str`	''
`true_target`	`str`	''
`false_target`	`str`	''

`GraphEdge(BaseModel)`¶

Field	Type	Default
`source`	`str`	required
`target`	`str`	required

`GraphInput(BaseModel)`¶

Field	Type	Default
`nodes`	`list[GraphNode]`	required
`edges`	`list[GraphEdge]`	required
`initial_state`	`dict[str, Any]`	required
`entry_point`	`str`	''
`thread_id`	`str`	'default'
`production_mode`	`bool`	False

Methods:

`nodes_not_empty(v: list[GraphNode]) -> list[GraphNode]`¶

`graph_references_are_valid() -> 'GraphInput'`¶

`GraphOutput(BaseModel)`¶

Field	Type	Default
`final_state`	`dict[str, Any]`	required
`execution_path`	`list[str]`	required
`n_steps`	`int`	required
`backend`	`str`	'pure_python'
`checkpoint_id`	`str \\| None`	None
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'qualified-draft'
`node_degradations`	`list[dict[str, Any]]`	Field(default_factory=list)

`AdaptLangGraphMCPBlock(AIBlock[MCPLangGraphInput, MCPLangGraphOutput, None])`¶

SQLite-backed LangGraph MCP block with 28 ops.

Field	Type	Default
`name`	`str`	'AdaptLangGraphMCPBlock'
`db_path`	`str`	':memory:'
`agentic_planner`	`object`	None

Methods:

`infer(inp: MCPLangGraphInput) -> Result[MCPLangGraphOutput]`¶

`MCPLangGraphInput(BaseModel)`¶

Field	Type	Default
`op`	`LangGraphOp`	'info'
`graph_id`	`str`	''
`name`	`str`	''
`nodes`	`list[dict]`	Field(default_factory=list)
`edges`	`list[dict]`	Field(default_factory=list)
`entry_point`	`str`	''
`initial_state`	`dict`	Field(default_factory=dict)
`backend`	`str`	'python'
`exec_id`	`str`	''
`checkpoint_id`	`str`	''
`step`	`int`	0
`node_name`	`str`	''
`state_patch`	`dict`	Field(default_factory=dict)
`op_type`	`str`	''
`template_id`	`str`	''
`field`	`str`	''
`value`	`Any`	None
`description`	`str`	''
`query`	`str`	''
`limit`	`int`	20
`max_steps`	`int`	10
`candidates`	`list[dict]`	Field(default_factory=list)

`MCPLangGraphOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	''
`key`	`str`	''
`value`	`Any`	None
`found`	`bool`	False
`count`	`int`	0
`records`	`list[dict]`	Field(default_factory=list)
`retrieved`	`list[Any]`	Field(default_factory=list)
`summary`	`dict`	Field(default_factory=dict)
`message`	`str`	''
`metadata`	`dict`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	''
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`dict`	Field(default_factory=dict)
`request_id`	`str`	''
`run_id`	`str`	''

`LangGraphStore`¶

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`save_graph(graph_id: str, name: str, nodes_json: str, edges_json: str, entry_point: str) -> None`¶

`retrieve_graph(graph_id: str) -> dict | None`¶

`list_graphs(limit: int = 20) -> list[dict]`¶

`delete_graph(graph_id: str) -> int`¶

`record_execution(exec_id: str, graph_id: str, final_state_json: str, path_json: str, n_steps: int, backend: str) -> None`¶

`query_executions(graph_id: str = '', limit: int = 20) -> list[dict]`¶

`summarize_executions(graph_id: str = '') -> dict`¶

`delete_executions(graph_id: str) -> int`¶

`save_checkpoint(checkpoint_id: str, graph_id: str, exec_id: str, step: int, node_name: str, state_json: str) -> None`¶

`load_checkpoint(checkpoint_id: str = '', graph_id: str = '', step: int = -1) -> dict | None`¶

`list_checkpoints(graph_id: str) -> list[dict]`¶

`delete_checkpoints(checkpoint_id: str = '', graph_id: str = '') -> int`¶

`register_template(template_id: str, name: str, op: str, field: str, value: str, description: str) -> None`¶

`list_templates(op_type: str = '') -> list[dict]`¶

`remove_template(template_id: str = '', name: str = '') -> int`¶

`get_latest_state(graph_id: str) -> dict`¶

`update_latest_state(graph_id: str, patch: dict) -> None`¶

`search(query: str, limit: int = 20) -> list[dict]`¶

`count_all() -> dict`¶

`info() -> dict`¶

`clear_all() -> None`¶

MCP Tools¶

Operation	Source
`run_graph`	langgraph_mcp
`run_with_checkpoint`	langgraph_mcp
`stream_graph`	langgraph_mcp
`validate_graph`	langgraph_mcp
`graph_status`	langgraph_mcp
`store_graph`	langgraph_mcp
`retrieve_graph`	langgraph_mcp
`list_graphs`	langgraph_mcp
`delete_graph`	langgraph_mcp
`record_execution`	langgraph_mcp
`query_executions`	langgraph_mcp
`summarize_executions`	langgraph_mcp
`delete_executions`	langgraph_mcp
`save_checkpoint`	langgraph_mcp
`load_checkpoint`	langgraph_mcp
`list_checkpoints`	langgraph_mcp
`delete_checkpoints`	langgraph_mcp
`register_template`	langgraph_mcp
`list_templates`	langgraph_mcp
`remove_template`	langgraph_mcp
`get_state`	langgraph_mcp
`update_state`	langgraph_mcp
`search`	langgraph_mcp
`info`	langgraph_mcp
`clear`	langgraph_mcp
`recommend_graph`	langgraph_mcp
`list_patterns`	langgraph_mcp
`capabilities`	langgraph_mcp

Agent Langgraph¶

Overview¶

Public API¶

AgentLangGraphBlock(LifecycleMixin, AIBlock[GraphInput, GraphOutput, None])¶

infer(data: GraphInput) -> Result[GraphOutput]¶

GraphNode(BaseModel)¶

GraphEdge(BaseModel)¶

GraphInput(BaseModel)¶

nodes_not_empty(v: list[GraphNode]) -> list[GraphNode]¶

graph_references_are_valid() -> 'GraphInput'¶

GraphOutput(BaseModel)¶

AdaptLangGraphMCPBlock(AIBlock[MCPLangGraphInput, MCPLangGraphOutput, None])¶

infer(inp: MCPLangGraphInput) -> Result[MCPLangGraphOutput]¶

MCPLangGraphInput(BaseModel)¶

MCPLangGraphOutput(BaseModel)¶

LangGraphStore¶

save_graph(graph_id: str, name: str, nodes_json: str, edges_json: str, entry_point: str) -> None¶

retrieve_graph(graph_id: str) -> dict | None¶

list_graphs(limit: int = 20) -> list[dict]¶

delete_graph(graph_id: str) -> int¶

record_execution(exec_id: str, graph_id: str, final_state_json: str, path_json: str, n_steps: int, backend: str) -> None¶

query_executions(graph_id: str = '', limit: int = 20) -> list[dict]¶

summarize_executions(graph_id: str = '') -> dict¶

delete_executions(graph_id: str) -> int¶

save_checkpoint(checkpoint_id: str, graph_id: str, exec_id: str, step: int, node_name: str, state_json: str) -> None¶

load_checkpoint(checkpoint_id: str = '', graph_id: str = '', step: int = -1) -> dict | None¶

list_checkpoints(graph_id: str) -> list[dict]¶

delete_checkpoints(checkpoint_id: str = '', graph_id: str = '') -> int¶

register_template(template_id: str, name: str, op: str, field: str, value: str, description: str) -> None¶

list_templates(op_type: str = '') -> list[dict]¶

remove_template(template_id: str = '', name: str = '') -> int¶

get_latest_state(graph_id: str) -> dict¶

update_latest_state(graph_id: str, patch: dict) -> None¶

search(query: str, limit: int = 20) -> list[dict]¶

count_all() -> dict¶

info() -> dict¶

clear_all() -> None¶

MCP Tools¶

`AgentLangGraphBlock(LifecycleMixin, AIBlock[GraphInput, GraphOutput, None])`¶

`infer(data: GraphInput) -> Result[GraphOutput]`¶

`GraphNode(BaseModel)`¶

`GraphEdge(BaseModel)`¶

`GraphInput(BaseModel)`¶

`nodes_not_empty(v: list[GraphNode]) -> list[GraphNode]`¶

`graph_references_are_valid() -> 'GraphInput'`¶

`GraphOutput(BaseModel)`¶

`AdaptLangGraphMCPBlock(AIBlock[MCPLangGraphInput, MCPLangGraphOutput, None])`¶

`infer(inp: MCPLangGraphInput) -> Result[MCPLangGraphOutput]`¶

`MCPLangGraphInput(BaseModel)`¶

`MCPLangGraphOutput(BaseModel)`¶

`LangGraphStore`¶

`save_graph(graph_id: str, name: str, nodes_json: str, edges_json: str, entry_point: str) -> None`¶

`retrieve_graph(graph_id: str) -> dict | None`¶

`list_graphs(limit: int = 20) -> list[dict]`¶

`delete_graph(graph_id: str) -> int`¶

`record_execution(exec_id: str, graph_id: str, final_state_json: str, path_json: str, n_steps: int, backend: str) -> None`¶

`query_executions(graph_id: str = '', limit: int = 20) -> list[dict]`¶

`summarize_executions(graph_id: str = '') -> dict`¶

`delete_executions(graph_id: str) -> int`¶

`save_checkpoint(checkpoint_id: str, graph_id: str, exec_id: str, step: int, node_name: str, state_json: str) -> None`¶

`load_checkpoint(checkpoint_id: str = '', graph_id: str = '', step: int = -1) -> dict | None`¶

`list_checkpoints(graph_id: str) -> list[dict]`¶

`delete_checkpoints(checkpoint_id: str = '', graph_id: str = '') -> int`¶

`register_template(template_id: str, name: str, op: str, field: str, value: str, description: str) -> None`¶

`list_templates(op_type: str = '') -> list[dict]`¶

`remove_template(template_id: str = '', name: str = '') -> int`¶

`get_latest_state(graph_id: str) -> dict`¶

`update_latest_state(graph_id: str, patch: dict) -> None`¶

`search(query: str, limit: int = 20) -> list[dict]`¶

`count_all() -> dict`¶

`info() -> dict`¶

`clear_all() -> None`¶