diff --git a/examples/widget_test/README.md b/examples/widget_test/README.md new file mode 100644 index 000000000..768c73002 --- /dev/null +++ b/examples/widget_test/README.md @@ -0,0 +1,135 @@ +# OpenAI Widget Demo + +This directory contains a demonstration of FastMCP's OpenAI widget support using the Pizzaz mapping library. + +## What's Here + +- **`pizzaz_server.py`**: Demo server with three widget examples showing different return patterns +- **`test_widgets.py`**: Test script that validates the auto-transformation works correctly + +## Setup + +This uses a local virtual environment with an editable install of FastMCP: + +```bash +# Already done, but for reference: +cd examples/widget_test +uv venv +source .venv/bin/activate # On Windows: .venv\Scripts\activate +uv pip install -e ../.. +``` + +## Running the Demo + +### Test the Widgets Locally + +```bash +source .venv/bin/activate +python test_widgets.py +``` + +This will test all three widget patterns: +- **dict return**: Structured data only (no narrative text) +- **str return**: Narrative text only (no structured data) +- **tuple[str, dict] return**: Both narrative text and structured data + +### Run the Server + +```bash +source .venv/bin/activate +python pizzaz_server.py +``` + +The server will start on `http://0.0.0.0:8080`. + +### Inspect the Server + +```bash +source .venv/bin/activate +fastmcp inspect pizzaz_server.py +``` + +Use `--format fastmcp` to see the full JSON including OpenAI metadata. + +## Testing with ChatGPT + +To test the widgets with ChatGPT: + +1. Expose your server with ngrok: + ```bash + ngrok http 8080 + ``` + +2. In ChatGPT, go to Settings → Developer Mode and add your server + +3. Try prompts like: + - "Show me a pizza map for pepperoni" + - "Track pizza order 12345" + - "What's the pizza status?" + +## Widget Examples + +### 1. Pizza Map (dict return) + +Returns structured data only, which the widget renders: + +```python +@app.ui.openai.widget( + name="pizza-map", + template_uri="ui://widget/pizza-map.html", + html=PIZZAZ_HTML, + invoking="Hand-tossing a map", + invoked="Served a fresh map", +) +def show_pizza_map(topping: str) -> dict: + return {"pizza_topping": topping.strip()} +``` + +### 2. Pizza Tracker (tuple return) + +Returns both narrative text and structured data: + +```python +@app.ui.openai.widget( + name="pizza-tracker", + template_uri="ui://widget/pizza-tracker.html", + html=PIZZAZ_HTML, + invoking="Tracking your pizza", + invoked="Pizza located!", +) +def track_pizza(order_id: str) -> tuple[str, dict]: + narrative = f"Tracking pizza order {order_id}..." + data = {"order_id": order_id, "status": "out_for_delivery"} + return narrative, data +``` + +### 3. Pizza Status (str return) + +Returns text only, no structured data: + +```python +@app.ui.openai.widget( + name="pizza-status", + template_uri="ui://widget/pizza-status.html", + html=PIZZAZ_HTML, +) +def pizza_status() -> str: + return "Pizza ovens are hot and ready! 🍕" +``` + +## How It Works + +The `@app.ui.openai.widget` decorator automatically: + +1. **Registers the HTML** as an MCP resource with MIME type `text/html+skybridge` +2. **Adds OpenAI metadata** to the tool including: + - `openai/outputTemplate`: Points to the widget HTML + - `openai/toolInvocation/invoking` and `invoked`: Status messages + - `openai.com/widget`: Embedded widget resource + - CSP configuration for security +3. **Auto-transforms return values** to OpenAI format: + - `dict` → `{"content": [], "structuredContent": {...}}` + - `str` → `{"content": [{"type": "text", "text": "..."}], "structuredContent": {}}` + - `tuple[str, dict]` → Both content and structuredContent + +You don't need to manually call `build_widget_tool_response()` or `register_decorated_widgets()` - it all happens automatically! diff --git a/examples/widget_test/pizzaz_server.py b/examples/widget_test/pizzaz_server.py new file mode 100644 index 000000000..649be00a6 --- /dev/null +++ b/examples/widget_test/pizzaz_server.py @@ -0,0 +1,100 @@ +"""Pizzaz Widget Demo Server + +This demonstrates FastMCP's OpenAI widget support using the Pizzaz mapping library. +The widget renders an interactive map when you ask about pizza toppings. + +Example usage: + Show me a pizza map for pepperoni + Create a pizza map with mushrooms +""" + +from fastmcp import FastMCP + +app = FastMCP("pizzaz-demo") + +# Pizzaz widget HTML with the library loaded from OpenAI's CDN +PIZZAZ_HTML = """ +
+ + +""" + + +@app.ui.openai.widget( + name="pizza-map", + template_uri="ui://widget/pizza-map.html", + html=PIZZAZ_HTML, + title="Pizza Map", + description="Show an interactive pizza map for a given topping", + invoking="Hand-tossing a map", + invoked="Served a fresh map", + widget_csp_resources=["https://persistent.oaistatic.com"], +) +def show_pizza_map(topping: str) -> dict: + """Show an interactive pizza map for the given topping. + + Args: + topping: The pizza topping to map (e.g., "pepperoni", "mushrooms") + + Returns: + Structured data for the widget to render + """ + return { + "pizza_topping": topping.strip(), + "map_type": "delicious", + } + + +@app.ui.openai.widget( + name="pizza-tracker", + template_uri="ui://widget/pizza-tracker.html", + html=PIZZAZ_HTML, + title="Pizza Tracker", + description="Track a pizza order with real-time updates", + invoking="Tracking your pizza", + invoked="Pizza located!", + widget_csp_resources=["https://persistent.oaistatic.com"], +) +def track_pizza(order_id: str) -> tuple[str, dict]: + """Track a pizza order by order ID. + + This demonstrates returning both narrative text and structured data. + + Args: + order_id: The pizza order ID to track + + Returns: + Tuple of (narrative text, structured data for widget) + """ + narrative = f"Tracking pizza order {order_id}. Your pizza is on the way!" + data = { + "order_id": order_id, + "status": "out_for_delivery", + "estimated_time": "15 minutes", + "driver_location": {"lat": 47.6062, "lng": -122.3321}, + } + return narrative, data + + +@app.ui.openai.widget( + name="pizza-status", + template_uri="ui://widget/pizza-status.html", + html=PIZZAZ_HTML, + title="Pizza Status", + description="Get the current status of your pizza", + widget_csp_resources=["https://persistent.oaistatic.com"], +) +def pizza_status() -> str: + """Get a simple status message about pizza availability. + + This demonstrates returning text only (no structured data). + + Returns: + Status message text + """ + return "Pizza ovens are hot and ready! 🍕 We can make any topping you'd like." + + +if __name__ == "__main__": + # Run with HTTP transport for testing with ChatGPT + app.run(transport="http", host="0.0.0.0", port=8080) diff --git a/src/fastmcp/server/context.py b/src/fastmcp/server/context.py index a0fcfc1c2..12d6b46ac 100644 --- a/src/fastmcp/server/context.py +++ b/src/fastmcp/server/context.py @@ -429,9 +429,13 @@ async def sample( ] elif isinstance(messages, Sequence): sampling_messages = [ - SamplingMessage(content=TextContent(text=m, type="text"), role="user") - if isinstance(m, str) - else m + ( + SamplingMessage( + content=TextContent(text=m, type="text"), role="user" + ) + if isinstance(m, str) + else m + ) for m in messages ] diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py index c758f64cb..e3eda0147 100644 --- a/src/fastmcp/server/server.py +++ b/src/fastmcp/server/server.py @@ -59,6 +59,7 @@ from fastmcp.tools import ToolManager from fastmcp.tools.tool import FunctionTool, Tool, ToolResult from fastmcp.tools.tool_transform import ToolTransformConfig +from fastmcp.ui import UIManager from fastmcp.utilities.cli import log_server_banner from fastmcp.utilities.components import FastMCPComponent from fastmcp.utilities.logging import get_logger, temporary_log_level @@ -192,6 +193,7 @@ def __init__( duplicate_behavior=on_duplicate_prompts, mask_error_details=mask_error_details, ) + self._ui_manager: UIManager | None = None self._tool_serializer = tool_serializer self._lifespan: LifespanCallable[LifespanResultT] = lifespan or default_lifespan @@ -357,6 +359,13 @@ def icons(self) -> list[mcp.types.Icon]: else: return list(self._mcp_server.icons) + @property + def ui(self) -> UIManager: + """Access UI components and integrations.""" + if self._ui_manager is None: + self._ui_manager = UIManager(self) + return self._ui_manager + @asynccontextmanager async def _lifespan_manager(self) -> AsyncIterator[None]: if self._lifespan_result_set: diff --git a/src/fastmcp/ui/__init__.py b/src/fastmcp/ui/__init__.py new file mode 100644 index 000000000..d93486423 --- /dev/null +++ b/src/fastmcp/ui/__init__.py @@ -0,0 +1,5 @@ +"""UI components and integrations.""" + +from fastmcp.ui.manager import UIManager + +__all__ = ["UIManager"] diff --git a/src/fastmcp/ui/manager.py b/src/fastmcp/ui/manager.py new file mode 100644 index 000000000..a55dbc229 --- /dev/null +++ b/src/fastmcp/ui/manager.py @@ -0,0 +1,26 @@ +"""UI manager for FastMCP.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from fastmcp.server.server import FastMCP + from fastmcp.ui.openai import OpenAIUIManager + + +class UIManager: + """Manager for UI-related components and integrations.""" + + def __init__(self, fastmcp: FastMCP[Any]) -> None: + self._fastmcp = fastmcp + self._openai: OpenAIUIManager | None = None + + @property + def openai(self) -> OpenAIUIManager: + """Access OpenAI-specific UI components.""" + if self._openai is None: + from fastmcp.ui.openai import OpenAIUIManager + + self._openai = OpenAIUIManager(self._fastmcp) + return self._openai diff --git a/src/fastmcp/ui/openai/__init__.py b/src/fastmcp/ui/openai/__init__.py new file mode 100644 index 000000000..02e3a613c --- /dev/null +++ b/src/fastmcp/ui/openai/__init__.py @@ -0,0 +1,15 @@ +"""OpenAI UI components.""" + +from fastmcp.ui.openai.manager import OpenAIUIManager +from fastmcp.ui.openai.response import ( + WidgetToolResponse, + build_widget_tool_response, + transform_widget_response, +) + +__all__ = [ + "OpenAIUIManager", + "WidgetToolResponse", + "build_widget_tool_response", + "transform_widget_response", +] diff --git a/src/fastmcp/ui/openai/manager.py b/src/fastmcp/ui/openai/manager.py new file mode 100644 index 000000000..71ea089d8 --- /dev/null +++ b/src/fastmcp/ui/openai/manager.py @@ -0,0 +1,444 @@ +"""OpenAI-specific UI components and widgets.""" + +from __future__ import annotations + +import asyncio +import functools +import inspect +from collections.abc import Callable, Sequence +from typing import TYPE_CHECKING, Any, cast, overload + +from mcp.types import AnyFunction, ToolAnnotations + +from fastmcp.tools.tool import FunctionTool +from fastmcp.ui.openai.metadata import ( + build_widget_resource_meta, + build_widget_tool_meta, + merge_resource_annotations, + merge_tool_annotations, +) +from fastmcp.ui.openai.response import transform_widget_response + +if TYPE_CHECKING: + from fastmcp.server.server import FastMCP + + +class OpenAIUIManager: + """Manager for OpenAI-specific UI components like widgets.""" + + def __init__(self, fastmcp: FastMCP[Any]) -> None: + self._fastmcp = fastmcp + + @overload + def widget( + self, + name_or_fn: AnyFunction, + *, + name: str | None = None, + template_uri: str, + html: str, + title: str | None = None, + description: str | None = None, + invoking: str | None = None, + invoked: str | None = None, + mime_type: str = "text/html+skybridge", + widget_description: str | None = None, + widget_prefers_border: bool = True, + widget_csp_resources: Sequence[str] | None = None, + widget_csp_connect: Sequence[str] | None = None, + widget_accessible: bool = True, + result_can_produce_widget: bool = True, + tags: set[str] | None = None, + annotations: ToolAnnotations | dict[str, Any] | None = None, + resource_annotations: dict[str, Any] | None = None, + meta: dict[str, Any] | None = None, + enabled: bool | None = None, + exclude_args: list[str] | None = None, + ) -> FunctionTool: ... + + @overload + def widget( + self, + name_or_fn: str | None = None, + *, + name: str | None = None, + template_uri: str, + html: str, + title: str | None = None, + description: str | None = None, + invoking: str | None = None, + invoked: str | None = None, + mime_type: str = "text/html+skybridge", + widget_description: str | None = None, + widget_prefers_border: bool = True, + widget_csp_resources: Sequence[str] | None = None, + widget_csp_connect: Sequence[str] | None = None, + widget_accessible: bool = True, + result_can_produce_widget: bool = True, + tags: set[str] | None = None, + annotations: ToolAnnotations | dict[str, Any] | None = None, + resource_annotations: dict[str, Any] | None = None, + meta: dict[str, Any] | None = None, + enabled: bool | None = None, + exclude_args: list[str] | None = None, + ) -> Callable[[AnyFunction], FunctionTool]: ... + + def widget( + self, + name_or_fn: str | AnyFunction | None = None, + *, + name: str | None = None, + template_uri: str | None = None, + html: str | None = None, + title: str | None = None, + description: str | None = None, + invoking: str | None = None, + invoked: str | None = None, + mime_type: str = "text/html+skybridge", + widget_description: str | None = None, + widget_prefers_border: bool = True, + widget_csp_resources: Sequence[str] | None = None, + widget_csp_connect: Sequence[str] | None = None, + widget_accessible: bool = True, + result_can_produce_widget: bool = True, + tags: set[str] | None = None, + annotations: ToolAnnotations | dict[str, Any] | None = None, + resource_annotations: dict[str, Any] | None = None, + meta: dict[str, Any] | None = None, + enabled: bool | None = None, + exclude_args: list[str] | None = None, + ) -> Callable[[AnyFunction], FunctionTool] | FunctionTool: + """Register an OpenAI widget as a tool. + + OpenAI widgets are interactive UI components that display in ChatGPT. + This decorator automatically: + 1. Registers widget HTML as an MCP resource + 2. Registers a tool with OpenAI metadata linking to the widget + 3. Auto-transforms function return values (str, dict, or tuple) to OpenAI format + + The widget function can return: + - `str`: Narrative text shown in conversation (no structured data) + - `dict`: Structured data passed to widget JavaScript (no narrative) + - `tuple[str, dict]`: Both narrative text and structured data + + This decorator supports multiple calling patterns: + - @server.ui.openai.widget(name=..., template_uri=..., html=...) + - @server.ui.openai.widget("name", template_uri=..., html=...) + + Args: + name_or_fn: Either a function (direct decoration) or string name + name: Tool name (defaults to function name or name_or_fn if str) + template_uri: Widget template URI (e.g. "ui://widget/pizza-map.html") + html: Widget HTML content (must include div and script tags) + title: Widget display title (defaults to name) + description: Tool description (defaults to function docstring) + invoking: Status message while tool executes (e.g. "Hand-tossing a map") + invoked: Status message after tool completes (e.g. "Served a fresh map") + mime_type: MIME type for widget HTML (default: "text/html+skybridge") + widget_description: Description of widget UI (defaults to "{title} widget UI.") + widget_prefers_border: Whether widget prefers a border (default: True) + widget_csp_resources: CSP resource_domains (default: ["https://persistent.oaistatic.com"]) + widget_csp_connect: CSP connect_domains (default: []) + widget_accessible: Whether widget is accessible (default: True) + result_can_produce_widget: Whether result can produce widget (default: True) + tags: Tags for categorizing the widget + annotations: Tool annotations (merged with widget defaults) + resource_annotations: Resource annotations for HTML resource + meta: Additional tool metadata (merged with widget metadata) + enabled: Whether widget is enabled (default: True) + exclude_args: Arguments to exclude from tool schema + + Returns: + Decorated function registered as a widget tool + + Examples: + Register a widget with data only: + ```python + @app.ui.openai.widget( + name="pizza-map", + template_uri="ui://widget/pizza-map.html", + html='
', + invoking="Hand-tossing a map", + invoked="Served a fresh map" + ) + def show_pizza_map(topping: str) -> dict: + \"\"\"Show a pizza map for the given topping.\"\"\" + return {"topping": topping} # Auto-transformed! + ``` + + Register a widget with both text and data: + ```python + @app.ui.openai.widget( + name="weather", + template_uri="ui://widget/weather.html", + html='
...' + ) + def show_weather(city: str) -> tuple[str, dict]: + return (f"Showing weather for {city}", {"city": city, "temp": 72}) + ``` + + Register a widget with text only: + ```python + @app.ui.openai.widget( + name="status", + template_uri="ui://widget/status.html", + html='
...' + ) + def show_status() -> str: + return "Status widget displayed" + ``` + """ + # Validate required parameters + if template_uri is None: + raise TypeError("widget() missing required keyword argument: 'template_uri'") + if html is None: + raise TypeError("widget() missing required keyword argument: 'html'") + + if isinstance(annotations, dict): + annotations = ToolAnnotations(**annotations) + + if isinstance(name_or_fn, classmethod): + raise ValueError( + inspect.cleandoc( + """ + To decorate a classmethod, first define the method and then call + widget() directly on the method instead of using it as a + decorator. See https://gofastmcp.com/patterns/decorating-methods + for examples and more information. + """ + ) + ) + + # Determine the widget name and function based on calling pattern + if inspect.isroutine(name_or_fn): + # Case 1: @widget(template_uri=..., html=...) - function passed directly + fn = name_or_fn + widget_name = name or fn.__name__ + + # Register immediately and return + return self._register_widget( + fn=fn, + identifier=widget_name, + template_uri=template_uri, + html=html, + title=title, + description=description, + invoking=invoking, + invoked=invoked, + mime_type=mime_type, + widget_description=widget_description, + widget_prefers_border=widget_prefers_border, + widget_csp_resources=widget_csp_resources, + widget_csp_connect=widget_csp_connect, + widget_accessible=widget_accessible, + result_can_produce_widget=result_can_produce_widget, + tags=tags, + annotations=cast(ToolAnnotations | None, annotations), + resource_annotations=resource_annotations, + meta=meta, + enabled=enabled, + exclude_args=exclude_args, + ) + + elif isinstance(name_or_fn, str): + # Case 2: @widget("custom_name", template_uri=..., html=...) + if name is not None: + raise TypeError( + "Cannot specify both a name as first argument and as keyword argument. " + f"Use either @widget('{name_or_fn}', ...) or @widget(name='{name}', ...), not both." + ) + widget_name = name_or_fn + elif name_or_fn is None: + # Case 3: @widget(name="name", template_uri=..., html=...) + widget_name = name + else: + raise TypeError( + f"First argument to @widget must be a function, string, or None, got {type(name_or_fn)}" + ) + + # Return decorator that will be applied to the function + def decorator(fn: AnyFunction) -> FunctionTool: + actual_name = widget_name or fn.__name__ + return self._register_widget( + fn=fn, + identifier=actual_name, + template_uri=template_uri, + html=html, + title=title, + description=description, + invoking=invoking, + invoked=invoked, + mime_type=mime_type, + widget_description=widget_description, + widget_prefers_border=widget_prefers_border, + widget_csp_resources=widget_csp_resources, + widget_csp_connect=widget_csp_connect, + widget_accessible=widget_accessible, + result_can_produce_widget=result_can_produce_widget, + tags=tags, + annotations=cast(ToolAnnotations | None, annotations), + resource_annotations=resource_annotations, + meta=meta, + enabled=enabled, + exclude_args=exclude_args, + ) + + return decorator + + def _register_widget( + self, + *, + fn: AnyFunction, + identifier: str, + template_uri: str, + html: str, + title: str | None, + description: str | None, + invoking: str | None, + invoked: str | None, + mime_type: str, + widget_description: str | None, + widget_prefers_border: bool, + widget_csp_resources: Sequence[str] | None, + widget_csp_connect: Sequence[str] | None, + widget_accessible: bool, + result_can_produce_widget: bool, + tags: set[str] | None, + annotations: ToolAnnotations | None, + resource_annotations: dict[str, Any] | None, + meta: dict[str, Any] | None, + enabled: bool | None, + exclude_args: list[str] | None, + ) -> FunctionTool: + """Internal method to register a widget. + + This orchestrates: + 1. Registering the widget HTML as a resource + 2. Wrapping the function to auto-transform return values + 3. Registering the wrapped function as a tool with OpenAI metadata + """ + # Use title or default to identifier + widget_title = title or identifier + + # Step 1: Register widget HTML as a resource + self._register_widget_resource( + template_uri=template_uri, + html=html, + title=widget_title, + mime_type=mime_type, + widget_description=widget_description, + widget_prefers_border=widget_prefers_border, + widget_csp_resources=widget_csp_resources, + widget_csp_connect=widget_csp_connect, + resource_annotations=resource_annotations, + ) + + # Step 2: Wrap function to auto-transform return values + wrapped_fn = self._wrap_widget_function(fn) + + # Step 3: Build OpenAI widget metadata for the tool + widget_meta = build_widget_tool_meta( + template_uri=template_uri, + html=html, + title=widget_title, + invoking=invoking, + invoked=invoked, + mime_type=mime_type, + widget_description=widget_description, + widget_prefers_border=widget_prefers_border, + widget_csp_resources=widget_csp_resources, + widget_csp_connect=widget_csp_connect, + widget_accessible=widget_accessible, + result_can_produce_widget=result_can_produce_widget, + additional_meta=meta, + ) + + # Merge tool annotations with widget defaults + merged_annotations = merge_tool_annotations( + annotations.model_dump() if annotations else None + ) + + # Step 4: Register as a tool with widget metadata + from fastmcp.tools.tool import Tool + + tool = Tool.from_function( + wrapped_fn, + name=identifier, + title=title, + description=description, + tags=tags, + annotations=ToolAnnotations(**merged_annotations), + exclude_args=exclude_args, + meta=widget_meta, + serializer=self._fastmcp._tool_serializer, + enabled=enabled, + ) + self._fastmcp.add_tool(tool) + return tool + + def _register_widget_resource( + self, + *, + template_uri: str, + html: str, + title: str, + mime_type: str, + widget_description: str | None, + widget_prefers_border: bool, + widget_csp_resources: Sequence[str] | None, + widget_csp_connect: Sequence[str] | None, + resource_annotations: dict[str, Any] | None, + ) -> None: + """Register widget HTML as an MCP resource.""" + # Build resource metadata + resource_meta = build_widget_resource_meta( + title=title, + widget_description=widget_description, + widget_prefers_border=widget_prefers_border, + widget_csp_resources=widget_csp_resources, + widget_csp_connect=widget_csp_connect, + ) + + # Merge resource annotations + merged_resource_annotations = merge_resource_annotations(resource_annotations) + + # Register the resource using FastMCP's resource decorator + @self._fastmcp.resource( + uri=template_uri, + name=title, + description=f"{title} widget markup", + mime_type=mime_type, + annotations=merged_resource_annotations, + meta=resource_meta, + ) + def widget_html_resource() -> str: + return html + + def _wrap_widget_function(self, fn: AnyFunction) -> AnyFunction: + """Wrap a widget function to auto-transform return values. + + The wrapper intercepts the function's return value and transforms it + to the OpenAI format: {"content": [...], "structuredContent": {...}} + + Supports sync and async functions. + """ + if asyncio.iscoroutinefunction(fn): + + @functools.wraps(fn) + async def async_wrapper(*args: Any, **kwargs: Any) -> dict[str, Any]: + result = await fn(*args, **kwargs) + return transform_widget_response(result) + + # Override return annotation to reflect actual return type + async_wrapper.__annotations__["return"] = dict[str, Any] + return async_wrapper + else: + + @functools.wraps(fn) + def sync_wrapper(*args: Any, **kwargs: Any) -> dict[str, Any]: + result = fn(*args, **kwargs) + return transform_widget_response(result) + + # Override return annotation to reflect actual return type + sync_wrapper.__annotations__["return"] = dict[str, Any] + return sync_wrapper diff --git a/src/fastmcp/ui/openai/metadata.py b/src/fastmcp/ui/openai/metadata.py new file mode 100644 index 000000000..1a0e37ad1 --- /dev/null +++ b/src/fastmcp/ui/openai/metadata.py @@ -0,0 +1,264 @@ +"""OpenAI-specific metadata builders for widgets.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Any + +# Default CSP resource domains for OpenAI widgets +DEFAULT_CSP_RESOURCES: tuple[str, ...] = ("https://persistent.oaistatic.com",) + +# Default resource annotations for widget HTML +DEFAULT_RESOURCE_ANNOTATIONS: dict[str, Any] = { + "readOnlyHint": True, + "idempotentHint": True, +} + +# Default tool annotations for widget tools +DEFAULT_TOOL_ANNOTATIONS: dict[str, Any] = { + "destructiveHint": False, + "openWorldHint": False, + "readOnlyHint": True, +} + + +def _normalize_sequence( + seq: Sequence[str] | None, default: Sequence[str] +) -> list[str]: + """Normalize a sequence to a list, using default if None.""" + return list(seq if seq is not None else default) + + +def build_widget_resource_meta( + *, + title: str, + widget_description: str | None = None, + widget_prefers_border: bool = True, + widget_csp_resources: Sequence[str] | None = None, + widget_csp_connect: Sequence[str] | None = None, +) -> dict[str, Any]: + """Build _meta dict for widget HTML resource. + + Args: + title: Widget title + widget_description: Description of widget UI (defaults to "{title} widget UI.") + widget_prefers_border: Whether widget prefers a border + widget_csp_resources: CSP resource_domains list + widget_csp_connect: CSP connect_domains list + + Returns: + Metadata dict with OpenAI widget configuration + + Examples: + Build basic widget metadata: + ```python + meta = build_widget_resource_meta(title="Pizza Map") + # { + # "openai/widgetDescription": "Pizza Map widget UI.", + # "openai/widgetPrefersBorder": True, + # "openai/widgetCSP": {...} + # } + ``` + """ + return { + "openai/widgetDescription": widget_description or f"{title} widget UI.", + "openai/widgetPrefersBorder": widget_prefers_border, + "openai/widgetCSP": { + "resource_domains": _normalize_sequence( + widget_csp_resources, DEFAULT_CSP_RESOURCES + ), + "connect_domains": _normalize_sequence(widget_csp_connect, ()), + }, + } + + +def build_embedded_widget_resource( + *, + template_uri: str, + html: str, + title: str, + mime_type: str = "text/html+skybridge", + widget_description: str | None = None, + widget_prefers_border: bool = True, + widget_csp_resources: Sequence[str] | None = None, + widget_csp_connect: Sequence[str] | None = None, +) -> dict[str, Any]: + """Build embedded widget resource structure for tool metadata. + + This creates the openai.com/widget embedded resource that ChatGPT + uses to render the widget. + + Args: + template_uri: Widget template URI (e.g. "ui://widget/pizza-map.html") + html: Widget HTML content + title: Widget title + mime_type: MIME type for the widget HTML + widget_description: Description of widget UI (defaults to "{title} widget UI.") + widget_prefers_border: Whether widget prefers a border + widget_csp_resources: CSP resource_domains list + widget_csp_connect: CSP connect_domains list + + Returns: + Embedded resource structure for tool metadata + + Examples: + Build embedded resource: + ```python + resource = build_embedded_widget_resource( + template_uri="ui://widget/map.html", + html="
...
", + title="Map Widget" + ) + ``` + """ + return { + "type": "resource", + "resource": { + "type": "text", + "uri": template_uri, + "mimeType": mime_type, + "text": html, + "title": title, + "_meta": build_widget_resource_meta( + title=title, + widget_description=widget_description, + widget_prefers_border=widget_prefers_border, + widget_csp_resources=widget_csp_resources, + widget_csp_connect=widget_csp_connect, + ), + }, + } + + +def build_widget_tool_meta( + *, + template_uri: str, + html: str, + title: str, + invoking: str | None = None, + invoked: str | None = None, + mime_type: str = "text/html+skybridge", + widget_description: str | None = None, + widget_prefers_border: bool = True, + widget_csp_resources: Sequence[str] | None = None, + widget_csp_connect: Sequence[str] | None = None, + widget_accessible: bool = True, + result_can_produce_widget: bool = True, + additional_meta: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Build _meta dict for widget tool. + + Args: + template_uri: Widget template URI (e.g. "ui://widget/pizza-map.html") + html: Widget HTML content + title: Widget title + invoking: Status message shown while tool is executing + invoked: Status message shown after tool completes + mime_type: MIME type for the widget HTML + widget_description: Description of widget UI (defaults to "{title} widget UI.") + widget_prefers_border: Whether widget prefers a border + widget_csp_resources: CSP resource_domains list + widget_csp_connect: CSP connect_domains list + widget_accessible: Whether widget is accessible + result_can_produce_widget: Whether result can produce a widget + additional_meta: Additional metadata to merge + + Returns: + Metadata dict with OpenAI widget tool configuration + + Examples: + Build widget tool metadata: + ```python + meta = build_widget_tool_meta( + template_uri="ui://widget/map.html", + html="
...
", + title="Map Widget", + invoking="Loading map...", + invoked="Map loaded!" + ) + ``` + """ + meta: dict[str, Any] = { + "openai.com/widget": build_embedded_widget_resource( + template_uri=template_uri, + html=html, + title=title, + mime_type=mime_type, + widget_description=widget_description, + widget_prefers_border=widget_prefers_border, + widget_csp_resources=widget_csp_resources, + widget_csp_connect=widget_csp_connect, + ), + "openai/outputTemplate": template_uri, + "openai/widgetAccessible": widget_accessible, + "openai/resultCanProduceWidget": result_can_produce_widget, + } + + # Add invocation status messages if provided + if invoking is not None: + meta["openai/toolInvocation/invoking"] = invoking + if invoked is not None: + meta["openai/toolInvocation/invoked"] = invoked + + # Merge additional metadata + if additional_meta: + meta.update(additional_meta) + + return meta + + +def merge_tool_annotations( + user_annotations: Mapping[str, Any] | None, +) -> dict[str, Any]: + """Merge user annotations with default widget tool annotations. + + Args: + user_annotations: User-provided annotations to merge + + Returns: + Combined annotations dict with defaults and user overrides + + Examples: + Merge annotations: + ```python + annotations = merge_tool_annotations({"customHint": True}) + # { + # "destructiveHint": False, + # "openWorldHint": False, + # "readOnlyHint": True, + # "customHint": True + # } + ``` + """ + combined = dict(DEFAULT_TOOL_ANNOTATIONS) + if user_annotations: + combined.update(user_annotations) + return combined + + +def merge_resource_annotations( + user_annotations: Mapping[str, Any] | None, +) -> dict[str, Any]: + """Merge user annotations with default widget resource annotations. + + Args: + user_annotations: User-provided annotations to merge + + Returns: + Combined annotations dict with defaults and user overrides + + Examples: + Merge annotations: + ```python + annotations = merge_resource_annotations({"customHint": True}) + # { + # "readOnlyHint": True, + # "idempotentHint": True, + # "customHint": True + # } + ``` + """ + combined = dict(DEFAULT_RESOURCE_ANNOTATIONS) + if user_annotations: + combined.update(user_annotations) + return combined diff --git a/src/fastmcp/ui/openai/response.py b/src/fastmcp/ui/openai/response.py new file mode 100644 index 000000000..72a1e1645 --- /dev/null +++ b/src/fastmcp/ui/openai/response.py @@ -0,0 +1,135 @@ +"""Response transformation utilities for OpenAI widgets.""" + +from __future__ import annotations + +from typing import Any + +WidgetToolResponse = dict[str, Any] + + +def build_widget_tool_response( + response_text: str | None = None, + structured_content: dict[str, Any] | None = None, +) -> WidgetToolResponse: + """Build a standardized OpenAI widget tool response. + + This helper is available for advanced use cases, but the widget decorator + automatically transforms simple return values (str, dict, tuple) so you + typically don't need to call this directly. + + Args: + response_text: Narrative text shown in the conversation + structured_content: Data passed to the widget JavaScript component + + Returns: + Formatted response dict with content and structuredContent + + Examples: + Build a response with both text and data: + ```python + return build_widget_tool_response( + response_text="Showing weather for Seattle", + structured_content={"city": "Seattle", "temp": 72} + ) + ``` + + Data only (no narrative): + ```python + return build_widget_tool_response( + structured_content={"data": [1, 2, 3]} + ) + ``` + """ + content = ( + [ + { + "type": "text", + "text": response_text, + } + ] + if response_text + else [] + ) + + return {"content": content, "structuredContent": structured_content or {}} + + +def transform_widget_response(result: Any) -> WidgetToolResponse: + """Auto-transform widget function return value to OpenAI format. + + Supports three return patterns: + - str: Narrative text only + - dict: Structured data for widget (no narrative) + - tuple[str, dict]: Both narrative and data + + Args: + result: The widget function's return value + + Returns: + Formatted response dict with content and structuredContent + + Raises: + TypeError: If return type is not str, dict, or tuple[str, dict] + + Examples: + Transform a string: + ```python + result = transform_widget_response("Showing map") + # {"content": [{"type": "text", "text": "Showing map"}], "structuredContent": {}} + ``` + + Transform a dict: + ```python + result = transform_widget_response({"lat": 47.6, "lon": -122.3}) + # {"content": [], "structuredContent": {"lat": 47.6, "lon": -122.3}} + ``` + + Transform a tuple: + ```python + result = transform_widget_response(("Showing map", {"lat": 47.6})) + # {"content": [{"type": "text", "text": "Showing map"}], "structuredContent": {"lat": 47.6}} + ``` + """ + if isinstance(result, str): + return build_widget_tool_response( + response_text=result if result else None, + structured_content=None, + ) + elif isinstance(result, dict): + return build_widget_tool_response( + response_text=None, + structured_content=result, + ) + elif isinstance(result, tuple): + if len(result) != 2: + raise TypeError( + f"Widget function returned tuple of length {len(result)}, expected 2 (text, data). " + f"Return either: str (text only), dict (data only), or tuple[str, dict] (both)" + ) + + text, data = result + + if text is not None and not isinstance(text, str): + raise TypeError( + f"First element of tuple must be str or None, got {type(text).__name__}. " + f"Return format: tuple[str | None, dict[str, Any]]" + ) + + if data is not None and not isinstance(data, dict): + raise TypeError( + f"Second element of tuple must be dict or None, got {type(data).__name__}. " + f"Return format: tuple[str | None, dict[str, Any]]" + ) + + return build_widget_tool_response( + response_text=text, + structured_content=data, + ) + else: + raise TypeError( + f"Widget function must return str, dict, or tuple[str, dict], got {type(result).__name__}. " + f"Examples:\n" + f" - return 'Showing map' # Text only\n" + f" - return {{'lat': 47.6}} # Data only\n" + f" - return ('Showing map', {{'lat': 47.6}}) # Both" + ) diff --git a/tests/ui/__init__.py b/tests/ui/__init__.py new file mode 100644 index 000000000..498a7a02f --- /dev/null +++ b/tests/ui/__init__.py @@ -0,0 +1 @@ +"""Tests for UI components.""" diff --git a/tests/ui/test_openai_widget.py b/tests/ui/test_openai_widget.py new file mode 100644 index 000000000..cd9e972a9 --- /dev/null +++ b/tests/ui/test_openai_widget.py @@ -0,0 +1,93 @@ +"""Tests for OpenAI widget decorator (basic usage).""" + +from fastmcp import FastMCP + + +async def test_widget_decorator_basic(): + """Test basic widget decorator usage.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="my_widget", + template_uri="ui://widget/test.html", + html="
Test Widget
", + ) + def my_widget(x: int) -> str: + return f"Result: {x}" + + # Widget should be registered as a tool + tools = await app.get_tools() + assert "my_widget" in tools + assert tools["my_widget"].name == "my_widget" + + +async def test_widget_decorator_with_name(): + """Test widget decorator with custom name.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="custom_widget", + template_uri="ui://widget/custom.html", + html="
Custom
", + ) + def my_function(x: int) -> str: + return f"Result: {x}" + + tools = await app.get_tools() + assert "custom_widget" in tools + assert tools["custom_widget"].name == "custom_widget" + + +async def test_widget_decorator_with_description(): + """Test widget decorator with description.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="my_widget", + template_uri="ui://widget/test.html", + html="
Test
", + description="This is a test widget", + ) + def my_widget(x: int) -> str: + return f"Result: {x}" + + tools = await app.get_tools() + assert "my_widget" in tools + assert tools["my_widget"].description == "This is a test widget" + + +async def test_widget_decorator_with_title(): + """Test widget decorator with title.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="my_widget", + template_uri="ui://widget/test.html", + html="
Test
", + title="My Widget Title", + ) + def my_widget(x: int) -> str: + return f"Result: {x}" + + tools = await app.get_tools() + assert "my_widget" in tools + assert tools["my_widget"].title == "My Widget Title" + + +async def test_widget_can_be_called(): + """Test that registered widgets can be called as tools.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="calculator", + template_uri="ui://widget/calc.html", + html="
Calc
", + ) + def calculator(x: int, y: int) -> dict: + return {"result": x + y} + + # Call the tool through the tool manager + result = await app._tool_manager.call_tool("calculator", {"x": 5, "y": 3}) + + # The result should be auto-transformed + assert '"structuredContent":{"result":8}' in result.content[0].text # type: ignore[attr-defined] diff --git a/tests/ui/test_openai_widget_full.py b/tests/ui/test_openai_widget_full.py new file mode 100644 index 000000000..69ada5b8f --- /dev/null +++ b/tests/ui/test_openai_widget_full.py @@ -0,0 +1,357 @@ +"""Comprehensive tests for OpenAI widget decorator with full integration.""" + +import pytest + +from fastmcp import FastMCP +from fastmcp.ui.openai import build_widget_tool_response + + +async def test_widget_with_dict_return(): + """Test widget that returns dict (structured data only).""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="test-widget", + template_uri="ui://widget/test.html", + html="
Test
", + ) + def test_widget(value: str) -> dict: + return {"value": value} + + # Tool should be registered + tools = await app.get_tools() + assert "test-widget" in tools + + # Call the tool + result = await app._tool_manager.call_tool("test-widget", {"value": "hello"}) + + # Should be transformed to OpenAI format (compact JSON) + assert '"content":[]' in result.content[0].text # type: ignore[attr-defined] + assert '"structuredContent":{"value":"hello"}' in result.content[0].text # type: ignore[attr-defined] + + +async def test_widget_with_str_return(): + """Test widget that returns str (text only).""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="text-widget", + template_uri="ui://widget/text.html", + html="
Text
", + ) + def text_widget() -> str: + return "Hello world" + + # Call the tool + result = await app._tool_manager.call_tool("text-widget", {}) + + # Should be transformed to OpenAI format with text in content (compact JSON) + assert '"type":"text"' in result.content[0].text # type: ignore[attr-defined] + assert '"text":"Hello world"' in result.content[0].text # type: ignore[attr-defined] + assert '"content":[{' in result.content[0].text # type: ignore[attr-defined] + + +async def test_widget_with_tuple_return(): + """Test widget that returns tuple (text and data).""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="combo-widget", + template_uri="ui://widget/combo.html", + html="
Combo
", + ) + def combo_widget(x: int) -> tuple[str, dict]: + return (f"Processing {x}", {"value": x, "doubled": x * 2}) + + # Call the tool + result = await app._tool_manager.call_tool("combo-widget", {"x": 5}) + + # Should have both content and structuredContent (compact JSON) + response_text = result.content[0].text # type: ignore[attr-defined] + assert '"text":"Processing 5"' in response_text + assert '"structuredContent":{"value":5,"doubled":10}' in response_text + + +async def test_widget_with_async_function(): + """Test widget decorator works with async functions.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="async-widget", + template_uri="ui://widget/async.html", + html="
Async
", + ) + async def async_widget(data: str) -> dict: + # Simulate async operation + import asyncio + await asyncio.sleep(0.001) + return {"processed": data.upper()} + + # Call the tool + result = await app._tool_manager.call_tool("async-widget", {"data": "test"}) + + # Should be transformed correctly (compact JSON) + assert '"structuredContent":{"processed":"TEST"}' in result.content[0].text # type: ignore[attr-defined] + + +async def test_widget_resource_registration(): + """Test that widget HTML is registered as a resource.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="widget-with-resource", + template_uri="ui://widget/test-resource.html", + html="
Widget HTML
", + title="Test Resource Widget", + ) + def widget_func() -> dict: + return {} + + # Resource should be registered + resources = await app.get_resources() + assert "ui://widget/test-resource.html" in resources + + # Read the resource + resource = resources["ui://widget/test-resource.html"] + assert resource.mime_type == "text/html+skybridge" + assert resource.name == "Test Resource Widget" + + +async def test_widget_metadata(): + """Test that widget has correct OpenAI metadata.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="meta-widget", + template_uri="ui://widget/meta.html", + html="
Meta
", + invoking="Loading widget...", + invoked="Widget loaded!", + ) + def meta_widget() -> dict: + return {} + + # Get the tool + tools = await app.get_tools() + tool = tools["meta-widget"] + + # Check OpenAI metadata + assert tool.meta is not None + assert "openai/outputTemplate" in tool.meta + assert tool.meta["openai/outputTemplate"] == "ui://widget/meta.html" + assert tool.meta["openai/toolInvocation/invoking"] == "Loading widget..." + assert tool.meta["openai/toolInvocation/invoked"] == "Widget loaded!" + assert tool.meta["openai/widgetAccessible"] is True + assert tool.meta["openai/resultCanProduceWidget"] is True + + +async def test_widget_csp_configuration(): + """Test widget CSP configuration in metadata.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="csp-widget", + template_uri="ui://widget/csp.html", + html="
CSP
", + widget_csp_resources=["https://custom.com", "https://another.com"], + widget_csp_connect=["wss://websocket.com"], + ) + def csp_widget() -> dict: + return {} + + # Get the resource to check its metadata + resources = await app.get_resources() + resource = resources["ui://widget/csp.html"] + + # Check CSP in resource metadata + assert resource.meta is not None + assert "openai/widgetCSP" in resource.meta + csp = resource.meta["openai/widgetCSP"] + assert "https://custom.com" in csp["resource_domains"] + assert "https://another.com" in csp["resource_domains"] + assert "wss://websocket.com" in csp["connect_domains"] + + +async def test_widget_with_title_and_description(): + """Test widget with custom title and description.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="titled-widget", + template_uri="ui://widget/titled.html", + html="
Title
", + title="My Custom Title", + description="Custom description for the widget", + ) + def titled_widget() -> dict: + return {} + + # Get the tool + tools = await app.get_tools() + tool = tools["titled-widget"] + + assert tool.title == "My Custom Title" + assert tool.description == "Custom description for the widget" + + +async def test_widget_with_tags(): + """Test widget with tags.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="tagged-widget", + template_uri="ui://widget/tagged.html", + html="
Tags
", + tags={"visualization", "data"}, + ) + def tagged_widget() -> dict: + return {} + + # Get the tool + tools = await app.get_tools() + tool = tools["tagged-widget"] + + assert tool.tags == {"visualization", "data"} + + +async def test_widget_with_enabled_false(): + """Test widget can be disabled.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="disabled-widget", + template_uri="ui://widget/disabled.html", + html="
Disabled
", + enabled=False, + ) + def disabled_widget() -> dict: + return {} + + # Get the tool + tools = await app.get_tools() + tool = tools["disabled-widget"] + + assert tool.enabled is False + + +async def test_widget_with_custom_name(): + """Test widget with custom name different from function name.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="custom-id", + template_uri="ui://widget/custom.html", + html="
Custom
", + ) + def some_function_name() -> dict: + return {"test": "data"} + + # Should be registered with custom name + tools = await app.get_tools() + assert "custom-id" in tools + assert "some_function_name" not in tools + + +async def test_widget_annotations(): + """Test widget has correct default annotations.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="annotated-widget", + template_uri="ui://widget/annotated.html", + html="
Annotated
", + ) + def annotated_widget() -> dict: + return {} + + # Get the tool + tools = await app.get_tools() + tool = tools["annotated-widget"] + + # Check default widget annotations + assert tool.annotations is not None + assert tool.annotations.destructiveHint is False + assert tool.annotations.readOnlyHint is True + + +async def test_build_widget_tool_response_helper(): + """Test the build_widget_tool_response helper function.""" + # Text only + result1 = build_widget_tool_response(response_text="Hello") + assert result1["content"] == [{"type": "text", "text": "Hello"}] + assert result1["structuredContent"] == {} + + # Data only + result2 = build_widget_tool_response(structured_content={"key": "value"}) + assert result2["content"] == [] + assert result2["structuredContent"] == {"key": "value"} + + # Both + result3 = build_widget_tool_response( + response_text="Processing...", + structured_content={"status": "done"} + ) + assert result3["content"] == [{"type": "text", "text": "Processing..."}] + assert result3["structuredContent"] == {"status": "done"} + + +async def test_widget_missing_required_params(): + """Test that widget decorator validates required parameters.""" + app = FastMCP("test") + + with pytest.raises(TypeError, match="missing required keyword argument: 'template_uri'"): + @app.ui.openai.widget( + html="
Missing URI
", + ) + def bad_widget1() -> dict: + return {} + + with pytest.raises(TypeError, match="missing required keyword argument: 'html'"): + @app.ui.openai.widget( + template_uri="ui://widget/test.html", + ) + def bad_widget2() -> dict: + return {} + + +async def test_widget_invalid_return_type(): + """Test that widget validates return types at runtime.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="bad-return-widget", + template_uri="ui://widget/bad.html", + html="
Bad
", + ) + def bad_return_widget() -> int: # Invalid return type + return 42 + + # Calling the tool should raise an error during response transformation + try: + await app._tool_manager.call_tool("bad-return-widget", {}) + assert False, "Should have raised TypeError" + except Exception as e: + # The error gets wrapped, so check the error message + assert "must return str, dict, or tuple" in str(e) + + +async def test_widget_with_exclude_args(): + """Test widget with excluded arguments.""" + app = FastMCP("test") + + @app.ui.openai.widget( + name="exclude-args-widget", + template_uri="ui://widget/exclude.html", + html="
Exclude
", + exclude_args=["secret_param"], + ) + def exclude_args_widget(public_param: str, secret_param: str = "hidden") -> dict: + return {"public": public_param, "secret": secret_param} + + # Get the tool + tools = await app.get_tools() + tool = tools["exclude-args-widget"] + + # Check that secret_param is excluded from schema + assert "secret_param" not in tool.parameters["properties"] + assert "public_param" in tool.parameters["properties"]