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

Commencez par lire le point d’entrée streamable_http_app() et les références à Context/session dans l’exemple, puis examinez comment add_handler() et SESSION_TERMINATE sont actuellement définis. L’issue n’identifie aucun fichier ni test existant. Pour considérer le travail terminé, il faudrait définir et documenter une approche pour coordonner les ressources entre les appels d’outils et gérer le nettoyage des sessions.