Add client subscription and server event broadcasting

Question

Hello,

I am trying to implement a pub/sub pattern for client event subscriptions. I need it to enable communication between AI agents via an MCP server. This is the message flow that I'm trying to achieve, but between two clients like this:

ClientA subscribe → MCP
- ClientA registers itself as a subscriber with the MCP server
- It's expressing interest in receiving certain types of events/messages
ClientB trigger event → MCP
- ClientB sends an event or message to the MCP server
- ClientB acts as a publisher/producer of events
MCP broadcast → subscribers(ClientA)
- The MCP server receives the event from ClientB
- It then broadcasts/forwards that event to all registered subscribers
- In this case, ClientA receives the event

I can send notifications through the context(this is pseudo code):

MCP Server:

async def logging_enabled_tool(ctx: Context) -> None:
    """Send back notifications to current session."""

    log_notification = LoggingMessageNotification(
        method="notifications/message",
        params={ "level": "info", "data": "Some message..."}
    )

    await ctx.session.send_notification(log_notification)

MCP Client:

async def handle_notifications(params: LoggingMessageNotificationParams) -> None:
    """Handle logging notifications."""
    print(f"Received log message: {params}")

 async with streamablehttp_client(
            f"{app_url}", auth=DatabricksOAuthClientProvider(workspace_client)
        ) as (read_stream, write_stream, _), ClientSession(
            read_stream, write_stream, logging_callback=handle_notifications
        ) as session:
             await session.initialize()

            result = await session.call_tool("logging_enabled_tool")
            print(f"Result from tool: {result.content}")

This example works because the notification is sent back to the client during tool execution. However, I need an option to subscribe multiple sessions to events that will be asynchronously broadcast by the server at any point in time.

ClientA:

client_session.call_tool("subscribe", { "event": "event-name" })
# response: successfully subscribed to event...

Server:

def subscribe(server, ctx: Context) -> str:
    # keep reference to all subscribed sessions
    server.subscribers.append(ctx.session)
    return "Subscribed to notifications."

async def broadcast_message(arguments: Dict[str, Any]) -> None:
    """Broadcast a message to all connected sessions."""
    # include arguments.get("event") in the LoggingMessageNotification
    log_notification = LoggingMessageNotification(...)

    for session in server.subscribers:
        await session.send_notification(log_notification)

ClientB:

client_session.call_tool("broadcast_message", { "event": "event-name" })
# response: broadcasted message to N clients...

For this to work, ClientA should await indefinitely; otherwise, the handle_notifications callback is not called:

 async with streamablehttp_client(...) as (...), ClientSession(
            read_stream, write_stream, logging_callback=handle_notifications
        ) as session:
           # initialize... 
           # subscribe...
          
           # and wait
           await asyncio.Future()

This approach doesn't work for my needs as I am initializing connection to the MCP from an AI agent and I need the agent to stay responsive, this blocks all execution.

Not sure if the streamablehttp transport is the best approach, I'm trying to follow the documentation as much as possible. Other suggestions are more than welcome. The core functionality that I need is async communication between Agents. The MCP is also required as we need it for other important tools. We thought we can also use it for communication between agents.

A little more context about the environment:

The Agents are MCP tool-calling LangGraph agents in Databricks - https://docs.databricks.com/aws/en/notebooks/source/generative-ai/langgraph-mcp-tool-calling-agent.html
The MCP is a custom MCP hosted on Databricks App - https://docs.databricks.com/aws/en/generative-ai/mcp/custom-m

Additional Context

No response

One design boundary I’d keep: don’t use MCP notifications as a cross-client message bus. Context/session notifications are scoped to the current client session; keeping ctx.session refs in server.subscribers gets fragile once clients reconnect, transports differ, or auth boundaries matter.

For agent-to-agent events, I’d put the pub/sub state behind an explicit broker or durable queue, then expose MCP tools like subscribe, publish, poll_events, or ack_event. The MCP calls manage the subscription and fetch events; the long-lived event stream should not block the agent loop.

If you stay with streamable HTTP, run the listener task outside the agent execution path and push events into a local queue that the agent can poll on its next turn. I’d also add TTL, consumer id, and ack/lease semantics so dead sessions don’t remain subscribers forever.

On 1.x the comment above holds: ctx.session belongs to the calling client, so a stashed-session subscriber list is fragile and durable pub/sub belongs behind a broker exposed through tools.

On main (v2, the 2026-07-28 protocol) the flow you sketched is subscriptions/listen: ClientA opens client.listen(resource_subscriptions=[uri]), ClientB's tool call runs await ctx.notify_resource_updated(uri), and the server fans that out to every stream subscribed to that URI, across sessions. Events are refetch cues, not payloads, so the message itself lives behind the resource; there is no replay either, so a stream that was down misses those events (keep a durable queue if delivery must survive that). Nothing needs to block: run the watcher as a task beside your main flow — see Watching without blocking and the server side.

Runnable sketch: two clients, one server

import anyio

from mcp import Client
from mcp.client.subscriptions import ResourceUpdated
from mcp.server.mcpserver import Context, MCPServer

mcp = MCPServer("Broker")
EVENTS: list[str] = []


@mcp.resource("events://feed")
def feed() -> str:
    return "\n".join(EVENTS)


@mcp.tool()
async def publish(event: str, ctx: Context) -> str:
    EVENTS.append(event)
    await ctx.notify_resource_updated("events://feed")
    return "published"


async def watch(a: Client, subscribed: anyio.Event) -> None:
    async with a.listen(resource_subscriptions=["events://feed"]) as sub:
        subscribed.set()  # acked: nothing published after this is missed
        async for event in sub:
            if isinstance(event, ResourceUpdated):
                print(await a.read_resource(event.uri))  # a cue: refetch
                return


async def main() -> None:
    async with Client(mcp) as a, Client(mcp) as b, anyio.create_task_group() as tg:
        subscribed = anyio.Event()
        tg.start_soon(watch, a, subscribed)  # watcher runs beside A's main flow
        await subscribed.wait()
        await b.call_tool("publish", {"event": "hello from B"})


anyio.run(main)

Closing this as part of a wider backlog cleanup around the v2 release. Feel free to reopen if this is still relevant.

_{AI Disclaimer}

Navigation Menu

Repository navigation

Add client subscription and server event broadcasting #1512

Question

Additional Context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Add client subscription and server event broadcasting #1512

Description

Question

Additional Context

Activity

tangweigang-jpg commented on Jun 29, 2026

maxisbey commented on Jul 24, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions