app.py

"""MCP Surface Lint - a Gradio app that is also an MCP server.

Paste a tool surface, get back the places where a caller could do the wrong
thing and receive no error for it. The same checks are exposed as MCP tools, so
an agent can lint an agent interface without going through the UI.

Launched with `mcp_server=True`; see the README for the endpoint.
"""

from __future__ import annotations

import json
from typing import Any, Iterator

import gradio as gr

import lint

SEVERITY_LABEL = {"high": "high", "medium": "medium", "low": "low"}
HEADERS = ["severity", "tool", "rule", "finding"]

EXAMPLE_LEAKY = json.dumps(
    {
        "tools": [
            {
                "name": "get_records",
                "description": "Gets records.",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "table": {"type": "string"},
                        "limit": {"type": "integer", "description": "How many rows."},
                        "output_format": {"type": "string", "description": "Response shape."},
                    },
                },
            },
            {
                "name": "get_record",
                "description": "Gets records.",
                "inputSchema": {"type": "object", "properties": {"id": {"description": "Row id."}}},
            },
            {
                "name": "delete_table",
                "description": "Removes a table and returns the remaining count.",
                "inputSchema": {
                    "type": "object",
                    "properties": {"table": {"type": "string", "description": "Table name."}},
                    "required": ["table"],
                },
            },
        ]
    },
    indent=2,
)

EXAMPLE_CLEAN = json.dumps(
    {
        "tools": [
            {
                "name": "search_documents",
                "description": (
                    "Search indexed documents by keyword. Returns an empty list when "
                    "nothing matches; raises an error when the index is unavailable."
                ),
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "Keyword expression matched against document text.",
                        },
                        "limit": {
                            "type": "integer",
                            "description": "Maximum documents to return.",
                            "minimum": 1,
                            "maximum": 100,
                        },
                    },
                    "required": ["query"],
                },
            }
        ]
    },
    indent=2,
)


def _summary(counts: dict[str, int], tool_count: int, done: bool) -> str:
    total = sum(counts.values())
    if not done:
        return f"Scanning {tool_count} tools... {total} findings so far."
    if total == 0:
        return (
            f"**{tool_count} tools, no findings.** Every tool declares its types, "
            "documents its parameters, and says what failure looks like."
        )
    return (
        f"**{total} findings across {tool_count} tools** - "
        f"{counts['high']} high, {counts['medium']} medium, {counts['low']} low."
    )


def analyse(raw: str, severities: list[str]) -> Iterator[tuple[Any, str, list[dict], Any]]:
    """Stream findings as they are produced.

    Yields (table rows, summary markdown, findings for state, detail visibility).
    """
    if not (raw or "").strip():
        yield [], "Paste a tool surface to begin.", [], gr.update(visible=False)
        return

    try:
        tools = lint.normalize(json.loads(raw))
    except json.JSONDecodeError as err:
        yield [], f"**Not valid JSON.** {err}", [], gr.update(visible=False)
        return
    except ValueError as err:
        yield [], f"**Unusable input.** {err}", [], gr.update(visible=False)
        return

    wanted = set(severities) or set(lint.SEVERITIES)
    rows: list[list[str]] = []
    kept: list[dict] = []
    counts = {s: 0 for s in lint.SEVERITIES}

    for finding in lint.lint_iter(tools):
        if finding.severity not in wanted:
            continue
        counts[finding.severity] += 1
        rows.append(finding.as_row())
        kept.append(finding.__dict__)
        # Emitting per finding keeps a large surface responsive instead of
        # showing nothing until the whole scan completes.
        yield rows[:], _summary(counts, len(tools), done=False), kept[:], gr.update(visible=False)

    yield (
        rows,
        _summary(counts, len(tools), done=True),
        kept,
        gr.update(visible=bool(kept)),
    )


def explain(findings: list[dict], event: gr.SelectData) -> str:
    """Show the reasoning behind whichever finding the user selected."""
    if not findings:
        return ""
    index = event.index[0] if isinstance(event.index, (list, tuple)) else event.index
    if index is None or index >= len(findings):
        return ""
    f = findings[index]
    return (
        f"### {f['rule']} - {f['tool']}\n\n"
        f"**{f['message']}**\n\n{f['detail']}"
    )


with gr.Blocks(title="MCP Surface Lint") as demo:
    gr.Markdown(
        "# MCP Surface Lint\n"
        "Static checks over an MCP tool surface. Every rule answers one question: "
        "**could a caller do the wrong thing and get no error back?**\n\n"
        "A tool that crashes on bad input is fine. A tool that accepts bad input and "
        "returns something plausible is the expensive kind of broken.\n\n"
        "This app is also an MCP server, so an agent can run these checks directly."
    )

    findings_state = gr.State([])

    with gr.Row():
        with gr.Column(scale=1):
            surface = gr.Code(
                label="Tool surface (a tools/list response, or a bare list of tools)",
                language="json",
                lines=22,
                value=EXAMPLE_LEAKY,
            )
            severity_filter = gr.CheckboxGroup(
                choices=list(lint.SEVERITIES),
                value=list(lint.SEVERITIES),
                label="Severities to report",
            )
            run = gr.Button("Lint surface", variant="primary")

        with gr.Column(scale=1):
            summary = gr.Markdown("Paste a tool surface to begin.")
            table = gr.Dataframe(
                headers=HEADERS,
                column_count=(len(HEADERS), "fixed"),
                interactive=False,
                wrap=True,
                label="Findings (select a row for the reasoning)",
            )
            detail = gr.Markdown(visible=False)

    gr.Examples(
        examples=[[EXAMPLE_LEAKY], [EXAMPLE_CLEAN]],
        inputs=[surface],
        label="Examples: a leaky surface, and one that passes",
    )

    # Keep the UI handlers off the MCP surface. Without this Gradio publishes
    # `analyse`, `analyse_1` and `explain` as tools: a duplicate name pair, plus a
    # handler taking a gr.State that cannot survive a tool call. This linter would
    # flag all three, so it had better not ship them.
    run.click(
        analyse,
        inputs=[surface, severity_filter],
        outputs=[table, summary, findings_state, detail],
        api_name=False,
    )
    severity_filter.change(
        analyse,
        inputs=[surface, severity_filter],
        outputs=[table, summary, findings_state, detail],
        api_name=False,
    )
    table.select(explain, inputs=[findings_state], outputs=[detail], api_name=False)

    # MCP-only surface. No UI component, but callable as a tool - which is the
    # point: the linter is itself something an agent can call.
    @gr.api
    def lint_tool_surface(surface_json: str) -> dict:
        """Lint an MCP tool surface for places a caller could fail silently.

        Returns an object with tool_count, counts by severity, and findings, each
        carrying the tool, rule id, severity, message and reasoning. Raises
        ValueError when the input is not valid JSON or is not a list of tools.

        Args:
            surface_json: JSON string: either a tools/list response (an object with a "tools" key) or a bare list of tool objects. Must not be empty.
        """
        tools = lint.normalize(json.loads(surface_json))
        findings = lint.lint(tools)
        return {
            "tool_count": len(tools),
            "counts": lint.severity_counts(findings),
            "findings": [f.__dict__ for f in findings],
        }

    @gr.api
    def list_rules() -> list[dict]:
        """List every check this linter applies, with its id and severity.

        Returns one object per rule with rule, severity and message, derived by
        running the rule set against a deliberately incomplete tool. Takes no
        arguments and never raises.
        """
        probe = {"name": "probe", "inputSchema": {"properties": {"arg": {}}}}
        seen: dict[str, dict] = {}
        for f in lint.lint([probe, {**probe, "name": "probes"}]):
            seen.setdefault(f.rule, {"rule": f.rule, "severity": f.severity, "message": f.message})
        return sorted(seen.values(), key=lambda r: r["rule"])


if __name__ == "__main__":
    demo.launch(mcp_server=True, theme=gr.themes.Soft())