How to Move an n8n Prototype into a LangGraph Production Agent

You already have an n8n workflow that functions. It receives a request, calls APIs, uses an LLM, makes decisions, updates a database, and returns a result. During the prototype stage, this is often enough. But as the workflow grows, things can start becoming harder to manage. - State is spread across multiple nodes - Agent decisions become difficult to trace - Retry logic becomes complicated - Long-running executions need persistence - Human approval needs pause/resume - Testing individual decisions becomes harder - Business logic gets tightly coupled to orchestration These are scenarios where moving the agent logic from an n8n prototype into LangGraph becomes purposeful. Replacing n8n just because LangGraph is newer is not quite the point. The goal is to move the parts that require stateful agent orchestration, explicit control flow, persistence, and long-running execution into a framework designed for those problems. This guide walks through that migration step by step. 1. Start With an Actual n8n Prototype Consider a real estate AI assistant that receives a buyer request: User Request ↓ Extract Requirements ↓ Search CRM ↓ Assign to AI Agent ↓ Call Property Search API ↓ Score Results ↓ Send Response ↓ Update CRM Consequently, an n8n workflow might contain: - Webhook node - Set/Edit Fields nodes - Code nodes - HTTP Request nodes - IF/Switch nodes - AI Agent node - CRM integration - WhatsApp/Email node - Error handling workflow This is a perfectly reasonable architecture for a prototype. However, the problem starts becoming noticeable when the workflow becomes something like: Webhook ↓ 20+ nodes ↓ Multiple IF branches ↓ AI Agent ↓ Multiple tool calls ↓ Retries ↓ Human approval ↓ CRM update ↓ Follow-up ↓ Scheduled continuation At this point, the workflow is doing more than simple automation. It is becoming an agentic state machine. This is the perfect stage to evaluate whether you should move the core agent logic into LangGraph. 2. Before Migrating: Separate the Workflow Into Responsibilities Don't start rewriting the entire n8n workflow immediately. First, inspect every node and determine what responsibility it actually performs. A useful mapping looks like this: | n8n Component | Responsibility | LangGraph Equivalent | |---|---|---| | Webhook | Receive input | API layer | | Set/Edit Fields | Transform data | Python function | | Code | Business logic | Python function | | IF/Switch | Routing | Conditional edge | | AI Agent | Reasoning | Agent/LLM node | | HTTP Request | External operation | Tool | | Database | Data persistence | DB/service | | Wait | Long-running state | Persistence/interrupt | | Human approval | Manual decision | interrupt() | | Error workflow | Recovery | Retry/recovery logic | This classification prevents one of the biggest migration mistakes: Rewriting the entire system when only the agent orchestration needs to change. The CRM, property database, or your external APIs don't necessarily need to move. The migration should focus on the orchestration layer. 3. Build the n8n Version First Before converting anything, define exactly what the existing workflow does. For example: Buyer Request ↓ Extract Requirements ↓ Search Properties ↓ Filter Results ↓ AI Ranker ↓ Return Recommendations A buyer might send: Looking for a 3-bedroom apartment in Dubai Marina under AED 2 million. The workflow needs to: - Extract the requirements - Search the property database - Filter out unsuitable properties - Rank the remaining properties - Return recommendations This gives us a clear baseline for the migration. 4. Define the LangGraph State This is one of the most important changes during the migration. In an n8n workflow, execution data naturally flows from one node to another. With LangGraph, you explicitly define the state shared across the graph. For example: from typing import TypedDict class AgentState(TypedDict): user_query: str requirements: dict candidates: list matches: list selected_property: dict | None error: str | None Now the agent has an explicit state contract. The workflow can move through: user_query ↓ requirements ↓ candidates ↓ matches ↓ selected_property This makes the state easier to inspect, test, persist, and reason about. A useful rule is: If a piece of information is required by multiple stages of the agent, consider making it part of the graph state. 5. Map n8n Nodes to LangGraph Nodes The next step is to convert individual workflow operations into graph nodes. The original n8n flow: Webhook ↓ Code ↓ HTTP Request ↓ AI Agent ↓ IF Can become: START ↓ normalize_request ↓ search_properties ↓ rank_properties ↓ route_result A basic graph can be created like this: from langgraph.graph import StateGraph, START, END builder = StateGraph(AgentState) builder.add_node("normalize_request", normalize_request) builder.add_node("search_properties", search_properties) builder.add_node("rank_properties", rank_properties) builder.add_edge(START, "normalize_request") builder.add_edge("normalize_request", "search_properties") builder.add_edge("search_properties", "rank_properties") builder.add_edge("rank_properties", END) graph = builder.compile() The important architectural difference is that the workflow is now represented explicitly as a graph. Each node has a defined responsibility. 6. Move n8n Tools Into Python Tools So far, we have identified the nodes' functionalities, and they are mapped to LangGraph nodes. Now, we move the tools. An n8n HTTP Request node might currently call a property API. So, instead of letting the agent directly deal with raw HTTP logic, wrap the operation as a tool. For example: from langchain_core.tools import tool @tool def search_properties( location: str, max_budget: int, bedrooms: int ): """ Search available properties. """ # Call property API or database results = property_service.search( location=location, max_budget=max_budget, bedrooms=bedrooms ) return results The tool should have: - Clear inputs - Clear outputs - Validation - Error handling - A single responsibility The important distinction is: The tool operates. The agent decides when to use it. This keeps the agent's reasoning separate from infrastructure code. 7. Replace n8n IF Nodes With Explicit Graph Routing This is another essential migration step. Suppose the n8n workflow contains: IF match_score > 0.8 ↓ Strong Match In LangGraph, make that routing explicit. def route_match(state: AgentState): if not state["matches"]: return "no_match" if state["matches"][0]["score"] >= 0.8: return "strong_match" return "weak_match" Then, connect the routes with: builderadd_conditional_edges( "rank_properties", route_match, { "strong_match": "send_recommendation", "weak_match": "request_more_preferences", "no_match": "fallback_search" } ) The resulting graph then becomes: rank_properties ↓ route_match / | \ / | \ strong_match weak_match no_match ↓ ↓ ↓ recommendation ask user fallback search This is much easier to reason when the number of branches increases. 8. Add Persistence Instead of Relying on Execution History A prototype often relies on execution history. A production agent cannot assume that the entire execution will always remain active. So, consider: Agent starts ↓ Search properties ↓ Human approval required ↓ Wait 6 hours ↓ Continue The agent needs to remember where it was and what state it had and here is where LangGraph persistence becomes important. Conceptually: Agent State ↓ Checkpoint ↓ Thread ↓ Resume Execution Compile the graph with a checkpointer: graph = builder.compile( checkpointer=checkpointer ) Then invoke it with a stable thread ID: config = { "configurable": { "thread_id": "lead-123" } } result = graph.invoke( initial_state, config=config ) The thread_id gives the execution a durable identity. This becomes particularly important for: - Long-running agents - Human approval - Multi-step conversations - Recovery - Stateful workflows - Resuming interrupted execution 9. Convert n8n Wait/Human Approval Into interrupt() Consider an n8n workflow: AI recommends property ↓ Wait ↓ Agent approval ↓ Continue A LangGraph implementation can model the same process using an interrupt: AI Recommendation ↓ interrupt() ↓ Human Decision ↓ Resume Graph For example: from langgraph.types import interrupt def approval_node(state): decision = interrupt({ "message": "Approve this property recommendation?", "property": state["selected_property"] }) return { "approval": decision } The important difference is that the agent doesn't need to remain continuously active while waiting. The state can be persisted and execution can resume when the human decision arrives. This is especially useful for workflows involving: - Financial approvals - Sensitive customer actions - Contract review - High-value sales - External side effects 10. Make External API Calls Idempotent This is one of the most significant production changes. Imagine the agent executes: send_whatsapp_message() Then, the process crashes immediately afterward. Next, when the graph resumes, the operation might run again. You could end up with: Message sent ↓ Process crashes ↓ Graph resumes ↓ Message sent again The result is a duplicate customer message. Instead, design external side effects to be idempotent. For example: Agent Decision ↓ Generate operation_id ↓ Check idempotency store ↓ Execute side effect ↓ Persist result A simple implementation might use: def send_message_once(operation_id, message): if already_processed(operation_id): return get_previous_result(operation_id) result = send_message(message) save_result( operation_id=operation_id, result=result ) return result This pattern is especially important for: - Payments - Emails - WhatsApp messages - CRM updates - Booking APIs - Ticket creation - Database writes A production agent should always assume that execution may be retried or resumed. 11. Move Retry Logic Out of the Prompt Don't rely on the LLM to decide: "If the API fails, try again." Retry behavior belongs in the application layer. For example: from t