MCP Events: High-level server API with @mcp.event() and emit_event()

[elijahr]

Summary

• declare_event() and @mcp.event() decorator for topic declaration
• emit_event() on FastMCP instances and Context objects for broadcast delivery
• Session registry with connect/disconnect lifecycle management
• Subscription/unsubscribe/list handlers via _receive_loop interception
• Parameterized topic pattern matching for retained value auto-detection
• Topic depth enforcement (max 8 segments)
• 71 tests including protocol-layer roundtrip and error path coverage

Context

High-level server API layer for MCP events SEP reference implementation.
Builds on python-sdk event types. Self-contains event types due to SDK version constraints.

Test plan

• uv run pytest tests/server/test_events.py -x -v

[gemini-code-assist]

Code Review

This pull request introduces an application-level event system to FastMCP, supporting MQTT-style wildcard subscriptions and retained events. Key changes include a new event registry, publishing capabilities in the Context and FastMCP classes, and JSON-RPC interception for event methods. Review feedback identifies a bug in the _receive_loop override that could cause runtime errors and suggests refactoring to avoid duplicating SDK logic. Further improvements are proposed to optimize session lookups using a dictionary and to deduplicate retained events during subscription.

[gemini-code-assist]

The _receive_loop override duplicates a significant amount of the SDK's internal message dispatch logic, which is fragile and has led to bugs in the duplicated code. Specifically, _handle_sdk_request and _handle_sdk_notification incorrectly call _handle_incoming with validated models or responders instead of the expected SessionMessage or Exception types, which will cause runtime errors.

Instead of duplicating the logic, you should intercept only the event-related methods and delegate all other messages to the base class's _handle_incoming method. This also makes the redundant _handle_sdk_request and _handle_sdk_notification methods unnecessary.

                async for message in self._read_stream:
                    if isinstance(message, Exception):
                        await self._handle_incoming(message)
                        continue

                    root = message.message.root
                    if isinstance(root, JSONRPCRequest) and root.method in self._EVENT_METHODS:
                        # Only handle events/* if the server has events capability
                        if self.fastmcp._event_topics:
                            await self._handle_event_request(
                                method=root.method,
                                params=root.params or {},
                                request_id=root.id,
                            )
                        else:
                            # No events capability: return -32601 Method not found
                            error_response = JSONRPCError(
                                jsonrpc="2.0",
                                id=root.id,
                                error=ErrorData(
                                    code=-32601,
                                    message=f"Method not found: {root.method}",
                                ),
                            )
                            session_message = SessionMessage(
                                message=JSONRPCMessage(error_response)
                            )
                            await self._write_stream.send(session_message)
                    else:
                        # Delegate to parent for standard SDK validation
                        await self._handle_incoming(message)

[gemini-code-assist]

Consider changing _active_sessions to a dictionary mapping session IDs to session objects. This allows for efficient $O(1)$ lookups during event emission instead of $O(N)$ iteration over all active sessions.

Suggested change

	self._active_sessions: set[MiddlewareServerSession] = set()
	self._active_sessions: dict[str, MiddlewareServerSession] = {}

[gemini-code-assist]

Update session registration to use the dictionary mapping suggested in FastMCP.

Suggested change

	self.fastmcp._active_sessions.add(session)
	self.fastmcp._active_sessions[session_id] = session

[gemini-code-assist]

Update session cleanup to use the dictionary mapping suggested in FastMCP.

Suggested change

	self.fastmcp._active_sessions.discard(session)
	self.fastmcp._active_sessions.pop(session_id, None)

[gemini-code-assist]

Broadcasting events by iterating over all active sessions and checking their IDs is inefficient ($O(N_{sessions})$). Since matching_session_ids already contains the IDs of sessions that should receive the event, it is much faster to look them up directly in the suggested _active_sessions dictionary ($O(N_{matches})$).

Suggested change

	for session in list(self._active_sessions):
	sid = getattr(session, "_fastmcp_event_session_id", None)
	if sid is None or sid not in matching_session_ids:
	continue
	# Broadcast to matching active sessions
	for sid in matching_session_ids:
	session = self._active_sessions.get(sid)
	if session:
	try:
	await session.send_notification(notification)
	except Exception:
	logger.warning(
	f"Failed to deliver event to session {sid}",
	exc_info=True,
	)

[gemini-code-assist]

The retained_events list can contain duplicate events if a client subscribes to multiple overlapping patterns (e.g., myapp/status and myapp/#) in a single request. It is recommended to deduplicate these events by their unique event_id before returning them to the client.

        retained_events: dict[str, RetainedEvent] = {}

        for pattern in topics:
            # Validate topic depth (max 8 segments)
            segments = pattern.split("/")
            if len(segments) > server._MAX_TOPIC_DEPTH:
                raise McpError(
                    mcp.types.ErrorData(
                        code=-32602,
                        message=(
                            f"Subscription pattern has {len(segments)} segments, "
                            f"maximum depth is {server._MAX_TOPIC_DEPTH}: {pattern!r}"
                        ),
                    )
                )

            # Check if the pattern matches any declared topic
            if not server._match_declared_topic(pattern):
                rejected.append(RejectedTopic(pattern=pattern, reason="unknown_topic"))
                continue

            await server._subscription_registry.add(session_id, pattern)
            subscribed.append(SubscribedTopic(pattern=pattern))

            # Deliver retained values for this pattern
            matching = await server._retained_store.get_matching(pattern)
            for event in matching:
                retained_events[event.event_id] = event

        result = EventSubscribeResult(
            subscribed=subscribed,
            rejected=rejected,
            retained=list(retained_events.values()),
        )
        return result.model_dump(exclude_none=True)

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request introduces a comprehensive event system to FastMCP, allowing servers to declare topics, publish events, and manage client subscriptions with MQTT-style wildcards. Key additions include a SubscriptionRegistry for session management, a RetainedValueStore for event persistence, and integration into the Context and FastMCP classes for easy event emission. Review feedback highlighted several areas for improvement: the mcp dependency should be updated to a stable release before production, type safety should be restored by replacing Any with specific types in protocol handlers and the responder parameter, and error handling should be enhanced by logging warnings instead of silently swallowing exceptions during expiry checks and JSON schema generation.

[gemini-code-assist]

The mcp dependency is pinned to a specific git branch. While this is acceptable for development, it's not suitable for production. Before this change is merged into a main or release branch, this should be updated to point to a released version of the mcp package on PyPI to ensure stable and reproducible builds.

[gemini-code-assist]

Catching ValueError and TypeError and returning False means that an invalid expires_at format is silently treated as "never expires". This can hide bugs. It would be better to log a warning when an invalid format is encountered so that developers are aware of the issue. You'll need to import a logger to do this.

[gemini-code-assist]

The type hint for the responder parameter has been removed, which reduces type safety and code clarity. Please restore the type hint. The RequestResponder type might have moved in the new SDK version, so you may need to adjust the import.

Suggested change

	responder,
	responder: "RequestResponder[mcp.types.ClientRequest, mcp.types.ServerResult]",

[gemini-code-assist]

The type hints for req and the return value are Any. Using specific types from fastmcp.server.events like EventSubscribeRequest and EventSubscribeResult would improve type safety and readability. This also applies to _handle_unsubscribe_events and _handle_list_events.

Suggested change

	self, req: Any
	) -> Any:
	self, req: "EventSubscribeRequest"
	) -> "EventSubscribeResult":

[gemini-code-assist]

The broad except Exception: pass can silently swallow important errors during JSON schema generation, making debugging difficult. At a minimum, this should log a warning when an exception occurs, so developers are aware that schema generation failed.

                except Exception as e:
                    logger.warning(
                        f"Failed to generate JSON schema for event topic '{pattern}': {e}",
                        exc_info=True,
                    )

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request introduces a comprehensive event system for FastMCP, enabling servers to declare, emit, and manage subscriptions for events with support for MQTT-style wildcards and retained values. Key components include a new SubscriptionRegistry for session-to-topic mapping, a RetainedValueStore for replaying the latest events to new subscribers, and protocol-level handlers for events/subscribe, events/unsubscribe, and events/list. Feedback suggests improving the stability of the project by updating the git-pinned mcp dependency to a released version and enhancing the robustness of the event expiration logic by logging warnings for malformed timestamps instead of failing silently.

[gemini-code-assist]

The mcp dependency is pinned to a git branch. While this is acceptable for development, it should be updated to a released version from PyPI before this is merged into a main or release branch to ensure stability and simplify dependency management.

    "mcp>=1.28.0,<2.0",

[gemini-code-assist]

The _is_expired method silently ignores invalid expires_at strings by returning False, effectively treating them as non-expiring. This could hide configuration errors or malformed data. It would be more robust to log a warning when the date string cannot be parsed.

To do this, you'll need to set up a logger at the module level, similar to other files in the project:

import logging

logger = logging.getLogger(__name__)

Then, you can modify the method to log the warning.

Suggested change

	def _is_expired(self, topic: str) -> bool:
	expires_at = self._expires.get(topic)
	if expires_at is None:
	return False
	try:
	expiry = datetime.fromisoformat(expires_at)
	if expiry.tzinfo is None:
	expiry = expiry.replace(tzinfo=timezone.utc)
	return datetime.now(timezone.utc) >= expiry
	except (ValueError, TypeError):
	return False
	def _is_expired(self, topic: str) -> bool:
	expires_at = self._expires.get(topic)
	if expires_at is None:
	return False
	try:
	expiry = datetime.fromisoformat(expires_at)
	if expiry.tzinfo is None:
	expiry = expiry.replace(tzinfo=timezone.utc)
	return datetime.now(timezone.utc) >= expiry
	except (ValueError, TypeError):
	logger.warning("Invalid expires_at value %r for topic %r", expires_at, topic)
	return False

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request implements a new event subscription and publishing system for FastMCP, introducing SubscriptionRegistry for MQTT-style topic matching and RetainedValueStore for retained events. The FastMCP server now supports event declaration, emission via ctx.emit_event, and handles event protocol requests (subscribe, unsubscribe, list). The mcp dependency is updated to a specific git branch, and python-ulid is added. Extensive tests for the new event features and server shutdown reliability are included. Feedback highlights the need to transition the mcp dependency to a stable PyPI release, address attribute naming inconsistencies in RetainedEvent and EventParams, improve datetime.fromisoformat compatibility, refine wildcard matching logic, optimize SubscriptionRegistry.match for efficiency, and implement concurrent event broadcasting to prevent head-of-line blocking.

[gemini-code-assist]

The RetainedEvent model from the MCP SDK uses snake_case for its attributes. Accessing event.eventId will likely raise an AttributeError. It should be event.event_id, which is consistent with the usage in the test suite (e.g., test_events.py line 138).

Suggested change

	eid = event.eventId
	eid = event.event_id

[gemini-code-assist]

The dependency on mcp is pointing to a specific git branch. While this is necessary during development of the events feature, it should be replaced with a versioned release from PyPI before merging to the main branch to ensure stability and reproducible builds.

[gemini-code-assist]

The match method can be optimized to avoid redundant regex matching and dictionary lookups. Currently, it iterates over all sessions and all their patterns, performing a dict lookup and regex match for each. A more efficient approach is to find all unique patterns that match the topic first, and then find sessions subscribed to those patterns.

Suggested change

	async def match(self, topic: str) -> set[str]:
	"""Return session IDs whose subscriptions match *topic*.
	Each session appears at most once (at-most-once delivery guarantee).
	"""
	async with self._lock:
	result: set[str] = set()
	for session_id, patterns in self._subscriptions.items():
	for pattern in patterns:
	regex = self._compile(pattern)
	if regex.match(topic):
	result.add(session_id)
	break # at-most-once per session
	return result
	async def match(self, topic: str) -> set[str]:
	"""Return session IDs whose subscriptions match *topic*.
	Each session appears at most once (at-most-once delivery guarantee).
	"""
	async with self._lock:
	# Find all unique patterns that match the topic
	matching_patterns = {
	pattern for pattern, regex in self._compiled.items()
	if regex.match(topic)
	}
	# Find sessions subscribed to any of those patterns
	result: set[str] = set()
	for session_id, patterns in self._subscriptions.items():
	if not patterns.isdisjoint(matching_patterns):
	result.add(session_id)
	return result

[gemini-code-assist]

datetime.fromisoformat() does not support the Z suffix for UTC in Python versions prior to 3.11. Since the MCP SDK supports Python 3.10, this will cause a ValueError when parsing ISO 8601 strings from clients that use the Z notation. Consider replacing Z with +00:00 for broader compatibility.

Suggested change

	expiry = datetime.fromisoformat(expires_at)
	expiry = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))

[gemini-code-assist]

The substitution of + and # with x is too aggressive. In MQTT, these are only wildcards if they occupy an entire segment of the topic. Replacing them anywhere in the string can lead to incorrect matches (e.g., a literal + in a segment being treated as a wildcard match). The substitution should be segment-aware.

Suggested change

	test_pattern = _re.sub(r"[+#]", "x", pattern)
	test_parts = [
	"x" if seg in ("+", "#") else seg
	for seg in pattern.split("/")
	]
	test_pattern = "/".join(test_parts)

[gemini-code-assist]

It is recommended to use snake_case attribute names when initializing Pydantic models from the MCP SDK to ensure consistency and avoid relying on populate_by_name configurations.

Suggested change

	eventId=event_id,
	event_id=event_id,

[gemini-code-assist]

It is recommended to use snake_case attribute names when initializing EventParams to ensure consistency with the rest of the codebase and the SDK's model definitions.

Suggested change

	eventId=event_id,
	payload=payload,
	retained=retained,
	source=source,
	correlationId=correlation_id,
	requestedEffects=requested_effects,
	expiresAt=expires_at,
	event_id=event_id,
	payload=payload,
	retained=retained,
	source=source,
	correlation_id=correlation_id,
	requested_effects=requested_effects,
	expires_at=expires_at,

[gemini-code-assist]

Broadcasting events sequentially can lead to head-of-line blocking. If one session's transport is slow or backpressured, it will delay delivery to all subsequent sessions. Using a TaskGroup to send notifications concurrently would improve performance and reliability of the broadcast.

        # Broadcast to matching active sessions concurrently
        import anyio
        async with anyio.create_task_group() as tg:
            for sid in matching_session_ids:
                session = self._active_sessions.get(sid)
                if session is None:
                    continue
                
                async def send_safe(s=session, id=sid):
                    try:
                        await s.send_notification(
                            cast(ServerNotification, notification)
                        )
                    except Exception:
                        logger.warning(
                            f"Failed to deliver event to session {id}",
                            exc_info=True,
                        )
                
                tg.start_soon(send_safe)

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request introduces application-level event support for FastMCP, enabling servers to declare topics, publish events, and manage client subscriptions using MQTT-style wildcards. Key additions include a SubscriptionRegistry for session-to-topic mapping, a RetainedValueStore for replaying events to new subscribers, and integration with the Context class for emitting events from tools. The implementation also includes protocol-level handlers for subscribing, unsubscribing, and listing topics. Review feedback identifies missing imports for 'cast' and 'logger' in several modules, suggests removing redundant asyncio.Lock instances and deduplication logic to optimize performance, and recommends adding timestamps to retained events to provide better context for clients.

[gemini-code-assist]

The logger object is used but not defined or imported in this module. This will result in a NameError. You should define a module-level logger: logger = logging.getLogger(__name__).

[gemini-code-assist]

The cast function is used here but is not included in the typing imports at the top of the file (line 104). This will raise a NameError at runtime.

[gemini-code-assist]

The asyncio.Lock is redundant here because none of the methods in SubscriptionRegistry contain await points within their critical sections (except for the lock acquisition itself). In a single-threaded asyncio environment, tasks only yield at await points, so these operations are already atomic relative to other tasks. Removing the lock will reduce overhead and simplify the code.

Suggested change

	self._lock = asyncio.Lock()
	# self._lock = asyncio.Lock() # Redundant in single-threaded asyncio without await points

[gemini-code-assist]

The match method performs an O(N*M) scan (sessions * patterns) for every event emission. While acceptable for small numbers of sessions, this will become a performance bottleneck as the server scales. Consider using a more efficient data structure for MQTT wildcard matching, such as a Topic Trie, or at least pre-filtering sessions by literal topic prefixes.

[gemini-code-assist]

Similar to SubscriptionRegistry, the asyncio.Lock in RetainedValueStore is redundant as there are no await points inside the methods that modify or read the store. Removing it will improve performance slightly and clarify that these operations are atomic in the current event loop.

Suggested change

	self._lock = asyncio.Lock()
	# self._lock = asyncio.Lock() # Redundant

[gemini-code-assist]

The deduplication logic using seen_event_ids inside get_matching is redundant. Since self._store is keyed by unique concrete topics, each iteration over self._store.items() will naturally yield unique events. Deduplication is only necessary when multiple patterns are processed in a single request, which is already handled in the protocol layer (mcp_operations.py).

Suggested change

	seen_event_ids: set[str] = set()
	expired_topics: list[str] = []
	for topic, event in self._store.items():
	if self._is_expired(topic):
	expired_topics.append(topic)
	continue
	if regex.match(topic):
	# Deduplicate: overlapping patterns can match same retained value
	eid = event.eventId
	if eid not in seen_event_ids:
	seen_event_ids.add(eid)
	result.append(event)
	for topic, event in self._store.items():
	if self._is_expired(topic):
	expired_topics.append(topic)
	continue
	if regex.match(topic):
	result.append(event)

[gemini-code-assist]

Retained events should ideally include a timestamp so that clients subscribing later know when the state was last updated. Setting timestamp=None misses this useful context. Consider using the current UTC time.

Suggested change

	timestamp=None,
	timestamp=datetime.now(timezone.utc).isoformat(),

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request implements a comprehensive Events system for FastMCP, enabling servers to push real-time notifications to clients via topic-based streams. It introduces mechanisms for declaring event topics using decorators or direct methods, emitting events from tools or background tasks, and supporting retained events for late subscribers. The implementation includes a subscription registry with MQTT-style wildcard support and protocol handlers for subscribing, unsubscribing, and listing events. Feedback suggests adding python-ulid as a direct dependency, removing redundant deduplication in the retained value store, and optimizing the topic matching logic to improve efficiency.

[gemini-code-assist]

The code in src/fastmcp/server/server.py now directly imports and uses ULID from the python-ulid package. This package should be added as a direct dependency in pyproject.toml to ensure it is available regardless of transitive dependency changes in the mcp SDK.

    "mcp @ git+https://github.com/axiomantic/python-sdk.git@mcp-events",
    "python-ulid>=3.1.0",

[gemini-code-assist]

The deduplication logic using seen_event_ids inside get_matching is redundant. Since self._store is a dictionary keyed by topic, iterating over its items already guarantees that each unique topic (and its associated event) is visited exactly once. Deduplication is correctly handled at a higher level in _handle_subscribe_events when processing multiple subscription patterns.

Suggested change

	seen_event_ids: set[str] = set()
	expired_topics: list[str] = []
	for topic, event in self._store.items():
	if self._is_expired(topic):
	expired_topics.append(topic)
	continue
	if regex.match(topic):
	# Deduplicate: overlapping patterns can match same retained value
	eid = event.eventId
	if eid not in seen_event_ids:
	seen_event_ids.add(eid)
	result.append(event)
	result: list[RetainedEvent] = []
	expired_topics: list[str] = []
	for topic, event in self._store.items():
	if self._is_expired(topic):
	expired_topics.append(topic)
	continue
	if regex.match(topic):
	result.append(event)

[gemini-code-assist]

The _match_declared_topic method is inefficient as it performs several redundant operations inside the loop over self._event_topics. Specifically, sub_regex and test_pattern should be computed once outside the loop. Additionally, the declared_regex and concrete_declared strings could be pre-calculated or cached to avoid repeated regex compilation and string manipulation during every subscription request.

        try:
            sub_regex = _pattern_to_regex(pattern)
        except ValueError:
            return False

        test_pattern = _re.sub(r"[+#]", "x", pattern)

        for declared_pattern in self._event_topics:
            # Forward: build regex from declared pattern's {param} placeholders
            declared_regex_parts = [
                "[^/]+" if s.startswith("{") and s.endswith("}") else _re.escape(s)
                for s in declared_pattern.split("/")
            ]
            declared_regex = _re.compile(
                "^" + "/".join(declared_regex_parts) + "$"
            )

            if declared_regex.match(test_pattern):
                return True

            # Reverse: does the declared pattern match the subscription pattern's MQTT regex?
            concrete_declared = _re.sub(r"\{[^}]+\}", "x", declared_pattern)
            if sub_regex.match(concrete_declared):
                return True

[elijahr]

@gemini-code-assist please re-review

[elijahr]

CI Failures Note

The failing tests in tests/server/tasks/ and tests/client/tasks/ are pre-existing upstream flakes, not caused by this PR:

1. Verified by running failing tests on main branch -- same failures occur
2. test_task_cancellation_interrupts_running_coroutine has a genuine hang upstream (times out even at 30s)
3. ~18 other task tests are timing-sensitive (pass at 30s, fail at 5s) due to CI slowness
4. Our changes to _received_request only intercept InitializeRequest, all other requests (including tasks) fall through unchanged
5. No interaction between our event code and the task system

Event tests (tests/server/test_events.py): 71/71 passing.

[gemini-code-assist]

Code Review

This pull request implements a comprehensive topic-based event system for FastMCP, enabling servers to broadcast real-time notifications to subscribed clients. Key features include a SubscriptionRegistry supporting MQTT-style wildcards, a RetainedValueStore for replaying the latest events to new subscribers, and new API methods in the FastMCP and Context classes for declaring and emitting events. The PR also adds protocol-level handlers, detailed documentation, and extensive test coverage. Review feedback suggests refactoring ISO timestamp parsing to remove unnecessary workarounds and moving local imports to the module level for better code organization.

[gemini-code-assist]

The use of fromisoformat with a manual replacement of 'Z' is a workaround for older Python versions. Since the project requires Python >= 3.10, datetime.fromisoformat already supports the 'Z' suffix natively. This replacement can be removed for cleaner code.

[gemini-code-assist]

The ulid library is imported inside the function. It is better practice to import it at the top of the module to avoid repeated imports and improve readability.

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request implements support for MCP events, enabling servers to broadcast real-time notifications to subscribed clients via topic-based streams with MQTT-style wildcard support. Key additions include a subscription registry, a retained value store for late subscribers, and new API methods for declaring and emitting events via the FastMCP and Context classes. The feedback identifies several performance optimization opportunities, specifically regarding the caching of compiled regex patterns for topic matching and the use of concurrent delivery when broadcasting events to multiple active sessions.

[gemini-code-assist]

The _pattern_to_regex function is called frequently during subscription matching and retained value lookups. Since topic patterns are often reused, adding an LRU cache will significantly improve performance by avoiding redundant string splitting and regex compilation.

import functools

@functools.lru_cache(maxsize=128)
def _pattern_to_regex(pattern: str) -> re.Pattern[str]:

[gemini-code-assist]

Regex compilation for declared patterns is happening inside a loop for every subscription request. This is inefficient, especially since FastMCP already provides a _declared_topic_regex_cache intended for this purpose. You should leverage the cache to avoid re-compiling the same patterns repeatedly.

Suggested change

	declared_regex_parts = []
	for segment in declared_pattern.split("/"):
	if segment.startswith("{") and segment.endswith("}"):
	declared_regex_parts.append("[^/]+")
	else:
	declared_regex_parts.append(_re.escape(segment))
	declared_regex = _re.compile(
	"^" + "/".join(declared_regex_parts) + "$"
	)
	if declared_pattern not in self._declared_topic_regex_cache:
	declared_regex_parts = []
	for segment in declared_pattern.split("/"):
	if segment.startswith("{") and segment.endswith("}"):
	declared_regex_parts.append("[^/]+")
	else:
	declared_regex_parts.append(_re.escape(segment))
	self._declared_topic_regex_cache[declared_pattern] = _re.compile(
	"^" + "/".join(declared_regex_parts) + "$"
	)
	declared_regex = self._declared_topic_regex_cache[declared_pattern]

[gemini-code-assist]

Broadcasting events to multiple sessions is currently sequential. If one session's transport is slow or experiences latency, it will delay delivery to all subsequent sessions and block the emit_event caller. Using asyncio.gather with error handling for each task would allow concurrent delivery and improve overall system responsiveness.

        async def _send_to_session(sid: str):
            session = self._active_sessions.get(sid)
            if session is None:
                return
            try:
                await session.send_notification(
                    cast(ServerNotification, notification)
                )
            except Exception:
                logger.warning(
                    f"Failed to deliver event to session {sid}",
                    exc_info=True,
                )

        await asyncio.gather(*[_send_to_session(sid) for sid in matching_session_ids])

[elijahr]

@gemini-code-assist please re-review

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request introduces a comprehensive Events system to FastMCP, enabling servers to publish real-time notifications to subscribed clients via topic-based streams with MQTT-style wildcards. Key features include the @mcp.event decorator and declare_event method for topic registration, emit_event for broadcasting (accessible via FastMCP or Context), and a RetainedValueStore to support late subscribers. The implementation also includes a session-based authorization system and automatic event source tracking for tool-initiated events. Feedback highlights potential performance bottlenecks in topic matching and descriptor lookups, suggesting the use of more efficient data structures like tries or caching for high-frequency event streams.

[gemini-code-assist]

The match method performs a linear scan over all sessions and all their subscription patterns for every event emission. In a server with many active sessions or complex subscription sets, this $O(N_{total_subscriptions})$ operation could become a performance bottleneck. Consider using a more efficient data structure, such as a trie (prefix tree) specifically designed for topic matching with wildcards, to accelerate this lookup.

[gemini-code-assist]

The _find_topic_descriptor method performs a linear scan over all declared topics for every event emission when a direct match is not found. For servers with many parameterized topics or high-frequency event streams, this can impact performance. Consider implementing a cache for concrete topic strings to their corresponding descriptors.

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request implements application-level MCP events, enabling servers to broadcast real-time notifications via topic-based streams. Key additions include a subscription registry with MQTT wildcard support, a retained event store for late subscribers, and new API methods on the FastMCP and Context classes. The implementation also features a flexible authorization system for session-scoped or custom-validated topics. Review feedback suggests returning protocol-compliant rejection lists for over-deep topic patterns instead of raising errors, refining topic matching logic to avoid false positives from synthetic literals, and ensuring exceptions during event broadcasting are properly logged.

[gemini-code-assist]

Raising an McpError immediately when a topic depth exceeds the limit will abort the entire subscription request. According to the MCP specification for events/subscribe, the server should return a result containing both subscribed and rejected lists. It would be more consistent with the protocol to add the over-deep pattern to the rejected list with an appropriate reason (e.g., "invalid_topic") rather than failing the whole batch.

            if len(segments) > server._MAX_TOPIC_DEPTH:
                rejected.append(
                    RejectedTopic(
                        pattern=pattern,
                        reason="invalid_topic",
                    )
                )
                continue

[gemini-code-assist]

The use of a synthetic literal 'x' to test subscription patterns against declared topic regexes might lead to false positives if 'x' is a meaningful part of a literal segment in a declared topic. A more robust approach would be to use a sentinel character that is guaranteed not to appear in valid topic segments, or to perform segment-by-segment matching directly.

[gemini-code-assist]

Using asyncio.gather with return_exceptions=True is good for ensuring that a failure in one session delivery doesn't stop others. However, since the results are ignored, any exceptions that occur during delivery (other than those caught and logged in _deliver) will be silently swallowed. Consider logging the results of gather if they contain exceptions.

        results = await asyncio.gather(
            *[_deliver(sid) for sid in matching_session_ids],
            return_exceptions=True,
        )
        for res in results:
            if isinstance(res, Exception):
                logger.error(f"Unexpected error during event broadcast: {res}", exc_info=res)

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request implements a topic-based event publishing and subscription system for FastMCP, including MQTT-style wildcard support and retained event persistence. Key components include a SubscriptionRegistry, a RetainedValueStore, and updated FastMCP and Context classes to handle event declaration and emission. Feedback identifies several performance optimization opportunities, specifically suggesting the use of Tries or regex caches to replace linear scans during topic matching and descriptor lookups. Additionally, it is recommended to use non-blocking background tasks for event delivery to prevent slow clients from delaying the broadcast process.

[gemini-code-assist]

The match method performs a linear scan over all active sessions and all of their subscription patterns for every event emission. In a server with many sessions or many subscriptions, this $O(N_{sessions} \times N_{patterns})$ complexity will become a significant performance bottleneck. Consider using a more efficient data structure for topic matching, such as a Trie (prefix tree), which is standard practice in MQTT brokers and other pub/sub systems.

[gemini-code-assist]

Similar to the subscription registry, get_matching performs a linear scan over all retained topics. As the number of retained topics grows, this will degrade the performance of new subscriptions. A Trie-based structure would also benefit this lookup.

[gemini-code-assist]

The _find_topic_descriptor method uses a manual segment-by-segment string comparison in a loop over all declared topics. This is inefficient and inconsistent with the regex-based matching used in _find_matching_declared_topics (in mcp_operations.py). This method should be refactored to leverage the _declared_topic_regex_cache for better performance and consistency.

[gemini-code-assist]

Using asyncio.gather with return_exceptions=True will wait for all delivery tasks to complete before returning. If one or more clients are slow or experiencing network backpressure, emit_event will block for the duration of the slowest delivery. This can be problematic if emit_event is called from a critical path or a background loop that needs to remain responsive. Consider using asyncio.create_task for a "fire-and-forget" broadcast or a task group that doesn't block the immediate caller.

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request introduces a comprehensive Events feature to FastMCP, enabling servers to publish real-time notifications via topic-based streams with MQTT-style wildcard support. Key components include a SubscriptionRegistry for managing session-to-topic mappings, a RetainedValueStore for replaying the latest events to new subscribers, and integrated authorization mechanisms. The implementation also includes extensive documentation and tests. Review feedback identifies critical naming inconsistencies between camelCase and snake_case for event identifiers (e.g., eventId vs event_id) in the code and tests, which could lead to runtime errors. Additionally, the use of a git-based dependency for the mcp SDK is flagged for replacement with a versioned release. Performance and concurrency improvements are suggested for the subscription matching logic and locking strategy, and topic pattern validation should be enhanced to reject empty segments during declaration.

[gemini-code-assist]

The mcp dependency is currently pointing to a specific git branch on GitHub. For a stable release and to ensure reproducible builds, this should be replaced with a versioned release from PyPI once the event support is officially published in the SDK.

[gemini-code-assist]

There is a naming inconsistency for the event ID field. Here it uses evt.eventId (camelCase), but the tests in tests/server/test_events.py (e.g., line 145) use e.event_id (snake_case). Additionally, RetainedEvent in server.py is constructed using eventId. This inconsistency between camelCase and snake_case for the same field across the codebase and tests will likely lead to AttributeError or validation failures. Please standardize on snake_case for Python attributes as per PEP 8.

            if evt.event_id not in seen_event_ids:
                seen_event_ids.add(evt.event_id)

[gemini-code-assist]

The EventParams constructor is being called with a mix of snake_case and camelCase arguments (e.g., topic vs eventId, correlationId). Standard Pydantic models in Python should use snake_case for all field names. Using camelCase names as keyword arguments will fail if the model does not explicitly define them as aliases with populate_by_name=True.

        notification = EventEmitNotification(
            params=EventParams(
                topic=topic,
                event_id=event_id,
                payload=payload,
                retained=retained,
                source=source,
                correlation_id=correlation_id,
                requested_effects=requested_effects,
                expires_at=expires_at,
            ),
        )

[gemini-code-assist]

The match method iterates over all active sessions and all of their subscription patterns for every event emitted. This results in $O(N_{sessions} \times N_{patterns})$ complexity. For servers with a large number of sessions or high event throughput, this nested loop (even with regex caching) will become a significant performance bottleneck and block the event loop. Consider using a more efficient data structure for wildcard matching, such as a Topic Trie.

[gemini-code-assist]

The SubscriptionRegistry uses a single asyncio.Lock for all operations, including the potentially long-running match iteration. Since match is called on every emit_event, this lock will cause significant contention. Given that match only reads from _subscriptions, consider using a more granular locking strategy or immutable data structures to allow concurrent matching while subscriptions are being added or removed.

[gemini-code-assist]

The declare_event method validates the maximum depth of the topic pattern but does not check for empty segments (e.g., myapp//status). While _topic_matches_pattern correctly ignores such patterns during matching, it is better to reject them early during declaration to provide immediate feedback to the developer.

        segments = pattern.split("/")
        if any(not s for s in segments):
            raise ValueError(f"Topic pattern contains empty segments: {pattern!r}")
        if len(segments) > self._MAX_TOPIC_DEPTH:
            raise ValueError(
                f"Topic pattern has {len(segments)} segments, "
                f"maximum depth is {self._MAX_TOPIC_DEPTH}: {pattern!r}"
            )

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request introduces support for MCP events, enabling servers to push real-time notifications to clients. The implementation includes subscription management, a retained value store, and authorization logic for topic patterns. I have reviewed the changes and provided a suggestion to improve the robustness of timestamp parsing in the event expiration logic. The performance of the topic matching lookup is noted as a potential area for future optimization, but no immediate action is required.

[gemini-code-assist]

The error handling for malformed expires_at timestamps in _is_expired could lead to unintended behavior. Currently, if datetime.fromisoformat fails, the except block returns False, effectively treating the event as non-expiring. This could cause an event with a malformed timestamp to be retained indefinitely.

It would be safer to fail-closed by treating malformed timestamps as expired. I suggest returning True from the except block and logging a warning.

Suggested change

	def _is_expired(self, topic: str) -> bool:
	expires_at = self._expires.get(topic)
	if expires_at is None:
	return False
	try:
	# Python 3.10 fromisoformat() doesn't support "Z" suffix;
	# replace with "+00:00" for compatibility.
	expiry = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
	if expiry.tzinfo is None:
	expiry = expiry.replace(tzinfo=timezone.utc)
	return datetime.now(timezone.utc) >= expiry
	except (ValueError, TypeError):
	return False
	def _is_expired(self, topic: str) -> bool:
	expires_at = self._expires.get(topic)
	if expires_at is None:
	return False
	try:
	# Python 3.10 fromisoformat() doesn't support "Z" suffix;
	# replace with "+00:00" for compatibility.
	expiry = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
	if expiry.tzinfo is None:
	expiry = expiry.replace(tzinfo=timezone.utc)
	return datetime.now(timezone.utc) >= expiry
	except (ValueError, TypeError):
	# Consider logging a warning here about the malformed timestamp.
	return True

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request implements support for MCP events in FastMCP, enabling servers to broadcast real-time notifications to subscribed clients. It introduces a subscription registry with MQTT-style wildcard support, a retained value store for late subscribers, and new API methods on the FastMCP and Context classes for declaring and emitting events. Comprehensive documentation and tests are also included. Review feedback identifies a critical attribute access error where camelCase was used instead of snake_case, which would lead to runtime errors, and suggests using idiomatic snake_case for constructor parameters to maintain consistency with the MCP SDK.

[gemini-code-assist]

The attribute access evt.eventId will raise an AttributeError because Pydantic models in the MCP SDK use snake_case for field names (e.g., event_id). While aliases like eventId can be used in constructors, they are not available as attributes on the instance. The tests in tests/server/test_events.py correctly use event_id, confirming this is the intended attribute name.

Suggested change

	if evt.eventId not in seen_event_ids:
	seen_event_ids.add(evt.eventId)
	for evt in matching:
	if evt.event_id not in seen_event_ids:
	seen_event_ids.add(evt.event_id)

[gemini-code-assist]

While Pydantic allows using aliases in constructors if configured, it is more idiomatic in Python to use snake_case field names. Using event_id instead of eventId aligns with the rest of the codebase and the tests.

Suggested change

	retained_event = RetainedEvent(
	topic=topic,
	eventId=event_id,
	timestamp=None,
	payload=payload,
	)
	if retained:
	retained_event = RetainedEvent(
	topic=topic,
	event_id=event_id,
	timestamp=None,
	payload=payload,
	)

[gemini-code-assist]

The EventParams constructor is using camelCase aliases (eventId, correlationId, requestedEffects, expiresAt). To maintain consistency with Python idioms and the snake_case attributes used in tests (e.g., notif.params.event_id), these should be updated to use snake_case field names.

Suggested change

	notification = EventEmitNotification(
	params=EventParams(
	topic=topic,
	eventId=event_id,
	payload=payload,
	retained=retained,
	source=source,
	correlationId=correlation_id,
	requestedEffects=requested_effects,
	expiresAt=expires_at,
	),
	)
	notification = EventEmitNotification(
	params=EventParams(
	topic=topic,
	event_id=event_id,
	payload=payload,
	retained=retained,
	source=source,
	correlation_id=correlation_id,
	requested_effects=requested_effects,
	expires_at=expires_at,
	),
	)

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request introduces a comprehensive event publishing system to FastMCP, allowing servers to broadcast real-time notifications to subscribed clients using topic-based streams with MQTT-style wildcards. Key additions include the @mcp.event decorator and ctx.emit_event for tools, a SubscriptionRegistry for managing client interests, and a RetainedValueStore for late-subscriber state recovery. The implementation also features session-scoped authorization and automatic event sourcing from tool contexts. Feedback focuses on improving error clarity for missing session contexts, optimizing broadcast performance for high session counts via bounded concurrency, and ensuring RetainedEvent models include valid timestamps to prevent potential client-side validation issues.

[gemini-code-assist]

Raising a generic Internal Error (-32603) when session_id is missing might be confusing for clients. Since this typically happens if the session wasn't properly initialized or registered in the registry, a more descriptive error or ensuring the session lifecycle is strictly managed before allowing subscriptions would be better. However, given the current architecture, this is a reasonable safety check.

[gemini-code-assist]

Broadcasting events using asyncio.gather with return_exceptions=True is efficient for a small number of sessions. However, if the server scales to hundreds or thousands of active sessions, creating a task for every single delivery might lead to performance spikes. Consider using a task group or a bounded concurrency mechanism if high session counts are expected.

[gemini-code-assist]

The RetainedEvent model is initialized with timestamp=None. Depending on the MCP SDK's strictness, a missing timestamp might cause validation issues on the client side if it expects a valid ISO 8601 string for all events. It is recommended to provide the current UTC timestamp if the SDK allows it.

Suggested change

	retained_event = RetainedEvent(
	topic=topic,
	eventId=event_id,
	timestamp=None,
	payload=payload,
	)
	if retained:
	from datetime import datetime, timezone
	retained_event = RetainedEvent(
	topic=topic,
	eventId=event_id,
	timestamp=datetime.now(timezone.utc).isoformat(),
	payload=payload,
	)

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request implements a comprehensive event publishing system for FastMCP, enabling servers to broadcast real-time notifications via topic-based streams with MQTT-style wildcards. Key features include event declaration decorators, support for retained events, and a session-aware authorization system. Review feedback identifies several improvement opportunities, specifically regarding potential memory leaks in internal regex caches and the O(N*M) complexity of the subscription matching logic, which could become a bottleneck at scale. Additionally, it is recommended to replace the direct git dependency for the core SDK with a stable release version to ensure build stability.

[gemini-code-assist]

Using a direct git URL for a core dependency like mcp is generally discouraged for a library or framework intended for public release. This can lead to non-deterministic builds and issues with dependency resolution for downstream users. If the required event features are not yet available in a stable release of the mcp SDK, consider waiting for a release or using a pre-release version if available on PyPI.

[gemini-code-assist]

The _compiled dictionary in SubscriptionRegistry grows without bound as clients subscribe to unique patterns. Since _pattern_to_regex is already decorated with lru_cache, this internal dictionary is redundant and introduces a memory leak. It is safer to call the cached function directly.

Suggested change

	self._compiled: dict[str, re.Pattern[str]] = {}
	# Remove self._compiled from __init__ and _compile method
	async def match(self, topic: str) -> set[str]:
	"""Return session IDs whose subscriptions match *topic*.
	Each session appears at most once (at-most-once delivery guarantee).
	"""
	async with self._lock:
	result: set[str] = set()
	for session_id, patterns in self._subscriptions.items():
	for pattern in patterns:
	regex = _pattern_to_regex(pattern)
	if regex.match(topic):
	result.add(session_id)
	break # at-most-once per session
	return result

[gemini-code-assist]

The match method iterates over all active sessions and all of their subscription patterns for every event emission. This results in $O(Sessions \times Patterns)$ complexity, which will become a significant performance bottleneck as the number of connected clients or subscriptions grows. Consider using a more efficient data structure for topic matching, such as a Trie or a specialized MQTT-style matcher.

[gemini-code-assist]

The _regex_cache in RetainedValueStore grows without bound and is redundant because _pattern_to_regex is already cached. Removing it prevents a potential memory leak.

Suggested change

	self._regex_cache: dict[str, re.Pattern[str]] = {}
	async def get_matching(self, pattern: str) -> list[RetainedEvent]:
	"""Return all non-expired retained events whose topic matches *pattern*."""
	async with self._lock:
	regex = _pattern_to_regex(pattern)
	result: list[RetainedEvent] = []
	expired_topics: list[str] = []
	for topic, event in self._store.items():
	if self._is_expired(topic):
	expired_topics.append(topic)
	continue
	if regex.match(topic):
	result.append(event)
	for topic in expired_topics:
	del self._store[topic]
	self._expires.pop(topic, None)
	return result

[gemini-code-assist]

The _declared_topic_regex_cache dictionary grows as new event topics are declared. While declared topics are typically fixed at startup, dynamic declarations could lead to unbounded growth. Consider using a function decorated with functools.lru_cache to handle the regex generation and caching for declared patterns.

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request introduces a comprehensive event publishing system to FastMCP, enabling servers to broadcast real-time notifications to subscribed clients via topic-based streams with MQTT-style wildcards. Key additions include the @mcp.event decorator and ctx.emit_event method, a SubscriptionRegistry for managing client subscriptions, and a RetainedValueStore for replaying the latest events to new subscribers. The implementation also features a built-in authorization mechanism using a {session_id} convention and custom callbacks. Feedback highlights the need to replace the git-based mcp dependency with a stable release, potential performance bottlenecks in the linear subscription matching logic, and the requirement to document or reconsider the restriction on topics with leading or trailing slashes.

[gemini-code-assist]

The mcp dependency is currently pointing to a specific git branch. While this might be necessary for development, it should be replaced with a stable versioned release before merging to production to ensure reproducible builds and avoid reliance on a mutable branch.

[gemini-code-assist]

The match method performs a linear scan over all sessions and all their subscription patterns. While acceptable for a small number of clients, this O(N*M) complexity could become a performance bottleneck if the server handles many sessions or high-frequency event streams. Consider using a more efficient data structure like a trie for topic matching if scalability becomes an issue.

[gemini-code-assist]

The check any(not s for s in concrete_parts) effectively bans topics with leading, trailing, or double slashes (e.g., /myapp, myapp/, myapp//status). While this simplifies the API, it deviates from MQTT behavior where these are valid and distinct topics. If this restriction is intentional for the high-level API, it should be explicitly documented.

[elijahr]

@gemini-code-assist please re-review

[gemini-code-assist]

Code Review

This pull request introduces support for MCP events in FastMCP, enabling servers to publish real-time notifications to subscribed clients via topic-based streams with MQTT-style wildcards. Key additions include a SubscriptionRegistry for managing session-to-topic mappings, a RetainedValueStore for replaying the latest events to new subscribers, and new API methods (declare_event, emit_event) on both the FastMCP server and the Context object. Feedback focuses on potential performance bottlenecks in topic matching and descriptor resolution, suggesting more efficient data structures or caching for high-concurrency scenarios.

[gemini-code-assist]

The match method iterates over all active sessions and all of their subscription patterns while holding an asyncio.Lock. For a server with a large number of concurrent sessions or complex subscription sets, this $O(N \times M)$ operation could become a performance bottleneck and block other registry operations (like add or remove). Consider using a more efficient data structure for topic matching, such as a trie or a specialized MQTT-style matcher, if high session counts are expected.

[gemini-code-assist]

The _find_topic_descriptor method performs a linear search over self._event_topics when a direct match is not found. While typically safe for a small number of declared topics, this $O(T)$ operation is called during every emit_event to determine retention policy. If a server declares a very large number of parameterized topics, consider caching the resolution of concrete topics to descriptors.