Skip to content

Update FastMCP constructor to respect Pydantic settings configuration #1244

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 79 additions & 51 deletions src/mcp/server/fastmcp/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@

import anyio
import pydantic_core
from pydantic import BaseModel
from pydantic import BaseModel, Field
from pydantic.networks import AnyUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
from starlette.applications import Starlette
Expand Down Expand Up @@ -68,42 +68,47 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
)

# Server settings
debug: bool
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
debug: bool = False
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"

# HTTP settings
host: str
port: int
mount_path: str
sse_path: str
message_path: str
streamable_http_path: str
host: str = "127.0.0.1"
port: int = 8000
mount_path: str = "/"
sse_path: str = "/sse"
message_path: str = "/messages/"
streamable_http_path: str = "/mcp"

# StreamableHTTP settings
json_response: bool
stateless_http: bool
json_response: bool = False
stateless_http: bool = False
"""Define if the server should create a new transport per request."""

# resource settings
warn_on_duplicate_resources: bool
warn_on_duplicate_resources: bool = True

# tool settings
warn_on_duplicate_tools: bool
warn_on_duplicate_tools: bool = True

# prompt settings
warn_on_duplicate_prompts: bool
warn_on_duplicate_prompts: bool = True

# TODO(Marcelo): Investigate if this is used. If it is, it's probably a good idea to remove it.
dependencies: list[str]
dependencies: list[str] = Field(
default_factory=list,
description="List of dependencies to install in the server environment",
)
"""A list of dependencies to install in the server environment."""

lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None
lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None = Field(
None, description="Lifespan context manager"
)
"""A async context manager that will be called when the server is started."""

auth: AuthSettings | None
auth: AuthSettings | None = None

# Transport security settings (DNS rebinding protection)
transport_security: TransportSecuritySettings | None
transport_security: TransportSecuritySettings | None = None


def lifespan_wrapper(
Expand All @@ -128,43 +133,66 @@ def __init__(
event_store: EventStore | None = None,
*,
tools: list[Tool] | None = None,
debug: bool = False,
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
host: str = "127.0.0.1",
port: int = 8000,
mount_path: str = "/",
sse_path: str = "/sse",
message_path: str = "/messages/",
streamable_http_path: str = "/mcp",
json_response: bool = False,
stateless_http: bool = False,
warn_on_duplicate_resources: bool = True,
warn_on_duplicate_tools: bool = True,
warn_on_duplicate_prompts: bool = True,
dependencies: Collection[str] = (),
debug: bool | None = None,
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None = None,
host: str | None = None,
port: int | None = None,
mount_path: str | None = None,
sse_path: str | None = None,
message_path: str | None = None,
streamable_http_path: str | None = None,
json_response: bool | None = None,
stateless_http: bool | None = None,
warn_on_duplicate_resources: bool | None = None,
warn_on_duplicate_tools: bool | None = None,
warn_on_duplicate_prompts: bool | None = None,
dependencies: Collection[str] | None = None,
lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None = None,
auth: AuthSettings | None = None,
transport_security: TransportSecuritySettings | None = None,
):
self.settings = Settings(
debug=debug,
log_level=log_level,
host=host,
port=port,
mount_path=mount_path,
sse_path=sse_path,
message_path=message_path,
streamable_http_path=streamable_http_path,
json_response=json_response,
stateless_http=stateless_http,
warn_on_duplicate_resources=warn_on_duplicate_resources,
warn_on_duplicate_tools=warn_on_duplicate_tools,
warn_on_duplicate_prompts=warn_on_duplicate_prompts,
dependencies=list(dependencies),
lifespan=lifespan,
auth=auth,
transport_security=transport_security,
)
# Collect all provided settings, which can be provided as explicit parameters,
# environment variables or .env via the Pydantic settings,
# or defaults from the Settings class.
settings_kwargs: dict[str, Any] = {}

if debug is not None:
settings_kwargs["debug"] = debug
if log_level is not None:
settings_kwargs["log_level"] = log_level
if host is not None:
settings_kwargs["host"] = host
if port is not None:
settings_kwargs["port"] = port
if mount_path is not None:
settings_kwargs["mount_path"] = mount_path
if sse_path is not None:
settings_kwargs["sse_path"] = sse_path
if message_path is not None:
settings_kwargs["message_path"] = message_path
if streamable_http_path is not None:
settings_kwargs["streamable_http_path"] = streamable_http_path
if json_response is not None:
settings_kwargs["json_response"] = json_response
if stateless_http is not None:
settings_kwargs["stateless_http"] = stateless_http
if warn_on_duplicate_resources is not None:
settings_kwargs["warn_on_duplicate_resources"] = warn_on_duplicate_resources
if warn_on_duplicate_tools is not None:
settings_kwargs["warn_on_duplicate_tools"] = warn_on_duplicate_tools
if warn_on_duplicate_prompts is not None:
settings_kwargs["warn_on_duplicate_prompts"] = warn_on_duplicate_prompts
if dependencies is not None:
settings_kwargs["dependencies"] = list(dependencies)
if lifespan is not None:
settings_kwargs["lifespan"] = lifespan
if auth is not None:
settings_kwargs["auth"] = auth
if transport_security is not None:
settings_kwargs["transport_security"] = transport_security

# Create settings - only specify the ones that were explicitly provided
self.settings: Settings[LifespanResultT] = Settings(**settings_kwargs)

self._mcp_server = MCPServer(
name=name or "FastMCP",
Expand Down
Loading