ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-24 06:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add FileUpload provider (#3669)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-03-28 19:45:43 -04:00 • committed by GitHub
commit 5338629474
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 792 additions and 209 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/apps/generative.mdx
View file

@ -123,6 +123,7 @@ The streaming renderer loads Pyodide from CDN in the browser. The CSP is configu
## Next Steps
- **[GenerativeUI Provider Reference](/apps/providers/generative)** — Configuration options and quick setup
- **[Prefab UI](/apps/prefab)** — The component library and state system the LLM writes code against
- **[Prefab Component Reference](https://prefab.prefect.io/docs/components)** — Full component library documentation
- **[Development](/apps/development)** — Preview generative UI tools locally with `fastmcp dev apps`

BIN
docs/apps/images/app-file-upload.png Normal file
View file

Binary file not shown.
Side by side

After

Width:  |  Height:  |  Size: 555 KiB

4
docs/apps/overview.mdx
View file

@ -148,7 +148,7 @@ mcp = FastMCP("Prefab Studio")
mcp.add_provider(GenerativeUI())
```
See [Generative UI](/apps/generative) for the full guide.
See [Generative UI](/apps/generative) for the full guide, or the [provider reference](/apps/providers/generative) for configuration options.
## Which Approach?
@ -160,6 +160,8 @@ When you want the LLM to design the UI at runtime, use **[Generative UI](/apps/g
When you need your own HTML/JS (maps, 3D, video), use **[Custom HTML](/apps/low-level)**.
FastMCP also includes ready-made **[app providers](/apps/providers/generative)** that add common capabilities with a single `add_provider()` call.
## Custom HTML Apps
All the approaches above use [Prefab UI](https://prefab.prefect.io) to build UIs in pure Python. If you need full control — your own HTML, CSS, JavaScript, a specific framework — you can use the [MCP Apps extension directly](/apps/low-level). You write the HTML yourself and communicate with the host via the [`@modelcontextprotocol/ext-apps`](https://github.com/modelcontextprotocol/ext-apps) SDK.

125
docs/apps/providers/file-upload.mdx Normal file
View file

@ -0,0 +1,125 @@
---
title: File Upload
sidebarTitle: File Upload
description: Drag-and-drop file upload for any MCP server
icon: upload
tag: NEW
---
import { VersionBadge } from '/snippets/version-badge.mdx'
<VersionBadge version="3.2.0" />
`FileUpload` adds drag-and-drop file upload to any server. Users upload files through an interactive UI, bypassing the LLM context window entirely. The LLM can then list and read uploaded files through model-visible tools.
<Frame>
<img src="/apps/images/app-file-upload.png" alt="The FileUpload provider shown in Goose, with a drag-and-drop zone for uploading files" />
</Frame>
```python
from fastmcp import FastMCP
from fastmcp.apps.file_upload import FileUpload
mcp = FastMCP("My Server")
mcp.add_provider(FileUpload())
```
This registers four tools:
| Tool | Visibility | Purpose |
|------|-----------|---------|
| `file_manager` | Model | Opens the drag-and-drop upload UI |
| `store_files` | App only | Called by the UI when the user clicks Upload |
| `list_files` | Model | Returns metadata for all uploaded files |
| `read_file` | Model | Returns a file's contents by name |
The LLM sees `file_manager`, `list_files`, and `read_file`. It calls `file_manager` to show the upload interface, then uses `list_files` and `read_file` to work with whatever the user uploaded. `store_files` is app-only — the UI calls it directly through the `___` routing mechanism and the LLM never needs to know about it.
## Configuration
```python
FileUpload(
name="Files", # App name (used in tool routing)
max_file_size=10 * 1024 * 1024, # 10 MB default, enforced server-side
title="File Upload", # Heading shown in the UI
description="Drop files to...", # Description text below the heading
drop_label="Drop files here", # Label inside the drop zone
)
```
The `max_file_size` limit is enforced both in the UI (the DropZone rejects oversized files) and on the server (the `store_files` tool validates before calling `on_store`).
## Storage Scoping
By default, files are stored in memory and scoped by MCP session ID. Each session gets its own isolated file store — files uploaded in one conversation aren't visible in another.
This works with **stdio**, **SSE**, and **stateful HTTP** transports, where sessions persist across requests. In **stateless HTTP** mode, each request creates a new session, so the default scoping won't work.
For stateless deployments, override `_get_scope_key` to return a stable identifier. For example, to scope files by authenticated user:
```python
from fastmcp.apps.file_upload import FileUpload
class UserScopedUpload(FileUpload):
def _get_scope_key(self, ctx):
return ctx.access_token["sub"]
```
For process-wide shared storage (all users see all files):
```python
class SharedUpload(FileUpload):
def _get_scope_key(self, ctx):
return "__shared__"
```
## Custom Storage
The default implementation stores files in memory for the lifetime of the server process. For persistent storage, subclass `FileUpload` and override three methods. Each receives the current `Context`, giving you access to session IDs, auth tokens, and request metadata for partitioning and authorization.
```python
import base64
from fastmcp.apps.file_upload import FileUpload
class S3Upload(FileUpload):
def on_store(self, files, ctx):
user_id = ctx.access_token["sub"]
for f in files:
s3.put_object(
Bucket="uploads",
Key=f"{user_id}/{f['name']}",
Body=base64.b64decode(f["data"]),
)
return self.on_list(ctx)
def on_list(self, ctx):
user_id = ctx.access_token["sub"]
objects = s3.list_objects(Bucket="uploads", Prefix=f"{user_id}/")
return [
{
"name": obj["Key"].split("/", 1)[1],
"type": "application/octet-stream",
"size": obj["Size"],
"size_display": f"{obj['Size']} B",
"uploaded_at": obj["LastModified"].isoformat(),
}
for obj in objects.get("Contents", [])
]
def on_read(self, name, ctx):
user_id = ctx.access_token["sub"]
obj = s3.get_object(Bucket="uploads", Key=f"{user_id}/{name}")
content = obj["Body"].read()
return {
"name": name,
"size": obj["ContentLength"],
"type": obj["ContentType"],
"uploaded_at": obj["LastModified"].isoformat(),
"content": content.decode("utf-8"),
}
```
Each file dict passed to `on_store` contains `name`, `size`, `type`, and `data` (base64-encoded content). The return value from `on_store` and `on_list` should be a list of summary dicts with `name`, `type`, `size`, `size_display`, and `uploaded_at` fields — these populate the file list in the UI.
`on_read` returns a dict with file metadata and either `content` (decoded text) or `content_base64` (a base64 preview for binary files).

49
docs/apps/providers/generative.mdx Normal file
View file

@ -0,0 +1,49 @@
---
title: Generative UI
sidebarTitle: Generative UI
description: Let the LLM generate custom UIs at runtime
icon: wand-magic-sparkles
tag: NEW
---
import { VersionBadge } from '/snippets/version-badge.mdx'
<VersionBadge version="3.2.0" />
`GenerativeUI` lets the LLM write Prefab Python code at runtime and render it as a streaming interactive UI. Instead of calling pre-built tools with fixed interfaces, the model creates tailored visualizations for whatever data it's working with.
```python
from fastmcp import FastMCP
from fastmcp.apps.generative import GenerativeUI
mcp = FastMCP("My Server")
mcp.add_provider(GenerativeUI())
```
This registers:
| Component | Type | Purpose |
|-----------|------|---------|
| `generate_prefab_ui` | Tool | Accepts Python code, executes in Pyodide sandbox, renders result |
| `search_prefab_components` | Tool | Lets the LLM discover available Prefab components |
| Generative renderer | Resource | `ui://` resource with browser-side Pyodide for streaming |
The LLM writes real Python — loops, f-strings, computation — using Prefab's component library (charts, tables, forms, cards, layout primitives). As the model generates tokens, the host streams partial code to the renderer via `ontoolinputpartial`, so the user watches the UI build up in real time.
## Configuration
```python
GenerativeUI(
tool_name="generate_prefab_ui", # Rename the generation tool
components_tool_name="search_prefab_components", # Rename the search tool
include_components_tool=True, # Set False to omit the search tool
)
```
## Requirements
Requires `fastmcp[apps]` (installs `prefab-ui`). The Pyodide sandbox for server-side validation requires Deno, which installs automatically on first use. The streaming renderer loads Pyodide from CDN in the browser — CSP is configured automatically by the provider.
## Learn More
See the full **[Generative UI guide](/apps/generative)** for details on how streaming works, what the LLM writes, how to pass data, and the component search tool.

4
docs/changelog.mdx
View file

@ -9,10 +9,10 @@ tag: NEW
**[v3.1.1: 'Tis But a Patch](https://github.com/PrefectHQ/fastmcp/releases/tag/v3.1.1)**
Pins `pydantic-monty<0.0.8` to fix a breaking change in Monty that affects code mode. Monty 0.0.8 removed the `external_functions` constructor parameter, causing `MontySandboxProvider` to fail. This patch caps the version so existing installs work correctly.
Pins `pydantic-monty` below 0.0.8 to fix a breaking change in Monty that affects code mode. Monty 0.0.8 removed the `external_functions` constructor parameter, causing `MontySandboxProvider` to fail. This patch caps the version so existing installs work correctly.
### Fixes 🐞
* Pin pydantic-monty<0.0.8 to fix code mode by [@jlowin](https://github.com/jlowin) in [#3497](https://github.com/PrefectHQ/fastmcp/pull/3497)
* Pin pydantic-monty below 0.0.8 to fix code mode by [@jlowin](https://github.com/jlowin) in [#3497](https://github.com/PrefectHQ/fastmcp/pull/3497)
**Full Changelog**: [v3.1.0...v3.1.1](https://github.com/PrefectHQ/fastmcp/compare/v3.1.0...v3.1.1)

59
docs/docs.json
View file

@ -159,19 +159,26 @@
},
{
"collapsed": true,
"group": "Authentication",
"icon": "key",
"group": "Auth",
"icon": "shield-check",
"pages": [
"servers/auth/authentication",
"servers/auth/token-verification",
"servers/auth/remote-oauth",
"servers/auth/oauth-proxy",
"servers/auth/oidc-proxy",
"servers/auth/full-oauth-server",
"servers/auth/multi-auth"
{
"collapsed": true,
"group": "Authentication",
"icon": "key",
"pages": [
"servers/auth/authentication",
"servers/auth/token-verification",
"servers/auth/remote-oauth",
"servers/auth/oauth-proxy",
"servers/auth/oidc-proxy",
"servers/auth/full-oauth-server",
"servers/auth/multi-auth"
]
},
"servers/authorization"
]
},
"servers/authorization",
{
"collapsed": true,
"group": "Deployment",
@ -189,16 +196,34 @@
"group": "Apps",
"pages": [
"apps/overview",
"apps/prefab",
"apps/interactive-apps",
"apps/generative",
"apps/development",
{
"collapsed": true,
"group": "Reference",
"icon": "book",
"group": "Building Apps",
"icon": "hammer",
"pages": [
"apps/patterns",
"apps/prefab",
"apps/interactive-apps",
"apps/generative",
"apps/patterns"
],
"tag": "NEW"
},
{
"collapsed": true,
"group": "Providers",
"icon": "layer-group",
"pages": [
"apps/providers/file-upload",
"apps/providers/generative"
],
"tag": "NEW"
},
{
"collapsed": true,
"group": "Advanced",
"icon": "gear",
"pages": [
"apps/development",
"apps/architecture",
"apps/low-level"
],

192
examples/apps/file_upload/file_upload_server.py
View file

@ -1,199 +1,13 @@
"""File upload — bypass the LLM context window to get files onto the server.
The most practical use of MCP Apps: letting users upload files directly to
the server without pushing bytes through the model's context. The LLM asks
the user to upload, the user drops files, clicks Upload, and the server
stores them. The LLM can then work with the files through backend tools.
Usage:
uv run python file_upload_server.py
"""
from __future__ import annotations
from fastmcp import FastMCP
from fastmcp.apps.file_upload import FileUpload
import base64
from datetime import datetime
from prefab_ui.actions import SetState, ShowToast
from prefab_ui.actions.mcp import CallTool
from prefab_ui.app import PrefabApp
from prefab_ui.components import (
H3,
Badge,
Button,
Card,
CardContent,
CardFooter,
CardHeader,
Column,
DropZone,
Muted,
Row,
Separator,
Small,
Text,
)
from prefab_ui.components.control_flow import Else, ForEach, If
from prefab_ui.rx import ERROR, RESULT, STATE, Rx
from fastmcp import FastMCP, FastMCPApp
# ---------------------------------------------------------------------------
# In-memory file store
# ---------------------------------------------------------------------------
_files: dict[str, dict] = {}
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
app = FastMCPApp("Files")
@app.tool()
def store_files(files: list[dict]) -> list[dict]:
"""Store uploaded files. Receives file objects with name, size, type, data (base64)."""
for f in files:
_files[f["name"]] = {
"name": f["name"],
"size": f["size"],
"type": f["type"],
"data": f["data"],
"uploaded_at": datetime.now().isoformat(timespec="seconds"),
}
return _file_summaries()
@app.tool(model=True)
def list_files() -> list[dict]:
"""List all uploaded files with metadata."""
return _file_summaries()
@app.tool(model=True)
def read_file(name: str) -> dict:
"""Read an uploaded file's contents by name."""
if name not in _files:
available = list(_files.keys())
raise ValueError(f"File {name!r} not found. Available: {available}")
entry = _files[name]
result = {
"name": entry["name"],
"size": entry["size"],
"type": entry["type"],
"uploaded_at": entry["uploaded_at"],
}
if entry["type"].startswith("text/") or entry["name"].endswith(
(".csv", ".json", ".txt", ".md", ".py", ".yaml", ".yml", ".toml")
):
try:
result["content"] = base64.b64decode(entry["data"]).decode("utf-8")
except (UnicodeDecodeError, Exception):
result["content_base64"] = entry["data"][:200] + "..."
else:
result["content_base64"] = entry["data"][:200] + "..."
return result
@app.ui()
def file_manager() -> PrefabApp:
"""Upload and manage files. Drop files here to send them to the server."""
with Card(css_class="max-w-2xl mx-auto") as view:
with CardHeader():
with Row(gap=2, align="center"):
H3("File Upload")
with If(STATE.stored.length()):
Badge(STATE.stored.length(), variant="secondary")
with CardContent():
with Column(gap=4):
Muted(
"Drop files to upload them to the server. "
"The model can then read and analyze them "
"without using the context window."
)
DropZone(
name="pending",
icon="inbox",
label="Drop files here",
description="Any file type, up to 10MB",
multiple=True,
max_size=10 * 1024 * 1024,
)
# Show pending files
with If(STATE.pending.length()):
with Column(gap=2):
with ForEach("pending") as (i, item):
with Row(gap=2, align="center"):
with Column(gap=0):
Small(item.name)
Muted(f"{item.type} · {item.size} bytes")
Button(
"Upload to Server",
on_click=CallTool(
"store_files",
arguments={"files": Rx("pending")},
on_success=[
SetState("stored", RESULT),
SetState("pending", []),
ShowToast("Files uploaded!", variant="success"),
],
on_error=ShowToast(ERROR, variant="error"),
),
)
# Show uploaded files
with If(STATE.stored.length()):
Separator()
Text("Uploaded", css_class="font-medium text-sm")
with ForEach("stored") as f:
with Row(gap=2, align="center", css_class="justify-between"):
with Column(gap=0):
Small(f.name)
Muted(f.uploaded_at)
with Row(gap=2):
Badge(f.type, variant="secondary")
Badge(f.size_display, variant="outline")
with CardFooter():
with Row(align="center", css_class="w-full"):
with If(STATE.stored.length()):
Muted(
f"{STATE.stored.length()}"
f" {STATE.stored.length().pluralize('file')} on server"
)
with Else():
Muted("No files uploaded yet")
return PrefabApp(view=view, state={"pending": [], "stored": _file_summaries()})
def _file_summaries() -> list[dict]:
summaries = []
for entry in _files.values():
size = entry["size"]
if size < 1024:
size_display = f"{size} B"
elif size < 1024 * 1024:
size_display = f"{size / 1024:.1f} KB"
else:
size_display = f"{size / (1024 * 1024):.1f} MB"
summaries.append(
{
"name": entry["name"],
"type": entry["type"],
"size_display": size_display,
"uploaded_at": entry["uploaded_at"],
}
)
return summaries
mcp = FastMCP("File Upload Server", providers=[app])
mcp = FastMCP("File Upload Server", providers=[FileUpload()])
if __name__ == "__main__":
mcp.run()

393
src/fastmcp/apps/file_upload.py Normal file
View file

@ -0,0 +1,393 @@
"""FileUpload — a Provider that adds drag-and-drop file upload to any server.
Lets users upload files directly to the server through an interactive UI,
bypassing the LLM context window entirely. The LLM can then read and work
with uploaded files through model-visible tools.
Requires ``fastmcp[apps]`` (prefab-ui).
Usage::
from fastmcp import FastMCP
from fastmcp.apps import FileUpload
mcp = FastMCP("My Server")
mcp.add_provider(FileUpload())
For custom persistence, override the storage methods::
class S3Upload(FileUpload):
def on_store(self, files, ctx):
# write to S3, return summaries
...
def on_list(self, ctx):
# list from S3
...
def on_read(self, name, ctx):
# read from S3
...
"""
from __future__ import annotations
try:
from prefab_ui.actions import SetState, ShowToast
from prefab_ui.actions.mcp import CallTool
from prefab_ui.app import PrefabApp
from prefab_ui.components import (
H3,
Badge,
Button,
Card,
CardContent,
CardFooter,
CardHeader,
Column,
DropZone,
Muted,
Row,
Separator,
Small,
Text,
)
from prefab_ui.components.control_flow import Else, ForEach, If
from prefab_ui.rx import ERROR, RESULT, STATE, Rx
except ImportError as _exc:
raise ImportError(
"FileUpload requires prefab-ui. Install with: pip install 'fastmcp[apps]'"
) from _exc
import base64
from datetime import datetime
from typing import Any
from fastmcp.apps.app import FastMCPApp
from fastmcp.server.context import Context
_TEXT_EXTENSIONS = frozenset(
(".csv", ".json", ".txt", ".md", ".py", ".yaml", ".yml", ".toml")
)
def _format_size(size: int) -> str:
if size < 1024:
return f"{size} B"
elif size < 1024 * 1024:
return f"{size / 1024:.1f} KB"
else:
return f"{size / (1024 * 1024):.1f} MB"
def _make_summary(entry: dict[str, Any]) -> dict[str, Any]:
return {
"name": entry["name"],
"type": entry["type"],
"size": entry["size"],
"size_display": _format_size(entry["size"]),
"uploaded_at": entry["uploaded_at"],
}
class FileUpload(FastMCPApp):
"""A Provider that adds file upload capabilities to a server.
Registers a drag-and-drop UI tool, a backend storage tool, and
model-visible tools for listing and reading uploaded files.
Files are scoped by MCP session and stored in memory by default.
Override ``on_store``, ``on_list``, and ``on_read`` for custom
persistence (filesystem, S3, database, etc.). Each method receives
the current ``Context``, giving access to session ID, auth tokens,
and request metadata for partitioning and authorization.
**Session scoping:** The default storage uses ``ctx.session_id`` to
isolate files by session. This works with stdio, SSE, and stateful
HTTP transports. In **stateless HTTP** mode, each request creates a
new session, so files won't persist across requests. For stateless
deployments, override the storage methods to partition by a stable
identifier from the auth context::
class UserScopedUpload(FileUpload):
def on_store(self, files, ctx):
user_id = ctx.access_token["sub"]
...
Example::
from fastmcp import FastMCP
from fastmcp.apps.file_upload import FileUpload
mcp = FastMCP("My Server")
mcp.add_provider(FileUpload())
"""
def __init__(
self,
name: str = "Files",
*,
max_file_size: int = 10 * 1024 * 1024,
title: str = "File Upload",
description: str = (
"Drop files to upload them to the server. "
"The model can then read and analyze them "
"without using the context window."
),
drop_label: str = "Drop files here",
) -> None:
super().__init__(name)
self._max_file_size = max_file_size
self._title = title
self._description = description
self._drop_label = drop_label
# Default in-memory store, keyed by session_id
self._store: dict[str, dict[str, dict[str, Any]]] = {}
self._register_tools()
def __repr__(self) -> str:
return f"FileUpload({self.name!r})"
# ------------------------------------------------------------------
# Storage interface — override these for custom persistence
# ------------------------------------------------------------------
def _get_scope_key(self, ctx: Context) -> str:
"""Return the key used to partition file storage.
Defaults to ``ctx.session_id``, which is stable for stdio, SSE,
and stateful HTTP. The default ``on_store``/``on_list``/``on_read``
implementations call this to partition the in-memory store.
Override to scope by user, tenant, or any other dimension::
def _get_scope_key(self, ctx):
return ctx.access_token["sub"]
"""
try:
return ctx.session_id
except RuntimeError:
return "__default__"
def on_store(
self,
files: list[dict[str, Any]],
ctx: Context,
) -> list[dict[str, Any]]:
"""Store uploaded files and return summaries.
Args:
files: List of file dicts, each with ``name``, ``size``,
``type``, and ``data`` (base64-encoded content).
ctx: The current request context. Use for session ID,
auth tokens, or any metadata needed for partitioning.
Override this method for custom persistence. The default
implementation stores files in memory, scoped by
``_get_scope_key(ctx)``.
Returns:
List of file summary dicts (``name``, ``type``, ``size``,
``size_display``, ``uploaded_at``).
"""
scope = self._get_scope_key(ctx)
session_files = self._store.setdefault(scope, {})
for f in files:
session_files[f["name"]] = {
"name": f["name"],
"size": f["size"],
"type": f["type"],
"data": f["data"],
"uploaded_at": datetime.now().isoformat(timespec="seconds"),
}
return [_make_summary(e) for e in session_files.values()]
def on_list(self, ctx: Context) -> list[dict[str, Any]]:
"""List all stored files.
Args:
ctx: The current request context.
Override this method for custom persistence. The default
implementation returns files from the current scope.
Returns:
List of file summary dicts.
"""
scope = self._get_scope_key(ctx)
session_files = self._store.get(scope, {})
return [_make_summary(e) for e in session_files.values()]
def on_read(self, name: str, ctx: Context) -> dict[str, Any]:
"""Read a file's contents by name.
Args:
name: The filename to read.
ctx: The current request context.
Override this method for custom persistence. The default
implementation reads from the current scope's in-memory store.
Text files are decoded from base64; binary files return a
truncated base64 preview.
Returns:
Dict with file metadata and ``content`` (text) or
``content_base64`` (binary preview).
Raises:
ValueError: If the file is not found.
"""
scope = self._get_scope_key(ctx)
session_files = self._store.get(scope, {})
if name not in session_files:
available = list(session_files.keys())
raise ValueError(f"File {name!r} not found. Available: {available}")
entry = session_files[name]
result: dict[str, Any] = {
"name": entry["name"],
"size": entry["size"],
"type": entry["type"],
"uploaded_at": entry["uploaded_at"],
}
is_text = entry["type"].startswith("text/") or any(
entry["name"].endswith(ext) for ext in _TEXT_EXTENSIONS
)
if is_text:
try:
result["content"] = base64.b64decode(entry["data"]).decode("utf-8")
except UnicodeDecodeError:
result["content_base64"] = entry["data"][:200] + "..."
else:
result["content_base64"] = entry["data"][:200] + "..."
return result
# ------------------------------------------------------------------
# Tool registration
# ------------------------------------------------------------------
def _register_tools(self) -> None:
provider = self
@self.tool()
def store_files(files: list[dict], ctx: Context) -> list[dict]:
"""Store uploaded files. Receives file objects with name, size, type, data (base64)."""
for f in files:
if f.get("size", 0) > provider._max_file_size:
raise ValueError(
f"File {f.get('name', '?')!r} exceeds max size "
f"({_format_size(f['size'])} > "
f"{_format_size(provider._max_file_size)})"
)
return provider.on_store(files, ctx)
@self.tool(model=True)
def list_files(ctx: Context) -> list[dict]:
"""List all uploaded files with metadata."""
return provider.on_list(ctx)
@self.tool(model=True)
def read_file(name: str, ctx: Context) -> dict:
"""Read an uploaded file's contents by name."""
return provider.on_read(name, ctx)
@self.ui()
def file_manager(ctx: Context) -> PrefabApp:
"""Upload and manage files. Drop files here to send them to the server."""
with Card(css_class="max-w-2xl mx-auto") as view:
with CardHeader(), Row(gap=2, align="center"):
H3(provider._title)
with If(STATE.stored.length()):
Badge(
STATE.stored.length(), # ty:ignore[invalid-argument-type]
variant="secondary",
)
with CardContent(), Column(gap=4):
Muted(provider._description)
DropZone(
name="pending",
icon="inbox",
label=provider._drop_label,
description=(
"Any file type, up to "
f"{_format_size(provider._max_file_size)}"
),
multiple=True,
max_size=provider._max_file_size,
)
with If(STATE.pending.length()), Column(gap=2):
with (
ForEach("pending"),
Row(gap=2, align="center"),
Column(gap=0),
):
Small(Rx("$item.name")) # ty:ignore[invalid-argument-type]
Muted(Rx("$item.type")) # ty:ignore[invalid-argument-type]
Button(
"Upload to Server",
on_click=CallTool(
"store_files",
arguments={
"files": Rx("pending"),
},
on_success=[
SetState("stored", RESULT),
SetState("pending", []),
ShowToast(
"Files uploaded!",
variant="success",
),
],
on_error=ShowToast(
ERROR, # ty:ignore[invalid-argument-type]
variant="error",
),
),
)
with If(STATE.stored.length()):
Separator()
Text(
"Uploaded",
css_class="font-medium text-sm",
)
with (
ForEach("stored") as f,
Row(
gap=2,
align="center",
css_class="justify-between",
),
):
with Column(gap=0):
Small(f.name) # ty:ignore[invalid-argument-type]
Muted(f.uploaded_at) # ty:ignore[invalid-argument-type]
with Row(gap=2):
Badge(f.type, variant="secondary") # ty:ignore[invalid-argument-type]
Badge(
f.size_display, # ty:ignore[invalid-argument-type]
variant="outline",
)
with CardFooter(), Row(align="center", css_class="w-full"):
with If(STATE.stored.length()):
Muted(
f"{STATE.stored.length()}"
f" {STATE.stored.length().pluralize('file')}"
" on server"
)
with Else():
Muted("No files uploaded yet")
return PrefabApp(
view=view,
state={
"pending": [],
"stored": provider.on_list(ctx),
},
)

0
tests/apps/__init__.py Normal file
View file

174
tests/apps/test_file_upload.py Normal file
View file

@ -0,0 +1,174 @@
"""Tests for the FileUpload provider."""
import base64
import pytest
from fastmcp import FastMCP
from fastmcp.apps.file_upload import FileUpload
def _make_file(
name: str = "test.txt",
content: str = "hello world",
mime_type: str = "text/plain",
) -> dict:
data = base64.b64encode(content.encode()).decode()
return {
"name": name,
"size": len(content),
"type": mime_type,
"data": data,
}
class TestFileUploadProvider:
async def test_basic_store_and_list(self):
server = FastMCP("test", providers=[FileUpload()])
files = [_make_file()]
result = await server.call_tool("Files___store_files", {"files": files})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "test.txt" in text
result = await server.call_tool("list_files", {})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "test.txt" in text
async def test_read_text_file(self):
server = FastMCP("test", providers=[FileUpload()])
files = [_make_file(content="DON'T PANIC")]
await server.call_tool("Files___store_files", {"files": files})
result = await server.call_tool("read_file", {"name": "test.txt"})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "DON'T PANIC" in text
async def test_read_binary_file(self):
server = FastMCP("test", providers=[FileUpload()])
data = base64.b64encode(b"\x00\x01\x02\xff").decode()
files = [{"name": "image.png", "size": 4, "type": "image/png", "data": data}]
await server.call_tool("Files___store_files", {"files": files})
result = await server.call_tool("read_file", {"name": "image.png"})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "content_base64" in text
async def test_read_missing_file_raises(self):
server = FastMCP("test", providers=[FileUpload()])
with pytest.raises(Exception, match="not found"):
await server.call_tool("read_file", {"name": "nope.txt"})
async def test_multiple_files(self):
server = FastMCP("test", providers=[FileUpload()])
files = [
_make_file("a.txt", "aaa"),
_make_file("b.txt", "bbb"),
]
await server.call_tool("Files___store_files", {"files": files})
result = await server.call_tool("list_files", {})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "a.txt" in text
assert "b.txt" in text
async def test_overwrite_file(self):
server = FastMCP("test", providers=[FileUpload()])
await server.call_tool(
"Files___store_files",
{"files": [_make_file(content="version 1")]},
)
await server.call_tool(
"Files___store_files",
{"files": [_make_file(content="version 2")]},
)
result = await server.call_tool("read_file", {"name": "test.txt"})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "version 2" in text
async def test_custom_name(self):
server = FastMCP("test", providers=[FileUpload(name="Uploads")])
tools = await server.list_tools()
tool_names = [t.name for t in tools]
assert "file_manager" in tool_names
# Routing uses the custom name
files = [_make_file()]
result = await server.call_tool("Uploads___store_files", {"files": files})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "test.txt" in text
async def test_ui_tool_visible_backend_hidden(self):
server = FastMCP("test", providers=[FileUpload()])
tools = await server.list_tools()
tool_names = [t.name for t in tools]
assert "file_manager" in tool_names
assert "list_files" in tool_names
assert "read_file" in tool_names
assert "store_files" not in tool_names
async def test_max_file_size_enforced_server_side(self):
server = FastMCP("test", providers=[FileUpload(max_file_size=100)])
big_file = _make_file(content="x" * 200)
with pytest.raises(Exception, match="exceeds max size"):
await server.call_tool("Files___store_files", {"files": [big_file]})
class TestFileUploadSubclass:
async def test_custom_storage(self):
"""Subclassing lets users provide their own persistence."""
stored: dict[str, dict] = {}
class MemoryUpload(FileUpload):
def on_store(self, files: list[dict], ctx) -> list[dict]:
for f in files:
stored[f["name"]] = f
return [
{
"name": f["name"],
"type": f["type"],
"size": f["size"],
"size_display": "?",
"uploaded_at": "now",
}
for f in files
]
def on_list(self, ctx) -> list[dict]:
return [
{
"name": f["name"],
"type": f["type"],
"size": f["size"],
"size_display": "?",
"uploaded_at": "now",
}
for f in stored.values()
]
def on_read(self, name: str, ctx) -> dict:
if name not in stored:
raise ValueError(f"Not found: {name}")
f = stored[name]
return {"name": f["name"], "content": "custom read"}
server = FastMCP("test", providers=[MemoryUpload()])
files = [_make_file()]
await server.call_tool("Files___store_files", {"files": files})
assert "test.txt" in stored
result = await server.call_tool("read_file", {"name": "test.txt"})
text = result.content[0].text # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
assert "custom read" in text