mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 07:09:11 +02:00
* Archive v3 docs under /v3 and publish v4 as the primary version * Label primary docs version v4.0.0 (alpha 1) * Add What's New in v4 page; fix upgrade-guide phrasing; point banner at What's New * Rewrite What's New around v4's new capabilities, not the sampling deprecation * Lead What's New with the SDK v2 engine swap and the SEPs it brings * State ships now (link Session State); tasks arrive next alpha * Exclude docs/v3 frozen snapshots from doc-example import validation
155 lines
4.7 KiB
Text
155 lines
4.7 KiB
Text
---
|
|
title: Notifications
|
|
sidebarTitle: Notifications
|
|
description: Handle server-sent notifications for list changes and other events.
|
|
icon: envelope
|
|
---
|
|
|
|
import { VersionBadge } from "/snippets/version-badge.mdx";
|
|
|
|
<VersionBadge version="2.9.1" />
|
|
|
|
Use this when you need to react to server-side changes like tool list updates or resource modifications.
|
|
|
|
MCP servers can send notifications to inform clients about state changes. The message handler provides a unified way to process these notifications.
|
|
|
|
## Handling Notifications
|
|
|
|
The simplest approach is a function that receives all messages and filters for the notifications you care about:
|
|
|
|
```python
|
|
from fastmcp import Client
|
|
|
|
async def message_handler(message):
|
|
"""Handle MCP notifications from the server."""
|
|
if hasattr(message, 'root'):
|
|
method = message.root.method
|
|
|
|
if method == "notifications/tools/list_changed":
|
|
print("Tools have changed - refresh tool cache")
|
|
elif method == "notifications/resources/list_changed":
|
|
print("Resources have changed")
|
|
elif method == "notifications/prompts/list_changed":
|
|
print("Prompts have changed")
|
|
|
|
client = Client(
|
|
"my_mcp_server.py",
|
|
message_handler=message_handler,
|
|
)
|
|
```
|
|
|
|
## MessageHandler Class
|
|
|
|
For fine-grained targeting, subclass `MessageHandler` to use specific hooks:
|
|
|
|
```python
|
|
from fastmcp import Client
|
|
from fastmcp.client.messages import MessageHandler
|
|
import mcp.types
|
|
|
|
class MyMessageHandler(MessageHandler):
|
|
async def on_tool_list_changed(
|
|
self, notification: mcp.types.ToolListChangedNotification
|
|
) -> None:
|
|
"""Handle tool list changes."""
|
|
print("Tool list changed - refreshing available tools")
|
|
|
|
async def on_resource_list_changed(
|
|
self, notification: mcp.types.ResourceListChangedNotification
|
|
) -> None:
|
|
"""Handle resource list changes."""
|
|
print("Resource list changed")
|
|
|
|
async def on_prompt_list_changed(
|
|
self, notification: mcp.types.PromptListChangedNotification
|
|
) -> None:
|
|
"""Handle prompt list changes."""
|
|
print("Prompt list changed")
|
|
|
|
client = Client(
|
|
"my_mcp_server.py",
|
|
message_handler=MyMessageHandler(),
|
|
)
|
|
```
|
|
|
|
### Handler Template
|
|
|
|
```python
|
|
from fastmcp.client.messages import MessageHandler
|
|
import mcp.types
|
|
|
|
class MyMessageHandler(MessageHandler):
|
|
async def on_message(self, message) -> None:
|
|
"""Called for ALL messages (requests and notifications)."""
|
|
pass
|
|
|
|
async def on_notification(
|
|
self, notification: mcp.types.ServerNotification
|
|
) -> None:
|
|
"""Called for notifications (fire-and-forget)."""
|
|
pass
|
|
|
|
async def on_tool_list_changed(
|
|
self, notification: mcp.types.ToolListChangedNotification
|
|
) -> None:
|
|
"""Called when the server's tool list changes."""
|
|
pass
|
|
|
|
async def on_resource_list_changed(
|
|
self, notification: mcp.types.ResourceListChangedNotification
|
|
) -> None:
|
|
"""Called when the server's resource list changes."""
|
|
pass
|
|
|
|
async def on_prompt_list_changed(
|
|
self, notification: mcp.types.PromptListChangedNotification
|
|
) -> None:
|
|
"""Called when the server's prompt list changes."""
|
|
pass
|
|
|
|
async def on_progress(
|
|
self, notification: mcp.types.ProgressNotification
|
|
) -> None:
|
|
"""Called for progress updates during long-running operations."""
|
|
pass
|
|
|
|
async def on_logging_message(
|
|
self, notification: mcp.types.LoggingMessageNotification
|
|
) -> None:
|
|
"""Called for log messages from the server."""
|
|
pass
|
|
```
|
|
|
|
## List Change Notifications
|
|
|
|
A practical example of maintaining a tool cache that refreshes when tools change:
|
|
|
|
```python
|
|
from fastmcp import Client
|
|
from fastmcp.client.messages import MessageHandler
|
|
import mcp.types
|
|
|
|
class ToolCacheHandler(MessageHandler):
|
|
def __init__(self):
|
|
self.cached_tools = []
|
|
|
|
async def on_tool_list_changed(
|
|
self, notification: mcp.types.ToolListChangedNotification
|
|
) -> None:
|
|
"""Clear tool cache when tools change."""
|
|
print("Tools changed - clearing cache")
|
|
self.cached_tools = [] # Force refresh on next access
|
|
|
|
client = Client("server.py", message_handler=ToolCacheHandler())
|
|
```
|
|
|
|
## Server Requests
|
|
|
|
While the message handler receives server-initiated requests, you should use dedicated callback parameters for most interactive scenarios:
|
|
|
|
- **Sampling requests**: Use [`sampling_handler`](/clients/sampling)
|
|
- **Elicitation requests**: Use [`elicitation_handler`](/clients/elicitation)
|
|
- **Progress updates**: Use [`progress_handler`](/clients/progress)
|
|
- **Log messages**: Use [`log_handler`](/clients/logging)
|
|
|
|
The message handler is primarily for monitoring and handling notifications rather than responding to requests.
|