Day 29/30: Deploying LangGraph + MCP

So you've finally gotten your LangGraph + MCP agent working on your local machine - it's responding to queries, adapting to new information, and generally behaving itself. But as soon as you try to deploy it to a server, things start to fall apart. The agent crashes after a few hours, or it starts producing bizarre, out-of-context responses. You're left scratching your head, wondering why your carefully crafted agent can't seem to stay up and running in production.

The problem often lies in the way we've structured our agent's lifecycle. When we're running locally, it's easy to just spin up the agent, test it out, and then shut it down when we're done. But in a production environment, our agent needs to be able to recover from failures, adapt to changing circumstances, and handle a constant stream of incoming requests. This requires a more robust approach to deployment, one that takes into account the realities of distributed systems and the unpredictability of the real world.

One key concept here is the idea of a "service" - a long-running process that provides some kind of functionality to the outside world. In our case, we want to deploy our LangGraph + MCP agent as a service that can receive incoming requests, process them, and respond accordingly. To do this, we'll need to use a combination of tools and techniques to ensure our agent stays up and running, even in the face of unexpected failures or changes.

Here's an example of how we might structure our agent as a service, using Python and the LangGraph + MCP APIs:

import langgraph
from mcp import ModelContextProtocol
import os
import signal
import sys

# Set up our LangGraph + MCP agent
agent = langgraph.Agent()
mcp = ModelContextProtocol()

# Define a function to handle incoming requests
def handle_request(request):
    # Use our agent to process the request and generate a response
    response = agent.process_request(request)
    return response

# Define a function to run our agent as a service
def run_service():
    # Set up a signal handler to catch SIGTERM and SIGINT
    def signal_handler(sig, frame):
        print("Received signal, shutting down...")
        agent.shutdown()
        sys.exit(0)

    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

    # Start our agent and begin listening for incoming requests
    agent.start()
    while True:
        request = mcp.receive_request()
        response = handle_request(request)
        mcp.send_response(response)

# Run our agent as a service
if __name__ == "__main__":
    run_service()

In this example, we define a run_service function that sets up our agent and begins listening for incoming requests. We use the signal module to catch SIGTERM and SIGINT signals, which allows us to cleanly shut down our agent when the service is terminated. We also define a handle_request function that uses our agent to process incoming requests and generate responses.

One practical gotcha to watch out for when deploying a LangGraph + MCP agent as a service is the need to properly handle checkpointing and recovery. If our agent crashes or is terminated unexpectedly, we'll need to be able to recover its state and resume processing from where we left off. This can be achieved using the langgraph.Checkpointer class, which allows us to save and restore our agent's state at regular intervals.

As we look to the future, we'll be exploring even more advanced techniques for deploying and managing LangGraph + MCP agents in production environments. With the ability to build robust, scalable agents that can adapt to changing circumstances, we'll be able to tackle even more complex and challenging problems in areas like natural language processing, computer vision, and beyond. Tomorrow, we'll be taking a closer look at how to integrate our agents with other systems and services, and how to use techniques like reinforcement learning to optimize their performance.

Comments

[Mads Hansen]

Checkpointing is only half of safe deployment; rolling upgrades make checkpoint compatibility the hard part. I’d persist a graph version, checkpoint-schema version, and tool-contract digest with every run. A new worker should explicitly migrate or reject incompatible state rather than deserialize it optimistically. On SIGTERM, first fail readiness and stop accepting work, then drain in-flight runs with a deadline; anything unfinished needs a durable interrupted/indeterminate outcome before shutdown. The production test I’d add is: kill the worker at every node boundary, deploy N+1 while N still has runs, replay the same request ID, and assert exactly one terminal outcome with no duplicated tool side effects. That validates recovery, not just serialization.