LangGraph

Define a LangGraph graph in a Python backend and invoke it from any Squid client.

Why Use LangGraph

Some AI workflows are not a single prompt. They branch, loop, call tools, and sometimes need to stop and wait for a human to approve something before they continue. Expressing that as a chain of agent calls means you own the state machine, the persistence, and the resume logic yourself.

LangGraph already models this as a graph of nodes with persisted state. Squid runs your graph inside your Python backend, checkpoints its state for you, and exposes it to every Squid client:

Backend code

# Declare the graph once in your Python backend.
@langgraph('approval-graph')
def define_approval_graph(self) -> StateGraph:
    ...

Client code

// Run it from the client, pause on interrupt, and resume when the user decides.
const started = await squid.langGraph('approval-graph').invoke({ input: { request: 'refund #123' } });
if (started.status === 'interrupted') {
  await squid.langGraph('approval-graph').resume(started.threadId, { approved: true });
}

Overview

A LangGraph graph is a state machine: nodes read and write a shared state object, and edges decide what runs next. Squid adds three things on top:

• Discovery. A method decorated with @langgraph in a Python backend service registers a graph under an id. You return an uncompiled StateGraph and Squid compiles it.
• Checkpointing. Squid supplies the checkpointer, so a graph's state persists between calls and survives an interrupt. State is grouped by a thread id.
• A client API. squid.langGraph(graphId) invokes, resumes, and inspects graphs from any Squid client, guarded by security rules.

When to use LangGraph

Use Case	Recommendation
Multi-step workflow with branching, loops, or human approval	✅ LangGraph
Answer a question, optionally with tools and knowledge bases	Use an AI agent
Expose one backend function to an agent	Use AI functions
Run code on a schedule	Use Schedulers
Call server-side logic from the client with no graph state	Use Executables

How it works

1. A Python backend service declares a graph with @langgraph('graph-id'), returning an uncompiled StateGraph.
2. A client calls squid.langGraph('graph-id').invoke({ input }). Squid runs the graph in your backend against a thread, creating a new thread when you do not name one.
3. The call resolves with the run's status, threadId, state, and next nodes.
4. If a node calls LangGraph's interrupt(), the run stops with status: 'interrupted' and its state is checkpointed. Calling resume(threadId, payload) continues from that point, and payload becomes the return value of the interrupt() call.

Quick Start

Prerequisites

• A Python Squid backend project. See Local development.
• The langgraph package installed in that backend's Python environment.
• The @squidcloud/client package installed in your frontend.

Step 1: Declare a graph

Decorate a method with @langgraph and return an uncompiled StateGraph. Do not call .compile(), because Squid compiles the graph with its own checkpointer.

Backend code

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

from squidcloud_backend import SquidService, langgraph


class GraphState(TypedDict):
    # `operator.add` makes `messages` append across nodes instead of being replaced.
    messages: Annotated[list[str], operator.add]


class ExampleService(SquidService):
    @langgraph('simple-graph')
    def define_simple_graph(self) -> StateGraph:
        def first_node(state: GraphState) -> dict:
            incoming = state.get('messages') or ['no input']
            return {'messages': [f'Processed: {incoming[0]}']}

        def second_node(state: GraphState) -> dict:
            return {'messages': ['Graph completed']}

        graph = StateGraph(GraphState)
        graph.add_node('first_node', first_node)
        graph.add_node('second_node', second_node)
        graph.add_edge(START, 'first_node')
        graph.add_edge('first_node', 'second_node')
        graph.add_edge('second_node', END)

        # Return the uncompiled graph. Squid compiles it with its checkpointer.
        return graph

Step 2: Allow clients to invoke it

Without an API key, a client call is rejected unless a security rule allows it. Add a @secure_langgraph rule:

Backend code

from squidcloud_backend import SquidService, secure_langgraph


class ExampleService(SquidService):
    @secure_langgraph('simple-graph')
    def allow_simple_graph(self, request: dict) -> bool:
        # request['operation'] is 'invoke', 'resume', 'getState', or 'deleteThread'.
        return self.is_authenticated()

See Securing LangGraph for the full request object.

Step 3: Deploy the backend

squid deploy

Step 4: Invoke the graph from the client

Client code

import { Squid } from '@squidcloud/client';

const squid = new Squid({
  appId: 'YOUR_APP_ID',
  region: 'us-east-1.aws',
  environmentId: 'dev',
});

const result = await squid.langGraph('simple-graph').invoke({
  input: { messages: ['Hello LangGraph!'] },
});

console.log(result.status); // 'completed'
console.log(result.threadId); // Server-generated id for this run
console.log(result.state); // { messages: ['Hello LangGraph!', 'Processed: ...', 'Graph completed'] }
console.log(result.next); // [] once the graph has finished

Core Concepts

Threads

A thread is the unit of persisted graph state. Every run belongs to one.

• Omit threadId and Squid creates a new thread, returning its id. Two runs that both omit it get different threads.
• Pass a threadId to run against a specific thread. Passing one that does not exist yet starts a fresh thread under that id, which lets you key a thread on something you already have, such as a conversation id or an order id.
• deleteThread(threadId) removes the thread's checkpoints. It is idempotent, so deleting a thread that does not exist still succeeds. After deletion, getState returns an empty state and the next invoke or resume on that id starts fresh.

Run results

invoke, resume, and getState all resolve to the same shape:

Field	Description
ok	Always true on success.
threadId	The thread the run executed against.
status	'completed' when the graph finished, 'interrupted' when it is awaiting a resume. getState does not return this field.
state	The final (or paused) state values, or null.
next	The nodes that run next. Empty when the graph has completed.

Interrupt and resume

Call LangGraph's interrupt() inside a node to pause the graph and hand a decision back to the caller. The value you pass to interrupt() is surfaced to the caller, and the value the caller later passes to resume() becomes the return value of that interrupt() call.

Backend code

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt

from squidcloud_backend import SquidService, langgraph


class ApprovalState(TypedDict):
    messages: Annotated[list[str], operator.add]
    approved: bool


class ExampleService(SquidService):
    @langgraph('approval-graph')
    def define_approval_graph(self) -> StateGraph:
        def request_approval(state: ApprovalState) -> dict:
            # Execution stops here and the state is checkpointed.
            response = interrupt({'kind': 'approval', 'question': 'Continue processing?'})
            # After resume(), `response` holds the payload the caller sent.
            approved = response.get('approved', False) if isinstance(response, dict) else False
            return {'messages': [f'Approval response: {approved}'], 'approved': approved}

        def finalize(state: ApprovalState) -> dict:
            status = 'approved' if state.get('approved') else 'rejected'
            return {'messages': [f'Graph completed with status: {status}']}

        graph = StateGraph(ApprovalState)
        graph.add_node('request_approval', request_approval)
        graph.add_node('finalize', finalize)
        graph.add_edge(START, 'request_approval')
        graph.add_edge('request_approval', 'finalize')
        graph.add_edge('finalize', END)

        return graph

Drive it from the client:

Client code

const graph = squid.langGraph('approval-graph');

const started = await graph.invoke({ input: { messages: [], approved: false } });
console.log(started.status); // 'interrupted'
console.log(started.next); // ['request_approval']

// Later, once a human has decided.
const finished = await graph.resume(started.threadId, { approved: true });
console.log(finished.status); // 'completed'
console.log(finished.state?.messages); // [..., 'Graph completed with status: approved']

Resuming a thread that already completed is not an error. The call returns the completed state with an empty next.

Long runs

Graph execution is capped at the backend IPC timeout, currently 4 minutes, for every method on this page.

invoke and resume hold a promise until the graph stops. For graphs that take long enough that holding a promise is awkward, use invokeAsync and resumeAsync. They return a job id immediately, which you pass to the Jobs API:

Client code

const jobId = await squid.langGraph('simple-graph').invokeAsync({
  input: { messages: ['Hello LangGraph!'] },
});

// Retrieve the result whenever you are ready.
const result = await squid.job().awaitJob(jobId);
console.log(result.status);

The async variants only free the caller from waiting. They do not raise the 4 minute execution cap.

Inspecting state

Client code

const state = await squid.langGraph('approval-graph').getState(threadId);
console.log(state.state); // Current state values, or null
console.log(state.next); // Nodes that would run next, empty if the graph completed

// Remove the thread's checkpoints when you no longer need them.
await squid.langGraph('approval-graph').deleteThread(threadId);

Securing LangGraph

A client without an API key can only reach a graph that a @secure_langgraph rule allows. Declare the rule with a graph id to secure one graph, or with no argument to cover every graph.

Backend code

from squidcloud_backend import SquidService, secure_langgraph


class ExampleService(SquidService):
    # Applies to every graph in the application.
    @secure_langgraph()
    def allow_all_graphs(self, request: dict) -> bool:
        return self.is_authenticated()

The rule receives a dict carrying:

Key	Description
operation	'invoke', 'resume', 'getState', or 'deleteThread'.
graphId	The graph being acted on.
threadId	The target thread, when the operation names one.
input	The initial state passed to invoke.
resumePayload	The payload passed to resume.

Use operation to hold destructive calls to a higher bar than reads:

Backend code

class ExampleService(SquidService):
    @secure_langgraph('approval-graph')
    def allow_approval_graph(self, request: dict) -> bool:
        auth = self.get_user_auth()
        if auth is None:
            return False
        if request.get('operation') == 'deleteThread':
            # Only the thread's owner may discard its history. This assumes threads are
            # named '<userId>:<something>' when they are created.
            thread_id = request.get('threadId') or ''
            return thread_id.startswith(f"{auth['userId']}:")
        return True

A permissive global rule does not widen a narrower per-graph rule. When a graph has its own @secure_langgraph('graph-id') rule, that rule still has to pass, so a graph-specific rule that only allows invoke rejects getState and deleteThread even while a global rule returns True.

For general guidance on security rules, see Security rules.

Error Handling

Error	Cause	Fix
... not found	No @langgraph method is registered under that graph id, or the backend holding it is not deployed.	Check the id string and redeploy the backend.
must return an uncompiled StateGraph	The @langgraph method returned something else, often a compiled graph.	Return the StateGraph itself. Do not call .compile().
An exception raised inside a node	The graph itself failed.	The message propagates to the caller. Handle it in the node if it is expected.
UNAUTHORIZED	No @secure_langgraph rule allowed the operation for this caller.	Add or widen the rule. See Securing LangGraph.

invoke({}) and invoke({ input: {} }) are both valid. A graph that needs no seed state can be started with no arguments at all.

Best Practices

• Do not compile the graph yourself. Squid compiles it with the checkpointer that makes threads and interrupts work.
• Key threads on something meaningful. Passing your own threadId, such as a conversation or order id, makes a graph resumable without you storing the generated id separately.
• Delete threads you are finished with. Checkpoints persist until deleteThread removes them.
• Keep node state JSON-serializable. State crosses a process boundary to be checkpointed.
• Use reducers for accumulating fields. Annotated[list[str], operator.add] appends across nodes, where a plain list[str] is replaced by whichever node wrote last.
• Secure by operation, not just by graph. deleteThread and resume change state, where getState only reads it.