diff --git a/docs/apps/generative.mdx b/docs/apps/generative.mdx
index 4046acd0f..cabdf1dfa 100644
--- a/docs/apps/generative.mdx
+++ b/docs/apps/generative.mdx
@@ -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`
diff --git a/docs/apps/images/app-file-upload.png b/docs/apps/images/app-file-upload.png
new file mode 100644
index 000000000..1178c09af
Binary files /dev/null and b/docs/apps/images/app-file-upload.png differ
diff --git a/docs/apps/overview.mdx b/docs/apps/overview.mdx
index c74482bd7..895c91301 100644
--- a/docs/apps/overview.mdx
+++ b/docs/apps/overview.mdx
@@ -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.
diff --git a/docs/apps/providers/file-upload.mdx b/docs/apps/providers/file-upload.mdx
new file mode 100644
index 000000000..a1dc0dc47
--- /dev/null
+++ b/docs/apps/providers/file-upload.mdx
@@ -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'
+
+
+
+`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.
+
+
+ 
+
+
+```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).
diff --git a/docs/apps/providers/generative.mdx b/docs/apps/providers/generative.mdx
new file mode 100644
index 000000000..9f5002221
--- /dev/null
+++ b/docs/apps/providers/generative.mdx
@@ -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'
+
+
+
+`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.
diff --git a/docs/changelog.mdx b/docs/changelog.mdx
index 3e21089af..d55783e9f 100644
--- a/docs/changelog.mdx
+++ b/docs/changelog.mdx
@@ -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)
diff --git a/docs/docs.json b/docs/docs.json
index 5e08f3069..323517825 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -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"
],
diff --git a/examples/apps/file_upload/file_upload_server.py b/examples/apps/file_upload/file_upload_server.py
index 73e994cc6..c567f7820 100644
--- a/examples/apps/file_upload/file_upload_server.py
+++ b/examples/apps/file_upload/file_upload_server.py
@@ -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()
diff --git a/src/fastmcp/apps/file_upload.py b/src/fastmcp/apps/file_upload.py
new file mode 100644
index 000000000..8cc91e039
--- /dev/null
+++ b/src/fastmcp/apps/file_upload.py
@@ -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),
+ },
+ )
diff --git a/tests/apps/__init__.py b/tests/apps/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/apps/test_file_upload.py b/tests/apps/test_file_upload.py
new file mode 100644
index 000000000..c8bd7470e
--- /dev/null
+++ b/tests/apps/test_file_upload.py
@@ -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