feat: inspector v1 with light/dark mode, tools, prompts, resources and ping functionality

.gitignore

@@ -0,0 +1,23 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+
+# Virtual Environments
+.venv/
+venv/
+env/
+.env
+
+# Distribution / Packaging
+dist/
+build/
+*.egg-info/
+
+# IDEs
+.vscode/
+.idea/
+
+# Misc
+.DS_Store
+.python-version

README.md

@@ -11,4 +11,41 @@ license: mit
 short_description: Debug and test remote MCP servers via Gradio.
 ---
 
+# Gradio MCP Inspector
+
+A powerful GUI-based inspector for the [Model Context Protocol (MCP)](https://modelcontextprotocol.io/), built with [Gradio](https://gradio.app/). This tool allows developers to interactively explore, debug, and test MCP servers.
+
+## Features
+
+- **🔌 Flexible Connection**: Support for both **Streamable HTTP** and **SSE** (Server-Sent Events) transports.
+- **📂 Resources**: Browse available resources, view templates, and inspect content.
+- **💬 Prompts**: List prompts and execute them with dynamically generated input forms.
+- **🛠️ Tools**: Discover tools and run them with schema-based argument inputs.
+- **🕵️ Debugging**: View raw JSON-RPC requests and responses for every interaction.
+- **⚙️ Configurable**: Adjustable timeouts and connection settings.
+
+## Installation
+
+1. Clone the repository:
+   ```bash
+   git clone https://github.com/yourusername/gradio-mcp-inspector.git
+   cd gradio-mcp-inspector
+   ```
+
+2. Install dependencies:
+   ```bash
+   pip install -r requirements.txt
+   ```
+
+## Usage
+
+Run the application locally:
+
+```bash
+python app.py
+```
+
+Open your browser and navigate to the provided URL (typically `http://127.0.0.1:7860`).
+
+
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,492 @@
+from __future__ import annotations
+
+import gradio as gr
+import json
+
+from handlers import (
+    connect,
+    disconnect,
+    list_resources_with_history,
+    list_resource_templates_with_history,
+    on_resource_select,
+    list_prompts_with_history,
+    invoke_prompt_with_history,
+    list_tools_with_history,
+    invoke_tool_with_history,
+    ping_with_history,
+    custom_request_with_history,
+    get_sampling_log,
+)
+
+with gr.Blocks(title="Gradio MCP Inspector") as app:
+
+    gr.Markdown("# Gradio MCP Inspector")
+    server_url_state = gr.State("")
+    history_state = gr.State([])
+    tools_state = gr.State([])
+    resources_state = gr.State([])
+    templates_state = gr.State([])
+    prompts_state = gr.State([])
+
+    with gr.Row(equal_height=True):
+        with gr.Column(scale=1, min_width=320):
+            transport_type = gr.Dropdown(
+                ["Streamable HTTP","SSE"],
+                label="Transport Type",
+                value="Streamable HTTP",
+            )
+            base_url_input = gr.Textbox(
+                label="URL",
+                placeholder="https://your-mcp-server.example/mcp",
+            )
+
+            with gr.Accordion("Authentication", open=False):
+                header_name = gr.Textbox(label="Header Name", placeholder="Authorization")
+                bearer_token = gr.Textbox(label="Bearer Token", placeholder="Bearer Token",type="password")
+
+            with gr.Accordion("Configuration", open=False):
+                request_timeout = gr.Number(label="Request Timeout", value=10000)
+                reset_timeout_on_progress = gr.Dropdown(
+                    label="Reset Timeout on Progress", 
+                    choices=["True", "False"], 
+                    value="True"
+                )
+                max_total_timeout = gr.Number(label="Maximum Total Timeout", value=60000)
+
+            status_badge = gr.Markdown("**Status:** 🔴 Disconnected.")
+            initial_connect_btn = gr.Button("Connect", variant="primary")
+            reconnect_btn = gr.Button("Reconnect", variant="primary", visible=False)
+            disconnect_btn = gr.Button("Disconnect", variant="stop", visible=False)
+
+            initial_connect_btn.click(
+                connect,
+                inputs=[transport_type, base_url_input, header_name, bearer_token, request_timeout, reset_timeout_on_progress, max_total_timeout],
+                outputs=[server_url_state, status_badge, initial_connect_btn, reconnect_btn, disconnect_btn],
+            )
+            reconnect_btn.click(
+                connect,
+                inputs=[transport_type, base_url_input, header_name, bearer_token, request_timeout, reset_timeout_on_progress, max_total_timeout],
+                outputs=[server_url_state, status_badge, initial_connect_btn, reconnect_btn, disconnect_btn],
+            )
+            disconnect_btn.click(
+                disconnect,
+                outputs=[server_url_state, status_badge, initial_connect_btn, reconnect_btn, disconnect_btn],
+            )
+
+            # Add spacing
+            gr.HTML("<div style='margin-top: 13em;'></div>")
+
+
+            theme_selector = gr.Radio(
+                ["Light", "Dark"],
+                value="Dark",
+                label="Theme",
+                interactive=True,
+            )
+            gr.Markdown("Connected indicator updates once you reconnect.")
+
+        with gr.Column(scale=3, min_width=640):
+            with gr.Tabs():
+                with gr.Tab("Resources"):
+                    with gr.Row():
+                        with gr.Column():
+                            gr.Markdown("### Resources")
+                            list_resources_btn = gr.Button("List Resources")
+                            clear_resources_btn = gr.Button("Clear")
+                            resource_list = gr.Radio(label="Available Resources", choices=[], interactive=True)
+                            resource_empty_msg = gr.Markdown("⚠️ No resources found.", visible=False)
+
+                        with gr.Column():
+                            gr.Markdown("### Resource Templates")
+                            list_templates_btn = gr.Button("List Templates")
+                            clear_templates_btn = gr.Button("Clear")
+                            template_list = gr.Radio(label="Available Templates", choices=[], interactive=True)
+                            template_empty_msg = gr.Markdown("⚠️ No templates found.", visible=False)
+
+                        with gr.Column():
+                            gr.Markdown("### Select a resource or template")
+                            # resource_content_info = gr.Markdown("Select a resource or template from the list to view its contents")
+                            # resource_content_view = gr.Code(label="Resource Content", language="json", visible=False)
+
+                    with gr.Accordion("Debug Info", open=False):
+                        resource_list_request = gr.Code(label="List Request", language="json")
+                        resource_list_response = gr.Code(label="List Response", language="json")
+                        with gr.Column(scale=2):
+                            @gr.render(inputs=[resource_list, resources_state])
+                            def render_resource_details(uri, resources):
+                                if not uri or not resources:
+                                    return
+                                res = next((r for r in resources if r["uri"] == uri), None)
+                                if res:
+                                    gr.Markdown(f"### {res.get('name', 'Resource')}")
+                                    if res.get("description"):
+                                        gr.Markdown(res["description"])
+                                    if res.get("mimeType"):
+                                        gr.Markdown(f"**MIME Type:** `{res['mimeType']}`")
+
+                            resource_content_view = gr.Code(label="Resource Content", language="json", visible=False)
+                            resource_read_request = gr.Code(label="Read Request", language="json", visible=False)
+                            resource_read_response = gr.Code(label="Read Response", language="json", visible=False)
+
+                    with gr.Accordion("Resource Templates", open=False):
+                        with gr.Row():
+                            template_list = gr.Radio(label="Templates", choices=[], visible=False)
+                            template_empty_msg = gr.Markdown("⚠️ No templates found.", visible=False)
+                        
+                        @gr.render(inputs=[template_list, templates_state])
+                        def render_template_details(uri_template, templates):
+                            if not uri_template or not templates:
+                                return
+                            tmpl = next((t for t in templates if t["uriTemplate"] == uri_template), None)
+                            if tmpl:
+                                gr.Markdown(f"### {tmpl.get('name', 'Template')}")
+                                if tmpl.get("description"):
+                                    gr.Markdown(tmpl["description"])
+                                if tmpl.get("mimeType"):
+                                    gr.Markdown(f"**MIME Type:** `{tmpl['mimeType']}`")
+                                gr.Markdown(f"**URI Template:** `{tmpl['uriTemplate']}`")
+
+                with gr.Tab("Prompts"):
+                    with gr.Row():
+                        with gr.Column(scale=1):
+                            gr.Markdown("### Prompts")
+                            list_prompts_btn = gr.Button("List Prompts")
+                            clear_prompts_btn = gr.Button("Clear")
+                            prompt_list = gr.Radio(label="Available Prompts", choices=[], interactive=True)
+                            prompt_empty_msg = gr.Markdown("⚠️ No prompts found.", visible=False)
+
+                        with gr.Column(scale=2):
+                            @gr.render(inputs=[prompt_list, prompts_state])
+                            def render_prompt_form(prompt_name, prompts):
+                                if not prompt_name or not prompts:
+                                    gr.Markdown("### Select a prompt")
+                                    gr.Markdown("Select a prompt from the list to view its details and invoke it.")
+                                    return
+                                
+                                prompt = next((p for p in prompts if p["name"] == prompt_name), None)
+                                if not prompt:
+                                    return
+
+                                gr.Markdown(f"### {prompt['name']}")
+                                gr.Markdown(prompt.get("description", "No description provided."))
+                                
+                                arguments = prompt.get("arguments", [])
+                                inputs = {}
+                                
+                                with gr.Group():
+                                    for arg in arguments:
+                                        arg_name = arg["name"]
+                                        label = arg_name + (" *" if arg.get("required") else "")
+                                        desc = arg.get("description", "")
+                                        # Prompts usually take string arguments
+                                        inputs[arg_name] = gr.Textbox(label=label, placeholder=desc)
+
+                                    run_btn = gr.Button("Get Prompt", variant="primary")
+                                
+                                async def wrapper(base_url, timeout, history, *form_values):
+                                    args = {}
+                                    keys = list(inputs.keys())
+                                    for i, key in enumerate(keys):
+                                        val = form_values[i]
+                                        if val:
+                                            args[key] = val
+                                    # Convert timeout (ms) to seconds for the handler
+                                    timeout_sec = float(timeout) / 1000.0 if timeout else 30.0
+                                    request, response, history, rendered = await invoke_prompt_with_history(base_url, timeout_sec, prompt_name, json.dumps(args), history)
+                                    return (
+                                        gr.update(value=response, visible=True),
+                                        history,
+                                        rendered
+                                    )
+
+                                run_btn.click(
+                                    wrapper,
+                                    inputs=[server_url_state, request_timeout, history_state] + list(inputs.values()),
+                                    outputs=[prompt_call_response, history_state, history_panel]
+                                )
+
+                            prompt_call_response = gr.Markdown(label="Result", visible=False)
+                    
+                    with gr.Accordion("Debug Info", open=False):
+                        prompt_list_request = gr.Code(label="List Request", language="json")
+                        prompt_list_response = gr.Code(label="List Response", language="json")
+                        prompt_call_request = gr.Code(label="Invocation Request", language="json")
+                with gr.Tab("Tools"):
+                    with gr.Row():
+                        with gr.Column(scale=1):
+                            gr.Markdown("### Tools")
+                            list_tools_btn = gr.Button("List Tools")
+                            clear_tools_btn = gr.Button("Clear")
+                            tool_list = gr.Radio(label="Available Tools", choices=[], interactive=True)
+                            tool_empty_msg = gr.Markdown("⚠️ No tools found.", visible=False)
+
+                        with gr.Column(scale=2):
+                            @gr.render(inputs=[tool_list, tools_state])
+                            def render_tool_form(tool_name, tools):
+                                if not tool_name or not tools:
+                                    gr.Markdown("### Select a tool")
+                                    gr.Markdown("Select a tool from the list to view its details and run it")
+                                    return
+                                
+                                tool = next((t for t in tools if t["name"] == tool_name), None)
+                                if not tool:
+                                    return
+
+                                gr.Markdown(f"### {tool['name']}")
+                                gr.Markdown(tool.get("description", "No description provided."))
+                                
+                                schema = tool.get("inputSchema", {})
+                                properties = schema.get("properties", {})
+                                required = schema.get("required", [])
+                                
+                                inputs = {}
+                                
+                                with gr.Group():
+                                    for prop_name, prop_schema in properties.items():
+                                        label = prop_name + (" *" if prop_name in required else "")
+                                        prop_type = prop_schema.get("type", "string")
+                                        desc = prop_schema.get("description", "")
+                                        default_val = prop_schema.get("default", None)
+                                        enum_vals = prop_schema.get("enum", None)
+                                        
+                                        if enum_vals:
+                                            inputs[prop_name] = gr.Dropdown(
+                                                label=label, 
+                                                choices=enum_vals, 
+                                                value=default_val, 
+                                                info=desc
+                                            )
+                                        elif prop_type == "string":
+                                            inputs[prop_name] = gr.Textbox(
+                                                label=label, 
+                                                placeholder=desc, 
+                                                value=default_val if default_val is not None else ""
+                                            )
+                                        elif prop_type == "integer" or prop_type == "number":
+                                            inputs[prop_name] = gr.Number(
+                                                label=label, 
+                                                info=desc, 
+                                                value=default_val
+                                            )
+                                        elif prop_type == "boolean":
+                                            inputs[prop_name] = gr.Checkbox(
+                                                label=label, 
+                                                info=desc, 
+                                                value=default_val if default_val is not None else False
+                                            )
+                                        else:
+                                            # For arrays/objects, try to show default as JSON if exists
+                                            def_json = json.dumps(default_val) if default_val is not None else "{}"
+                                            inputs[prop_name] = gr.Code(
+                                                label=label + " (JSON)", 
+                                                language="json", 
+                                                value=def_json
+                                            )
+
+                                    run_btn = gr.Button("Run Tool", variant="primary")
+                                
+                                async def wrapper(base_url, timeout, history, *form_values):
+                                    args = {}
+                                    keys = list(inputs.keys())
+                                    for i, key in enumerate(keys):
+                                        val = form_values[i]
+                                        # Only include if it has a value or is required? 
+                                        # For now, include if not None/Empty string to avoid sending empty strings for optional fields
+                                        if val is not None and val != "":
+                                            args[key] = val
+                                    
+                                    # Convert timeout (ms) to seconds
+                                    timeout_sec = float(timeout) / 1000.0 if timeout else 30.0
+                                    return await invoke_tool_with_history(base_url, timeout_sec, tool_name, json.dumps(args), history)
+
+                                run_btn.click(
+                                    wrapper,
+                                    inputs=[server_url_state, request_timeout, history_state] + list(inputs.values()),
+                                    outputs=[tool_call_request, tool_call_response, history_state, history_panel]
+                                )
+                            tool_call_response = gr.Markdown(label="Tool Result")
+                    with gr.Accordion("Debug Info", open=False):
+                        tool_list_request = gr.Code(label="List Request", language="json")
+                        tool_list_response = gr.Code(label="List Response", language="json")
+                        tool_call_request = gr.Code(label="Invocation Request", language="json") # Moved outside render block
+
+                with gr.Tab("Ping"):
+                    with gr.Row():
+                        gr.Column(scale=1)
+                        with gr.Column(scale=0, min_width=150):
+                            ping_btn = gr.Button("Ping Server", variant="primary")
+                        gr.Column(scale=1)
+                with gr.Tab("Sampling"):
+                    gr.Markdown("When the server requests LLM sampling, requests will appear here for approval.")
+                    
+                    gr.Markdown("### Recent Requests")
+                    
+                    sampling_log_display = gr.Markdown("No pending requests")
+                    
+                    with gr.Row():
+                        refresh_log_btn = gr.Button("Refresh", variant="primary")
+                    
+                    refresh_log_btn.click(
+                        lambda: get_sampling_log(),
+                        outputs=[sampling_log_display]
+                    )
+
+                with gr.Tab("Roots"):
+                    gr.Markdown("Display root resources reported by the server here.")
+
+                with gr.Tab("Auth"):
+                    gr.Markdown("Use the Authentication accordion on the left to manage credentials.")
+
+            with gr.Row():
+                with gr.Column():
+                    gr.Markdown("### History")
+                    history_panel = gr.HTML(value="<p><em>No calls yet</em></p>")
+                with gr.Column():
+                    gr.Markdown("### Server Notifications")
+                    notifications_panel = gr.Markdown(value="_No notifications yet_")
+
+    # Wiring (done after layout so every component is defined)
+    list_resources_btn.click(
+        list_resources_with_history,
+        inputs=[server_url_state, request_timeout, history_state],
+        outputs=[
+            resource_list_request,
+            resource_list_response,
+            resource_list,
+            resource_empty_msg,
+            resources_state,
+            history_state,
+            history_panel,
+        ],
+    )
+
+    list_templates_btn.click(
+        list_resource_templates_with_history,
+        inputs=[server_url_state, request_timeout, history_state],
+        outputs=[
+            resource_list_request,
+            resource_list_response,
+            template_list,
+            template_empty_msg,
+            templates_state,
+            history_state,
+            history_panel,
+        ],
+    )
+
+    resource_list.change(
+        on_resource_select,
+        inputs=[server_url_state, request_timeout, resource_list, history_state],
+        outputs=[
+            resource_read_request,
+            resource_read_response,
+            resource_content_view,
+            history_state,
+            history_panel,
+        ],
+    )
+
+    clear_resources_btn.click(
+        lambda: (gr.update(choices=[], value=None, visible=True), gr.update(value="", visible=False), gr.update(visible=False)),
+        outputs=[resource_list, resource_content_view, resource_empty_msg]
+    )
+
+    clear_templates_btn.click(
+        lambda: (gr.update(choices=[], value=None, visible=True), gr.update(visible=False)),
+        outputs=[template_list, template_empty_msg]
+    )
+
+    list_prompts_btn.click(
+        list_prompts_with_history,
+        inputs=[server_url_state, request_timeout, history_state],
+        outputs=[
+            prompt_list_request,
+            prompt_list_response,
+            prompt_list,
+            prompt_empty_msg,
+            prompts_state,
+            history_state,
+            history_panel,
+        ],
+    )
+
+    # prompt_list.change(
+    #     lambda: (gr.update(visible=False), gr.update(visible=True)),
+    #     outputs=[prompt_instruction, prompt_details]
+    # )
+
+    # invoke_prompt_btn.click(
+    #     invoke_prompt_with_history,
+    #     inputs=[
+    #         server_url_state,
+    #         timeout_seconds,
+    #         prompt_list,
+    #         prompt_args,
+    #         history_state,
+    #     ],
+    #     outputs=[
+    #         prompt_call_request,
+    #         prompt_call_response,
+    #         history_state,
+    #         history_panel,
+    #     ],
+    # )
+
+    clear_prompts_btn.click(
+        lambda: (gr.update(choices=[], value=None), gr.update(visible=False)),
+        outputs=[prompt_list, prompt_empty_msg],
+    )
+
+    list_tools_btn.click(
+        list_tools_with_history,
+        inputs=[server_url_state, request_timeout, history_state],
+        outputs=[
+            tool_list_request,
+            tool_list_response,
+            tool_list,
+            tool_empty_msg,
+            tools_state,
+            history_state,
+            history_panel,
+        ],
+    )
+
+    clear_tools_btn.click(
+        lambda: (gr.update(choices=[], value=None), gr.update(visible=False)),
+        outputs=[tool_list, tool_empty_msg],
+    )
+
+    ping_btn.click(
+        ping_with_history,
+        inputs=[server_url_state, request_timeout, history_state],
+        outputs=[history_state, history_panel],
+    )
+
+    theme_selector.change(
+        None,
+        inputs=theme_selector,
+        js="""(theme) => {
+            if (theme === "Dark") {
+                document.body.classList.add('dark');
+            } else {
+                document.body.classList.remove('dark');
+            }
+        }"""
+    )
+
+    # send_btn.click(
+    #     custom_request_with_history,
+    #     inputs=[server_url_state, method, params, timeout_seconds, history_state],
+    #     outputs=[
+    #         custom_request_box,
+    #         custom_response_box,
+    #         history_state,
+    #         history_panel,
+    #     ],
+    # )
+
+from theme import CustomLightTheme
+
+if __name__ == "__main__":
+    app.launch(theme=gr.themes.Soft())
+    # app.launch(theme=CustomLightTheme())

handlers.py

@@ -0,0 +1,247 @@
+from __future__ import annotations
+
+import gradio as gr
+import json
+from fastmcp.client.sampling import SamplingMessage, SamplingParams, RequestContext
+
+from mcp_client import (
+    connect as mcp_connect,
+    disconnect as mcp_disconnect,
+    invoke_prompt,
+    invoke_tool,
+    list_prompts,
+    list_resources,
+    list_resource_templates,
+    list_tools,
+    ping_server,
+    read_resource,
+    send_custom_request,
+)
+
+
+def _render_history(history: list[dict]) -> str:
+    if not history:
+        return "_No calls yet_"
+
+    html_parts = []
+    total_items = len(history)
+    for idx, entry in enumerate(history):
+        display_num = total_items - idx
+        method = entry["method"]
+        request = entry["request"]
+        response = entry["response"]
+        
+        # Create an accordion-style HTML using details/summary
+        html = f"""
+        <details style="margin-bottom: 10px; border: 1px solid #ddd; border-radius: 4px; padding: 10px;">
+            <summary style="cursor: pointer; font-weight: bold; user-select: none;">
+                {display_num}. {method}
+            </summary>
+            <div style="margin-top: 10px;">
+                <div style="margin-bottom: 10px;">
+                    <strong>Request:</strong>
+                    <pre style="background: #333333; color: #ffffff; padding: 10px; border-radius: 4px; overflow-x: auto; max-height: 300px;">{request}</pre>
+                </div>
+                <div>
+                    <strong>Response:</strong>
+                    <pre style="background: #333333; color: #ffffff; padding: 10px; border-radius: 4px; overflow-x: auto; max-height: 300px;">{response}</pre>
+                </div>
+            </div>
+        </details>
+        """
+        html_parts.append(html)
+    
+    return "\n".join(html_parts)
+
+
+def _update_history(history: list[dict] | None, method: str, request: str, response: str):
+    history = history or []
+    history.insert(
+        0,
+        {
+            "method": method,
+            "request": request,
+            "response": response,
+        },
+    )
+    return history, _render_history(history)
+
+
+async def connect(transport: str, url: str, header_name: str, token: str, request_timeout: float, reset_timeout: str, max_timeout: float):
+    cleaned_url = url.strip()
+    if not cleaned_url:
+        return (
+            "",
+            "**Status:** ⚠️ Provide a server URL.",
+            gr.update(visible=True),
+            gr.update(visible=False),
+            gr.update(visible=False),
+        )
+
+    try:
+        # Convert ms to seconds for the client
+        timeout_sec = float(request_timeout) / 1000.0 if request_timeout else 10.0
+        # Establish persistent connection
+        await mcp_connect(cleaned_url, timeout_sec, transport, sampling_handler)
+    except Exception as e:
+        return (
+            cleaned_url,
+            f"**Status:** 🔴 Connection failed: {str(e)}",
+            gr.update(visible=True),
+            gr.update(visible=False),
+            gr.update(visible=False),
+        )
+
+    badge = f"**Status:** 🟢 Connected via {transport}."
+    if token.strip():
+        badge += " (bearer token applied)"
+    elif header_name.strip():
+        badge += f" ({header_name} header applied)"
+    return (
+        cleaned_url,
+        badge,
+        gr.update(visible=False),
+        gr.update(visible=True),
+        gr.update(visible=True),
+    )
+
+
+async def disconnect():
+    await mcp_disconnect()
+    return (
+        "",
+        "**Status:** 🔴 Disconnected.",
+        gr.update(visible=True),
+        gr.update(visible=False),
+        gr.update(visible=False),
+    )
+
+
+async def custom_request_with_history(base_url, method, params, timeout, history):
+    request, response = await send_custom_request(base_url, method, params, timeout)
+    history, rendered = _update_history(history, method or "custom", request, response)
+    return request, response, history, rendered
+
+
+async def list_tools_with_history(base_url, timeout, history):
+    request, response, tools_data = await list_tools(base_url, timeout, sampling_handler)
+    names = [t["name"] for t in tools_data]
+    update = gr.update(choices=names, value=None, visible=bool(names))
+    empty_msg_update = gr.update(visible=not names)
+    history, rendered = _update_history(history, "tools/list", request, response)
+    return request, response, update, empty_msg_update, tools_data, history, rendered
+
+
+async def invoke_tool_with_history(base_url, timeout, tool_name, args, history):
+    request, response = await invoke_tool(base_url, timeout, tool_name, args, sampling_handler)
+    history, rendered = _update_history(history, "tools/call", request, response)
+    return request, response, history, rendered
+
+
+async def list_prompts_with_history(base_url, timeout, history):
+    request, response, prompts_data = await list_prompts(base_url, timeout, sampling_handler)
+    names = [p["name"] for p in prompts_data]
+    update = gr.update(choices=names, value=None, visible=bool(names))
+    empty_msg_update = gr.update(visible=not names)
+    history, rendered = _update_history(history, "prompts/list", request, response)
+    return request, response, update, empty_msg_update, prompts_data, history, rendered
+
+
+async def invoke_prompt_with_history(base_url, timeout, prompt_name, args, history):
+    request, response = await invoke_prompt(base_url, timeout, prompt_name, args, sampling_handler)
+    history, rendered = _update_history(history, "prompts/invoke", request, response)
+    return request, response, history, rendered
+
+
+async def list_resources_with_history(base_url, timeout, history):
+    request, response, resources_data = await list_resources(base_url, timeout, sampling_handler)
+    names = [r["uri"] for r in resources_data]
+    update = gr.update(choices=names, value=None, visible=bool(names))
+    empty_msg_update = gr.update(visible=not names)
+    history, rendered = _update_history(history, "resources/list", request, response)
+    return request, response, update, empty_msg_update, resources_data, history, rendered
+
+
+async def list_resource_templates_with_history(base_url, timeout, history):
+    request, response, templates_data = await list_resource_templates(base_url, timeout, sampling_handler)
+    names = [t["uriTemplate"] for t in templates_data]
+    update = gr.update(choices=names, value=None, visible=bool(names))
+    empty_msg_update = gr.update(visible=not names)
+    history, rendered = _update_history(history, "resources/templates/list", request, response)
+    return request, response, update, empty_msg_update, templates_data, history, rendered
+
+
+async def read_resource_with_history(base_url, timeout, resource_uri, history):
+    request, response = await read_resource(base_url, timeout, resource_uri, sampling_handler)
+    history, rendered = _update_history(history, "resources/read", request, response)
+    return request, response, history, rendered
+
+
+async def ping_with_history(base_url, timeout, history):
+    request, response = await ping_server(base_url, timeout, sampling_handler)
+    history, rendered = _update_history(history, "ping", request, response)
+    return history, rendered
+
+
+# Sampling handler that will be set dynamically
+sampling_log = []
+
+
+async def sampling_handler(
+    messages: list[SamplingMessage], params: SamplingParams, context: RequestContext
+) -> str:
+    """Simple sampling handler that logs requests and returns a mock response."""
+    # Extract message content
+    conversation = []
+    for message in messages:
+        content = (
+            message.content.text
+            if hasattr(message.content, "text")
+            else str(message.content)
+        )
+        conversation.append(f"{message.role}: {content}")
+
+    # Log the sampling request
+    log_entry = {
+        "messages": conversation,
+        "system_prompt": params.systemPrompt if hasattr(params, "systemPrompt") else None,
+        "temperature": params.temperature if hasattr(params, "temperature") else None,
+        "max_tokens": params.maxTokens if hasattr(params, "maxTokens") else None,
+    }
+    sampling_log.insert(0, log_entry)
+
+    # Return a simple response
+    return "This is a mock response from the sampling handler. In a real implementation, this would call an LLM API."
+
+
+def get_sampling_log():
+    """Get the current sampling log."""
+    if not sampling_log:
+        return "No sampling requests yet."
+    
+    log_text = []
+    for idx, entry in enumerate(sampling_log[:10], 1):  # Show last 10 entries
+        log_text.append(f"### Request {idx}")
+        log_text.append(f"**Messages:**")
+        for msg in entry["messages"]:
+            log_text.append(f"- {msg}")
+        if entry["system_prompt"]:
+            log_text.append(f"**System Prompt:** {entry['system_prompt']}")
+        if entry["temperature"] is not None:
+            log_text.append(f"**Temperature:** {entry['temperature']}")
+        if entry["max_tokens"]:
+            log_text.append(f"**Max Tokens:** {entry['max_tokens']}")
+        log_text.append("---")
+    
+    return "\n".join(log_text)
+
+async def on_resource_select(base_url, timeout, resource_uri, history):
+    request, response = await read_resource(base_url, timeout, resource_uri, sampling_handler)
+    history, rendered = _update_history(history, "resources/read", request, response)
+    return (
+        gr.update(value=request, visible=True),
+        gr.update(value=response, visible=True),
+        gr.update(value=response, visible=True),
+        history,
+        rendered
+    )

mcp_client.py

@@ -0,0 +1,206 @@
+from __future__ import annotations
+
+import json
+from typing import Tuple
+from contextlib import AsyncExitStack
+
+from fastmcp import Client
+from fastmcp.client import SSETransport, StreamableHttpTransport
+import mcp.types as types
+import httpx
+
+JsonStrPair = Tuple[str, str]
+
+# Global client state
+_client_session: Client | None = None
+_client_exit_stack: AsyncExitStack | None = None
+
+
+async def connect(base_url: str, timeout_seconds: float, transport_type: str, sampling_handler=None):
+    global _client_session, _client_exit_stack
+    await disconnect()
+    
+    _client_exit_stack = AsyncExitStack()
+    
+    if transport_type == "Streamable HTTP":
+        transport = StreamableHttpTransport(base_url)
+    else:
+        # Default to SSE
+        transport = SSETransport(base_url)
+        
+    client = Client(transport=transport, timeout=timeout_seconds, sampling_handler=sampling_handler)
+    # Enter the async context to establish connection
+    _client_session = await _client_exit_stack.enter_async_context(client)
+    return _client_session
+
+
+async def disconnect():
+    global _client_session, _client_exit_stack
+    if _client_exit_stack:
+        await _client_exit_stack.aclose()
+    _client_session = None
+    _client_exit_stack = None
+
+
+def _get_client() -> Client:
+    if _client_session is None:
+        raise RuntimeError("Client not connected. Please connect first.")
+    return _client_session
+
+
+async def list_resources(base_url: str, timeout_seconds: float, sampling_handler=None) -> tuple[str, str, list[str]]:
+    try:
+        client = _get_client()
+        resources = await client.list_resources()
+        # Convert Pydantic models to list of dicts then to formatted JSON
+        resources_data = [r.model_dump() for r in resources]
+        return "list_resources()", json.dumps(resources_data, indent=2), resources_data
+    except Exception as e:
+        return "Error", str(e), []
+
+
+async def list_resource_templates(base_url: str, timeout_seconds: float, sampling_handler=None) -> tuple[str, str, list[str]]:
+    try:
+        client = _get_client()
+        templates = await client.list_resource_templates()
+        templates_data = [t.model_dump() for t in templates]
+        return "list_resource_templates()", json.dumps(templates_data, indent=2), templates_data
+    except Exception as e:
+        return "Error", str(e), []
+
+
+async def read_resource(base_url: str, timeout_seconds: float, resource_uri: str, sampling_handler=None) -> JsonStrPair:
+    if not resource_uri:
+        return "", "Select a resource first."
+    try:
+        client = _get_client()
+        resource = await client.read_resource(resource_uri)
+        return f"read_resource({resource_uri})", resource.model_dump_json(indent=2)
+    except Exception as e:
+        return "Error", str(e)
+
+
+async def list_prompts(base_url: str, timeout_seconds: float, sampling_handler=None) -> tuple[str, str, list[str]]:
+    try:
+        client = _get_client()
+        prompts = await client.list_prompts()
+        prompts_data = [p.model_dump() for p in prompts]
+        return "list_prompts()", json.dumps(prompts_data, indent=2), prompts_data
+    except Exception as e:
+        return "Error", str(e), []
+
+
+async def invoke_prompt(
+    base_url: str, timeout_seconds: float, prompt_name: str, arguments_text: str, sampling_handler=None
+) -> JsonStrPair:
+    if not prompt_name:
+        return "", "Select a prompt first."
+    try:
+        args = json.loads(arguments_text) if arguments_text.strip() else {}
+    except ValueError as exc:
+        return "", f"Invalid JSON arguments: {exc}"
+
+    try:
+        client = _get_client()
+        # Use manual request to ensure empty arguments dict is sent (workaround for potential fastmcp/mcp issue)
+        result = await client.session.send_request(
+            types.GetPromptRequest(
+                method="prompts/get",
+                params=types.GetPromptRequestParams(
+                    name=prompt_name,
+                    arguments=args
+                )
+            ),
+            types.GetPromptResult
+        )
+        # Extract text from messages
+        messages_text = []
+        if hasattr(result, "messages"):
+            for msg in result.messages:
+                if hasattr(msg, "content"):
+                    content = msg.content
+                    if hasattr(content, "text"):
+                        messages_text.append(content.text)
+                    elif isinstance(content, dict) and "text" in content:
+                        messages_text.append(content["text"])
+        
+        final_result = "\n\n".join(messages_text) if messages_text else "No text content found in prompt result."
+        return f"get_prompt({prompt_name})", final_result
+    except Exception as e:
+        return "Error", str(e)
+
+
+async def list_tools(base_url: str, timeout_seconds: float, sampling_handler=None) -> tuple[str, str, list[dict]]:
+    try:
+        client = _get_client()
+        tools = await client.list_tools()
+        # Return list of tool dictionaries (including schema)
+        tools_data = [t.model_dump() for t in tools]
+        return "list_tools()", json.dumps(tools_data, indent=2), tools_data
+    except Exception as e:
+        return "Error", str(e), []
+
+
+async def invoke_tool(
+    base_url: str, timeout_seconds: float, tool_name: str, arguments_text: str, sampling_handler=None
+) -> JsonStrPair:
+    if not tool_name:
+        return "", "Select a tool first."
+    try:
+        args = json.loads(arguments_text) if arguments_text.strip() else {}
+    except ValueError as exc:
+        return "", f"Invalid JSON arguments: {exc}"
+
+    try:
+        client = _get_client()
+        result = await client.call_tool(tool_name, arguments=args)
+        
+        # Extract content from the result
+        content = []
+        if hasattr(result, "content"):
+            for item in result.content:
+                if hasattr(item, "text"):
+                    content.append(item.text)
+                elif isinstance(item, dict) and "text" in item:
+                    content.append(item["text"])
+                else:
+                    content.append(str(item))
+        
+        final_result = "\n".join(content) if content else "No content returned."
+        return f"call_tool({tool_name})", final_result
+    except Exception as e:
+        return "Error", str(e)
+
+
+async def ping_server(base_url: str, timeout_seconds: float, sampling_handler=None) -> JsonStrPair:
+    try:
+        client = _get_client()
+        await client.ping()
+        return "ping()", json.dumps({"status": "Pong"}, indent=2)
+    except Exception as e:
+        return "Error", str(e)
+
+
+async def send_custom_request(
+    base_url: str, method: str, params_text: str, timeout_seconds: float
+) -> JsonStrPair:
+    # Fallback to manual httpx for custom requests since FastMCP Client is high-level
+    try:
+        params = json.loads(params_text) if params_text.strip() else {}
+    except ValueError as exc:
+        return "", f"Invalid JSON params: {exc}"
+    
+    payload = {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": method,
+        "params": params,
+    }
+    
+    try:
+        async with httpx.AsyncClient(timeout=timeout_seconds) as client:
+            response = await client.post(base_url, json=payload)
+            response.raise_for_status()
+            return json.dumps(payload, indent=2), json.dumps(response.json(), indent=2)
+    except Exception as exc:
+        return json.dumps(payload, indent=2), f"Error: {exc}"

pyproject.toml

@@ -0,0 +1,11 @@
+[project]
+name = "gradio-mcp-inspector"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "fastmcp>=2.13.1",
+    "gradio>=6.0.1",
+    "httpx>=0.28.1",
+]

requirements.txt

@@ -0,0 +1,3 @@
+fastmcp>=2.13.1
+gradio>=5.0.0
+httpx>=0.28.1

theme.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+from typing import Iterable
+import gradio as gr
+from gradio.themes.base import Base
+from gradio.themes.utils import colors, fonts, sizes
+
+class CustomLightTheme(Base):
+    def __init__(
+        self,
+        *,
+        primary_hue: colors.Color | str = colors.indigo,
+        secondary_hue: colors.Color | str = colors.blue,
+        neutral_hue: colors.Color | str = colors.slate,
+        spacing_size: sizes.Size | str = sizes.spacing_md,
+        radius_size: sizes.Size | str = sizes.radius_lg,
+        text_size: sizes.Size | str = sizes.text_md,
+        font: fonts.Font | str | Iterable[fonts.Font | str] = (
+            fonts.GoogleFont("Inter"),
+            "ui-sans-serif",
+            "system-ui",
+            "sans-serif",
+        ),
+        font_mono: fonts.Font | str | Iterable[fonts.Font | str] = (
+            fonts.GoogleFont("JetBrains Mono"),
+            "ui-monospace",
+            "Consolas",
+            "monospace",
+        ),
+    ):
+        super().__init__(
+            primary_hue=primary_hue,
+            secondary_hue=secondary_hue,
+            neutral_hue=neutral_hue,
+            spacing_size=spacing_size,
+            radius_size=radius_size,
+            text_size=text_size,
+            font=font,
+            font_mono=font_mono,
+        )
+        super().set(
+            body_background_fill="white",
+            body_background_fill_dark="*neutral_950",
+            body_text_color="*neutral_950",
+            body_text_color_dark="*neutral_50",
+            background_fill_primary="white",
+            background_fill_primary_dark="*neutral_900",
+            background_fill_secondary="#f9fafb",
+            background_fill_secondary_dark="*neutral_800",
+            border_color_primary="#e5e7eb",
+            border_color_primary_dark="*neutral_700",
+            block_background_fill="white",
+            block_background_fill_dark="*neutral_900",
+            block_label_background_fill="white",
+            block_label_background_fill_dark="*neutral_900",
+            input_background_fill="white",
+            input_background_fill_dark="*neutral_800",
+            button_primary_background_fill="*primary_600",
+            button_primary_background_fill_hover="*primary_700",
+            button_primary_text_color="white",
+            button_secondary_background_fill="white",
+            button_secondary_background_fill_hover="#f3f4f6",
+            button_secondary_text_color="*neutral_800",
+            block_title_text_color="*neutral_800",
+            block_title_text_color_dark="*neutral_200",
+            block_label_text_color="*neutral_500",
+            block_label_text_color_dark="*neutral_400",
+            input_border_color="#e5e7eb",
+            input_border_color_dark="*neutral_700",
+            input_border_width="1px",
+            input_shadow="0 1px 2px 0 rgb(0 0 0 / 0.05)",
+            input_shadow_focus="0 0 0 2px *primary_500",
+        )