Create an MCP server with lifecycle management

Introduction

When you run an MCP server in production, startup failures and uncleaned resources at shutdown can silently corrupt tool state or leave client connections hanging. A bare server.run() call gives you no structured place to initialize databases, register metrics, or ensure teardown happens in the right order. By the end of this lesson, you'll be able to wrap any MCP server in a lifecycle manager that tracks discrete states—created, starting, running, stopping, stopped—and fires async hooks at each transition, letting your initialization and teardown logic execute at exactly the right moment.

Key Terminology

• ServerState — An Enum whose six variants (CREATED, STARTING, RUNNING, STOPPING, STOPPED, ERROR) represent every discrete phase a managed MCP server can occupy, making the server's current condition unambiguous at any point during its lifecycle.
• LifecycleHooks — A dataclass holding four optional async callbacks—on_start, on_ready, on_shutdown, and on_stop—that are invoked at specific ServerState transitions so callers can inject initialization or teardown logic without modifying the core run loop.
• ManagedServer — A wrapper dataclass that combines the SDK Server instance, the current ServerState, a LifecycleHooks configuration, and a private asyncio.Event to drive the full startup and shutdown sequence.
• shutdown event — The asyncio.Event stored as _shutdown_event on ManagedServer that acts as the coordination point between an external signal handler and the running event loop; it is set inside shutdown() to trigger teardown.
• managed_server context manager — An asynccontextmanager-decorated generator that yields a ManagedServer and guarantees shutdown() is called in its finally block, preventing transport resources from leaking even when surrounding code raises an exception.
• ERROR state — The ServerState.ERROR variant entered when start()'s run loop raises an unhandled exception, preserving a meaningful state value so callers and log readers can distinguish a crash from a clean stop.

Concepts

Why a Bare server.run() Call Isn't Enough

Calling server.run() directly gives you a blocking coroutine that accepts MCP requests until it exits — but no structured extension points around it. There is no guaranteed slot for acquiring a database connection before the first tool call arrives, registering observability metrics, or flushing open resources when the process receives a shutdown signal. If the run loop exits because of an unhandled exception, the calling code has no way to distinguish a crash from a clean stop without scattering ad-hoc try/except guards at every call site.

The lifecycle manager pattern moves those concerns out of scattered call-site code and into a single wrapper that knows when each phase starts and what to do at each transition.

The Lifecycle State Machine

A state machine is the right mental model here because the server is never in an ambiguous condition: it is always in exactly one of CREATED, STARTING, RUNNING, STOPPING, STOPPED, or ERROR. Each transition is driven by a deliberate action — start() advances CREATED → STARTING → RUNNING; shutdown() moves RUNNING → STOPPING; _cleanup() lands on STOPPED; any unhandled exception in the run loop drops the state to ERROR instead of leaving it frozen at RUNNING.

The ERROR variant is especially important: without it, a crash would leave state == ServerState.RUNNING even though the run loop is gone, causing any subsequent guard that checks if self.state != ServerState.RUNNING — such as shutdown() itself — to behave incorrectly. Naming the error condition explicitly prevents that class of silent corruption (see Code Walkthrough).

Hook Points as Ordered Extension Joints

The four hooks — on_start, on_ready, on_shutdown, and on_stop — map to different moments in the state machine, and the ordering matters. on_start fires before the run loop begins, making it the right place for one-time setup that must complete before any tool request can arrive (opening a database pool, warming a cache). on_ready fires after ServerState.RUNNING is set but still before server.run() blocks, making it the right place for emitting a readiness signal or logging that the server is accepting work.

On the teardown side, on_shutdown fires while the server is still in STOPPING — before the event loop has exited — so async cleanup operations (draining a queue, flushing metrics) can await properly. on_stop fires last, after _cleanup() is reached via the finally block, confirming teardown is complete. All four are typed as Optional[Callable[[], Awaitable[None]]] and default to None, so callers supply only the hooks they actually need.

The Context Manager as a Safety Net

The managed_server async context manager adds one guarantee on top of ManagedServer: shutdown() is called in a finally block regardless of how the async with block exits. Without this, an exception in the caller's code could leave the server in RUNNING state with transport streams still open and no on_stop hook ever fired.

Because shutdown() is a no-op when state != ServerState.RUNNING, calling it from the context manager's finally is safe even if the server never reached RUNNING — there is no double-teardown risk. This makes managed_server the recommended entry point: it pairs resource acquisition with resource release at the language level, the same contract that asynccontextmanager enforces for any other async resource.

Code Walkthrough

Now that you understand how lifecycle states and hook points fit together, the implementation ties them into three layers: a ServerState enum, a LifecycleHooks dataclass, and a ManagedServer wrapper that coordinates the full startup and shutdown sequence.

Code snippetpython
1import asyncio
2import logging
3from dataclasses import dataclass, field
4from typing import Optional, Callable, Awaitable
5from contextlib import asynccontextmanager
6from enum import Enum, auto
7from mcp.server import Server
8
9logger = logging.getLogger(__name__)
10
11class ServerState(Enum):
12    """Server lifecycle states."""
13    CREATED = auto()
14    STARTING = auto()
15    RUNNING = auto()
16    STOPPING = auto()
17    STOPPED = auto()
18    ERROR = auto()
19
20@dataclass
21class LifecycleHooks:
22    """Callbacks for server lifecycle events."""
23    on_start: Optional[Callable[[], Awaitable[None]]] = None
24    on_ready: Optional[Callable[[], Awaitable[None]]] = None
25    on_shutdown: Optional[Callable[[], Awaitable[None]]] = None
26    on_stop: Optional[Callable[[], Awaitable[None]]] = None
27
28@dataclass
29class ManagedServer:
30    """MCP server with lifecycle management."""
31    server: Server
32    state: ServerState = ServerState.CREATED
33    hooks: LifecycleHooks = field(default_factory=LifecycleHooks)
34    _shutdown_event: asyncio.Event = field(default_factory=asyncio.Event)

ServerState maps every phase the server can occupy; the ERROR variant captures unhandled exceptions without leaving state ambiguous. LifecycleHooks collects four optional async callbacks—defaulting to None so callers only supply what they need. ManagedServer holds the SDK Server instance, the current state, the hooks, and a private asyncio.Event that coordinates the shutdown signal between the caller and the running event loop.

The behavioral methods and context manager complete the pattern:

Code snippetpython
1    async def start(self, read_stream, write_stream, init_options) -> None:
2        self.state = ServerState.STARTING
3        logger.info(f"Starting server: {self.server.name}")
4        if self.hooks.on_start:
5            await self.hooks.on_start()
6        try:
7            self.state = ServerState.RUNNING
8            if self.hooks.on_ready:
9                await self.hooks.on_ready()
10            logger.info(f"Server ready: {self.server.name}")
11            await self.server.run(read_stream, write_stream, init_options)
12        except Exception as e:
13            self.state = ServerState.ERROR
14            logger.error(f"Server error: {e}")
15            raise
16        finally:
17            await self._cleanup()
18
19    async def shutdown(self) -> None:
20        if self.state != ServerState.RUNNING:
21            return
22        self.state = ServerState.STOPPING
23        logger.info(f"Shutting down server: {self.server.name}")
24        if self.hooks.on_shutdown:
25            await self.hooks.on_shutdown()
26        self._shutdown_event.set()
27
28    async def _cleanup(self) -> None:
29        if self.hooks.on_stop:
30            await self.hooks.on_stop()
31        self.state = ServerState.STOPPED
32        logger.info(f"Server stopped: {self.server.name}")
33
34@asynccontextmanager
35async def managed_server(
36    server: Server,
37    hooks: Optional[LifecycleHooks] = None
38):
39    """Context manager for server lifecycle."""
40    managed = ManagedServer(server=server, hooks=hooks or LifecycleHooks())
41    try:
42        yield managed
43    finally:
44        if managed.state == ServerState.RUNNING:
45            await managed.shutdown()

start drives the state machine from STARTING to RUNNING, fires on_start before the run loop and on_ready once the loop is accepting requests, and always calls _cleanup through the finally block whether the run loop exits normally or raises. shutdown is a no-op unless the server is currently RUNNING, making it safe to call from a signal handler without an additional guard at the call site. The managed_server context manager guarantees shutdown runs even if the surrounding code raises an exception, so no transport resources are silently leaked.

Confirm that ServerState.RUNNING appears in your logs before the first tool call arrives and that ServerState.STOPPED appears after the process receives a shutdown signal.

Do's and Don'ts

Having walked through the material above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do place _cleanup() inside the finally block of start() — Without this guarantee, an unhandled exception in server.run() exits the state machine without firing on_stop or advancing to STOPPED, leaving transport resources held and ManagedServer.state permanently stuck in RUNNING or STARTING.
2. ✓Do rely on shutdown()'s built-in no-op guard (if self.state != ServerState.RUNNING: return) — Because the guard makes shutdown() safe to call from OS signal handlers without knowing current state, adding a redundant check at the call site creates drift and masks double-invocation bugs rather than preventing them.
3. ✓Do wrap server runs in the managed_server async context manager rather than calling ManagedServer.start() directly — The context manager's finally block invokes shutdown() even when the surrounding coroutine raises, guaranteeing on_shutdown and on_stop hooks execute and no client connections are silently abandoned.

Don'ts

1. ✗Don't call server.run() as a bare statement without a ManagedServer wrapper — A bare call gives you no structured hook point to initialize databases or register metrics before the run loop starts (on_start) or confirm readiness after it begins accepting requests (on_ready), and no guaranteed teardown order when the loop crashes.
2. ✗Don't fire on_ready before transitioning state to ServerState.RUNNING — The on_ready callback is the signal that the server is accepting tool calls; invoking it while still in STARTING gives any downstream readiness probe a false positive and breaks logic that guards on managed.state == ServerState.RUNNING.
3. ✗Don't omit the ServerState.ERROR transition inside the except block of start() — If an exception from server.run() is re-raised without first setting state = ServerState.ERROR, the state remains RUNNING, causing a subsequent shutdown() call to attempt teardown on a server that has already crashed and potentially firing on_shutdown against uninitialized resources.