ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Support logging/setLevel and add client_log_level setting (#3491)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-03-14 10:36:01 -04:00 • committed by GitHub
commit e2bdc9288b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 306 additions and 116 deletions
Download patch file Download diff file Expand all files Collapse all files

3
docs/more/settings.mdx
View file

@ -21,8 +21,9 @@ You can change which `.env` file is loaded by setting the `FASTMCP_ENV_FILE` env
| Environment Variable | Type | Default | Description |
|---|---|---|---|
| `FASTMCP_LOG_LEVEL` | `Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]` | `INFO` | Log level. Case-insensitive. |
| `FASTMCP_LOG_LEVEL` | `Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]` | `INFO` | Log level for FastMCP's own logging output. Case-insensitive. |
| `FASTMCP_LOG_ENABLED` | `bool` | `true` | Enable or disable FastMCP logging entirely. |
| `FASTMCP_CLIENT_LOG_LEVEL` | `Literal["debug", "info", "notice", "warning", "error", "critical", "alert", "emergency"]` | None | Default minimum log level for messages sent to MCP clients via `context.log()`. When set, messages below this level are suppressed. Individual clients can override this per-session using the MCP `logging/setLevel` request. |
| `FASTMCP_ENABLE_RICH_LOGGING` | `bool` | `true` | Use rich formatting for log output. Set to `false` for plain Python logging. |
| `FASTMCP_ENABLE_RICH_TRACEBACKS` | `bool` | `true` | Use rich tracebacks for errors. |
| `FASTMCP_DEPRECATION_WARNINGS` | `bool` | `true` | Show deprecation warnings. |

263
docs/servers/server.mdx
View file

@ -11,36 +11,105 @@ The `FastMCP` class is the central piece of every FastMCP application. It acts a
## Creating a Server
Instantiate a server by providing a name that identifies it in client applications and logs. You can also provide instructions that help clients understand the server's purpose.
At its simplest, a FastMCP server just needs a name. Everything else has sensible defaults.
```python
from fastmcp import FastMCP
mcp = FastMCP(name="MyAssistantServer")
mcp = FastMCP("MyServer")
```
# Instructions help clients understand how to interact with the server
mcp_with_instructions = FastMCP(
name="HelpfulAssistant",
instructions="""
This server provides data analysis tools.
Call get_average() to analyze numerical data.
""",
Instructions help clients (and the LLMs behind them) understand what your server does and how to use it effectively.
```python
mcp = FastMCP(
"DataAnalysis",
instructions="Provides tools for analyzing numerical datasets. Start with get_summary() for an overview.",
)
```
The `FastMCP` constructor accepts several configuration options. The most commonly used parameters control server identity, authentication, and component behavior.
## Components
<Card icon="code" title="FastMCP Constructor Parameters">
FastMCP servers expose three types of components to clients, each serving a distinct role in the MCP protocol.
**Tools** are functions that clients invoke to perform actions or access external systems.
```python
@mcp.tool
def multiply(a: float, b: float) -> float:
"""Multiplies two numbers together."""
return a * b
```
**Resources** expose data that clients can read — passive data sources rather than invocable functions.
```python
@mcp.resource("data://config")
def get_config() -> dict:
return {"theme": "dark", "version": "1.0"}
```
**Prompts** are reusable message templates that guide LLM interactions.
```python
@mcp.prompt
def analyze_data(data_points: list[float]) -> str:
formatted_data = ", ".join(str(point) for point in data_points)
return f"Please analyze these data points: {formatted_data}"
```
Each component type has detailed documentation: [Tools](/servers/tools), [Resources](/servers/resources) (including [Resource Templates](/servers/resources#resource-templates)), and [Prompts](/servers/prompts).
## Running the Server
Start your server by calling `mcp.run()`. The `if __name__` guard ensures compatibility with MCP clients that launch your server as a subprocess.
```python
from fastmcp import FastMCP
mcp = FastMCP("MyServer")
@mcp.tool
def greet(name: str) -> str:
"""Greet a user by name."""
return f"Hello, {name}!"
if __name__ == "__main__":
mcp.run()
```
FastMCP supports several transports:
- **STDIO** (default): For local integrations and CLI tools
- **HTTP**: For web services using the Streamable HTTP protocol
- **SSE**: Legacy web transport (deprecated)
```python
# Run with HTTP transport
mcp.run(transport="http", host="127.0.0.1", port=9000)
```
The server can also be run using the FastMCP CLI. For detailed information on transports and deployment, see [Running Your Server](/deployment/running-server).
## Configuration Reference
The `FastMCP` constructor accepts parameters organized into four categories: identity, composition, behavior, and handlers.
### Identity
These parameters control how your server presents itself to clients.
<Card>
<ParamField body="name" type="str" default="FastMCP">
A human-readable name for your server
A human-readable name for your server, shown in client applications and logs
</ParamField>
<ParamField body="instructions" type="str | None">
Description of how to interact with this server. These instructions help clients understand the server's purpose and available functionality
Description of how to interact with this server. Clients surface these instructions to help LLMs understand the server's purpose and available functionality
</ParamField>
<ParamField body="version" type="str | None">
Version string for your server. If not provided, defaults to the FastMCP library version
Version string for your server. Defaults to the FastMCP library version if not provided
</ParamField>
<ParamField body="website_url" type="str | None">
@ -52,108 +121,106 @@ The `FastMCP` constructor accepts several configuration options. The most common
<ParamField body="icons" type="list[Icon] | None">
<VersionBadge version="2.13.0" />
List of icon representations for your server. Icons help users visually identify your server in client applications. See [Icons](/servers/icons) for detailed examples
List of icon representations for your server. See [Icons](/servers/icons) for details
</ParamField>
</Card>
### Composition
These parameters control what your server is built from — its components, middleware, providers, and lifecycle.
<Card>
<ParamField body="tools" type="list[Tool | Callable] | None">
Tools to register on the server. An alternative to the `@mcp.tool` decorator when you need to add tools programmatically
</ParamField>
<ParamField body="auth" type="OAuthProvider | TokenVerifier | None">
Authentication provider for securing HTTP-based transports. See [Authentication](/servers/auth/authentication) for configuration options
Authentication provider for securing HTTP-based transports. See [Authentication](/servers/auth/authentication) for configuration
</ParamField>
<ParamField body="lifespan" type="Lifespan | AsyncContextManager | None">
Server-level setup and teardown logic. See [Lifespans](/servers/lifespan) for composable lifespans
<ParamField body="middleware" type="list[Middleware] | None">
[Middleware](/servers/middleware) that intercepts and transforms every MCP message flowing through the server — requests, responses, and notifications in both directions. Use for cross-cutting concerns like logging, error handling, and rate limiting
</ParamField>
<ParamField body="tools" type="list[Tool | Callable] | None">
A list of tools (or functions to convert to tools) to add to the server. In some cases, providing tools programmatically may be more convenient than using the `@mcp.tool` decorator
<ParamField body="providers" type="list[Provider] | None">
[Providers](/servers/providers) that supply tools, resources, and prompts dynamically. Providers are queried at request time, so they can serve components from databases, APIs, or other external sources
</ParamField>
<ParamField body="transforms" type="list[Transform] | None">
<VersionBadge version="3.1.0" />
Server-level [transforms](/servers/transforms/transforms) to apply to all components. Transforms modify how tools, resources, and prompts are presented to clients — for example, [search transforms](/servers/transforms/tool-search) replace large catalogs with on-demand discovery, and [CodeMode](/servers/transforms/code-mode) lets LLMs write scripts that chain tool calls in a sandbox
Server-level [transforms](/servers/transforms/transforms) to apply to all components. Transforms modify how tools, resources, and prompts are presented to clients — for example, [search transforms](/servers/transforms/tool-search) replace large catalogs with on-demand discovery
</ParamField>
<ParamField body="lifespan" type="Lifespan | AsyncContextManager | None">
Server-level setup and teardown logic that runs when the server starts and stops. See [Lifespans](/servers/lifespan) for composable lifespans
</ParamField>
</Card>
### Behavior
These parameters tune how the server processes requests and communicates with clients.
<Card>
<ParamField body="on_duplicate" type='Literal["warn", "error", "replace", "ignore"]' default="warn">
How to handle duplicate component registrations
</ParamField>
<ParamField body="strict_input_validation" type="bool" default="False">
<VersionBadge version="2.13.0" />
Controls how tool input parameters are validated. When `False` (default), FastMCP uses Pydantic's flexible validation that coerces compatible inputs (e.g., `"10"` to `10` for int parameters). When `True`, uses the MCP SDK's JSON Schema validation to validate inputs against the exact schema before passing them to your function, rejecting any type mismatches. The default mode improves compatibility with LLM clients while maintaining type safety. See [Input Validation Modes](/servers/tools#input-validation-modes) for details
When `False` (default), FastMCP uses Pydantic's flexible validation that coerces compatible inputs (e.g., `"10"` → `10` for int parameters). When `True`, validates inputs against the exact JSON Schema before calling your function, rejecting type mismatches. See [Input Validation Modes](/servers/tools#input-validation-modes) for details
</ParamField>
<ParamField body="mask_error_details" type="bool | None">
When `True`, replaces internal error details in tool/resource responses with a generic message to avoid leaking implementation details to clients. Defaults to the `FASTMCP_MASK_ERROR_DETAILS` environment variable
</ParamField>
<ParamField body="list_page_size" type="int | None" default="None">
<VersionBadge version="3.0.0" />
Maximum number of items per page for list operations (`tools/list`, `resources/list`, etc.). When `None` (default), all results are returned in a single response. When set, responses are paginated and include a `nextCursor` for fetching additional pages. See [Pagination](/servers/pagination) for details
Maximum items per page for list operations (`tools/list`, `resources/list`, etc.). When `None`, all results are returned in a single response. See [Pagination](/servers/pagination) for details
</ParamField>
<ParamField body="tasks" type="bool | None" default="False">
Enable background task support. When `True`, tools and resources can return `CreateTaskResult` to run work asynchronously while the client polls for results
</ParamField>
<ParamField body="client_log_level" type="LoggingLevel | None">
<VersionBadge version="3.2.0" />
Default minimum log level for messages sent to MCP clients via `context.log()`. When set, messages below this level are suppressed. Individual clients can override this per-session using the MCP `logging/setLevel` request. One of `"debug"`, `"info"`, `"notice"`, `"warning"`, `"error"`, `"critical"`, `"alert"`, or `"emergency"`
</ParamField>
<ParamField body="dereference_schemas" type="bool" default="True">
Automatically dereference `$ref` pointers in JSON schemas generated from complex Pydantic models. Most clients require flat schemas without `$ref`, so this should usually stay enabled
</ParamField>
</Card>
## Components
### Handlers and Storage
FastMCP servers expose three types of components to clients. Each type serves a distinct purpose in the MCP protocol.
These parameters provide custom handlers for MCP capabilities and persistent storage for session state.
### Tools
<Card>
<ParamField body="sampling_handler" type="SamplingHandler | None">
Custom handler for MCP sampling requests (server-initiated LLM calls). See [Sampling](/servers/sampling) for details
</ParamField>
Tools are functions that clients can invoke to perform actions or access external systems. They're the primary way clients interact with your server's capabilities.
<ParamField body="sampling_handler_behavior" type='Literal["always", "fallback"] | None' default="fallback">
When `"fallback"`, the sampling handler is used only when no tool-specific handler exists. When `"always"`, this handler is used for all sampling requests
</ParamField>
```python
@mcp.tool
def multiply(a: float, b: float) -> float:
"""Multiplies two numbers together."""
return a * b
```
<ParamField body="session_state_store" type="AsyncKeyValue | None">
Persistent key-value store for session state that survives across requests. Defaults to an in-memory store. Provide a custom implementation for persistence across server restarts
</ParamField>
</Card>
See [Tools](/servers/tools) for detailed documentation.
### Resources
Resources expose data that clients can read. Unlike tools, resources are passive data sources that clients pull from rather than invoke.
```python
@mcp.resource("data://config")
def get_config() -> dict:
"""Provides the application configuration."""
return {"theme": "dark", "version": "1.0"}
```
See [Resources](/servers/resources) for detailed documentation.
### Resource Templates
Resource templates are parameterized resources. The client provides values for template parameters in the URI, and the server returns data specific to those parameters.
```python
@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
"""Retrieves a user's profile by ID."""
return {"id": user_id, "name": f"User {user_id}", "status": "active"}
```
See [Resource Templates](/servers/resources#resource-templates) for detailed documentation.
### Prompts
Prompts are reusable message templates that guide LLM interactions. They help establish consistent patterns for how clients should frame requests.
```python
@mcp.prompt
def analyze_data(data_points: list[float]) -> str:
"""Creates a prompt asking for analysis of numerical data."""
formatted_data = ", ".join(str(point) for point in data_points)
return f"Please analyze these data points: {formatted_data}"
```
See [Prompts](/servers/prompts) for detailed documentation.
## Tag-Based Filtering
<VersionBadge version="2.8.0" />
Tags let you categorize components and selectively expose them based on configurable include/exclude sets. This is useful for creating different views of your server for different environments or user types.
Components can be tagged when defined using the `tags` parameter. A component can have multiple tags, and filtering operates on tag membership.
Tags let you categorize components and selectively expose them. This is useful for creating different views of your server for different environments or user types.
```python
@mcp.tool(tags={"public", "utility"})
@ -174,8 +241,6 @@ The filtering logic works as follows:
To ensure a component is never exposed, you can set `enabled=False` on the component itself. See the component-specific documentation for details.
</Tip>
Configure tag-based filtering after creating your server.
```python
# Only expose components tagged with "public"
mcp = FastMCP()
@ -192,38 +257,9 @@ mcp.enable(tags={"admin"}, only=True).disable(tags={"deprecated"})
This filtering applies to all component types (tools, resources, resource templates, and prompts) and affects both listing and access.
## Running the Server
FastMCP servers communicate with clients through transport mechanisms. Start your server by calling `mcp.run()`, typically within an `if __name__ == "__main__":` block. This pattern ensures compatibility with various MCP clients.
```python
from fastmcp import FastMCP
mcp = FastMCP(name="MyServer")
@mcp.tool
def greet(name: str) -> str:
"""Greet a user by name."""
return f"Hello, {name}!"
if __name__ == "__main__":
# Defaults to STDIO transport
mcp.run()
# Or use HTTP transport
# mcp.run(transport="http", host="127.0.0.1", port=9000)
```
FastMCP supports several transports:
- **STDIO** (default): For local integrations and CLI tools
- **HTTP**: For web services using the Streamable HTTP protocol
- **SSE**: Legacy web transport (deprecated)
The server can also be run using the FastMCP CLI. For detailed information on transports and configuration, see the [Running Your Server](/deployment/running-server) guide.
## Custom Routes
When running with HTTP transport, you can add custom web routes alongside your MCP endpoint using the `@custom_route` decorator. This is useful for auxiliary endpoints like health checks.
When running with HTTP transport, you can add custom web routes alongside your MCP endpoint using the `@custom_route` decorator.
```python
from fastmcp import FastMCP
@ -240,9 +276,4 @@ if __name__ == "__main__":
mcp.run(transport="http") # Health check at http://localhost:8000/health
```
Custom routes are served alongside your MCP endpoint and are useful for:
- Health check endpoints for monitoring
- Simple status or info endpoints
- Basic webhooks or callbacks
For more complex web applications, consider [mounting your MCP server into a FastAPI or Starlette app](/deployment/http#integration-with-web-frameworks).
Custom routes are useful for health checks, status endpoints, and simple webhooks. For more complex web applications, consider [mounting your MCP server into a FastAPI or Starlette app](/deployment/http#integration-with-web-frameworks).

19
src/fastmcp/server/context.py
View file

@ -1356,6 +1356,18 @@ class Context:
await _reset_visibility(self)
_MCP_LEVEL_SEVERITY: dict[LoggingLevel, int] = {
"debug": 0,
"info": 1,
"notice": 2,
"warning": 3,
"error": 4,
"critical": 5,
"alert": 6,
"emergency": 7,
}
async def _log_to_server_and_client(
data: LogData,
session: ServerSession,
@ -1364,6 +1376,13 @@ async def _log_to_server_and_client(
related_request_id: str | None = None,
) -> None:
"""Log a message to the server and client."""
from fastmcp.server.low_level import MiddlewareServerSession
if isinstance(session, MiddlewareServerSession):
min_level = session._minimum_logging_level or session.fastmcp.client_log_level
if min_level is not None:
if _MCP_LEVEL_SEVERITY[level] < _MCP_LEVEL_SEVERITY[min_level]:
return
msg_prefix = f"Sending {level.upper()} to client"

4
src/fastmcp/server/low_level.py
View file

@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Any
import anyio
import mcp.types
from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
from mcp import McpError
from mcp import LoggingLevel, McpError
from mcp.server.lowlevel.server import (
LifespanResultT,
NotificationOptions,
@ -41,6 +41,8 @@ class MiddlewareServerSession(ServerSession):
self._fastmcp_ref: weakref.ref[FastMCP] = weakref.ref(fastmcp)
# Task group for subscription tasks (set during session run)
self._subscription_task_group: anyio.TaskGroup | None = None # type: ignore[valid-type]
# Minimum logging level requested by the client via logging/setLevel
self._minimum_logging_level: LoggingLevel | None = None
@property
def fastmcp(self) -> FastMCP:

19
src/fastmcp/server/mixins/mcp_operations.py
View file

@ -79,6 +79,7 @@ class MCPOperationsMixin:
)
self._mcp_server.read_resource()(self._read_resource_mcp)
self._mcp_server.get_prompt()(self._get_prompt_mcp)
self._mcp_server.set_logging_level()(self._set_logging_level_mcp)
# Register SEP-1686 task protocol handlers
self._setup_task_protocol_handlers()
@ -352,3 +353,21 @@ class MCPOperationsMixin:
raise NotFoundError(f"Unknown prompt: {name!r}") from e
except NotFoundError:
raise
async def _set_logging_level_mcp(self, level: mcp.types.LoggingLevel) -> None:
"""Handle MCP 'logging/setLevel' requests.
Stores the requested minimum log level on the session so that
subsequent log messages below this level are suppressed.
"""
from fastmcp.server.low_level import MiddlewareServerSession
server = cast("FastMCP", self)
logger.debug(f"[{server.name}] Handler called: set_logging_level %s", level)
try:
ctx = server._mcp_server.request_context
session = ctx.session
if isinstance(session, MiddlewareServerSession):
session._minimum_logging_level = level
except LookupError:
pass

7
src/fastmcp/server/server.py
View file

@ -240,6 +240,7 @@ class FastMCP(
session_state_store: AsyncKeyValue | None = None,
sampling_handler: SamplingHandler | None = None,
sampling_handler_behavior: Literal["always", "fallback"] | None = None,
client_log_level: mcp.types.LoggingLevel | None = None,
**kwargs: Any,
):
_check_removed_kwargs(kwargs)
@ -332,6 +333,12 @@ class FastMCP(
else fastmcp.settings.strict_input_validation
)
self.client_log_level: mcp.types.LoggingLevel | None = (
client_log_level
if client_log_level is not None
else fastmcp.settings.client_log_level
)
self.middleware: list[Middleware] = list(middleware or [])
if dereference_schemas:

18
src/fastmcp/settings.py
View file

@ -21,6 +21,10 @@ ENV_FILE = os.getenv("FASTMCP_ENV_FILE", ".env")
LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
MCP_LOG_LEVEL = Literal[
"debug", "info", "notice", "warning", "error", "critical", "alert", "emergency"
]
DuplicateBehavior = Literal["warn", "error", "replace", "ignore"]
TEN_MB_IN_BYTES = 1024 * 1024 * 10
@ -254,6 +258,20 @@ class Settings(BaseSettings):
),
] = False
client_log_level: Annotated[
MCP_LOG_LEVEL | None,
Field(
description=inspect.cleandoc(
"""
Default minimum log level for messages sent to MCP clients.
When set, log messages below this level are suppressed.
Individual clients can override this per-session using the
MCP logging/setLevel request.
"""
),
),
] = None
strict_input_validation: Annotated[
bool,
Field(

93
tests/client/test_logs.py
View file

@ -90,6 +90,99 @@ class TestClientLogs:
assert caplog.records[1].levelname == "WARNING"
class TestSetLoggingLevel:
async def test_set_logging_level(self, fastmcp_server: FastMCP):
"""Client can set the minimum log level and lower-level messages are suppressed."""
log_handler = LogHandler()
async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
await client.set_logging_level("warning")
await client.call_tool(
"echo_log", {"message": "debug msg", "level": "debug"}
)
await client.call_tool("echo_log", {"message": "info msg", "level": "info"})
await client.call_tool(
"echo_log", {"message": "warning msg", "level": "warning"}
)
await client.call_tool(
"echo_log", {"message": "error msg", "level": "error"}
)
assert len(log_handler.logs) == 2
assert log_handler.logs[0].data["msg"] == "warning msg"
assert log_handler.logs[1].data["msg"] == "error msg"
async def test_set_logging_level_debug_allows_all(self, fastmcp_server: FastMCP):
"""Setting level to debug allows all messages through."""
log_handler = LogHandler()
async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
await client.set_logging_level("debug")
await client.call_tool(
"echo_log", {"message": "debug msg", "level": "debug"}
)
await client.call_tool("echo_log", {"message": "info msg", "level": "info"})
assert len(log_handler.logs) == 2
async def test_default_level_allows_all(self, fastmcp_server: FastMCP):
"""Without calling set_logging_level, all messages are sent."""
log_handler = LogHandler()
async with Client(fastmcp_server, log_handler=log_handler.handle_log) as client:
await client.call_tool(
"echo_log", {"message": "debug msg", "level": "debug"}
)
await client.call_tool("echo_log", {"message": "info msg", "level": "info"})
assert len(log_handler.logs) == 2
async def test_server_default_client_log_level(self):
"""Server-wide client_log_level filters messages for all sessions."""
mcp = FastMCP(client_log_level="error")
@mcp.tool
async def echo_log(
message: str, context: Context, level: LoggingLevel | None = None
) -> None:
await context.log(message=message, level=level)
log_handler = LogHandler()
async with Client(mcp, log_handler=log_handler.handle_log) as client:
await client.call_tool("echo_log", {"message": "info msg", "level": "info"})
await client.call_tool(
"echo_log", {"message": "warning msg", "level": "warning"}
)
await client.call_tool(
"echo_log", {"message": "error msg", "level": "error"}
)
assert len(log_handler.logs) == 1
assert log_handler.logs[0].data["msg"] == "error msg"
async def test_session_level_overrides_server_default(self):
"""Per-session setLevel overrides the server's client_log_level."""
mcp = FastMCP(client_log_level="error")
@mcp.tool
async def echo_log(
message: str, context: Context, level: LoggingLevel | None = None
) -> None:
await context.log(message=message, level=level)
log_handler = LogHandler()
async with Client(mcp, log_handler=log_handler.handle_log) as client:
await client.set_logging_level("warning")
await client.call_tool("echo_log", {"message": "info msg", "level": "info"})
await client.call_tool(
"echo_log", {"message": "warning msg", "level": "warning"}
)
await client.call_tool(
"echo_log", {"message": "error msg", "level": "error"}
)
assert len(log_handler.logs) == 2
assert log_handler.logs[0].data["msg"] == "warning msg"
assert log_handler.logs[1].data["msg"] == "error msg"
class TestDefaultLogHandler:
"""Tests for default_log_handler with data as any JSON-serializable type."""