Where can I find a working python example that show how to raise ToolListChangedNotification at the server and process the notification on the Client

[juacostab]

Pre-submission Checklist

• I have checked that this question would not be more appropriate as an issue in a specific repository
• I have searched existing discussions and documentation for answers

Question Category

• Protocol Specification
• SDK Usage
• Server Implementation
• General Implementation
• Documentation
• Other

Your Question

I want to use the Python SDK functionality that allows a server sending notifications/tools/list_changed to a client and have the client process the notification and re-request the tool list.

I have experimented with what its available in the repo and the user guides (only shows how it is done with JavaScript).

It seems the only way to do this is to create a subclass of ClientSession and provide an override to

    async def _received_notification(self, notification: ReceiveNotificationT) -> None:

Any help will be gladly appreciated

[juacostab]

@jspahrsummers any guidancefor the Python SDK ?

[Finndersen]

I'm also interested in how this works!

[abhishekbhakat]

from mcp.server.session import ServerSession
ServerSession.send_tool_list_changed()

[juacostab]

Can you share a bit more where this call must be done calling it like this fails

        """Raise notification list changed """
        logger.info("Dispatching tool notification")
        if self.session is None:
            logger.error("Session is not initialized")
            return

        try:
            result = await ServerSession.send_tool_list_changed()

with an error indicating self is missing

[abhishekbhakat]

https://github.com/abhishekbhakat/airflow-mcp-server/blob/main/src/airflow_mcp_server/hierarchical_manager.py#L112-L126

This is what I did to make send updated tool list. It's not that straightforward. I used context to get the session.
See if that helps you.

[juacostab]

Thanks a lot for the response. I thought the call was as simple as that. I implemented an earlier solution where I do use the FastMCP instance to get the contexts and then the session

[MischaPanch]

I also encountered this and since it's far from trivial to get the Session object from anywhere outside of the running mcp app, I post my solution (for stdio servers, can be extended for the other transports as well) as a reference:

import anyio
import janus
from mcp import stdio_server, types
from mcp.server.fastmcp import server
from mcp.server.fastmcp.server import FastMCP, Settings
from mcp.server.fastmcp.tools.base import Tool as MCPTool
from mcp.shared.message import ServerMessageMetadata, SessionMessage
from mcp.shared.session import RequestId, SendNotificationT
from mcp.types import JSONRPCMessage, JSONRPCNotification

class FastMCPWithExternalCommandExecution(FastMCP):
    """
    Extends FastMCP to support executing tools in an external process.
    """

    @dataclass
    class ExternalCall:
        method_name: str
        args: dict[str, Any]

    def run(
        self,
        transport: Literal["stdio", "sse", "streamable-http"] = "stdio",
        mount_path: str | None = None,
    ) -> None:
        """Run the FastMCP server with cross-thread queue support.
        We need this to be able to run tool executions and other operations in
        the main thread.
        """
        self._external_calls_queue: janus.Queue[FastMCPWithExternalCommandExecution.ExternalCall] = janus.Queue()

        async def with_external_calls_queue() -> None:
            # Start background queue consumer
            async with anyio.create_task_group() as tg:
                tg.start_soon(self._process_external_calls_queue_loop)

                # Run the normal transport startup
                match transport:
                    case "stdio":
                        await self.run_stdio_async()
                    case "sse":
                        await self.run_sse_async(mount_path)
                    case "streamable-http":
                        await self.run_streamable_http_async()

        anyio.run(with_external_calls_queue)

    async def _process_external_calls_queue_loop(self) -> None:
        """Continuously process items from the external calls queue."""
        while True:
            external_call: FastMCPWithExternalCommandExecution.ExternalCall = await self._external_calls_queue.async_q.get()
            try:
                method = getattr(self, external_call.method_name)
                await method(**external_call.args)
            except Exception as e:
                log.error(f"Error running external call {external_call}: {e}")

    def update_tool_list(self) -> None:
        """Sync method to request tool list update from other threads."""
        update_tool_list_call = FastMCPWithExternalCommandExecution.ExternalCall(
            method_name="send_tool_list_changed",
            args={},
        )
        self._external_calls_queue.sync_q.put(update_tool_list_call)
        
    # For demo purposes, we aren't using this now
    def call_tool_sync(self, tool_name: str, parameters: dict[str, Any]) -> None:
        """Sync method to call a tool from other threads."""
        call_tool = FastMCPWithExternalCommandExecution.ExternalCall(
            method_name="call_tool",
            args={
                "tool_name": tool_name,
                "parameters": parameters,
            },
        )
        self._external_calls_queue.sync_q.put(call_tool)

    # Copied from mcp.server.session.Session since we don't have access to it here, but
    # we hacked our way to have access to the write stream.
    async def send_notification(
        self,
        notification: SendNotificationT,
        related_request_id: RequestId | None = None,
    ) -> None:
        """
        Emits a notification, which is a one-way message that does not expect
        a response.
        """
        # Some transport implementations may need to set the related_request_id
        # to attribute to the notifications to the request that triggered them.
        jsonrpc_notification = JSONRPCNotification(
            jsonrpc="2.0",
            **notification.model_dump(by_alias=True, mode="json", exclude_none=True),
        )
        session_message = SessionMessage(
            message=JSONRPCMessage(jsonrpc_notification),
            metadata=ServerMessageMetadata(related_request_id=related_request_id) if related_request_id else None,
        )
        await self._write_stream.send(session_message)

    # also copied from mcp.server.session.Session
    async def send_tool_list_changed(self) -> None:
        """Send a tool list changed notification."""
        await self.send_notification(
            types.ServerNotification(
                types.ToolListChangedNotification(
                    method="notifications/tools/list_changed",
                )
            )
        )

    # override to save the read and write streams, which permits talking to (stdio) clients outside of request handling
    # by using the external calls queue
    async def run_stdio_async(self) -> None:
        """Run the server using stdio transport."""
        async with stdio_server() as (read_stream, write_stream):
            self._read_stream = read_stream
            self._write_stream = write_stream
            await self._mcp_server.run(
                read_stream,
                write_stream,
                self._mcp_server.create_initialization_options(),
            )