durable-mcp 0.3.0

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 DurableContext, DurableMCP
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: DurableContext) -> int:
    """Add two numbers and also store result in `SortedMap`."""
    result = a + b
    await SortedMap.ref("adds").Insert(
        context,
        entries={f"{a} + {b}": f"{result}".encode()},
    )
    return result


async def main():
    # Reboot application that runs everything necessary for `DurableMCP`.
    await mcp.application().run()


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

You can run the server via:

rbt dev run --python --application=path/to/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 --no-generate-watch
dev run --python --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 that creates a streamable HTTP client
    # and session using the MCP SDK. You can also write a client the
    # direclty uses the MCP SDK 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())

Performing a side-effect "at least once"

Within your tools (and eventually within your prompts and resources too), you can try and perform a side-effect that it is safe to try one or more times until success using at_least_once. Usually what makes it safe to perform one or more times is that you can somehow do it idempotently, e.g., passing an idempotency key as part of an API call. Use at_least_once for this, for example:

from reboot.aio.workflows import at_least_once
from reboot.mcp.server import DurableContext, DurableMCP


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


@mcp.tool()
async def add(a: int, b: int, context: DurableContext) -> int:

    async def do_side_effect_idempotently() -> int:
        """
        Pretend that we are doing a side-effect that we can try
        more than once because we can do it idempotently, hence using
        `at_least_once`.
        """
        return a + b

    result = await at_least_once(
        "Do side-effect _idempotently_",
        context,
        do_side_effect_idempotently,
        type=int,
    )

    # ...

Performing a side-effect "at most once"

Within your tools (and eventually within your prompts and resources too), you can try and perform a side-effect that can only be tried once using at_most_once. If you can attempt to perform the side-effect more than once always prefer at_least_once. Here's an example of at_most_once:

from reboot.aio.workflows import at_least_once
from reboot.mcp.server import DurableContext, DurableMCP


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


@mcp.tool()
async def add(a: int, b: int, context: DurableContext) -> int:

    async def do_side_effect() -> int:
        """
        Pretend that we are doing a side-effect that we can only
        try to do once because it is not able to be performed
        idempotently, hence using `at_most_once`.
        """
        return a + b

    # NOTE: if we reboot, e.g., due to a hardware failure, within
    # `do_side_effect()` then `at_most_once` will forever raise with
    # `AtMostOnceFailedBeforeCompleting` and you will need to handle
    # appropriately.
    result = await at_most_once(
        "Do side-effect",
        context,
        do_side_effect,
        type=int,
    )

    # ...

Debugging

Start by enabling debug logging:

mcp = DurableMCP(path="/mcp", log_level="DEBUG")

Then consider wrapping your tool/prompt/etc functions in a try/except because the MCP SDK will "swallow" errors and all that the client gets back is just that the request failed.

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:

• Add examples of using at_least_once and at_most_once
• Add examples of how to test via Reboot().start/up/down/stop()
• Add example of rebooting server using MCP Inspector version 0.16.7 which includes modelcontextprotocol/inspector#787
• Auth
• Docs at docs.reboot.dev
• yapf
• 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