Skip to content

types

Concrete resource implementations.

TextResource

Bases: Resource

A resource that reads from a string.

Source code in src/mcp/server/mcpserver/resources/types.py 
22
23
24
25
26
27
28
29
class TextResource(Resource):
    """A resource that reads from a string."""

    text: str = Field(description="Text content of the resource")

    async def read(self) -> str:
        """Read the text content."""
        return self.text  # pragma: no cover

read async 

read() -> str

Read the text content.

Source code in src/mcp/server/mcpserver/resources/types.py 
27
28
29
async def read(self) -> str:
    """Read the text content."""
    return self.text  # pragma: no cover

BinaryResource

Bases: Resource

A resource that reads from bytes.

Source code in src/mcp/server/mcpserver/resources/types.py 
32
33
34
35
36
37
38
39
class BinaryResource(Resource):
    """A resource that reads from bytes."""

    data: bytes = Field(description="Binary content of the resource")

    async def read(self) -> bytes:
        """Read the binary content."""
        return self.data  # pragma: no cover

read async 

read() -> bytes

Read the binary content.

Source code in src/mcp/server/mcpserver/resources/types.py 
37
38
39
async def read(self) -> bytes:
    """Read the binary content."""
    return self.data  # pragma: no cover

FunctionResource

Bases: Resource

A resource that defers data loading by wrapping a function.

The function is only called when the resource is read, allowing for lazy loading of potentially expensive data. This is particularly useful when listing resources, as the function won't be called until the resource is actually accessed.

The function can return: - str for text content (default) - bytes for binary content - other types will be converted to JSON

Source code in src/mcp/server/mcpserver/resources/types.py 
 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
class FunctionResource(Resource):
    """A resource that defers data loading by wrapping a function.

    The function is only called when the resource is read, allowing for lazy loading
    of potentially expensive data. This is particularly useful when listing resources,
    as the function won't be called until the resource is actually accessed.

    The function can return:
    - str for text content (default)
    - bytes for binary content
    - other types will be converted to JSON
    """

    fn: Callable[[], Any] = Field(exclude=True)

    async def read(self) -> str | bytes:
        """Read the resource by calling the wrapped function."""
        try:
            fn = self.fn
            if is_async_callable(fn):
                result = await fn()
            else:
                result = await anyio.to_thread.run_sync(self.fn)

            if isinstance(result, Resource):  # pragma: no cover
                return await result.read()
            elif isinstance(result, bytes):
                return result
            elif isinstance(result, str):
                return result
            else:
                return pydantic_core.to_json(result, fallback=str, indent=2).decode()
        except Exception as e:
            raise ValueError(f"Error reading resource {self.uri}: {e}")

    @classmethod
    def from_function(
        cls,
        fn: Callable[..., Any],
        uri: str,
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        mime_type: str | None = None,
        icons: list[Icon] | None = None,
        annotations: Annotations | None = None,
        meta: dict[str, Any] | None = None,
    ) -> FunctionResource:
        """Create a FunctionResource from a function."""
        func_name = name or fn.__name__
        if func_name == "<lambda>":  # pragma: no cover
            raise ValueError("You must provide a name for lambda functions")

        # ensure the arguments are properly cast
        fn = validate_call(fn)

        return cls(
            uri=uri,
            name=func_name,
            title=title,
            description=description or fn.__doc__ or "",
            mime_type=mime_type or "text/plain",
            fn=fn,
            icons=icons,
            annotations=annotations,
            meta=meta,
        )

read async 

read() -> str | bytes

Read the resource by calling the wrapped function.

Source code in src/mcp/server/mcpserver/resources/types.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def read(self) -> str | bytes:
    """Read the resource by calling the wrapped function."""
    try:
        fn = self.fn
        if is_async_callable(fn):
            result = await fn()
        else:
            result = await anyio.to_thread.run_sync(self.fn)

        if isinstance(result, Resource):  # pragma: no cover
            return await result.read()
        elif isinstance(result, bytes):
            return result
        elif isinstance(result, str):
            return result
        else:
            return pydantic_core.to_json(result, fallback=str, indent=2).decode()
    except Exception as e:
        raise ValueError(f"Error reading resource {self.uri}: {e}")

from_function classmethod 

from_function(
    fn: Callable[..., Any],
    uri: str,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
) -> FunctionResource

Create a FunctionResource from a function.

Source code in src/mcp/server/mcpserver/resources/types.py 
 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
@classmethod
def from_function(
    cls,
    fn: Callable[..., Any],
    uri: str,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
) -> FunctionResource:
    """Create a FunctionResource from a function."""
    func_name = name or fn.__name__
    if func_name == "<lambda>":  # pragma: no cover
        raise ValueError("You must provide a name for lambda functions")

    # ensure the arguments are properly cast
    fn = validate_call(fn)

    return cls(
        uri=uri,
        name=func_name,
        title=title,
        description=description or fn.__doc__ or "",
        mime_type=mime_type or "text/plain",
        fn=fn,
        icons=icons,
        annotations=annotations,
        meta=meta,
    )

FileResource

Bases: Resource

A resource that reads from a file.

Set is_binary=True to read the file as binary data instead of text.

Source code in src/mcp/server/mcpserver/resources/types.py 
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
class FileResource(Resource):
    """A resource that reads from a file.

    Set is_binary=True to read the file as binary data instead of text.
    """

    path: Path = Field(description="Path to the file")
    is_binary: bool = Field(
        default=False,
        description="Whether to read the file as binary data",
    )
    mime_type: str = Field(
        default="text/plain",
        description="MIME type of the resource content",
    )

    @pydantic.field_validator("path")
    @classmethod
    def validate_absolute_path(cls, path: Path) -> Path:
        """Ensure path is absolute."""
        if not path.is_absolute():
            raise ValueError("Path must be absolute")
        return path

    @pydantic.field_validator("is_binary")
    @classmethod
    def set_binary_from_mime_type(cls, is_binary: bool, info: ValidationInfo) -> bool:
        """Set is_binary based on mime_type if not explicitly set."""
        if is_binary:
            return True
        mime_type = info.data.get("mime_type", "text/plain")
        return not mime_type.startswith("text/")

    async def read(self) -> str | bytes:
        """Read the file content."""
        try:
            if self.is_binary:
                return await anyio.to_thread.run_sync(self.path.read_bytes)
            return await anyio.to_thread.run_sync(self.path.read_text)
        except Exception as e:
            raise ValueError(f"Error reading file {self.path}: {e}")

validate_absolute_path classmethod 

validate_absolute_path(path: Path) -> Path

Ensure path is absolute.

Source code in src/mcp/server/mcpserver/resources/types.py 
127
128
129
130
131
132
133
@pydantic.field_validator("path")
@classmethod
def validate_absolute_path(cls, path: Path) -> Path:
    """Ensure path is absolute."""
    if not path.is_absolute():
        raise ValueError("Path must be absolute")
    return path

set_binary_from_mime_type classmethod 

set_binary_from_mime_type(
    is_binary: bool, info: ValidationInfo
) -> bool

Set is_binary based on mime_type if not explicitly set.

Source code in src/mcp/server/mcpserver/resources/types.py 
135
136
137
138
139
140
141
142
@pydantic.field_validator("is_binary")
@classmethod
def set_binary_from_mime_type(cls, is_binary: bool, info: ValidationInfo) -> bool:
    """Set is_binary based on mime_type if not explicitly set."""
    if is_binary:
        return True
    mime_type = info.data.get("mime_type", "text/plain")
    return not mime_type.startswith("text/")

read async 

read() -> str | bytes

Read the file content.

Source code in src/mcp/server/mcpserver/resources/types.py 
144
145
146
147
148
149
150
151
async def read(self) -> str | bytes:
    """Read the file content."""
    try:
        if self.is_binary:
            return await anyio.to_thread.run_sync(self.path.read_bytes)
        return await anyio.to_thread.run_sync(self.path.read_text)
    except Exception as e:
        raise ValueError(f"Error reading file {self.path}: {e}")

HttpResource

Bases: Resource

A resource that reads from an HTTP endpoint.

Source code in src/mcp/server/mcpserver/resources/types.py 
154
155
156
157
158
159
160
161
162
163
164
165
class HttpResource(Resource):
    """A resource that reads from an HTTP endpoint."""

    url: str = Field(description="URL to fetch content from")
    mime_type: str = Field(default="application/json", description="MIME type of the resource content")

    async def read(self) -> str | bytes:
        """Read the HTTP content."""
        async with httpx.AsyncClient() as client:  # pragma: no cover
            response = await client.get(self.url)
            response.raise_for_status()
            return response.text

read async 

read() -> str | bytes

Read the HTTP content.

Source code in src/mcp/server/mcpserver/resources/types.py 
160
161
162
163
164
165
async def read(self) -> str | bytes:
    """Read the HTTP content."""
    async with httpx.AsyncClient() as client:  # pragma: no cover
        response = await client.get(self.url)
        response.raise_for_status()
        return response.text

DirectoryResource

Bases: Resource

A resource that lists files in a directory.

Source code in src/mcp/server/mcpserver/resources/types.py 
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
class DirectoryResource(Resource):
    """A resource that lists files in a directory."""

    path: Path = Field(description="Path to the directory")
    recursive: bool = Field(default=False, description="Whether to list files recursively")
    pattern: str | None = Field(default=None, description="Optional glob pattern to filter files")
    mime_type: str = Field(default="application/json", description="MIME type of the resource content")

    @pydantic.field_validator("path")
    @classmethod
    def validate_absolute_path(cls, path: Path) -> Path:  # pragma: no cover
        """Ensure path is absolute."""
        if not path.is_absolute():
            raise ValueError("Path must be absolute")
        return path

    def list_files(self) -> list[Path]:  # pragma: no cover
        """List files in the directory."""
        if not self.path.exists():
            raise FileNotFoundError(f"Directory not found: {self.path}")
        if not self.path.is_dir():
            raise NotADirectoryError(f"Not a directory: {self.path}")

        try:
            if self.pattern:
                return list(self.path.glob(self.pattern)) if not self.recursive else list(self.path.rglob(self.pattern))
            return list(self.path.glob("*")) if not self.recursive else list(self.path.rglob("*"))
        except Exception as e:
            raise ValueError(f"Error listing directory {self.path}: {e}")

    async def read(self) -> str:  # Always returns JSON string  # pragma: no cover
        """Read the directory listing."""
        try:
            files = await anyio.to_thread.run_sync(self.list_files)
            file_list = [str(f.relative_to(self.path)) for f in files if f.is_file()]
            return json.dumps({"files": file_list}, indent=2)
        except Exception as e:
            raise ValueError(f"Error reading directory {self.path}: {e}")

validate_absolute_path classmethod 

validate_absolute_path(path: Path) -> Path

Ensure path is absolute.

Source code in src/mcp/server/mcpserver/resources/types.py 
176
177
178
179
180
181
182
@pydantic.field_validator("path")
@classmethod
def validate_absolute_path(cls, path: Path) -> Path:  # pragma: no cover
    """Ensure path is absolute."""
    if not path.is_absolute():
        raise ValueError("Path must be absolute")
    return path

list_files 

list_files() -> list[Path]

List files in the directory.

Source code in src/mcp/server/mcpserver/resources/types.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
def list_files(self) -> list[Path]:  # pragma: no cover
    """List files in the directory."""
    if not self.path.exists():
        raise FileNotFoundError(f"Directory not found: {self.path}")
    if not self.path.is_dir():
        raise NotADirectoryError(f"Not a directory: {self.path}")

    try:
        if self.pattern:
            return list(self.path.glob(self.pattern)) if not self.recursive else list(self.path.rglob(self.pattern))
        return list(self.path.glob("*")) if not self.recursive else list(self.path.rglob("*"))
    except Exception as e:
        raise ValueError(f"Error listing directory {self.path}: {e}")

read async 

read() -> str

Read the directory listing.

Source code in src/mcp/server/mcpserver/resources/types.py 
198
199
200
201
202
203
204
205
async def read(self) -> str:  # Always returns JSON string  # pragma: no cover
    """Read the directory listing."""
    try:
        files = await anyio.to_thread.run_sync(self.list_files)
        file_list = [str(f.relative_to(self.path)) for f in files if f.is_file()]
        return json.dumps({"files": file_list}, indent=2)
    except Exception as e:
        raise ValueError(f"Error reading directory {self.path}: {e}")