durable-mcp 0.2.1

Durable Model Context Protocol (MCP) helpers for building resumable MCP servers on top of Reboot durable workflows.

Reboot Durable MCP

A framework for building durable MCP servers.

• Takes advantage of the protocols ability to resume after disconnection, e.g., due to the server getting rebooted.

• Any existing requests will be retried safely using Reboot workflows.

• Using Reboot you can run multiple replicas of your server, and session messages will always be routed to the same replica.

Requirements

• Linux and macOS
• Python >= 3.12.11
• Docker

Install

We recommend using uv, as it will manage the version of Python for you. For example, to start a new project in the directory foo:

uv init --python 3.12.11 .
uv add durable-mcp

Activate the venv:

source .venv/bin/activate

Make sure you have Docker running:

docker ps

Building an MCP server

Instead of using FastMCP from the MCP SDK, you use DurableMCP. Here is a simple server to get you started:

import asyncio
from reboot.aio.applications import Application
from reboot.mcp.server import DurableMCP, ToolContext
from reboot.std.collections.v1.sorted_map import SortedMap

# `DurableMCP` server which will handle HTTP requests at path "/mcp".
mcp = DurableMCP(path="/mcp")


@mcp.tool()
async def add(a: int, b: int, context: ToolContext) -> int:
    """Add two numbers and also store result in `SortedMap`."""
    await SortedMap.ref("adds").Insert(
        context,
        entries={f"{a} + {b}": f"{a + b}".encode()},
    )
    return a + b


async def main():
    # Reboot application that runs everything necessary for `DurableMCP`.
    application = Application(servicers=mcp.servicers())

    # Mounts the server at the path specified.
    application.http.mount(mcp.path, factory=mcp.streamable_http_app_factory)

    await application.run()


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

You can run the server via:

rbt dev run --python --application=path/to/your/main.py --working-directory=. --no-generate-watch

While developing you can tell rbt to restart your server when you modify files by adding one or more --watch=path/to/**/*.py to the above command line.

We recommend you move all of your command line args to a .rbtrc:

# This file will aggregate all of the command line args
# into a single command line that will be run when you
# use `rbt`.
#
# For example, to add args for running `rbt dev run`
# you can add lines that start with `dev run`. You can add
# one or more args to each line.

dev run --python
dev run --application=path/to/your/main.py
dev run --watch=path/to/**/*.py --watch=different/path/to/**/*.py

Then you can just run:

rbt dev run

Testing your MCP server

You can use the MCP Inspector to test out the server, or create a simple client.

import asyncio
from reboot.mcp.client import connect, reconnect

URL = "http://localhost:9991"


async def main():
    # Connect is a helper around creating a streamable HTTP client and
    # session from the MCP SDK, you can also use those if you prefer!
    async with connect(URL + "/mcp") as (
        session, session_id, protocol_version
    ):
        print(await session.list_tools())
        print(await session.call_tool("add", arguments={"a": 5, "b": 3}))


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

Supported client --> server requests:

• initialize
• tools/call
• tools/list
• prompts/get
• prompts/list
• resources/list
• resources/read
• resources/templates/list
• resources/subscribe
• resources/unsubscribe
• completion/complete
• logging/setLevel

Supported client --> server notifications:

• notifications/initialized
• notifications/progress (for server initiated requests, e.g., elicitation)
• notifications/roots/list_changed

Supported client <-- server requests:

• elicitation/create
• roots/list
• sampling/createMessage

Supported client <-- server notifications:

• notifications/progress
• notifications/message
• notifications/prompts/list_changed
• notifications/resources/list_changed
• notifications/tools/list_changed
• notifications/resources/updated

Supported client <--> server notifications:

• notifications/cancelled

TODO:

• EventStore support for resumability
• Add examples of how to test via Reboot().start/up/down/stop()
• Auth
• Docs
• yapf
• Push to durable-mcp in pypi.org
• Pydantic state for each session

Contributing

First grab all dependencies:

uv sync

Activate the venv:

source .venv/bin/activate

Generate code:

rbt generate

Make sure you have Docker running:

docker ps

Make your changes and run the tests:

pytest tests