Python

1. Setup your project

   uv init --python 3.12.11
   uv add durable-mcp

2. Activate your venv

   source .venv/bin/activate

3. Make sure Docker is running:

   docker ps

4. Building your MCP server

   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())

5. Run your server

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

6. Setup your .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

7. Then just run:

   rbt dev run

8. Test your MCP server

   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
       # directly 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())

9. Perform a side-effect "at least 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_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,
       )
   
       # ...

10. Performing a side-effect "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,
        )
    
        # ...

11. Debugging

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

Next steps

Well done! You made it! Here are some great next steps:

• Every tool (and eventually prompt, and resource) is a generic Reboot workflow! Check out the docs for the other things you can do in a workflow.
• Consider creating your own Reboot durable data structures to use from your durable MCP servers!