	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server Disaster Recovery and Failover

MCP Server Disaster Recovery and Failover

Author: Venkata Sudhakar

Production MCP servers must remain available even when individual backend services fail or a regional outage occurs. A well-designed disaster recovery strategy includes health checks to detect failures, automatic failover to a standby endpoint, and graceful degradation that returns meaningful error messages rather than crashing the server and breaking all agent workflows.

ShopMax India deploys its core MCP server across two regions - asia-south1 (Mumbai) as primary and asia-south2 (Delhi) as standby. The MCP server implements a health-checked connection pool that automatically switches to the standby backend if the primary becomes unavailable, ensuring agents continue operating during regional incidents with minimal disruption.

The below example shows an MCP server with an active health check loop and automatic failover between primary and standby database endpoints, with graceful degradation on total failure.

# Install: pip install mcp asyncpg
import asyncio
import time
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import asyncpg

# Primary and standby endpoints
ENDPOINTS = [
    {"name": "primary-mumbai",  "dsn": "postgresql://agent:pass@10.1.0.5:5432/shopmax"},
    {"name": "standby-delhi",   "dsn": "postgresql://agent:pass@10.2.0.5:5432/shopmax"}
]

app = Server("shopmax-ha-mcp")
active_pool = None
active_endpoint = None
last_health_check = 0
HEALTH_INTERVAL = 30   # seconds

async def connect_to_endpoint(ep):
    try:
        pool = await asyncpg.create_pool(ep["dsn"], min_size=1, max_size=5,
                                          command_timeout=5)
        await pool.fetchval("SELECT 1")   # verify connectivity
        return pool
    except Exception:
        return None

async def get_pool():
    global active_pool, active_endpoint, last_health_check
    now = time.time()
    # Periodic health check
    if now - last_health_check > HEALTH_INTERVAL or active_pool is None:
        last_health_check = now
        # Try to verify current pool is still healthy
        if active_pool:
            try:
                await active_pool.fetchval("SELECT 1")
                return active_pool   # still healthy
            except Exception:
                await active_pool.close()
                active_pool = None
        # Try endpoints in order
        for ep in ENDPOINTS:
            pool = await connect_to_endpoint(ep)
            if pool:
                active_pool = pool
                active_endpoint = ep["name"]
                print("Connected to: " + active_endpoint)
                return active_pool
        active_pool = None
        active_endpoint = None
    return active_pool

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="get_order",
            description="Fetch order details by order ID",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"}
                },
                "required": ["order_id"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    pool = await get_pool()
    if pool is None:
        return [types.TextContent(type="text",
            text="Service unavailable - all database endpoints are down. Please retry later.")]

if name == "get_order":
        try:
            row = await pool.fetchrow(
                "SELECT order_id, customer_id, status, total_rs FROM orders WHERE order_id = $1",
                arguments["order_id"]
            )
            if row:
                return [types.TextContent(type="text",
                    text=str(dict(row)) + " [via " + active_endpoint + "]")]
            return [types.TextContent(type="text", text="Order not found")]
        except Exception as e:
            return [types.TextContent(type="text", text="Query failed: " + str(e))]

async def main():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

It gives the following output,

# Normal operation - primary available:
Connected to: primary-mumbai
Tool: get_order({order_id: "ORD-7701"})
{"order_id": "ORD-7701", "customer_id": "C-1201", "status": "SHIPPED",
 "total_rs": 78000} [via primary-mumbai]

# Primary goes down - automatic failover:
Connected to: standby-delhi
Tool: get_order({order_id: "ORD-7702"})
{"order_id": "ORD-7702", "customer_id": "C-3302", "status": "CONFIRMED",
 "total_rs": 34500} [via standby-delhi]

# Both endpoints down - graceful degradation:
Service unavailable - all database endpoints are down. Please retry later.

For Cloud Run deployments, use Global External Load Balancer with backend services in multiple regions so failover happens at the network layer before the MCP server even receives the request. Test failover regularly by simulating primary failures in staging - agents should experience only a brief delay, not a hard failure. Pair this pattern with Cloud Monitoring uptime checks on MCP server health endpoints so on-call engineers are alerted before agents start returning degraded responses.

Send your comments, suggestions or queries regarding this site to [email protected].