Spaces:
Running
Running
File size: 5,789 Bytes
957b451 c7184ac 957b451 c7184ac 0fdb55c c7184ac 957b451 c7184ac 957b451 c7184ac 226fdf7 c7184ac 957b451 7be2f51 2a88273 0fdb55c 957b451 2e693f4 957b451 2e693f4 957b451 7be2f51 8d16d26 c9434ed 8d16d26 c9434ed 0fdb55c c7184ac abc4a6d c7184ac 596281e 226fdf7 2a88273 a9a1a98 | 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 | from __future__ import annotations
from collections.abc import Sequence
from typing import Annotated, Any, TypedDict, TypeVar
from pydantic import BeforeValidator, Field, PrivateAttr
from typing_extensions import Self
import fastmcp
from fastmcp.utilities.types import FastMCPBaseModel
T = TypeVar("T")
class FastMCPMeta(TypedDict, total=False):
tags: list[str]
def _convert_set_default_none(maybe_set: set[T] | Sequence[T] | None) -> set[T]:
"""Convert a sequence to a set, defaulting to an empty set if None."""
if maybe_set is None:
return set()
if isinstance(maybe_set, set):
return maybe_set
return set(maybe_set)
class FastMCPComponent(FastMCPBaseModel):
"""Base class for FastMCP tools, prompts, resources, and resource templates."""
name: str = Field(
description="The name of the component.",
)
title: str | None = Field(
default=None,
description="The title of the component for display purposes.",
)
description: str | None = Field(
default=None,
description="The description of the component.",
)
tags: Annotated[set[str], BeforeValidator(_convert_set_default_none)] = Field(
default_factory=set,
description="Tags for the component.",
)
meta: dict[str, Any] | None = Field(
default=None, description="Meta information about the component"
)
enabled: bool = Field(
default=True,
description="Whether the component is enabled.",
)
_key: str | None = PrivateAttr()
def __init__(self, *, key: str | None = None, **kwargs: Any) -> None:
super().__init__(**kwargs)
self._key = key
@property
def key(self) -> str:
"""
The key of the component. This is used for internal bookkeeping
and may reflect e.g. prefixes or other identifiers. You should not depend on
keys having a certain value, as the same tool loaded from different
hierarchies of servers may have different keys.
"""
return self._key or self.name
def get_meta(
self, include_fastmcp_meta: bool | None = None
) -> dict[str, Any] | None:
"""
Get the meta information about the component.
If include_fastmcp_meta is True, a `_fastmcp` key will be added to the
meta, containing a `tags` field with the tags of the component.
"""
if include_fastmcp_meta is None:
include_fastmcp_meta = fastmcp.settings.include_fastmcp_meta
meta = self.meta or {}
if include_fastmcp_meta:
fastmcp_meta = FastMCPMeta(tags=sorted(self.tags))
# overwrite any existing _fastmcp meta with keys from the new one
if upstream_meta := meta.get("_fastmcp"):
fastmcp_meta = upstream_meta | fastmcp_meta
meta["_fastmcp"] = fastmcp_meta
return meta or None
def model_copy(
self,
*,
update: dict[str, Any] | None = None,
deep: bool = False,
key: str | None = None,
) -> Self:
"""
Create a copy of the component.
Args:
update: A dictionary of fields to update.
deep: Whether to deep copy the component.
key: The key to use for the copy.
"""
# `model_copy` has an `update` parameter but it doesn't work for certain private attributes
# https://github.com/pydantic/pydantic/issues/12116
# So we manually set the private attribute here instead, such as _key
copy = super().model_copy(update=update, deep=deep)
if key is not None:
copy._key = key
return copy
def __eq__(self, other: object) -> bool:
if type(self) is not type(other):
return False
if not isinstance(other, type(self)):
return False
return self.model_dump() == other.model_dump()
def __repr__(self) -> str:
return f"{self.__class__.__name__}(name={self.name!r}, title={self.title!r}, description={self.description!r}, tags={self.tags}, enabled={self.enabled})"
def enable(self) -> None:
"""Enable the component."""
self.enabled = True
def disable(self) -> None:
"""Disable the component."""
self.enabled = False
def copy(self) -> Self:
"""Create a copy of the component."""
return self.model_copy()
class MirroredComponent(FastMCPComponent):
"""Base class for components that are mirrored from a remote server.
Mirrored components cannot be enabled or disabled directly. Call copy() first
to create a local version you can modify.
"""
_mirrored: bool = PrivateAttr(default=False)
def __init__(self, *, _mirrored: bool = False, **kwargs: Any) -> None:
super().__init__(**kwargs)
self._mirrored = _mirrored
def enable(self) -> None:
"""Enable the component."""
if self._mirrored:
raise RuntimeError(
f"Cannot enable mirrored component '{self.name}'. "
f"Create a local copy first with {self.name}.copy() and add it to your server."
)
super().enable()
def disable(self) -> None:
"""Disable the component."""
if self._mirrored:
raise RuntimeError(
f"Cannot disable mirrored component '{self.name}'. "
f"Create a local copy first with {self.name}.copy() and add it to your server."
)
super().disable()
def copy(self) -> Self:
"""Create a copy of the component that can be modified."""
# Create a copy and mark it as not mirrored
copied = self.model_copy()
copied._mirrored = False
return copied
|