How can I manage resource across multiple tool calls during one session?

Description

I was messing around multiple agents running in parallel, and suffered from some kind of race condition. I had to write a lock for them.

It looks like each agent has their own session, but I don't have any idea to handle how they create and terminate.

The code can be like below, assuming it is a streamable-http app:

import asyncio

import mcp
from mcp.server.fastmthemcp import Context
import mcp.events # some mcp protocol events, maybe session creation or termination.

mcp_server = mcp.server.FastMCP("demo-locking")
lock = None

async def acquire_lock(ctx: Context):
    # assume ctx.session.session_id = mcp-session-id acquired from headers, however not defined in this field.
    while lock is not None and lock != ctx.session.session_id:
        await asyncio.sleep(1)
        ctx.info("Waiting for the lock")
    lock = ctx.session.session_id


@mcp_server.tool()
async def tool1(ctx: Context):
    await acquire_lock(ctx)
    # ... do tool things

@mcp_server.tool()
async def tool2(ctx: Context):
    await acquire_lock(ctx)
    # ... do tool things

async def release_lock(ctx: Context):
    if lock == ctx.session.session_id:
        lock = None

mcp_server.add_handler(mcp.events.SESSION_TERMINATE, release_lock) # should be something like this to register a session cleanup.

app = mcp_server.streamable_http_app()

import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

References

No response

Inizia leggendo l’entry point streamable_http_app() e i riferimenti a Context/session nell’esempio, quindi esamina come sono attualmente definiti add_handler() e SESSION_TERMINATE. L’issue non identifica file o test esistenti. Per considerarlo completato sarebbe necessario decidere e documentare un approccio per coordinare le risorse tra le chiamate agli strumenti e gestire la pulizia delle sessioni.