Upload folder using huggingface_hub (part 7)

e1a9145 verified 7 days ago

64.3 kB

	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