File size: 11,561 Bytes
48f7143
ef5e55f
177d39a
0bf61b1
cfb4777
927e044
48f7143
0345194
927e044
b310c16
 
927e044
 
b310c16
 
ea8c89c
b8df5e6
905120e
 
 
 
ef5e55f
b8df5e6
2c1fc32
fa364d7
 
927e044
e2a36a0
 
 
 
 
 
927e044
 
 
 
 
 
 
 
 
 
 
e2a36a0
905120e
927e044
 
 
 
 
 
 
 
 
 
 
8c5bdf2
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
b8df5e6
48f7143
 
927e044
 
48f7143
 
7bdb963
 
8c5bdf2
48f7143
 
8c5bdf2
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
927e044
 
 
 
 
 
 
 
 
2101c78
 
927e044
 
 
 
 
 
 
2101c78
 
 
 
 
 
 
905120e
2101c78
 
 
 
cfb4777
 
48f7143
1db82ae
48f7143
1db82ae
 
 
 
 
 
 
 
8c5bdf2
 
acef890
 
 
 
 
 
 
 
 
 
b45adb7
9dbf7f8
 
 
 
 
 
 
 
 
 
 
 
 
 
177d39a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
b45adb7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
b056b3b
 
 
 
 
 
 
48f7143
6cbbd70
48f7143
c219134
48f7143
c219134
b8df5e6
 
85e9cfd
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
49e6154
 
 
 
48f7143
4c1f058
 
 
 
 
 
b310c16
3959209
 
ccc6631
 
 
3959209
 
 
 
ccc6631
 
 
 
 
3959209
 
 
ccc6631
 
 
 
b310c16
1175ac0
23dba76
1175ac0
 
 
 
 
23dba76
 
1175ac0
 
 
 
 
23dba76
1175ac0
 
 
 
 
23dba76
 
1175ac0
 
 
 
 
957b451
 
 
 
 
 
 
 
 
 
 
 
 
 
 
28c234f
 
 
 
 
 
 
 
 
 
 
 
 
 
48f7143
0bf61b1
 
 
 
 
 
 
 
 
 
 
52d423c
0bf61b1
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
from __future__ import annotations as _annotations

import inspect
import warnings
from pathlib import Path
from typing import Annotated, Any, Literal

from pydantic import Field, field_validator
from pydantic.fields import FieldInfo
from pydantic_settings import (
    BaseSettings,
    EnvSettingsSource,
    PydanticBaseSettingsSource,
    SettingsConfigDict,
)
from typing_extensions import Self

from fastmcp.utilities.logging import get_logger

logger = get_logger(__name__)

LOG_LEVEL = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

DuplicateBehavior = Literal["warn", "error", "replace", "ignore"]


class ExtendedEnvSettingsSource(EnvSettingsSource):
    """
    A special EnvSettingsSource that allows for multiple env var prefixes to be used.

    Raises a deprecation warning if the old `FASTMCP_SERVER_` prefix is used.
    """

    def get_field_value(
        self, field: FieldInfo, field_name: str
    ) -> tuple[Any, str, bool]:
        if prefixes := self.config.get("env_prefixes"):
            for prefix in prefixes:
                self.env_prefix = prefix
                env_val, field_key, value_is_complex = super().get_field_value(
                    field, field_name
                )
                if env_val is not None:
                    if prefix == "FASTMCP_SERVER_":
                        # Deprecated in 2.8.0
                        logger.warning(
                            "Using `FASTMCP_SERVER_` environment variables is deprecated. Use `FASTMCP_` instead.",
                        )
                    return env_val, field_key, value_is_complex

        return super().get_field_value(field, field_name)


class ExtendedSettingsConfigDict(SettingsConfigDict, total=False):
    env_prefixes: list[str] | None


class ExperimentalSettings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="FASTMCP_EXPERIMENTAL_",
        extra="ignore",
    )

    enable_new_openapi_parser: Annotated[
        bool,
        Field(
            description=inspect.cleandoc(
                """
                Whether to use the new OpenAPI parser. This parser was introduced
                for testing in 2.11 and will become the default soon.
                """
            ),
        ),
    ] = False


class Settings(BaseSettings):
    """FastMCP settings."""

    model_config = ExtendedSettingsConfigDict(
        env_prefixes=["FASTMCP_", "FASTMCP_SERVER_"],
        env_file=".env",
        extra="ignore",
        env_nested_delimiter="__",
        nested_model_default_partial_update=True,
        validate_assignment=True,
    )

    def get_setting(self, attr: str) -> Any:
        """
        Get a setting. If the setting contains one or more `__`, it will be
        treated as a nested setting.
        """
        settings = self
        while "__" in attr:
            parent_attr, attr = attr.split("__", 1)
            if not hasattr(settings, parent_attr):
                raise AttributeError(f"Setting {parent_attr} does not exist.")
            settings = getattr(settings, parent_attr)
        return getattr(settings, attr)

    def set_setting(self, attr: str, value: Any) -> None:
        """
        Set a setting. If the setting contains one or more `__`, it will be
        treated as a nested setting.
        """
        settings = self
        while "__" in attr:
            parent_attr, attr = attr.split("__", 1)
            if not hasattr(settings, parent_attr):
                raise AttributeError(f"Setting {parent_attr} does not exist.")
            settings = getattr(settings, parent_attr)
        setattr(settings, attr, value)

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> tuple[PydanticBaseSettingsSource, ...]:
        # can remove this classmethod after deprecated FASTMCP_SERVER_ prefix is
        # removed
        return (
            init_settings,
            ExtendedEnvSettingsSource(settings_cls),
            dotenv_settings,
            file_secret_settings,
        )

    @property
    def settings(self) -> Self:
        """
        This property is for backwards compatibility with FastMCP < 2.8.0,
        which accessed fastmcp.settings.settings
        """
        # Deprecated in 2.8.0
        logger.warning(
            "Using fastmcp.settings.settings is deprecated. Use fastmcp.settings instead.",
        )
        return self

    home: Path = Path.home() / ".fastmcp"

    test_mode: bool = False

    log_level: LOG_LEVEL = "INFO"

    @field_validator("log_level", mode="before")
    @classmethod
    def normalize_log_level(cls, v):
        if isinstance(v, str):
            return v.upper()
        return v

    experimental: ExperimentalSettings = ExperimentalSettings()

    enable_rich_tracebacks: Annotated[
        bool,
        Field(
            description=inspect.cleandoc(
                """
                If True, will use rich tracebacks for logging.
                """
            )
        ),
    ] = True

    deprecation_warnings: Annotated[
        bool,
        Field(
            description=inspect.cleandoc(
                """
                Whether to show deprecation warnings. You can completely reset
                Python's warning behavior by running `warnings.resetwarnings()`.
                Note this will NOT apply to deprecation warnings from the
                settings class itself.
                """,
            )
        ),
    ] = True

    client_raise_first_exceptiongroup_error: Annotated[
        bool,
        Field(
            default=True,
            description=inspect.cleandoc(
                """
                Many MCP components operate in anyio taskgroups, and raise
                ExceptionGroups instead of exceptions. If this setting is True, FastMCP Clients
                will `raise` the first error in any ExceptionGroup instead of raising
                the ExceptionGroup as a whole. This is useful for debugging, but may
                mask other errors.
                """
            ),
        ),
    ] = True

    resource_prefix_format: Annotated[
        Literal["protocol", "path"],
        Field(
            default="path",
            description=inspect.cleandoc(
                """
                When perfixing a resource URI, either use path formatting (resource://prefix/path)
                or protocol formatting (prefix+resource://path). Protocol formatting was the default in FastMCP < 2.4;
                path formatting is current default.
                """
            ),
        ),
    ] = "path"

    client_init_timeout: Annotated[
        float | None,
        Field(
            description="The timeout for the client's initialization handshake, in seconds. Set to None or 0 to disable.",
        ),
    ] = None

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

    # error handling
    mask_error_details: Annotated[
        bool,
        Field(
            default=False,
            description=inspect.cleandoc(
                """
                If True, error details from user-supplied functions (tool, resource, prompt)
                will be masked before being sent to clients. Only error messages from explicitly
                raised ToolError, ResourceError, or PromptError will be included in responses.
                If False (default), all error details will be included in responses, but prefixed
                with appropriate context.
                """
            ),
        ),
    ] = False

    server_dependencies: list[str] = Field(
        default_factory=list,
        description="List of dependencies to install in the server environment",
    )

    # StreamableHTTP settings
    json_response: bool = False
    stateless_http: bool = (
        False  # If True, uses true stateless mode (new transport per request)
    )

    # Auth settings
    server_auth: Annotated[
        str | None,
        Field(
            description=inspect.cleandoc(
                """
                Configure the authentication provider for the server. Auth
                providers are registered with a specific key, and providing that
                key here will cause the server to automatically configure the
                provider from the environment.

                If None, no automatic configuration will take place.

                This setting is *always* overriden by any auth provider passed to the
                FastMCP constructor.

                Note that most auth providers require additional configuration
                that must be provided via env vars.
                """
            ),
        ),
    ] = None

    include_tags: Annotated[
        set[str] | None,
        Field(
            default=None,
            description=inspect.cleandoc(
                """
                If provided, only components that match these tags will be
                exposed to clients. A component is considered to match if ANY of
                its tags match ANY of the tags in the set.
                """
            ),
        ),
    ] = None
    exclude_tags: Annotated[
        set[str] | None,
        Field(
            default=None,
            description=inspect.cleandoc(
                """
                If provided, components that match these tags will be excluded
                from the server. A component is considered to match if ANY of
                its tags match ANY of the tags in the set.
                """
            ),
        ),
    ] = None

    include_fastmcp_meta: Annotated[
        bool,
        Field(
            default=True,
            description=inspect.cleandoc(
                """
                Whether to include FastMCP meta in the server's MCP responses.
                If True, a `_fastmcp` key will be added to the `meta` field of
                all MCP component responses. This key will contain a dict of
                various FastMCP-specific metadata, such as tags. 
                """
            ),
        ),
    ] = True

    mounted_components_raise_on_load_error: Annotated[
        bool,
        Field(
            default=False,
            description=inspect.cleandoc(
                """
                If True, errors encountered when loading mounted components (tools, resources, prompts)
                will be raised instead of logged as warnings. This is useful for debugging
                but will interrupt normal operation.
                """
            ),
        ),
    ] = False


def __getattr__(name: str):
    """
    Used to deprecate the module-level Image class; can be removed once it is no longer imported to root.
    """
    if name == "settings":
        import fastmcp

        settings = fastmcp.settings
        # Deprecated in 2.10.2
        if settings.deprecation_warnings:
            warnings.warn(
                "`from fastmcp.settings import settings` is deprecated. use `fastmcp.settings` instead.",
                DeprecationWarning,
                stacklevel=2,
            )
        return settings

    raise AttributeError(f"module '{__name__}' has no attribute '{name}'")