Skip to content

Commit 890cacc

Browse files
committed
Update FastMCP constructor to respect Pydantic settings configuration
1 parent e68e513 commit 890cacc

File tree

1 file changed

+79
-51
lines changed

1 file changed

+79
-51
lines changed

src/mcp/server/fastmcp/server.py

Lines changed: 79 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010

1111
import anyio
1212
import pydantic_core
13-
from pydantic import BaseModel
13+
from pydantic import BaseModel, Field
1414
from pydantic.networks import AnyUrl
1515
from pydantic_settings import BaseSettings, SettingsConfigDict
1616
from starlette.applications import Starlette
@@ -68,42 +68,47 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
6868
)
6969

7070
# Server settings
71-
debug: bool
72-
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
71+
debug: bool = False
72+
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
7373

7474
# HTTP settings
75-
host: str
76-
port: int
77-
mount_path: str
78-
sse_path: str
79-
message_path: str
80-
streamable_http_path: str
75+
host: str = "127.0.0.1"
76+
port: int = 8000
77+
mount_path: str = "/"
78+
sse_path: str = "/sse"
79+
message_path: str = "/messages/"
80+
streamable_http_path: str = "/mcp"
8181

8282
# StreamableHTTP settings
83-
json_response: bool
84-
stateless_http: bool
83+
json_response: bool = False
84+
stateless_http: bool = False
8585
"""Define if the server should create a new transport per request."""
8686

8787
# resource settings
88-
warn_on_duplicate_resources: bool
88+
warn_on_duplicate_resources: bool = True
8989

9090
# tool settings
91-
warn_on_duplicate_tools: bool
91+
warn_on_duplicate_tools: bool = True
9292

9393
# prompt settings
94-
warn_on_duplicate_prompts: bool
94+
warn_on_duplicate_prompts: bool = True
9595

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

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

103-
auth: AuthSettings | None
108+
auth: AuthSettings | None = None
104109

105110
# Transport security settings (DNS rebinding protection)
106-
transport_security: TransportSecuritySettings | None
111+
transport_security: TransportSecuritySettings | None = None
107112

108113

109114
def lifespan_wrapper(
@@ -128,43 +133,66 @@ def __init__(
128133
event_store: EventStore | None = None,
129134
*,
130135
tools: list[Tool] | None = None,
131-
debug: bool = False,
132-
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
133-
host: str = "127.0.0.1",
134-
port: int = 8000,
135-
mount_path: str = "/",
136-
sse_path: str = "/sse",
137-
message_path: str = "/messages/",
138-
streamable_http_path: str = "/mcp",
139-
json_response: bool = False,
140-
stateless_http: bool = False,
141-
warn_on_duplicate_resources: bool = True,
142-
warn_on_duplicate_tools: bool = True,
143-
warn_on_duplicate_prompts: bool = True,
144-
dependencies: Collection[str] = (),
136+
debug: bool | None = None,
137+
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | None = None,
138+
host: str | None = None,
139+
port: int | None = None,
140+
mount_path: str | None = None,
141+
sse_path: str | None = None,
142+
message_path: str | None = None,
143+
streamable_http_path: str | None = None,
144+
json_response: bool | None = None,
145+
stateless_http: bool | None = None,
146+
warn_on_duplicate_resources: bool | None = None,
147+
warn_on_duplicate_tools: bool | None = None,
148+
warn_on_duplicate_prompts: bool | None = None,
149+
dependencies: Collection[str] | None = None,
145150
lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None = None,
146151
auth: AuthSettings | None = None,
147152
transport_security: TransportSecuritySettings | None = None,
148153
):
149-
self.settings = Settings(
150-
debug=debug,
151-
log_level=log_level,
152-
host=host,
153-
port=port,
154-
mount_path=mount_path,
155-
sse_path=sse_path,
156-
message_path=message_path,
157-
streamable_http_path=streamable_http_path,
158-
json_response=json_response,
159-
stateless_http=stateless_http,
160-
warn_on_duplicate_resources=warn_on_duplicate_resources,
161-
warn_on_duplicate_tools=warn_on_duplicate_tools,
162-
warn_on_duplicate_prompts=warn_on_duplicate_prompts,
163-
dependencies=list(dependencies),
164-
lifespan=lifespan,
165-
auth=auth,
166-
transport_security=transport_security,
167-
)
154+
# Collect all provided settings, which can be provided as explicit parameters,
155+
# environment variables or .env via the Pydantic settings,
156+
# or defaults from the Settings class.
157+
settings_kwargs: dict[str, Any] = {}
158+
159+
if debug is not None:
160+
settings_kwargs["debug"] = debug
161+
if log_level is not None:
162+
settings_kwargs["log_level"] = log_level
163+
if host is not None:
164+
settings_kwargs["host"] = host
165+
if port is not None:
166+
settings_kwargs["port"] = port
167+
if mount_path is not None:
168+
settings_kwargs["mount_path"] = mount_path
169+
if sse_path is not None:
170+
settings_kwargs["sse_path"] = sse_path
171+
if message_path is not None:
172+
settings_kwargs["message_path"] = message_path
173+
if streamable_http_path is not None:
174+
settings_kwargs["streamable_http_path"] = streamable_http_path
175+
if json_response is not None:
176+
settings_kwargs["json_response"] = json_response
177+
if stateless_http is not None:
178+
settings_kwargs["stateless_http"] = stateless_http
179+
if warn_on_duplicate_resources is not None:
180+
settings_kwargs["warn_on_duplicate_resources"] = warn_on_duplicate_resources
181+
if warn_on_duplicate_tools is not None:
182+
settings_kwargs["warn_on_duplicate_tools"] = warn_on_duplicate_tools
183+
if warn_on_duplicate_prompts is not None:
184+
settings_kwargs["warn_on_duplicate_prompts"] = warn_on_duplicate_prompts
185+
if dependencies is not None:
186+
settings_kwargs["dependencies"] = list(dependencies)
187+
if lifespan is not None:
188+
settings_kwargs["lifespan"] = lifespan
189+
if auth is not None:
190+
settings_kwargs["auth"] = auth
191+
if transport_security is not None:
192+
settings_kwargs["transport_security"] = transport_security
193+
194+
# Create settings - only specify the ones that were explicitly provided
195+
self.settings: Settings[LifespanResultT] = Settings(**settings_kwargs)
168196

169197
self._mcp_server = MCPServer(
170198
name=name or "FastMCP",

0 commit comments

Comments
 (0)