Abubakar Abid
58088ad1d7
* Self-host frontend assets so Gradio works offline. Bundle theme fonts, Bokeh, FFmpeg, and iframe-resizer locally instead of loading from external CDNs, and document what still requires network access. Co-authored-by: Cursor <cursoragent@cursor.com> * add changeset * Address Copilot review feedback on offline asset bundling. Align LocalFont default weights with bundled fonts, fix Bokeh CDN fallback and download URLs, use importlib.resources for font detection, and support subpath-mounted static assets. Co-authored-by: Cursor <cursoragent@cursor.com> * Bundle the installed Bokeh version for offline use. gradio's Plot component sends `bokeh.__version__` of the installed package to the frontend, which then loads /static/bokeh/{version}/. The download script only bundled a hardcoded list, so any other installed version 404'd locally and fell back to the CDN (breaking offline). Always include the installed version. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Fix Bokeh local asset URL resolving to undefined at init. The local/CDN script URLs were reactive (`$:`) declarations, but `load_bokeh()` consumes `main_src` synchronously during component init -- before Svelte runs reactive statements. This set the script src to `undefined`, 404'd on /undefined, and silently fell back to the CDN (breaking offline). Compute the URLs as plain consts since `bokeh_version` is a creation-time prop. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * Use certifi CA bundle when downloading offline assets. python.org Python on macOS ships without a usable system CA bundle, so the default SSL context fails to verify TLS when fetching fonts/bokeh assets during the frontend build. Prefer certifi's bundle (already a transitive Gradio dependency) when available. This still performs full certificate verification. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * add changeset * Add offline-support regression tests. Guards the offline contract so future changes fail loudly if they reintroduce a runtime CDN dependency in the default experience: - every built-in theme bundles its fonts locally (no external stylesheet URLs, font CSS points only at static/fonts/) - the app/SPA HTML shells load scripts/stylesheets from /static, not a CDN - the MCP landing page references fonts/scripts locally Intentional CDN *fallbacks* (GoogleFont for user fonts, Bokeh plugins) are not forbidden -- the contract is that the built-in path resolves offline. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * changes * changes --------- Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: gradio-pr-bot <gradio-pr-bot@users.noreply.github.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1633 lines
63 KiB
Python
1633 lines
63 KiB
Python
import base64
|
||
import contextlib
|
||
import copy
|
||
import html
|
||
import json
|
||
import os
|
||
import re
|
||
import tempfile
|
||
import warnings
|
||
from collections.abc import AsyncIterator, Sequence
|
||
from io import BytesIO
|
||
from pathlib import Path
|
||
from typing import TYPE_CHECKING, Any, Optional, cast
|
||
from urllib.parse import unquote, urlparse
|
||
|
||
import gradio_client.utils as client_utils
|
||
import httpx
|
||
from anyio.to_thread import run_sync
|
||
from gradio_client import Client, handle_file
|
||
from gradio_client.utils import Status, StatusUpdate
|
||
from PIL import Image
|
||
from pydantic import AnyUrl
|
||
from starlette.applications import Starlette
|
||
from starlette.requests import Request
|
||
from starlette.responses import HTMLResponse, JSONResponse, Response
|
||
from starlette.routing import Mount, Route
|
||
from starlette.types import Receive, Scope, Send
|
||
|
||
from gradio import processing_utils, route_utils, utils
|
||
from gradio.blocks import BlockFunction
|
||
from gradio.components import State
|
||
from gradio.route_utils import Header
|
||
from gradio.state_holder import SessionState
|
||
|
||
if TYPE_CHECKING:
|
||
from mcp import types # noqa: F401
|
||
from mcp.server import Server # noqa: F401
|
||
from mcp.server.lowlevel.helper_types import ReadResourceContents # noqa: F401
|
||
|
||
from gradio.blocks import BlockContext, Blocks
|
||
from gradio.components import Component
|
||
|
||
|
||
DEFAULT_TEMP_DIR = os.environ.get("GRADIO_TEMP_DIR") or str(
|
||
Path(tempfile.gettempdir()) / "gradio"
|
||
)
|
||
|
||
|
||
# Landing page served on a plain browser GET to the MCP endpoint. The
|
||
# `__SERVER_URL__` / `__DOCS_URL__` (JSON-encoded, for the inline <script>) and
|
||
# `__SERVER_URL_TEXT__` (HTML-escaped, for display) placeholders are filled in
|
||
# by `GradioMCPServer._landing_page_html`.
|
||
_MCP_LANDING_PAGE_TEMPLATE = """<!DOCTYPE html>
|
||
<html lang="en">
|
||
<head>
|
||
<meta charset="utf-8" />
|
||
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
||
<title>Gradio MCP Server</title>
|
||
<link href="__STATIC_PREFIX__/fonts/SourceSans3/SourceSans3-Regular.woff2" rel="preload" as="font" type="font/woff2" crossorigin />
|
||
<link href="__STATIC_PREFIX__/fonts/SourceSans3/SourceSans3-SemiBold.woff2" rel="preload" as="font" type="font/woff2" crossorigin />
|
||
<link href="__STATIC_PREFIX__/fonts/IBMPlexMono/IBMPlexMono-Regular.woff2" rel="preload" as="font" type="font/woff2" crossorigin />
|
||
<style>
|
||
@font-face {
|
||
font-family: 'Source Sans 3';
|
||
src: url('__STATIC_PREFIX__/fonts/SourceSans3/SourceSans3-Regular.woff2') format('woff2');
|
||
font-weight: 400;
|
||
font-style: normal;
|
||
}
|
||
@font-face {
|
||
font-family: 'Source Sans 3';
|
||
src: url('__STATIC_PREFIX__/fonts/SourceSans3/SourceSans3-SemiBold.woff2') format('woff2');
|
||
font-weight: 600;
|
||
font-style: normal;
|
||
}
|
||
@font-face {
|
||
font-family: 'IBM Plex Mono';
|
||
src: url('__STATIC_PREFIX__/fonts/IBMPlexMono/IBMPlexMono-Regular.woff2') format('woff2');
|
||
font-weight: 400;
|
||
font-style: normal;
|
||
}
|
||
@font-face {
|
||
font-family: 'IBM Plex Mono';
|
||
src: url('__STATIC_PREFIX__/fonts/IBMPlexMono/IBMPlexMono-Medium.woff2') format('woff2');
|
||
font-weight: 500;
|
||
font-style: normal;
|
||
}
|
||
:root {
|
||
--primary: #ff7c00;
|
||
--primary-soft: #fff2e5;
|
||
--text: #1f2937;
|
||
--text-muted: #6b7280;
|
||
--border: #e5e7eb;
|
||
--bg: #f6f7f8;
|
||
--card: #ffffff;
|
||
--radius: 12px;
|
||
--mono: "IBM Plex Mono", ui-monospace, SFMono-Regular, Menlo, Consolas, monospace;
|
||
--sans: "Source Sans 3", -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif;
|
||
}
|
||
* { box-sizing: border-box; }
|
||
body {
|
||
font-family: var(--sans);
|
||
line-height: 1.6;
|
||
color: var(--text);
|
||
background: var(--bg);
|
||
margin: 0;
|
||
padding: 3rem 1.25rem 4rem;
|
||
}
|
||
.wrap { max-width: 46rem; margin: 0 auto; }
|
||
header { margin-bottom: 1.75rem; }
|
||
h1 { font-size: 1.7rem; font-weight: 700; margin: 0 0 0.4rem; letter-spacing: -0.01em; }
|
||
header p { margin: 0; color: var(--text-muted); }
|
||
.card {
|
||
background: var(--card);
|
||
border: 1px solid var(--border);
|
||
border-radius: var(--radius);
|
||
padding: 1.4rem 1.5rem;
|
||
margin-bottom: 1.25rem;
|
||
box-shadow: 0 1px 2px rgba(0, 0, 0, 0.04);
|
||
}
|
||
.card h2 {
|
||
font-size: 0.78rem;
|
||
font-weight: 600;
|
||
text-transform: uppercase;
|
||
letter-spacing: 0.06em;
|
||
color: var(--primary);
|
||
margin: 0 0 0.85rem;
|
||
}
|
||
.url-box {
|
||
display: flex;
|
||
align-items: center;
|
||
gap: 0.5rem;
|
||
border: 1px solid var(--border);
|
||
border-radius: 8px;
|
||
background: #fafafa;
|
||
padding: 0.7rem 0.85rem;
|
||
}
|
||
.url-box code {
|
||
flex: 1;
|
||
font-family: var(--mono);
|
||
font-size: 0.95rem;
|
||
color: var(--text);
|
||
word-break: break-all;
|
||
background: none;
|
||
}
|
||
.copy-btn {
|
||
flex-shrink: 0;
|
||
display: inline-flex;
|
||
align-items: center;
|
||
justify-content: center;
|
||
gap: 0.35rem;
|
||
border: 1px solid var(--border);
|
||
background: #fff;
|
||
color: var(--text-muted);
|
||
border-radius: 7px;
|
||
padding: 0.35rem 0.55rem;
|
||
font-family: var(--sans);
|
||
font-size: 0.8rem;
|
||
cursor: pointer;
|
||
transition: all 0.12s ease;
|
||
}
|
||
.copy-btn:hover { border-color: var(--primary); color: var(--primary); }
|
||
.copy-btn.copied { border-color: var(--primary); color: var(--primary); background: var(--primary-soft); }
|
||
.tabs {
|
||
display: flex;
|
||
flex-wrap: wrap;
|
||
gap: 0.25rem 1.25rem;
|
||
border-bottom: 1px solid var(--border);
|
||
margin-bottom: 1.1rem;
|
||
}
|
||
.tab {
|
||
appearance: none;
|
||
border: none;
|
||
background: none;
|
||
font-family: var(--sans);
|
||
font-size: 0.95rem;
|
||
color: var(--text-muted);
|
||
padding: 0.5rem 0;
|
||
cursor: pointer;
|
||
border-bottom: 2px solid transparent;
|
||
margin-bottom: -1px;
|
||
transition: color 0.12s ease;
|
||
}
|
||
.tab:hover { color: var(--text); }
|
||
.tab.active { color: var(--primary); border-bottom-color: var(--primary); font-weight: 600; }
|
||
.instruction { margin: 0 0 0.7rem; font-size: 0.95rem; }
|
||
.instruction code { font-family: var(--mono); font-size: 0.85em; background: var(--primary-soft); color: #b45309; padding: 0.1rem 0.35rem; border-radius: 5px; }
|
||
.code-block { position: relative; }
|
||
.code-block pre {
|
||
margin: 0;
|
||
background: #0b1221;
|
||
color: #e2e8f0;
|
||
border-radius: 10px;
|
||
padding: 1rem 1.1rem;
|
||
overflow-x: auto;
|
||
font-family: var(--mono);
|
||
font-size: 0.88rem;
|
||
line-height: 1.55;
|
||
}
|
||
.code-block .copy-btn {
|
||
position: absolute;
|
||
top: 0.6rem;
|
||
right: 0.6rem;
|
||
background: rgba(255, 255, 255, 0.08);
|
||
border-color: rgba(255, 255, 255, 0.15);
|
||
color: #cbd5e1;
|
||
}
|
||
.code-block .copy-btn:hover { background: rgba(255, 124, 0, 0.18); border-color: var(--primary); color: #fff; }
|
||
.code-block .copy-btn.copied { background: rgba(255, 124, 0, 0.25); border-color: var(--primary); color: #fff; }
|
||
footer { color: var(--text-muted); font-size: 0.9rem; margin-top: 0.5rem; }
|
||
footer a { color: var(--primary); text-decoration: none; }
|
||
footer a:hover { text-decoration: underline; }
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="wrap">
|
||
<header>
|
||
<h1>\U0001f6e0️ Gradio MCP Server</h1>
|
||
<p>This is the <strong>Model Context Protocol (MCP)</strong> endpoint for a Gradio app. It is meant to be connected to by an MCP client (such as an AI assistant), not opened directly in a browser.</p>
|
||
</header>
|
||
|
||
<section class="card">
|
||
<h2>MCP Server URL</h2>
|
||
<div class="url-box">
|
||
<code id="server-url">__SERVER_URL_TEXT__</code>
|
||
<button class="copy-btn" data-copy-target="server-url" type="button">Copy</button>
|
||
</div>
|
||
</section>
|
||
|
||
<section class="card">
|
||
<h2>Connect your AI assistant</h2>
|
||
<div class="tabs" id="tabs"></div>
|
||
<p class="instruction" id="instruction"></p>
|
||
<div class="code-block">
|
||
<button class="copy-btn" data-copy-target="snippet" type="button">Copy</button>
|
||
<pre><code id="snippet"></code></pre>
|
||
</div>
|
||
</section>
|
||
|
||
<footer>
|
||
Need help? Read the <a id="docs-link" href="#">Gradio MCP guide</a>.
|
||
</footer>
|
||
</div>
|
||
|
||
<script>
|
||
const SERVER_URL = __SERVER_URL__;
|
||
const DOCS_URL = __DOCS_URL__;
|
||
document.getElementById("docs-link").href = DOCS_URL;
|
||
|
||
const CLIENTS = [
|
||
{
|
||
id: "cursor",
|
||
name: "Cursor",
|
||
instruction: 'Add this to your <code>~/.cursor/mcp.json</code>:',
|
||
config: { mcpServers: { gradio: { url: SERVER_URL } } },
|
||
},
|
||
{
|
||
id: "claude-desktop",
|
||
name: "Claude Desktop",
|
||
instruction: 'Add this to your <code>claude_desktop_config.json</code> (requires Node.js):',
|
||
config: { mcpServers: { gradio: { command: "npx", args: ["mcp-remote", SERVER_URL] } } },
|
||
},
|
||
{
|
||
id: "claude-code",
|
||
name: "Claude Code",
|
||
instruction: "Run this command in your terminal:",
|
||
command: "claude mcp add --transport http gradio " + SERVER_URL,
|
||
},
|
||
{
|
||
id: "vscode",
|
||
name: "VS Code",
|
||
instruction: 'Add this to <code>.vscode/mcp.json</code> in your workspace:',
|
||
config: { servers: { gradio: { type: "http", url: SERVER_URL } } },
|
||
},
|
||
{
|
||
id: "cline",
|
||
name: "Cline",
|
||
instruction: 'Add this to your <code>cline_mcp_settings.json</code>:',
|
||
config: { mcpServers: { gradio: { type: "streamableHttp", url: SERVER_URL } } },
|
||
},
|
||
{
|
||
id: "windsurf",
|
||
name: "Windsurf",
|
||
instruction: 'Add this to your <code>~/.codeium/windsurf/mcp_config.json</code>:',
|
||
config: { mcpServers: { gradio: { serverUrl: SERVER_URL } } },
|
||
},
|
||
];
|
||
|
||
const tabsEl = document.getElementById("tabs");
|
||
const instructionEl = document.getElementById("instruction");
|
||
const snippetEl = document.getElementById("snippet");
|
||
let active = CLIENTS[0].id;
|
||
|
||
function snippetFor(client) {
|
||
return client.command ? client.command : JSON.stringify(client.config, null, 2);
|
||
}
|
||
|
||
function render() {
|
||
tabsEl.innerHTML = "";
|
||
CLIENTS.forEach(function (client) {
|
||
const btn = document.createElement("button");
|
||
btn.type = "button";
|
||
btn.className = "tab" + (client.id === active ? " active" : "");
|
||
btn.textContent = client.name;
|
||
btn.addEventListener("click", function () {
|
||
active = client.id;
|
||
render();
|
||
});
|
||
tabsEl.appendChild(btn);
|
||
});
|
||
const client = CLIENTS.find(function (c) { return c.id === active; });
|
||
instructionEl.innerHTML = client.instruction;
|
||
snippetEl.textContent = snippetFor(client);
|
||
}
|
||
|
||
function copyText(text, btn) {
|
||
const done = function () {
|
||
const original = btn.textContent;
|
||
btn.textContent = "Copied!";
|
||
btn.classList.add("copied");
|
||
setTimeout(function () {
|
||
btn.textContent = original;
|
||
btn.classList.remove("copied");
|
||
}, 1500);
|
||
};
|
||
if (navigator.clipboard && navigator.clipboard.writeText) {
|
||
navigator.clipboard.writeText(text).then(done, function () {});
|
||
} else {
|
||
const ta = document.createElement("textarea");
|
||
ta.value = text;
|
||
document.body.appendChild(ta);
|
||
ta.select();
|
||
try { document.execCommand("copy"); done(); } catch (e) {}
|
||
document.body.removeChild(ta);
|
||
}
|
||
}
|
||
|
||
document.addEventListener("click", function (event) {
|
||
const btn = event.target.closest("[data-copy-target]");
|
||
if (!btn) return;
|
||
const target = document.getElementById(btn.getAttribute("data-copy-target"));
|
||
if (target) copyText(target.textContent, btn);
|
||
});
|
||
|
||
render();
|
||
</script>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
|
||
class GradioMCPServer:
|
||
"""
|
||
A class for creating an MCP server around a Gradio app. This class
|
||
requires `mcp` to be installed.
|
||
|
||
Args:
|
||
blocks: The Blocks app to create the MCP server for.
|
||
"""
|
||
|
||
# Imports are here to avoid needing to install `mcp` when not using this class.
|
||
# This way, we are able to export `gr.tool`, `gr.resource`, etc. to `__init__.py`
|
||
# without the user needing to have `mcp` installed.
|
||
try:
|
||
from mcp import types
|
||
from mcp.server import Server
|
||
from mcp.server.lowlevel.helper_types import ReadResourceContents
|
||
from mcp.server.sse import SseServerTransport
|
||
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
|
||
except ImportError:
|
||
pass
|
||
|
||
def __init__(self, blocks: "Blocks"):
|
||
try:
|
||
import mcp # noqa: F401
|
||
except ImportError as e:
|
||
raise ImportError(
|
||
"The `mcp` package is required to use the Gradio MCP integration. Please install it with the `mcp` extra: `pip install gradio[mcp]`."
|
||
) from e
|
||
|
||
self.blocks = blocks
|
||
self.api_info = self.blocks.get_api_info()
|
||
self.mcp_server = self.create_mcp_server()
|
||
self.root_path = ""
|
||
space_id = utils.get_space()
|
||
self.tool_prefix = space_id.split("/")[-1] + "_" if space_id else ""
|
||
self.tool_to_endpoint = self.get_tool_to_endpoint()
|
||
self.warn_about_state_inputs()
|
||
self._local_url: str | None = None
|
||
self._client_instance: Client | None = None
|
||
|
||
manager = self.StreamableHTTPSessionManager( # type: ignore
|
||
app=self.mcp_server, json_response=False, stateless=True
|
||
)
|
||
|
||
async def handle_streamable_http(
|
||
scope: Scope, receive: Receive, send: Send
|
||
) -> None:
|
||
path = scope.get("path", "")
|
||
if not path.endswith(
|
||
(
|
||
"/gradio_api/mcp",
|
||
"/gradio_api/mcp/",
|
||
"/gradio_api/mcp/http",
|
||
"/gradio_api/mcp/http/",
|
||
)
|
||
):
|
||
response = Response(
|
||
content=f"Path '{path}' not found. The MCP HTTP transport is available at /gradio_api/mcp.",
|
||
status_code=404,
|
||
)
|
||
await response(scope, receive, send)
|
||
return
|
||
|
||
# MCP clients connect using the streamable HTTP transport, which
|
||
# requires an `Accept: text/event-stream` header. A plain browser
|
||
# GET (e.g. a user clicking the MCP link printed in the terminal)
|
||
# would otherwise receive a raw JSON-RPC "Not Acceptable" error, so
|
||
# serve a human-friendly landing page instead.
|
||
if scope.get("method") == "GET":
|
||
accept = b""
|
||
for key, value in scope.get("headers", []):
|
||
if key.lower() == b"accept":
|
||
accept = value
|
||
break
|
||
if b"text/event-stream" not in accept:
|
||
server_url = self._server_url_from_scope(scope)
|
||
response = HTMLResponse(content=self._landing_page_html(server_url))
|
||
await response(scope, receive, send)
|
||
return
|
||
|
||
await manager.handle_request(scope, receive, send)
|
||
|
||
@contextlib.asynccontextmanager
|
||
async def lifespan(app: Starlette) -> AsyncIterator[None]: # noqa: ARG001
|
||
"""Context manager for managing session manager lifecycle."""
|
||
async with manager.run():
|
||
try:
|
||
yield
|
||
finally:
|
||
pass
|
||
|
||
self.lifespan = lifespan
|
||
self.manager = manager
|
||
self.handle_streamable_http = handle_streamable_http
|
||
|
||
@property
|
||
def local_url(self) -> str | None:
|
||
return self._local_url
|
||
|
||
@staticmethod
|
||
def _server_url_from_scope(scope: Scope) -> str:
|
||
"""Reconstruct the public URL of the MCP endpoint from an ASGI scope.
|
||
|
||
Honours ``X-Forwarded-Proto``/``X-Forwarded-Host`` so that the URL
|
||
shown on the landing page matches what the user typed into the browser
|
||
even when Gradio is behind a proxy (e.g. a Hugging Face Space).
|
||
"""
|
||
headers = {key.lower(): value for key, value in scope.get("headers", []) or []}
|
||
|
||
scheme = scope.get("scheme", "http")
|
||
if (forwarded_proto := headers.get(b"x-forwarded-proto")) is not None:
|
||
scheme = forwarded_proto.decode("latin-1").split(",")[0].strip()
|
||
|
||
host = b""
|
||
if (forwarded_host := headers.get(b"x-forwarded-host")) is not None:
|
||
host = forwarded_host.split(b",")[0].strip()
|
||
elif (host_header := headers.get(b"host")) is not None:
|
||
host = host_header
|
||
host_str = host.decode("latin-1") or "localhost"
|
||
|
||
# Normalise to the canonical streamable HTTP path with a trailing slash,
|
||
# matching the URL Gradio prints to the terminal at launch.
|
||
path = scope.get("path", "/gradio_api/mcp")
|
||
path = "/" + path.strip("/")
|
||
if path.endswith(("/http", "/http/")):
|
||
path = path[: path.rfind("/http")]
|
||
path = path.rstrip("/") + "/"
|
||
|
||
return f"{scheme}://{host_str}{path}"
|
||
|
||
@staticmethod
|
||
def _landing_page_html(server_url: str) -> str:
|
||
"""HTML landing page shown when a browser navigates to the MCP endpoint.
|
||
|
||
MCP clients connect using the streamable HTTP transport (which requires
|
||
an ``Accept: text/event-stream`` header). A plain browser ``GET`` would
|
||
otherwise receive a raw JSON-RPC "Not Acceptable" error, so we serve
|
||
this informational page instead. It surfaces the server URL and ready
|
||
to paste configuration snippets for popular MCP clients.
|
||
"""
|
||
docs_url = "https://www.gradio.app/guides/building-mcp-server-with-gradio"
|
||
server_url_js = json.dumps(server_url)
|
||
docs_url_js = json.dumps(docs_url)
|
||
parsed = urlparse(server_url)
|
||
path = parsed.path
|
||
if "/gradio_api/mcp" in path:
|
||
root_path = path.split("/gradio_api/mcp")[0]
|
||
else:
|
||
root_path = path.rstrip("/")
|
||
static_prefix = f"{root_path}/static"
|
||
return (
|
||
_MCP_LANDING_PAGE_TEMPLATE.replace("__SERVER_URL__", server_url_js)
|
||
.replace("__DOCS_URL__", docs_url_js)
|
||
.replace("__SERVER_URL_TEXT__", html.escape(server_url))
|
||
.replace("__STATIC_PREFIX__", static_prefix)
|
||
)
|
||
|
||
def get_route_path(self, request: Request) -> str: # type: ignore
|
||
"""
|
||
Gets the route path of the MCP server based on the incoming request.
|
||
Can be different depending on whether the request is coming from the MCP SSE transport or the HTTP transport.
|
||
"""
|
||
url = httpx.URL(str(request.url))
|
||
url = url.copy_with(query=None)
|
||
url = str(url).rstrip("/")
|
||
if url.endswith("/gradio_api/mcp/messages"):
|
||
return "/gradio_api/mcp/messages"
|
||
else:
|
||
return "/gradio_api/mcp"
|
||
|
||
def get_selected_tools_from_request(self) -> list[str] | None:
|
||
"""
|
||
Extract the selected tools from the request query parameters and return the full tool names (with the tool prefix).
|
||
Returns None if no tools parameter is specified (meaning all tools are available).
|
||
"""
|
||
context_request: Request | None = self.mcp_server.request_context.request
|
||
if context_request is None:
|
||
return None
|
||
query_params = dict(getattr(context_request, "query_params", {}))
|
||
if "tools" in query_params:
|
||
tools = query_params["tools"].split(",")
|
||
full_tool_names = [self.tool_prefix + tool for tool in tools]
|
||
return full_tool_names
|
||
return None
|
||
|
||
@staticmethod
|
||
def valid_and_unique_tool_name(
|
||
tool_name: str, existing_tool_names: set[str]
|
||
) -> str:
|
||
"""
|
||
Sanitizes a tool name to make it a valid MCP tool name (only
|
||
alphanumeric characters, underscores, <= 128 characters)
|
||
and is unique among the existing tool names.
|
||
"""
|
||
tool_name = re.sub(r"[^a-zA-Z0-9]", "_", tool_name)
|
||
tool_name = tool_name[:120] # Leave room for suffix if needed
|
||
tool_name_base = tool_name
|
||
suffix = 1
|
||
while tool_name in existing_tool_names:
|
||
tool_name = tool_name_base + f"_{suffix}"
|
||
suffix += 1
|
||
return tool_name
|
||
|
||
def get_tool_to_endpoint(self) -> dict[str, str]:
|
||
"""
|
||
Gets all of the tools that are exposed by the Gradio app and also
|
||
creates a mapping from the tool names to the endpoint names in the API docs.
|
||
"""
|
||
tool_to_endpoint = {}
|
||
for endpoint_name in self.api_info["named_endpoints"]:
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
if block_fn is None or block_fn.fn is None:
|
||
continue
|
||
fn_name = (
|
||
getattr(block_fn.fn, "__name__", None)
|
||
or (
|
||
hasattr(block_fn.fn, "__class__")
|
||
and getattr(block_fn.fn.__class__, "__name__", None)
|
||
)
|
||
or endpoint_name.lstrip("/")
|
||
)
|
||
tool_name = self.tool_prefix + fn_name
|
||
tool_name = self.valid_and_unique_tool_name(
|
||
tool_name, set(tool_to_endpoint.keys())
|
||
)
|
||
tool_to_endpoint[tool_name] = endpoint_name
|
||
return tool_to_endpoint
|
||
|
||
def warn_about_state_inputs(self) -> None:
|
||
"""
|
||
Warn about tools that have gr.State inputs.
|
||
"""
|
||
for _, endpoint_name in self.tool_to_endpoint.items():
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
if block_fn and any(isinstance(input, State) for input in block_fn.inputs):
|
||
warnings.warn(
|
||
"This MCP server includes a tool that has a gr.State input, which will not be "
|
||
"updated between tool calls. The original, default value of the State will be "
|
||
"used each time."
|
||
)
|
||
|
||
def _get_or_create_client(self) -> Client:
|
||
if self._client_instance is None:
|
||
context_request: Request | None = self.mcp_server.request_context.request
|
||
if context_request is None:
|
||
raise ValueError(
|
||
"Could not find the request object in the MCP server context. This is not expected to happen. Please raise an issue: https://github.com/gradio-app/gradio."
|
||
)
|
||
route_path = self.get_route_path(context_request)
|
||
root_url = route_utils.get_root_url(
|
||
request=context_request,
|
||
route_path=route_path,
|
||
root_path=self.root_path,
|
||
)
|
||
self._client_instance = Client(
|
||
self.local_url or root_url,
|
||
download_files=False,
|
||
verbose=False,
|
||
analytics_enabled=False,
|
||
ssl_verify=False,
|
||
_skip_components=False,
|
||
headers={"x-gradio-user": "mcp"},
|
||
)
|
||
return self._client_instance
|
||
|
||
def _prepare_tool_call_args(
|
||
self, name: str, arguments: dict[str, Any]
|
||
) -> tuple[str, list[Any], dict[str, str], "BlockFunction"]:
|
||
"""
|
||
Prepare and validate arguments for a tool call.
|
||
|
||
Returns:
|
||
A tuple of (endpoint_name, processed_args, request_headers, block_fn)
|
||
"""
|
||
selected_tools = self.get_selected_tools_from_request()
|
||
_, filedata_positions = self.get_input_schema(name)
|
||
processed_kwargs = self.convert_strings_to_filedata(
|
||
arguments, filedata_positions
|
||
)
|
||
endpoint_name = self.tool_to_endpoint.get(name)
|
||
if endpoint_name is None:
|
||
raise ValueError(f"Unknown tool for this Gradio app: {name}")
|
||
|
||
if selected_tools is not None and name not in selected_tools:
|
||
raise ValueError(f"Tool '{name}' is not in the selected tools list")
|
||
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
assert block_fn is not None # noqa: S101
|
||
|
||
if endpoint_name in self.api_info["named_endpoints"]:
|
||
parameters_info = self.api_info["named_endpoints"][endpoint_name][
|
||
"parameters"
|
||
]
|
||
processed_args = client_utils.construct_args(
|
||
parameters_info,
|
||
(),
|
||
processed_kwargs,
|
||
)
|
||
else:
|
||
processed_args = []
|
||
|
||
context_request: Request | None = self.mcp_server.request_context.request
|
||
if context_request is None:
|
||
raise ValueError(
|
||
"Could not find the request object in the MCP server context. This is not expected to happen. Please raise an issue: https://github.com/gradio-app/gradio."
|
||
)
|
||
request_headers = dict(context_request.headers.items())
|
||
request_headers.pop("content-length", None)
|
||
request_headers.pop("x-gradio-user", None)
|
||
|
||
return endpoint_name, processed_args, request_headers, block_fn
|
||
|
||
async def _execute_tool_without_progress(self, job: Any) -> list[Any]:
|
||
"""
|
||
Execute a tool call without progress tracking (fast path).
|
||
|
||
Calls job.result() to get the final output without processing
|
||
intermediate status updates.
|
||
|
||
Returns:
|
||
The output data as a list.
|
||
"""
|
||
result = await run_sync(job.result)
|
||
return [result]
|
||
|
||
@staticmethod
|
||
def _format_progress_message(update: StatusUpdate) -> str | None:
|
||
"""
|
||
Format a status update into a human-readable progress message.
|
||
|
||
Returns:
|
||
A formatted message string, or None if no message should be shown.
|
||
"""
|
||
if update.code in [Status.JOINING_QUEUE, Status.STARTING]:
|
||
return "Joined server queue."
|
||
elif update.code in [Status.IN_QUEUE]:
|
||
message = f"In queue. Position {update.rank} out of {update.queue_size}."
|
||
if update.eta is not None:
|
||
message += f" Estimated time remaining: {update.eta} seconds."
|
||
return message
|
||
elif update.code in [Status.PROGRESS]:
|
||
for progress_unit in update.progress_data or []:
|
||
title = (
|
||
"Progress"
|
||
if progress_unit.desc is None
|
||
else f"Progress {progress_unit.desc}"
|
||
)
|
||
if progress_unit.index is not None and progress_unit.length is not None:
|
||
return (
|
||
f"{title}: Step {progress_unit.index} of {progress_unit.length}"
|
||
)
|
||
elif progress_unit.index is not None and progress_unit.length is None:
|
||
return f"{title}: Step {progress_unit.index}"
|
||
elif update.code in [Status.PROCESSING, Status.ITERATING]:
|
||
return "Processing"
|
||
return None
|
||
|
||
async def _execute_tool_with_progress( # type: ignore
|
||
self, job: Any, progress_token: str | int
|
||
) -> dict[str, Any]:
|
||
"""
|
||
Execute a tool call with progress tracking (streaming path).
|
||
|
||
Iterates through job updates to send progress notifications to the client.
|
||
|
||
Returns:
|
||
The output data as a list.
|
||
"""
|
||
step = 0
|
||
async for update in job:
|
||
if update.type == "status":
|
||
update = cast(StatusUpdate, update)
|
||
message = self._format_progress_message(update)
|
||
|
||
await (
|
||
self.mcp_server.request_context.session.send_progress_notification(
|
||
progress_token=progress_token,
|
||
progress=step,
|
||
message=message, # type: ignore
|
||
related_request_id=str(
|
||
self.mcp_server.request_context.request_id
|
||
),
|
||
)
|
||
)
|
||
step += 1
|
||
elif update.type == "output" and update.final:
|
||
output = update.outputs
|
||
if not update.success:
|
||
error_title = output.get("title")
|
||
error_message = output.get("error")
|
||
if error_title and error_message:
|
||
msg = f"{error_title}: {error_message}"
|
||
elif error_message:
|
||
msg = error_message
|
||
elif error_title:
|
||
msg = error_title
|
||
else:
|
||
msg = "Error!"
|
||
raise RuntimeError(msg)
|
||
if job.exception():
|
||
raise job.exception()
|
||
return output["data"]
|
||
|
||
def create_mcp_server(self) -> "Server":
|
||
"""
|
||
Create an MCP server for the given Gradio Blocks app.
|
||
|
||
Parameters:
|
||
blocks: The Blocks app to create the MCP server for.
|
||
|
||
Returns:
|
||
The MCP server.
|
||
"""
|
||
server = self.Server(str(self.blocks.title or "Gradio App")) # type: ignore
|
||
|
||
@server.call_tool()
|
||
async def call_tool(
|
||
name: str, arguments: dict[str, Any]
|
||
) -> self.types.CallToolResult: # type: ignore
|
||
"""
|
||
Call a tool on the Gradio app.
|
||
|
||
Args:
|
||
name: The name of the tool to call.
|
||
arguments: The arguments to pass to the tool.
|
||
"""
|
||
endpoint_name, processed_args, request_headers, block_fn = (
|
||
self._prepare_tool_call_args(name, arguments)
|
||
)
|
||
processed_args = self.insert_empty_state(block_fn.inputs, processed_args)
|
||
|
||
if not block_fn.queue:
|
||
# Fast path for non-queued events: call blocks.process_api()
|
||
# directly instead of the HTTP loopback through gradio_client.
|
||
# This eliminates thread dispatches, TCP round-trips, and SSE
|
||
# overhead — reducing MCP tool-call latency significantly.
|
||
session_state = SessionState(self.blocks)
|
||
raw_output = await self.blocks.process_api(
|
||
block_fn=block_fn,
|
||
inputs=processed_args,
|
||
state=session_state,
|
||
request=self.mcp_server.request_context.request,
|
||
)
|
||
output_data = raw_output["data"]
|
||
else:
|
||
# Queued path: use the HTTP loopback to preserve streaming
|
||
# updates, progress notifications, and queue-based features.
|
||
progress_token = None
|
||
if self.mcp_server.request_context.meta is not None:
|
||
progress_token = self.mcp_server.request_context.meta.progressToken
|
||
|
||
client = await run_sync(self._get_or_create_client)
|
||
job = client.submit(
|
||
*processed_args,
|
||
api_name=endpoint_name,
|
||
headers=request_headers,
|
||
)
|
||
|
||
if progress_token is None:
|
||
output_data = await self._execute_tool_without_progress(job)
|
||
else:
|
||
output_data = await self._execute_tool_with_progress(
|
||
job,
|
||
progress_token,
|
||
)
|
||
|
||
output_data = self.pop_returned_state(block_fn.outputs, output_data)
|
||
|
||
context_request: Request | None = self.mcp_server.request_context.request
|
||
route_path = self.get_route_path(context_request) # type: ignore
|
||
root_url = route_utils.get_root_url( # type: ignore
|
||
request=context_request, # type: ignore
|
||
route_path=route_path, # type: ignore
|
||
root_path=self.root_path, # type: ignore
|
||
)
|
||
content = self.postprocess_output_data(output_data, root_url)
|
||
if getattr(block_fn.fn, "_mcp_structured_output", False):
|
||
structured_content = {"result": content}
|
||
else:
|
||
structured_content = None
|
||
return self.types.CallToolResult( # type: ignore
|
||
content=content, # type: ignore
|
||
structuredContent=structured_content, # type: ignore
|
||
_meta=getattr(block_fn.fn, "_mcp_meta", None), # type: ignore
|
||
)
|
||
|
||
@server.list_tools()
|
||
async def list_tools() -> list[self.types.Tool]: # type: ignore
|
||
"""
|
||
List all tools on the Gradio app.
|
||
"""
|
||
selected_tools = self.get_selected_tools_from_request()
|
||
|
||
tools = []
|
||
for tool_name, endpoint_name in self.tool_to_endpoint.items():
|
||
if selected_tools is not None and tool_name not in selected_tools:
|
||
continue
|
||
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
if (
|
||
block_fn is None
|
||
or block_fn.fn is None
|
||
or (
|
||
hasattr(block_fn.fn, "_mcp_type")
|
||
and block_fn.fn._mcp_type != "tool"
|
||
)
|
||
):
|
||
continue
|
||
|
||
description, parameters = self.get_fn_description(block_fn, tool_name)
|
||
schema, _ = self.get_input_schema(tool_name, parameters)
|
||
tool_meta = getattr(block_fn.fn, "_mcp_meta", None)
|
||
|
||
tools.append(
|
||
self.types.Tool( # type: ignore
|
||
name=tool_name,
|
||
description=description,
|
||
inputSchema=schema,
|
||
_meta=tool_meta, # type: ignore
|
||
)
|
||
)
|
||
return tools
|
||
|
||
@server.list_resources()
|
||
async def list_resources() -> list[self.types.Resource]: # type: ignore
|
||
"""
|
||
List all available resources.
|
||
"""
|
||
resources = []
|
||
|
||
selected_tools = self.get_selected_tools_from_request()
|
||
for tool_name, endpoint_name in self.tool_to_endpoint.items():
|
||
if selected_tools is not None and tool_name not in selected_tools:
|
||
continue
|
||
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
if (
|
||
block_fn
|
||
and block_fn.fn
|
||
and hasattr(block_fn.fn, "_mcp_type")
|
||
and block_fn.fn._mcp_type == "resource"
|
||
):
|
||
uri_template = block_fn.fn._mcp_uri_template # type: ignore
|
||
parameters = re.findall(r"\{([^}]+)\}", uri_template)
|
||
description, parameters, _ = utils.get_function_description(
|
||
block_fn.fn
|
||
)
|
||
if not parameters:
|
||
resources.append(
|
||
self.types.Resource( # type: ignore
|
||
uri=uri_template,
|
||
name=block_fn.fn.__name__, # type: ignore
|
||
description=description,
|
||
mimeType=block_fn.fn._mcp_mime_type, # type: ignore
|
||
)
|
||
)
|
||
return resources
|
||
|
||
@server.list_resource_templates()
|
||
async def list_resource_templates() -> list[self.types.ResourceTemplate]: # type: ignore
|
||
"""
|
||
List all available resource templates.
|
||
"""
|
||
templates = []
|
||
selected_tools = self.get_selected_tools_from_request()
|
||
for tool_name, endpoint_name in self.tool_to_endpoint.items():
|
||
if selected_tools is not None and tool_name not in selected_tools:
|
||
continue
|
||
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
if (
|
||
block_fn
|
||
and block_fn.fn
|
||
and hasattr(block_fn.fn, "_mcp_type")
|
||
and block_fn.fn._mcp_type == "resource"
|
||
):
|
||
uri_template = block_fn.fn._mcp_uri_template # type: ignore
|
||
parameters = re.findall(r"\{([^}]+)\}", uri_template)
|
||
description, parameters, _ = utils.get_function_description(
|
||
block_fn.fn
|
||
)
|
||
if parameters:
|
||
templates.append(
|
||
self.types.ResourceTemplate( # type: ignore
|
||
uriTemplate=uri_template,
|
||
name=block_fn.fn.__name__, # type: ignore
|
||
description=description,
|
||
mimeType=block_fn.fn._mcp_mime_type, # type: ignore
|
||
)
|
||
)
|
||
return templates
|
||
|
||
@server.read_resource()
|
||
async def read_resource(uri: AnyUrl | str) -> list[self.ReadResourceContents]: # type: ignore
|
||
"""
|
||
Read a specific resource by URI.
|
||
"""
|
||
uri = str(uri)
|
||
client = await run_sync(self._get_or_create_client)
|
||
for endpoint_name in self.tool_to_endpoint.values():
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
|
||
if (
|
||
block_fn
|
||
and block_fn.fn
|
||
and hasattr(block_fn.fn, "_mcp_type")
|
||
and block_fn.fn._mcp_type == "resource"
|
||
):
|
||
uri_template = block_fn.fn._mcp_uri_template # type: ignore
|
||
parameters = re.findall(r"\{([^}]+)\}", uri_template) # type: ignore
|
||
|
||
kwargs = {}
|
||
matched = False
|
||
|
||
if parameters:
|
||
pattern = re.escape(uri_template)
|
||
for param in parameters:
|
||
pattern = pattern.replace(
|
||
f"\\{{{param}\\}}", f"(?P<{param}>[^/]+)"
|
||
)
|
||
match = re.match(f"^{pattern}$", uri)
|
||
if match:
|
||
kwargs = match.groupdict()
|
||
matched = True
|
||
elif uri_template == uri:
|
||
matched = True
|
||
|
||
if matched:
|
||
if endpoint_name in self.api_info["named_endpoints"]:
|
||
parameters_info = self.api_info["named_endpoints"][
|
||
endpoint_name
|
||
]["parameters"]
|
||
processed_args = client_utils.construct_args(
|
||
parameters_info,
|
||
(),
|
||
kwargs,
|
||
)
|
||
else:
|
||
processed_args = list(kwargs.values())
|
||
|
||
async for update in client.submit(
|
||
*processed_args, api_name=endpoint_name
|
||
):
|
||
if update.type == "output" and update.final: # type: ignore
|
||
output = update.outputs # type: ignore
|
||
result = output["data"][0]
|
||
break
|
||
|
||
mime_type = block_fn.fn._mcp_mime_type # type: ignore
|
||
if mime_type and not mime_type.startswith("text/"):
|
||
result = base64.b64decode(result.encode("ascii"))
|
||
return [
|
||
self.ReadResourceContents( # type: ignore
|
||
content=result, mime_type=mime_type
|
||
)
|
||
]
|
||
|
||
raise ValueError(f"Resource not found: {uri}")
|
||
|
||
@server.list_prompts()
|
||
async def list_prompts() -> list[self.types.Prompt]: # type: ignore
|
||
"""
|
||
List all available prompts.
|
||
"""
|
||
prompts = []
|
||
selected_tools = self.get_selected_tools_from_request()
|
||
for tool_name, endpoint_name in self.tool_to_endpoint.items():
|
||
if selected_tools is not None and tool_name not in selected_tools:
|
||
continue
|
||
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
if (
|
||
block_fn
|
||
and block_fn.fn
|
||
and hasattr(block_fn.fn, "_mcp_type")
|
||
and block_fn.fn._mcp_type == "prompt"
|
||
):
|
||
description, parameters, _ = utils.get_function_description(
|
||
block_fn.fn
|
||
)
|
||
function_params = utils.get_function_params(block_fn.fn)
|
||
arguments = [
|
||
self.types.PromptArgument( # type: ignore
|
||
name=param_name,
|
||
description=parameters.get(param_name, ""),
|
||
required=not has_default,
|
||
)
|
||
for param_name, has_default, _, _ in function_params
|
||
]
|
||
prompts.append(
|
||
self.types.Prompt( # type: ignore
|
||
name=tool_name,
|
||
description=description,
|
||
arguments=arguments,
|
||
)
|
||
)
|
||
return prompts
|
||
|
||
@server.get_prompt()
|
||
async def get_prompt(
|
||
name: str, arguments: dict[str, Any] | None = None
|
||
) -> self.types.GetPromptResult: # type: ignore
|
||
"""
|
||
Get a specific prompt with filled-in arguments.
|
||
"""
|
||
client = await run_sync(self._get_or_create_client)
|
||
|
||
endpoint_name = None
|
||
for endpoint_name in self.tool_to_endpoint.values():
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
if (
|
||
block_fn
|
||
and block_fn.fn
|
||
and hasattr(block_fn.fn, "_mcp_type")
|
||
and block_fn.fn._mcp_type == "prompt"
|
||
and block_fn.fn._mcp_name == name # type: ignore
|
||
):
|
||
break
|
||
|
||
if not endpoint_name:
|
||
raise ValueError(f"Prompt not found: {name}")
|
||
|
||
arguments = arguments or {}
|
||
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
assert block_fn is not None # noqa: S101
|
||
|
||
if endpoint_name in self.api_info["named_endpoints"]:
|
||
parameters_info = self.api_info["named_endpoints"][endpoint_name][
|
||
"parameters"
|
||
]
|
||
processed_args = client_utils.construct_args(
|
||
parameters_info,
|
||
(),
|
||
arguments,
|
||
)
|
||
else:
|
||
processed_args = list(arguments.values())
|
||
|
||
async for update in client.submit(*processed_args, api_name=endpoint_name):
|
||
if update.type == "output" and update.final: # type: ignore
|
||
output = update.outputs # type: ignore
|
||
result = output["data"][0]
|
||
break
|
||
|
||
return self.types.GetPromptResult( # type: ignore
|
||
messages=[
|
||
self.types.PromptMessage( # type: ignore
|
||
role="user",
|
||
content=self.types.TextContent(type="text", text=str(result)), # type: ignore
|
||
)
|
||
]
|
||
)
|
||
|
||
return server
|
||
|
||
def launch_mcp_on_sse(self, app: Starlette, subpath: str, root_path: str) -> None:
|
||
"""
|
||
Launch the MCP server on the SSE transport.
|
||
|
||
Parameters:
|
||
app: The Gradio app to mount the MCP server on.
|
||
subpath: The subpath to mount the MCP server on. E.g. "/gradio_api/mcp"
|
||
root_path: The root path of the Gradio Blocks app.
|
||
"""
|
||
messages_path = "/messages/"
|
||
sse = self.SseServerTransport(messages_path) # type: ignore
|
||
self.root_path = root_path
|
||
|
||
async def handle_sse(request):
|
||
try:
|
||
async with sse.connect_sse(
|
||
request.scope, request.receive, request._send
|
||
) as streams:
|
||
await self.mcp_server.run(
|
||
streams[0],
|
||
streams[1],
|
||
self.mcp_server.create_initialization_options(),
|
||
)
|
||
return Response()
|
||
except Exception as e:
|
||
print(f"MCP SSE connection error: {str(e)}")
|
||
raise
|
||
|
||
app.mount(
|
||
subpath,
|
||
Starlette(
|
||
routes=[
|
||
Route(
|
||
"/schema",
|
||
endpoint=self.get_complete_schema, # Not required for MCP but used by the Hugging Face MCP server to get the schema for MCP Spaces without needing to establish an SSE connection
|
||
),
|
||
Route("/sse", endpoint=handle_sse),
|
||
Mount("/messages/", app=sse.handle_post_message),
|
||
Mount("/", app=self.handle_streamable_http),
|
||
],
|
||
),
|
||
)
|
||
|
||
def get_block_fn_from_endpoint_name(
|
||
self, endpoint_name: str
|
||
) -> "BlockFunction | None":
|
||
"""
|
||
Get the BlockFunction for a given endpoint name (e.g. "/predict").
|
||
|
||
Parameters:
|
||
endpoint_name: The name of the endpoint to get the BlockFunction for.
|
||
|
||
Returns:
|
||
The BlockFunction for the given endpoint name, or None if it is not found.
|
||
"""
|
||
block_fn = next(
|
||
(
|
||
fn
|
||
for fn in self.blocks.fns.values()
|
||
if fn.api_name == endpoint_name.lstrip("/")
|
||
),
|
||
None,
|
||
)
|
||
return block_fn
|
||
|
||
@property
|
||
def _file_data_tool_description(self) -> str:
|
||
"""
|
||
Sentence prompting the agent to use the upload_file_to_gradio tool if a file is passed as an input.
|
||
"""
|
||
return " If a user passes a file as an input, use the upload_file_to_gradio tool, if present, to upload the file to the gradio app and create a Gradio File Input. Then use the returned path as the input to the tool"
|
||
|
||
def get_fn_description(
|
||
self, block_fn: "BlockFunction", tool_name: str
|
||
) -> tuple[str, dict[str, str]]:
|
||
"""
|
||
Get the description of a function, which is used to describe the tool in the MCP server.
|
||
Also returns the description of each parameter of the function as a dictionary.
|
||
"""
|
||
description, parameters, returns = utils.get_function_description(block_fn.fn) # type: ignore
|
||
_, filedata_positions = self.get_input_schema(tool_name, parameters)
|
||
if block_fn.api_description is False:
|
||
description = ""
|
||
elif block_fn.api_description is None:
|
||
if len(filedata_positions) > 0:
|
||
description += self._file_data_tool_description
|
||
if returns:
|
||
description += (
|
||
("" if description.endswith(".") else ".")
|
||
+ " Returns: "
|
||
+ ", ".join(returns)
|
||
)
|
||
else:
|
||
description = block_fn.api_description
|
||
if len(filedata_positions) > 0:
|
||
description += self._file_data_tool_description # type: ignore
|
||
assert isinstance(description, str) # noqa: S101
|
||
return description, parameters
|
||
|
||
@staticmethod
|
||
def insert_empty_state(
|
||
inputs: Sequence["Component | BlockContext"], data: list
|
||
) -> list:
|
||
"""
|
||
Insert None placeholder values for any State input components, as State inputs
|
||
are not included in the endpoint schema.
|
||
"""
|
||
for i, input_component_type in enumerate(inputs):
|
||
if isinstance(input_component_type, State):
|
||
data.insert(i, None)
|
||
return data
|
||
|
||
@staticmethod
|
||
def pop_returned_state(
|
||
components: Sequence["Component | BlockContext"], data: Any
|
||
) -> list:
|
||
"""
|
||
Remove any values corresponding to State output components from the data
|
||
as State outputs are not included in the endpoint schema.
|
||
"""
|
||
for i, component_type in enumerate(components):
|
||
if isinstance(component_type, State):
|
||
data.pop(i)
|
||
return data
|
||
|
||
def get_input_schema(
|
||
self,
|
||
tool_name: str,
|
||
parameters: dict[str, str] | None = None,
|
||
) -> tuple[dict[str, Any], list[list[str | int]]]:
|
||
"""
|
||
Get the input schema of the Gradio app API, appropriately formatted for MCP.
|
||
|
||
Parameters:
|
||
tool_name: The name of the tool to get the schema for, e.g. "predict"
|
||
parameters: The description and parameters of the tool to get the schema for.
|
||
Returns:
|
||
- The input schema of the Gradio app API.
|
||
- A list of positions of FileData objects in the input schema.
|
||
"""
|
||
endpoint_name = self.tool_to_endpoint.get(tool_name)
|
||
if endpoint_name is None:
|
||
raise ValueError(f"Unknown tool for this Gradio app: {tool_name}")
|
||
named_endpoints = self.api_info["named_endpoints"]
|
||
endpoint_info = named_endpoints.get(endpoint_name)
|
||
assert endpoint_info is not None # noqa: S101
|
||
|
||
schema = {
|
||
"type": "object",
|
||
"properties": {
|
||
p["parameter_name"]: {
|
||
**p["type"],
|
||
**(
|
||
{"description": parameters[p["parameter_name"]]}
|
||
if parameters and p["parameter_name"] in parameters
|
||
else {}
|
||
),
|
||
**(
|
||
{"default": p["parameter_default"]}
|
||
if "parameter_default" in p and p["parameter_default"]
|
||
else {}
|
||
),
|
||
}
|
||
for p in endpoint_info["parameters"]
|
||
},
|
||
}
|
||
return self.simplify_filedata_schema(schema)
|
||
|
||
async def get_complete_schema(self, request) -> JSONResponse:
|
||
"""
|
||
Get the complete schema of the Gradio app API. For debugging purposes, also used by
|
||
the Hugging Face MCP server to get the schema for MCP Spaces without needing to
|
||
establish an SSE connection.
|
||
|
||
Parameters:
|
||
request: The Starlette request object.
|
||
|
||
Returns:
|
||
A JSONResponse containing a dictionary mapping tool names to their input schemas.
|
||
"""
|
||
if not self.api_info:
|
||
return JSONResponse({})
|
||
|
||
query_params = dict(getattr(request, "query_params", {}))
|
||
selected_tools = None
|
||
if "tools" in query_params:
|
||
tools = query_params["tools"].split(",")
|
||
selected_tools = set(tools)
|
||
|
||
file_data_present = False
|
||
|
||
schemas = []
|
||
for tool_name, endpoint_name in self.tool_to_endpoint.items():
|
||
if selected_tools is not None and tool_name not in selected_tools:
|
||
continue
|
||
block_fn = self.get_block_fn_from_endpoint_name(endpoint_name)
|
||
assert block_fn is not None and block_fn.fn is not None # noqa: S101
|
||
|
||
description, parameters = self.get_fn_description(block_fn, tool_name)
|
||
schema, filedata_positions = self.get_input_schema(tool_name, parameters)
|
||
if len(filedata_positions) > 0 and not file_data_present:
|
||
file_data_present = True
|
||
|
||
type_hints = utils.get_type_hints(block_fn.fn)
|
||
required_headers = []
|
||
for param_name, type_hint in type_hints.items():
|
||
if type_hint is Header or type_hint is Optional[Header]:
|
||
header_name = param_name.replace("_", "-").lower()
|
||
required_headers.append(header_name)
|
||
|
||
mcp_type = "tool" # Default
|
||
if hasattr(block_fn.fn, "_mcp_type"):
|
||
mcp_type = block_fn.fn._mcp_type
|
||
|
||
meta = {
|
||
"file_data_present": file_data_present,
|
||
"mcp_type": mcp_type,
|
||
"endpoint_name": block_fn.api_name,
|
||
}
|
||
if required_headers:
|
||
meta["headers"] = required_headers
|
||
|
||
info = {
|
||
"name": tool_name,
|
||
"description": description,
|
||
"inputSchema": schema,
|
||
"meta": meta,
|
||
}
|
||
schemas.append(info)
|
||
|
||
return JSONResponse(schemas)
|
||
|
||
def simplify_filedata_schema(
|
||
self, schema: dict[str, Any]
|
||
) -> tuple[dict[str, Any], list[list[str | int]]]:
|
||
"""
|
||
Parses a schema of a Gradio app API to identify positions of FileData objects. Replaces them with base64
|
||
strings while keeping track of their positions so that they can be converted back to FileData objects
|
||
later.
|
||
|
||
Parameters:
|
||
schema: The original schema of the Gradio app API.
|
||
|
||
Returns:
|
||
A tuple containing the simplified schema and the positions of the FileData objects.
|
||
"""
|
||
|
||
def is_gradio_filedata(obj: Any, defs: dict[str, Any]) -> bool:
|
||
if not isinstance(obj, dict):
|
||
return False
|
||
|
||
if "$ref" in obj:
|
||
ref = obj["$ref"]
|
||
if ref.startswith("#/$defs/"):
|
||
key = ref.split("/")[-1]
|
||
obj = defs.get(key, {})
|
||
else:
|
||
return False
|
||
|
||
props = obj.get("properties", {})
|
||
meta = props.get("meta", {})
|
||
|
||
if "$ref" in meta:
|
||
ref = meta["$ref"]
|
||
if ref.startswith("#/$defs/"):
|
||
key = ref.split("/")[-1]
|
||
meta = defs.get(key, {})
|
||
else:
|
||
return False
|
||
|
||
type_field = meta.get("properties", {}).get("_type", {})
|
||
default_type = meta.get("default", {}).get("_type")
|
||
return (
|
||
type_field.get("const") == "gradio.FileData"
|
||
or default_type == "gradio.FileData"
|
||
)
|
||
|
||
def traverse(
|
||
node: Any,
|
||
path: list[str | int] | None = None,
|
||
defs: dict[str, Any] | None = None,
|
||
) -> Any:
|
||
if path is None:
|
||
path = []
|
||
if defs is None:
|
||
defs = {}
|
||
# Deep copy the node to avoid modifying the original node
|
||
node = copy.deepcopy(node)
|
||
|
||
if isinstance(node, dict):
|
||
if "$defs" in node:
|
||
defs.update(node["$defs"])
|
||
|
||
if is_gradio_filedata(node, defs):
|
||
filedata_positions.append(path.copy())
|
||
for key in ["properties", "additional_description", "$defs"]:
|
||
node.pop(key, None)
|
||
node["type"] = "string"
|
||
node["format"] = "Gradio File Input - a http or https url to a file"
|
||
|
||
result = {}
|
||
is_schema_root = "type" in node and "properties" in node
|
||
for key, value in node.items():
|
||
if is_schema_root and key == "properties":
|
||
result[key] = traverse(value, path, defs)
|
||
else:
|
||
path.append(key)
|
||
result[key] = traverse(value, path, defs)
|
||
path.pop()
|
||
return result
|
||
|
||
elif isinstance(node, list):
|
||
result = []
|
||
for i, item in enumerate(node):
|
||
path.append(i)
|
||
result.append(traverse(item, path, defs))
|
||
path.pop()
|
||
return result
|
||
|
||
return node
|
||
|
||
filedata_positions: list[list[str | int]] = []
|
||
simplified_schema = traverse(schema)
|
||
return simplified_schema, filedata_positions
|
||
|
||
def convert_strings_to_filedata(
|
||
self, value: Any, filedata_positions: list[list[str | int]]
|
||
) -> Any:
|
||
"""
|
||
Convert specific string values back to FileData objects based on their positions.
|
||
This is used to convert string values (as base64 encoded strings) to FileData
|
||
dictionaries so that they can be passed into .preprocess() logic of a Gradio app.
|
||
|
||
Parameters:
|
||
value: The input data to process, which can be an arbitrary nested data structure
|
||
that may or may not contain strings that should be converted to FileData objects.
|
||
filedata_positions: List of paths to positions in the input data that should be converted to FileData objects.
|
||
|
||
Returns:
|
||
The processed data with strings converted to FileData objects where appropriate. Base64
|
||
encoded strings are first saved to a temporary file and then converted to a FileData object.
|
||
|
||
Example:
|
||
>>> convert_strings_to_filedata(
|
||
{"image": "data:image/jpeg;base64,..."},
|
||
[["image"]]
|
||
)
|
||
>>> {'image': FileData(path='<temporary file path>')},
|
||
"""
|
||
|
||
def traverse(node: Any, path: list[str | int] | None = None) -> Any:
|
||
if path is None:
|
||
path = []
|
||
|
||
if isinstance(node, dict):
|
||
return {
|
||
key: traverse(value, path + [key]) for key, value in node.items()
|
||
}
|
||
elif isinstance(node, list):
|
||
return [traverse(item, path + [i]) for i, item in enumerate(node)]
|
||
elif isinstance(node, str) and path in filedata_positions:
|
||
if node.startswith("data:"):
|
||
# Even though base64 is not officially part of our schema, some MCP clients
|
||
# might return base64 encoded strings, so try to save it to a temporary file.
|
||
return handle_file(
|
||
processing_utils.save_base64_to_cache(node, DEFAULT_TEMP_DIR)
|
||
)
|
||
elif node.startswith(("http://", "https://")):
|
||
return handle_file(node)
|
||
else:
|
||
raise ValueError(
|
||
f"Invalid file data format, provide a url ('http://...' or 'https://...'). Received: {node}"
|
||
)
|
||
return node
|
||
|
||
return traverse(value)
|
||
|
||
@staticmethod
|
||
def get_image(file_path: str) -> Image.Image | None:
|
||
"""
|
||
If a filepath is a valid image, returns a PIL Image object. Otherwise returns None.
|
||
"""
|
||
if not os.path.exists(file_path):
|
||
return None
|
||
ext = os.path.splitext(file_path.lower())[1]
|
||
if ext not in Image.registered_extensions():
|
||
return None
|
||
try:
|
||
return Image.open(file_path)
|
||
except Exception:
|
||
return None
|
||
|
||
@staticmethod
|
||
def get_svg(file_data: Any) -> bytes | None:
|
||
"""
|
||
If a file_data is a valid FileDataDict with a url that is a data:image/svg+xml, returns bytes of the svg. Otherwise returns None.
|
||
"""
|
||
if isinstance(file_data, dict) and (url := file_data.get("url")):
|
||
if isinstance(url, str) and url.startswith("data:image/svg"):
|
||
return unquote(url.split(",", 1)[1]).encode()
|
||
else:
|
||
return None
|
||
else:
|
||
return None
|
||
|
||
@staticmethod
|
||
def get_base64_data(image: Image.Image, format: str) -> str:
|
||
"""
|
||
Returns a base64 encoded string of the image.
|
||
"""
|
||
buffer = BytesIO()
|
||
image.save(buffer, format=format)
|
||
return base64.b64encode(buffer.getvalue()).decode("utf-8")
|
||
|
||
def postprocess_output_data(
|
||
self, data: Any, root_url: str
|
||
) -> list["types.TextContent | types.ImageContent"]:
|
||
"""
|
||
Postprocess the output data from the Gradio app to convert FileData objects back to base64 encoded strings.
|
||
|
||
Parameters:
|
||
data: The output data to postprocess.
|
||
"""
|
||
return_values = []
|
||
data = processing_utils.add_root_url(data, root_url, None)
|
||
for output in data:
|
||
if svg_bytes := self.get_svg(output):
|
||
base64_data = base64.b64encode(svg_bytes).decode("utf-8")
|
||
mimetype = "image/svg+xml"
|
||
svg_path = processing_utils.save_bytes_to_cache(
|
||
svg_bytes, f"{output['orig_name']}", DEFAULT_TEMP_DIR
|
||
)
|
||
svg_url = f"{root_url}/gradio_api/file={svg_path}"
|
||
return_value = [
|
||
self.types.ImageContent( # type: ignore
|
||
type="image", data=base64_data, mimeType=mimetype
|
||
),
|
||
self.types.TextContent( # type: ignore
|
||
type="text",
|
||
text=f"SVG Image URL: {svg_url}",
|
||
),
|
||
]
|
||
elif client_utils.is_file_obj_with_meta(output):
|
||
if image := self.get_image(output["path"]):
|
||
image_format = image.format or "png"
|
||
base64_data = self.get_base64_data(image, image_format)
|
||
mimetype = f"image/{image_format.lower()}"
|
||
return_value = [
|
||
self.types.ImageContent( # type: ignore
|
||
type="image", data=base64_data, mimeType=mimetype
|
||
),
|
||
self.types.TextContent( # type: ignore
|
||
type="text",
|
||
text=f"Image URL: {output['url'] or output['path']}",
|
||
),
|
||
]
|
||
else:
|
||
return_value = [
|
||
self.types.TextContent( # type: ignore
|
||
type="text", text=str(output["url"] or output["path"])
|
||
)
|
||
]
|
||
else:
|
||
return_value = [self.types.TextContent(type="text", text=str(output))] # type: ignore
|
||
return_values.extend(return_value)
|
||
return return_values
|
||
|
||
|
||
######################################################
|
||
### MCP decorators that add metadata to functions.
|
||
######################################################
|
||
|
||
|
||
def resource(
|
||
uri_template: str, description: str | None = None, mime_type: str | None = None
|
||
):
|
||
"""Decorator to mark a function as an MCP resource."""
|
||
|
||
def decorator(fn):
|
||
fn._mcp_type = "resource"
|
||
fn._mcp_uri_template = uri_template
|
||
fn._mcp_description = description
|
||
fn._mcp_mime_type = mime_type or "text/plain"
|
||
return fn
|
||
|
||
return decorator
|
||
|
||
|
||
def prompt(name: str | None = None, description: str | None = None):
|
||
"""Decorator to mark a function as an MCP prompt."""
|
||
|
||
def decorator(fn):
|
||
fn._mcp_type = "prompt"
|
||
fn._mcp_name = name or fn.__name__
|
||
fn._mcp_description = description
|
||
return fn
|
||
|
||
return decorator
|
||
|
||
|
||
def tool(
|
||
name: str | None = None,
|
||
description: str | None = None,
|
||
structured_output: bool = False,
|
||
_meta: dict[str, Any] | None = None,
|
||
):
|
||
"""
|
||
Decorator to mark a function as an MCP tool (optional, since functions are registered as tools by default).
|
||
Can be used to configure various aspects of the tool.
|
||
|
||
Parameters:
|
||
name: The name of the tool. Overrides the default name of the function.
|
||
description: The description of the tool. Overrides the default description from the function's docstring.
|
||
structured_output: Whether the tool should return structured output (implementation is quite limited at the moment). If True, the output will be wrapped in a dictionary with the key "result" and the value being the output of the function. Recommended to keep this False unless you have a specific reason to need the structured output.
|
||
_meta: Additional metadata for the tool.
|
||
"""
|
||
|
||
def decorator(fn):
|
||
fn._mcp_type = "tool"
|
||
fn._mcp_name = name
|
||
fn._mcp_structured_output = structured_output
|
||
fn._mcp_description = description
|
||
fn._mcp_meta = _meta
|
||
return fn
|
||
|
||
return decorator
|