--- title: "LangChain and LangGraph" description: "LangChain create_agent builds a tool-calling agent on LangGraph. The Dex adapter runs a create_agent agent or a custom LangGraph graph inside a durable session. It is available for Python." canonical_url: "https://docs.superdurable.io/agent-frameworks/langchain-langgraph/" source: "docs/content/agent-frameworks/langchain-langgraph.mdx" locale: "en" --- # LangChain and LangGraph LangChain **create_agent** builds a tool-calling agent on LangGraph. The Dex adapter runs a **create_agent** agent or a custom LangGraph graph inside a durable session. It is available for Python. ## When to use it Use it when your agent already uses LangChain **create_agent** or a LangGraph graph. Keep the graph, its middleware, and its tools, and let Dex own the session. ## How a session runs **DurableToolCallMiddleware** stops each tool call with a LangGraph interrupt, so no tool runs inside **RunModel**. The **RunModel** Step runs the graph to its next tool calls, usually with one model call, and stores the graph checkpoint. Each tool call runs in its own Step, in parallel: **CallTool** for a read or an idempotent write, **CallSingleAttemptTool** for any other write. A call that needs approval waits in **AwaitToolApproval** first. The tool Step validates the arguments and calls the tool once. The next **RunModel** restores the checkpoint and resumes the graph with the recorded results. A custom graph builds its tool node with **durable_tool_node** instead of **ToolNode**. Its state must hold the conversation as LangChain messages. A turn that looks up an order runs **RunModel**, **CallTool**, **ResumeTurn**, and **RunModel**. ## Install Install the adapter with its framework. Each adapter brings in the Dex Agents core. **Python** ```bash pip install dex-agents-langgraph langchain-openai ``` Example: [agents/python/langgraph/README.md](https://github.com/superdurable/dex/tree/main/agents/python/langgraph/README.md) ## The example The example is the [order support agent](https://docs.superdurable.io/agent-frameworks.md#the-examples) that every framework page uses. The Flow type is **LangChainOrderSupport**. The full example is under [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses). ### Agent and tools The tools are LangChain tools. A factory builds the agent with **DurableToolCallMiddleware**. Only a Worker calls the factory, so a Client process needs no provider key. The tool policies name each tool. **Python** ```python def build_langchain_order_support_flow( order_desk: OrderDesk, model: str | BaseChatModel, slack_connection: slack.Connection | None, ) -> AgentSessionFlow: """Builds the session Flow; ``model`` is a LangChain model name, such as "openai:gpt-5.2".""" # The tools are closures, so they call the OrderDesk passed in here. @tool def look_up_order(order_id: str) -> str: """Look up one order by its order ID and return its items and status.""" return order_desk.describe_order(order_id) @tool def cancel_order(order_id: str, reason: str) -> str: """Cancel one order that has not shipped yet.""" return order_desk.cancel_order(order_id, reason) # Only a Worker calls the factory, so a Client process needs no provider key. def build_order_support_agent() -> Pregel[Any, Any, Any, Any]: return create_agent( model, tools=[look_up_order, cancel_order], system_prompt=ORDER_SUPPORT_INSTRUCTIONS, middleware=[DurableToolCallMiddleware()], ) return langchain_agent_session_flow( build_order_support_agent, flow_type=LANGCHAIN_ORDER_SUPPORT_FLOW_TYPE, tool_sources=build_slack_tool_sources(slack_connection), tool_policies={ "look_up_order": ToolPolicy(ToolEffect.READ), "cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE), }, ) ``` Example: [examples/python/agent-harnesses/agent_harness_examples/langchain_order_support.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/langchain_order_support.py) ### Worker One Worker serves the session Flows of every harness in the example. Agent session Steps are async, so the Python Registry allows async handlers. Build the same session Flow in every process that starts sessions or calls their RPCs. **Python** ```python # Agent session Steps are async, so the Registry allows async handlers. self.registry = Registry( tuple(example.session_flow for example in self.examples), allow_async_handlers=True ) config.blob_cache_dir.mkdir(parents=True, exist_ok=True) self.blob_cache = open_blob_cache(BlobCacheConfig(str(config.blob_cache_dir), 1 << 30)) self.worker = AsyncWorker( self.registry, self.blob_cache, WorkerOptions( server_address=config.server_address, pull=PullWorkerOptions(max_concurrent_tasks=MAX_CONCURRENT_TASKS), ), ) self.client = AsyncClient( self.registry, self.blob_cache, ClientOptions(server_address=config.server_address) ) # ... def _new_langchain_example( config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None ) -> AgentHarnessExample: order_desk = OrderDesk() model = _langchain_model(config.langchain_model) flow = build_langchain_order_support_flow(order_desk, model, slack_connection) return AgentHarnessExample("langchain", flow, order_desk) # ... def _langchain_model(model: str) -> str | BaseChatModel: return LangChainDemoChatModel() if is_demo_model(model) else model ``` Example: [examples/python/agent-harnesses/agent_harness_examples/app.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/app.py) ### Session routes Every harness serves these same routes. [Session routes](https://docs.superdurable.io/agent-frameworks.md#session-routes) explains them and the session token. **Python** ```python @blueprint.post("/sessions") async def start_session() -> Response: # The server picks the session ID, so a caller cannot claim an existing session. session_id = f"{harness_name}-order-support-{uuid.uuid4()}" await sessions.start_session(session_id) session_token = sign_session_id(app_state.config.session_key, session_id) return jsonify(session_id=session_id, session_token=session_token) @blueprint.post("/messages") async def send_user_message() -> Response: body, session_id = await session_request() accepted = await sessions.send_user_message(session_id, _required_string(body, "text")) return jsonify(accepted=accepted) @blueprint.post("/interrupts/resolve") async def resolve_interrupt() -> Response: body, session_id = await session_request() interrupt_id = _required_string(body, "interrupt_id") if not isinstance(approved := body.get("approved"), bool): raise BadRequest("approved must be a boolean") rejection_message = body.get("rejection_message") if rejection_message is not None and not isinstance(rejection_message, str): raise BadRequest("rejection_message must be a string") accepted = await sessions.resolve_interrupt( session_id, interrupt_id, approved=approved, rejection_message=rejection_message ) return jsonify(accepted=accepted) @blueprint.get("/snapshot") async def get_agent_snapshot() -> Response: session_id = request.args.get("session_id", "") if not session_id: raise BadRequest("session_id is required") authorize_session(session_id) try: snapshot = await read_agent_event_stream_snapshot( app_state.client, example.session_flow, session_id ) except FlowNotActiveOrNotFoundError as error: raise NotFound("the agent session is missing or no longer retained") from error return jsonify(snapshot.to_json()) ``` Example: [examples/python/agent-harnesses/agent_harness_examples/http_routes.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/http_routes.py) ## Run it locally ### Start Dex and the example In a first terminal, render the session graph, then start Dex. Dex Web reads the graph when it starts, to show the session and its approval controls. **Python** ```bash cd examples/python/agent-harnesses uv sync --locked mkdir -p definitions uv run --frozen dex-agents-session-graph --flow-type LangChainOrderSupport \ --permission order-support.operate --out definitions/langchain-langgraph.json dexcli dev --flow-rendering-dir ./definitions ``` Example: [examples/python/agent-harnesses/README.md](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/README.md) **dexcli dev** stays in the foreground. In a second terminal, start the example: **Python** ```bash cd examples/python/agent-harnesses uv run --frozen python -m agent_harness_examples ``` Example: [examples/python/agent-harnesses/README.md](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/README.md) The app serves on **127.0.0.1:8081**. Dex Web is at ****. Without **DEX_LANGCHAIN_MODEL**, the agent uses **mock/order-support**, a local demo model that needs no key. It calls **look_up_order** for "look up order 1001" and **cancel_order** for "cancel order 1002". | Variable | Default | Meaning | | ------------------------- | ---------------------- | --------------------------------------------------- | | **DEX_LANGCHAIN_MODEL** | **mock/order-support** | A LangChain model name, such as **openai:gpt-5.2**. | | **OPENAI_API_KEY** | none | The provider key, for an OpenAI model. | | **DEX_AGENT_SESSION_KEY** | a random key | The key that signs session tokens. | ### Call the session Start a session, ask the agent to cancel order 1002, and approve the call: **Python** ```bash base=http://127.0.0.1:8081/products/agent-harnesses/langchain session=$(curl -s -X POST "$base/sessions" -H 'Content-Type: application/json' -d '{}') id=$(echo "$session" | jq -r .session_id) token=$(echo "$session" | jq -r .session_token) curl -s -X POST "$base/messages" -H 'Content-Type: application/json' \ -H "X-Agent-Session-Token: $token" -d "{\"session_id\": \"$id\", \"text\": \"Cancel order 1002\"}" until interrupt=$(curl -s "$base/snapshot?session_id=$id" -H "X-Agent-Session-Token: $token" \ | jq -re '.snapshot.pending_interrupts[0].interrupt_id'); do sleep 1; done curl -s -X POST "$base/interrupts/resolve" -H 'Content-Type: application/json' \ -H "X-Agent-Session-Token: $token" \ -d "{\"session_id\": \"$id\", \"interrupt_id\": \"$interrupt\", \"approved\": true}" ``` Example: [examples/python/agent-harnesses/README.md](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/README.md) The loop reads the snapshot until the agent asks to run **cancel_order**. That call waits for approval, then runs once. Read the snapshot again to see the agent's answer. ### Approve in Dex Web Instead of the last command, you can approve the call in Dex Web. Follow the steps in [Approve in Dex Web](https://docs.superdurable.io/agent-frameworks.md#approve-in-dex-web). ## Connect Connectors The agent can call [Connector](https://docs.superdurable.io/concepts/connector.md) operations as tools, as [Connectors as tools](https://docs.superdurable.io/agent-frameworks.md#connectors-as-tools) describes. The example adds the Slack tool **slack_post_thread_reply**, which replies in a Slack thread. ### Offer the tool Install the **connectors** extra of **dex-agents** and the Connector package, such as **dex-connector-slack**. The install commands are in [Connector tools](https://docs.superdurable.io/reference/agents/connector-tools.md#install). The Worker opens the Slack connection **support-slack** from the project configuration. **connector_tool** turns the operation into a tool, and **ConnectorToolSource** holds it. Without the connection, the agent gets no Slack tool. **Python** ```python def build_slack_tool_sources(connection: slack.Connection | None) -> list[ToolSource]: """Returns the Slack tool source, or no source when Slack is not configured.""" if connection is None: return [] return [ ConnectorToolSource([ # The manifest does not ask for approval; this agent asks before every post. connector_tool( connection.post_thread_reply(), connection.reference, requires_approval=True ), ]), ] # ... async def open_slack_project_connection( environment: Mapping[str, str] = os.environ, ) -> SlackProjectConnection | None: # ... # An incomplete or invalid configuration raises here, so the Worker fails at startup. try: project = await load_from_environment(environment) except FileNotFoundError: _log_slack_not_set_up("the connector configuration file does not exist") return None # ... try: connection = slack.new_project_connection(project, SLACK_CONNECTION_NAME) except BaseException: await project.close() raise return SlackProjectConnection(project, connection) ``` Example: [examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py) The Worker opens the connection once, at startup, and every agent uses it: **Python** ```python async def main() -> None: config = AgentHarnessExamplesConfig.from_env() slack_setup = await open_slack_project_connection() try: app_state = AgentHarnessExamplesApp(config, slack_setup.connection if slack_setup else None) # ... finally: if slack_setup is not None: await slack_setup.aclose() ``` Example: [examples/python/agent-harnesses/agent_harness_examples/\_\_main\_\_.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/__main__.py) The agent passes the tool source next to its own tools: **Python** ```python return langchain_agent_session_flow( build_order_support_agent, flow_type=LANGCHAIN_ORDER_SUPPORT_FLOW_TYPE, tool_sources=build_slack_tool_sources(slack_connection), # ... ) ``` Example: [examples/python/agent-harnesses/agent_harness_examples/langchain_order_support.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/langchain_order_support.py) ### How a call runs Each call of **slack_post_thread_reply** runs once, in its own **CallSingleAttemptTool** Step, after a person approves it. [The Slack Connector tool](https://docs.superdurable.io/agent-frameworks.md#slack-connector-tool) explains why. **DurableToolCallMiddleware** adds the Connector tools to every model request of a **create_agent** agent, after the agent's own tools. A custom graph binds **get_current_tool_source_tools()** next to its own tools in its model node, and **durable_tool_node** defers their calls like any other. The tool bodies never run inside the graph. ### Configure the connection [Connections](https://docs.superdurable.io/reference/agents/connector-tools.md#connector-tools-connection) explains where the Worker finds the **support-slack** connection and what the Slack app needs. To try it, start the example with **DEX_CONNECTOR_CONFIG_FILE** set. Then ask the agent to reply in a Slack thread, with the channel ID and the thread timestamp: ```text Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday. ``` The post waits for approval, like **cancel_order**. ## Limits - A graph whose nodes make several model calls before a tool node makes them all in one **RunModel**. Summarization middleware and a planner node are examples. A retry repeats them all. - Provider-run tools repeat when **RunModel** reruns. - History is append-only. A graph that edits or removes earlier messages does not change the stored history. - Only the message list survives between turns. Other state keys start from their defaults on every turn. - Tools run outside the graph. They cannot use **ToolRuntime**, **InjectedState**, or **InjectedStore**, and cannot return a graph **Command**. **InjectedToolCallId** works. - The graph may raise no interrupts of its own. The adapter README lists every limit. See [Limits for long sessions](https://docs.superdurable.io/reference/agents/session-limits.md) for the server limits. ## Related - [Agent Frameworks](https://docs.superdurable.io/agent-frameworks.md) - [Agents reference](https://docs.superdurable.io/reference/agents.md) - [Python adapter README](https://github.com/superdurable/dex/tree/main/agents/python/langgraph)