Jeremiah Lowin commited on
Commit
c6466c2
·
1 Parent(s): 4229596

Create common base class for components

Browse files
src/fastmcp/prompts/prompt.py CHANGED
@@ -5,13 +5,13 @@ from __future__ import annotations as _annotations
5
  import inspect
6
  from abc import ABC, abstractmethod
7
  from collections.abc import Awaitable, Callable, Sequence
8
- from typing import TYPE_CHECKING, Annotated, Any
9
 
10
  import pydantic_core
11
  from mcp.types import EmbeddedResource, ImageContent, PromptMessage, Role, TextContent
12
  from mcp.types import Prompt as MCPPrompt
13
  from mcp.types import PromptArgument as MCPPromptArgument
14
- from pydantic import BeforeValidator, Field, TypeAdapter, validate_call
15
 
16
  from fastmcp.exceptions import PromptError
17
  from fastmcp.server.dependencies import get_context
@@ -19,7 +19,7 @@ from fastmcp.utilities.json_schema import compress_schema
19
  from fastmcp.utilities.logging import get_logger
20
  from fastmcp.utilities.types import (
21
  FastMCPBaseModel,
22
- _convert_set_defaults,
23
  find_kwarg_by_type,
24
  get_cached_typeadapter,
25
  )
@@ -66,26 +66,13 @@ class PromptArgument(FastMCPBaseModel):
66
  )
67
 
68
 
69
- class Prompt(FastMCPBaseModel, ABC):
70
  """A prompt template that can be rendered with parameters."""
71
 
72
- name: str = Field(description="Name of the prompt")
73
- description: str | None = Field(
74
- default=None, description="Description of what the prompt does"
75
- )
76
- tags: Annotated[set[str], BeforeValidator(_convert_set_defaults)] = Field(
77
- default_factory=set, description="Tags for the prompt"
78
- )
79
  arguments: list[PromptArgument] | None = Field(
80
  default=None, description="Arguments that can be passed to the prompt"
81
  )
82
 
83
- def __eq__(self, other: object) -> bool:
84
- if type(self) is not type(other):
85
- return False
86
- assert isinstance(other, type(self))
87
- return self.model_dump() == other.model_dump()
88
-
89
  def to_mcp_prompt(self, **overrides: Any) -> MCPPrompt:
90
  """Convert the prompt to an MCP prompt."""
91
  arguments = [
 
5
  import inspect
6
  from abc import ABC, abstractmethod
7
  from collections.abc import Awaitable, Callable, Sequence
8
+ from typing import TYPE_CHECKING, Any
9
 
10
  import pydantic_core
11
  from mcp.types import EmbeddedResource, ImageContent, PromptMessage, Role, TextContent
12
  from mcp.types import Prompt as MCPPrompt
13
  from mcp.types import PromptArgument as MCPPromptArgument
14
+ from pydantic import Field, TypeAdapter, validate_call
15
 
16
  from fastmcp.exceptions import PromptError
17
  from fastmcp.server.dependencies import get_context
 
19
  from fastmcp.utilities.logging import get_logger
20
  from fastmcp.utilities.types import (
21
  FastMCPBaseModel,
22
+ FastMCPComponent,
23
  find_kwarg_by_type,
24
  get_cached_typeadapter,
25
  )
 
66
  )
67
 
68
 
69
+ class Prompt(FastMCPComponent, ABC):
70
  """A prompt template that can be rendered with parameters."""
71
 
 
 
 
 
 
 
 
72
  arguments: list[PromptArgument] | None = Field(
73
  default=None, description="Arguments that can be passed to the prompt"
74
  )
75
 
 
 
 
 
 
 
76
  def to_mcp_prompt(self, **overrides: Any) -> MCPPrompt:
77
  """Convert the prompt to an MCP prompt."""
78
  arguments = [
src/fastmcp/resources/resource.py CHANGED
@@ -22,7 +22,7 @@ from pydantic import (
22
  from fastmcp.server.dependencies import get_context
23
  from fastmcp.utilities.types import (
24
  FastMCPBaseModel,
25
- _convert_set_defaults,
26
  find_kwarg_by_type,
27
  )
28
 
@@ -42,7 +42,7 @@ class Resource(FastMCPBaseModel, abc.ABC):
42
  description: str | None = Field(
43
  default=None, description="Description of the resource"
44
  )
45
- tags: Annotated[set[str], BeforeValidator(_convert_set_defaults)] = Field(
46
  default_factory=set, description="Tags for the resource"
47
  )
48
  mime_type: str = Field(
 
22
  from fastmcp.server.dependencies import get_context
23
  from fastmcp.utilities.types import (
24
  FastMCPBaseModel,
25
+ _convert_set_default_none,
26
  find_kwarg_by_type,
27
  )
28
 
 
42
  description: str | None = Field(
43
  default=None, description="Description of the resource"
44
  )
45
+ tags: Annotated[set[str], BeforeValidator(_convert_set_default_none)] = Field(
46
  default_factory=set, description="Tags for the resource"
47
  )
48
  mime_type: str = Field(
src/fastmcp/resources/template.py CHANGED
@@ -5,12 +5,11 @@ from __future__ import annotations
5
  import inspect
6
  import re
7
  from collections.abc import Callable
8
- from typing import Annotated, Any
9
  from urllib.parse import unquote
10
 
11
  from mcp.types import ResourceTemplate as MCPResourceTemplate
12
  from pydantic import (
13
- BeforeValidator,
14
  Field,
15
  field_validator,
16
  validate_call,
@@ -20,8 +19,7 @@ from fastmcp.resources.types import Resource
20
  from fastmcp.server.dependencies import get_context
21
  from fastmcp.utilities.json_schema import compress_schema
22
  from fastmcp.utilities.types import (
23
- FastMCPBaseModel,
24
- _convert_set_defaults,
25
  find_kwarg_by_type,
26
  get_cached_typeadapter,
27
  )
@@ -51,17 +49,12 @@ def match_uri_template(uri: str, uri_template: str) -> dict[str, str] | None:
51
  return None
52
 
53
 
54
- class ResourceTemplate(FastMCPBaseModel):
55
  """A template for dynamically creating resources."""
56
 
57
  uri_template: str = Field(
58
  description="URI template with parameters (e.g. weather://{city}/current)"
59
  )
60
- name: str = Field(description="Name of the resource")
61
- description: str | None = Field(description="Description of what the resource does")
62
- tags: Annotated[set[str], BeforeValidator(_convert_set_defaults)] = Field(
63
- default_factory=set, description="Tags for the resource"
64
- )
65
  mime_type: str = Field(
66
  default="text/plain", description="MIME type of the resource content"
67
  )
 
5
  import inspect
6
  import re
7
  from collections.abc import Callable
8
+ from typing import Any
9
  from urllib.parse import unquote
10
 
11
  from mcp.types import ResourceTemplate as MCPResourceTemplate
12
  from pydantic import (
 
13
  Field,
14
  field_validator,
15
  validate_call,
 
19
  from fastmcp.server.dependencies import get_context
20
  from fastmcp.utilities.json_schema import compress_schema
21
  from fastmcp.utilities.types import (
22
+ FastMCPComponent,
 
23
  find_kwarg_by_type,
24
  get_cached_typeadapter,
25
  )
 
49
  return None
50
 
51
 
52
+ class ResourceTemplate(FastMCPComponent):
53
  """A template for dynamically creating resources."""
54
 
55
  uri_template: str = Field(
56
  description="URI template with parameters (e.g. weather://{city}/current)"
57
  )
 
 
 
 
 
58
  mime_type: str = Field(
59
  default="text/plain", description="MIME type of the resource content"
60
  )
src/fastmcp/tools/tool.py CHANGED
@@ -5,21 +5,20 @@ import json
5
  from abc import ABC, abstractmethod
6
  from collections.abc import Callable
7
  from dataclasses import dataclass
8
- from typing import TYPE_CHECKING, Annotated, Any
9
 
10
  import pydantic_core
11
  from mcp.types import EmbeddedResource, ImageContent, TextContent, ToolAnnotations
12
  from mcp.types import Tool as MCPTool
13
- from pydantic import BeforeValidator, Field
14
 
15
  import fastmcp
16
  from fastmcp.server.dependencies import get_context
17
  from fastmcp.utilities.json_schema import compress_schema
18
  from fastmcp.utilities.logging import get_logger
19
  from fastmcp.utilities.types import (
20
- FastMCPBaseModel,
21
  Image,
22
- _convert_set_defaults,
23
  find_kwarg_by_type,
24
  get_cached_typeadapter,
25
  )
@@ -34,17 +33,10 @@ def default_serializer(data: Any) -> str:
34
  return pydantic_core.to_json(data, fallback=str, indent=2).decode()
35
 
36
 
37
- class Tool(FastMCPBaseModel, ABC):
38
  """Internal tool registration info."""
39
 
40
- name: str = Field(description="Name of the tool")
41
- description: str | None = Field(
42
- default=None, description="Description of what the tool does"
43
- )
44
  parameters: dict[str, Any] = Field(description="JSON schema for tool parameters")
45
- tags: Annotated[set[str], BeforeValidator(_convert_set_defaults)] = Field(
46
- default_factory=set, description="Tags for the tool"
47
- )
48
  annotations: ToolAnnotations | None = Field(
49
  default=None, description="Additional annotations about the tool"
50
  )
 
5
  from abc import ABC, abstractmethod
6
  from collections.abc import Callable
7
  from dataclasses import dataclass
8
+ from typing import TYPE_CHECKING, Any
9
 
10
  import pydantic_core
11
  from mcp.types import EmbeddedResource, ImageContent, TextContent, ToolAnnotations
12
  from mcp.types import Tool as MCPTool
13
+ from pydantic import Field
14
 
15
  import fastmcp
16
  from fastmcp.server.dependencies import get_context
17
  from fastmcp.utilities.json_schema import compress_schema
18
  from fastmcp.utilities.logging import get_logger
19
  from fastmcp.utilities.types import (
20
+ FastMCPComponent,
21
  Image,
 
22
  find_kwarg_by_type,
23
  get_cached_typeadapter,
24
  )
 
33
  return pydantic_core.to_json(data, fallback=str, indent=2).decode()
34
 
35
 
36
+ class Tool(FastMCPComponent, ABC):
37
  """Internal tool registration info."""
38
 
 
 
 
 
39
  parameters: dict[str, Any] = Field(description="JSON schema for tool parameters")
 
 
 
40
  annotations: ToolAnnotations | None = Field(
41
  default=None, description="Additional annotations about the tool"
42
  )
src/fastmcp/utilities/types.py CHANGED
@@ -2,24 +2,55 @@
2
 
3
  import base64
4
  import inspect
5
- from collections.abc import Callable
6
  from functools import lru_cache
7
  from pathlib import Path
8
  from types import UnionType
9
  from typing import Annotated, TypeVar, Union, get_args, get_origin
10
 
11
  from mcp.types import ImageContent
12
- from pydantic import BaseModel, ConfigDict, TypeAdapter
13
 
14
  T = TypeVar("T")
15
 
16
 
 
 
 
 
 
 
 
 
 
17
  class FastMCPBaseModel(BaseModel):
18
  """Base model for FastMCP models."""
19
 
20
  model_config = ConfigDict(extra="forbid")
21
 
22
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
23
  @lru_cache(maxsize=5000)
24
  def get_cached_typeadapter(cls: T) -> TypeAdapter[T]:
25
  """
@@ -80,15 +111,6 @@ def find_kwarg_by_type(fn: Callable, kwarg_type: type) -> str | None:
80
  return None
81
 
82
 
83
- def _convert_set_defaults(maybe_set: set[T] | list[T] | None) -> set[T]:
84
- """Convert a set or list to a set, defaulting to an empty set if None."""
85
- if maybe_set is None:
86
- return set()
87
- if isinstance(maybe_set, set):
88
- return maybe_set
89
- return set(maybe_set)
90
-
91
-
92
  class Image:
93
  """Helper class for returning images from tools."""
94
 
 
2
 
3
  import base64
4
  import inspect
5
+ from collections.abc import Callable, Sequence
6
  from functools import lru_cache
7
  from pathlib import Path
8
  from types import UnionType
9
  from typing import Annotated, TypeVar, Union, get_args, get_origin
10
 
11
  from mcp.types import ImageContent
12
+ from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, TypeAdapter
13
 
14
  T = TypeVar("T")
15
 
16
 
17
+ def _convert_set_default_none(maybe_set: set[T] | Sequence[T] | None) -> set[T]:
18
+ """Convert a sequence to a set, defaulting to an empty set if None."""
19
+ if maybe_set is None:
20
+ return set()
21
+ if isinstance(maybe_set, set):
22
+ return maybe_set
23
+ return set(maybe_set)
24
+
25
+
26
  class FastMCPBaseModel(BaseModel):
27
  """Base model for FastMCP models."""
28
 
29
  model_config = ConfigDict(extra="forbid")
30
 
31
 
32
+ class FastMCPComponent(FastMCPBaseModel):
33
+ """Base class for FastMCP tools, prompts, resources, and resource templates."""
34
+
35
+ name: str = Field(
36
+ description="The name of the component.",
37
+ )
38
+ description: str | None = Field(
39
+ default=None,
40
+ description="The description of the component.",
41
+ )
42
+ tags: Annotated[set[str], BeforeValidator(_convert_set_default_none)] = Field(
43
+ default_factory=set,
44
+ description="Tags for the component.",
45
+ )
46
+
47
+ def __eq__(self, other: object) -> bool:
48
+ if type(self) is not type(other):
49
+ return False
50
+ assert isinstance(other, type(self))
51
+ return self.model_dump() == other.model_dump()
52
+
53
+
54
  @lru_cache(maxsize=5000)
55
  def get_cached_typeadapter(cls: T) -> TypeAdapter[T]:
56
  """
 
111
  return None
112
 
113
 
 
 
 
 
 
 
 
 
 
114
  class Image:
115
  """Helper class for returning images from tools."""
116
 
tests/server/test_proxy.py CHANGED
@@ -140,7 +140,7 @@ class TestTools:
140
  assert proxy_result[0].text == "3" # type: ignore[attr-defined]
141
 
142
  async def test_error_tool_raises_error(self, proxy_server):
143
- with pytest.raises(ToolError, match=""):
144
  async with Client(proxy_server) as client:
145
  await client.call_tool("error_tool", {})
146
 
 
140
  assert proxy_result[0].text == "3" # type: ignore[attr-defined]
141
 
142
  async def test_error_tool_raises_error(self, proxy_server):
143
+ with pytest.raises(ToolError, match="This is a test error"):
144
  async with Client(proxy_server) as client:
145
  await client.call_tool("error_tool", {})
146