Cordum + LangGraph: Governed State Machine

LangGraph makes observable governance outcomes explicit graph state. This example calls the real cordum.workflow.run tool advertised by the bridge and routes completed calls and tool errors to different nodes. Approval holds stay inside the blocking bridge call until an operator resolves them or the call times out.

Before you start

• Python 3.10+.
• A workflow identifier in your tenant.
• Cordum bootstrapped with ./tools/scripts/quickstart.sh.
• cordum-mcp-bridge on PATH.
• The TLS environment from Framework integrations.

1. Install

python -m venv .venv
source .venv/bin/activate
pip install "cordum-adapters[langchain]" langgraph

2. Build the graph around the MCP result contract

Save as governed_graph.py:

import os
from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from cordum_agent_adapters.mcp_client import McpStdioClient, McpToolError

TOOL_NAME = "cordum.workflow.run"


class RunState(TypedDict, total=False):
    workflow_id: str
    input: dict[str, object]
    status: str
    result: dict[str, object]
    error: str


def require_tool(client: McpStdioClient, name: str) -> None:
    names = {
        str(tool.get("name"))
        for tool in client.list_tools()
        if isinstance(tool.get("name"), str)
    }
    if name not in names:
        raise RuntimeError(f"{name} is not advertised; available={sorted(names)}")


def build_graph(client: McpStdioClient):
    def invoke(state: RunState) -> RunState:
        try:
            result = client.call_tool(
                TOOL_NAME,
                {
                    "workflow_id": state["workflow_id"],
                    "input": state.get("input", {}),
                    "dry_run": True,
                },
            )
            return {**state, "status": "completed", "result": result}
        except McpToolError as exc:
            return {**state, "status": "blocked", "error": str(exc)}

    def outcome(state: RunState) -> str:
        return state["status"]

    def preserve(state: RunState) -> RunState:
        return state

    graph = StateGraph(RunState)
    graph.add_node("invoke", invoke)
    graph.add_node("complete", preserve)
    graph.add_node("blocked", preserve)
    graph.add_edge(START, "invoke")
    graph.add_conditional_edges(
        "invoke",
        outcome,
        {
            "completed": "complete",
            "blocked": "blocked",
        },
    )
    graph.add_edge("complete", END)
    graph.add_edge("blocked", END)
    return graph.compile()


with McpStdioClient(
    command=["cordum-mcp-bridge"],
    env=os.environ.copy(),
) as client:
    require_tool(client, TOOL_NAME)
    app = build_graph(client)
    final = app.invoke({"workflow_id": "demo-workflow", "input": {}})
    print(final)

Run it with a real workflow identifier:

python governed_graph.py

The printed state is the actual bridge response. No scripted language model or static dashboard row is used as proof.

3. Exercise approval routing

Upload a v1 policy bundle that holds this capability:

version: v1
rules:
  - id: review-workflow-run
    match:
      capabilities:
        - cordum.workflow.run
    decision: require_approval
    reason: "Workflow run requires review"

The bridge submits a Cordum job and leaves the original tool call waiting while that job is pending approval. Set a timeout long enough for the review window before starting the process:

export CORDUM_MCP_CALL_TIMEOUT=5m
python governed_graph.py

Resolve the pending job in the Cordum dashboard from another window. Approval lets the same bridge call continue; there is no approval-reference retry. If the review is denied or the timeout expires, the bridge returns a tool error and the graph routes to blocked.

4. Exercise the hard-deny route

Change the rule decision to deny and upload a new bundle version. A hard deny returns an MCP result with isError: true; McpStdioClient raises McpToolError, and the graph routes to blocked.

Visibility boundary

The graph records only what this process observes. Cordum governs calls made through the generated bridge tool, while LangGraph owns application routing. Verify actual policy and invocation events in the dashboard or gateway API.

Next steps

• Put a durable checkpoint before the blocking invoke node.
• Choose CORDUM_MCP_CALL_TIMEOUT from the documented operator response window.
• Use separate graph nodes for other tools returned by client.list_tools().
• Read the framework integration contracts.