test_geocode_mcp.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Exhaustive test of the `geocode` tool exposed by the geocontext MCP server over
HTTP (Streamable HTTP transport).

Goal: understand why ChatGPT "loses" the tool-call result and renders an empty
`multimodal_text` block with blank `display_url` / `display_title` instead of
the actual geocoding result.

The script:
  1. Opens an MCP session (initialize + notifications/initialized).
  2. Fetches the tool definition (inputSchema, outputSchema, annotations, _meta).
  3. Calls `geocode` with several parameter sets.
  4. Dissects the raw response (content[], structuredContent, _meta, isError...).
  5. Diagnoses anything that could confuse the ChatGPT client.

Usage:
    .venv\\Scripts\\python.exe test_geocode_mcp.py
    .venv\\Scripts\\python.exe test_geocode_mcp.py --url https://geollm.beta.ign.fr/geocontext/mcp
"""

from __future__ import annotations

import argparse
import json
import sys
from typing import Any

import httpx

DEFAULT_URL = "https://geollm.beta.ign.fr/geocontext/mcp"
PROTOCOL_VERSION = "2025-06-18"


# ANSI colors (can be disabled)
class C:
    RESET = "\033[0m"
    BOLD = "\033[1m"
    DIM = "\033[2m"
    RED = "\033[31m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    BLUE = "\033[34m"
    CYAN = "\033[36m"
    MAGENTA = "\033[35m"


def hr(title: str = "", char: str = "=") -> None:
    width = 78
    if title:
        title = f" {title} "
        pad = (width - len(title)) // 2
        print(f"{C.CYAN}{char * pad}{title}{char * (width - pad - len(title))}{C.RESET}")
    else:
        print(f"{C.CYAN}{char * width}{C.RESET}")


def pretty(obj: Any) -> str:
    return json.dumps(obj, ensure_ascii=False, indent=2)


class MCPHttpClient:
    """Minimal client for MCP Streamable HTTP (JSON-RPC over POST)."""

    def __init__(self, url: str, timeout: float = 30.0):
        self.url = url
        self.session_id: str | None = None
        self.protocol_version: str | None = None
        self._id = 0
        self._client = httpx.Client(timeout=timeout, follow_redirects=True)

    def _next_id(self) -> int:
        self._id += 1
        return self._id

    def _headers(self) -> dict[str, str]:
        h = {
            "Content-Type": "application/json",
            # ChatGPT/MCP clients advertise accepting both content types
            "Accept": "application/json, text/event-stream",
        }
        if self.session_id:
            h["Mcp-Session-Id"] = self.session_id
        if self.protocol_version:
            h["MCP-Protocol-Version"] = self.protocol_version
        return h

    @staticmethod
    def _parse_body(resp: httpx.Response) -> dict[str, Any]:
        """Handle plain JSON OR SSE stream (text/event-stream)."""
        ctype = resp.headers.get("content-type", "")
        text = resp.text
        if "text/event-stream" in ctype:
            # Concatenate the data: lines, keep the last JSON-RPC message.
            messages: list[dict[str, Any]] = []
            for raw_line in text.splitlines():
                line = raw_line.strip()
                if line.startswith("data:"):
                    payload = line[len("data:"):].strip()
                    if not payload:
                        continue
                    try:
                        messages.append(json.loads(payload))
                    except json.JSONDecodeError:
                        pass
            if messages:
                return messages[-1]
            return {}
        # Plain JSON
        try:
            return resp.json()
        except json.JSONDecodeError:
            return {"_raw_text": text}

    def _post(self, method: str, params: dict[str, Any] | None, *, notify: bool = False) -> dict[str, Any]:
        body: dict[str, Any] = {"jsonrpc": "2.0", "method": method}
        if not notify:
            body["id"] = self._next_id()
        if params is not None:
            body["params"] = params

        resp = self._client.post(self.url, headers=self._headers(), json=body)

        # Capture the session id returned by initialize
        sid = resp.headers.get("mcp-session-id") or resp.headers.get("Mcp-Session-Id")
        if sid:
            self.session_id = sid

        if notify:
            return {"_status": resp.status_code}

        if resp.status_code >= 400:
            return {
                "_http_error": resp.status_code,
                "_body": self._parse_body(resp),
                "_headers": dict(resp.headers),
            }
        return self._parse_body(resp)

    def initialize(self) -> dict[str, Any]:
        result = self._post(
            "initialize",
            {
                "protocolVersion": PROTOCOL_VERSION,
                "capabilities": {},
                "clientInfo": {"name": "geocode-debug-client", "version": "1.0.0"},
            },
        )
        # Remember the negotiated protocol version
        pv = (result.get("result") or {}).get("protocolVersion")
        if pv:
            self.protocol_version = pv
        # Notify that initialization is complete
        self._post("notifications/initialized", {}, notify=True)
        return result

    def list_tools(self) -> dict[str, Any]:
        return self._post("tools/list", {})

    def call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
        return self._post("tools/call", {"name": name, "arguments": arguments})

    def close(self) -> None:
        self._client.close()


def analyze_tool_definition(tool: dict[str, Any]) -> None:
    hr("geocode TOOL DEFINITION", "-")
    print(f"{C.BOLD}name{C.RESET}        : {tool.get('name')}")
    print(f"{C.BOLD}title{C.RESET}       : {tool.get('title')}")
    desc = tool.get("description") or ""
    print(f"{C.BOLD}description{C.RESET} : {desc[:200]}{'…' if len(desc) > 200 else ''}")
    print(f"\n{C.BOLD}inputSchema{C.RESET} :\n{pretty(tool.get('inputSchema'))}")

    out_schema = tool.get("outputSchema")
    if out_schema is not None:
        print(f"\n{C.BOLD}{C.GREEN}outputSchema PRESENT{C.RESET} :\n{pretty(out_schema)}")
        print(
            f"\n{C.YELLOW}>> When a tool declares an outputSchema, the server MUST return "
            f"structuredContent, and clients validate that content.{C.RESET}"
        )
    else:
        print(f"\n{C.DIM}outputSchema : (absent){C.RESET}")

    if "annotations" in tool:
        print(f"\n{C.BOLD}annotations{C.RESET} :\n{pretty(tool['annotations'])}")
    if "_meta" in tool:
        print(f"\n{C.BOLD}_meta{C.RESET} :\n{pretty(tool['_meta'])}")


def analyze_call_result(label: str, response: dict[str, Any]) -> dict[str, Any]:
    """Dissect a tools/call result and return a diagnostic dict."""
    hr(f"CALL: {label}", "-")
    diag: dict[str, Any] = {"label": label, "issues": []}

    if "_http_error" in response:
        print(f"{C.RED}HTTP ERROR {response['_http_error']}{C.RESET}")
        print(pretty(response.get("_body")))
        diag["issues"].append(f"HTTP {response['_http_error']}")
        return diag

    if "error" in response:
        print(f"{C.RED}JSON-RPC ERROR{C.RESET} :\n{pretty(response['error'])}")
        diag["issues"].append("JSON-RPC error")
        return diag

    result = response.get("result", {})
    print(f"{C.DIM}Full JSON-RPC response:{C.RESET}\n{pretty(response)}\n")

    # --- content[] ---
    content = result.get("content", [])
    print(f"{C.BOLD}content[]{C.RESET} : {len(content)} block(s)")
    for i, block in enumerate(content):
        btype = block.get("type")
        print(f"  [{i}] type={C.MAGENTA}{btype}{C.RESET}")
        if btype == "text":
            txt = block.get("text", "")
            print(f"      length={len(txt)} characters")
            print(f"      content:\n{C.DIM}{txt}{C.RESET}")
            # Is the text valid JSON?
            try:
                parsed = json.loads(txt)
                print(f"      {C.GREEN}-> text = valid JSON{C.RESET}")
                diag["text_is_json"] = True
                diag["text_parsed"] = parsed
            except json.JSONDecodeError:
                print(f"      {C.YELLOW}-> text = raw text (not JSON){C.RESET}")
                diag["text_is_json"] = False
        else:
            print(f"      {pretty(block)}")

    # --- structuredContent ---
    if "structuredContent" in result:
        sc = result["structuredContent"]
        print(f"\n{C.BOLD}{C.GREEN}structuredContent PRESENT{C.RESET} :\n{pretty(sc)}")
        diag["has_structured_content"] = True
    else:
        print(f"\n{C.DIM}structuredContent : (absent){C.RESET}")
        diag["has_structured_content"] = False

    # --- isError ---
    is_error = result.get("isError", False)
    print(f"\n{C.BOLD}isError{C.RESET} : {is_error}")
    diag["is_error"] = is_error

    # --- _meta ---
    if "_meta" in result:
        print(f"\n{C.BOLD}_meta{C.RESET} :\n{pretty(result['_meta'])}")

    # --- Detect the fields ChatGPT looks for when building citations ---
    print(f"\n{C.BOLD}Looking for ChatGPT citation fields (display_url/display_title){C.RESET}")
    blob = json.dumps(result, ensure_ascii=False)
    for field in ("display_url", "display_title", "url", "title"):
        present = f'"{field}"' in blob
        marker = C.GREEN + "present" if present else C.DIM + "absent"
        print(f"  {field:14s}: {marker}{C.RESET}")

    return diag


def main() -> int:
    parser = argparse.ArgumentParser(description="Exhaustive geocode MCP HTTP test")
    parser.add_argument("--url", default=DEFAULT_URL)
    parser.add_argument("--no-color", action="store_true")
    args = parser.parse_args()

    if args.no_color or not sys.stdout.isatty():
        for attr in dir(C):
            if not attr.startswith("_"):
                setattr(C, attr, "")

    hr("CONNECTING TO MCP SERVER")
    print(f"URL: {args.url}\n")

    client = MCPHttpClient(args.url)
    try:
        # 1) initialize
        init = client.initialize()
        if "_http_error" in init or "error" in init:
            print(f"{C.RED}Initialization failed:{C.RESET}\n{pretty(init)}")
            return 1
        server_info = (init.get("result") or {}).get("serverInfo", {})
        caps = (init.get("result") or {}).get("capabilities", {})
        print(f"{C.GREEN}Session opened{C.RESET}")
        print(f"  serverInfo       : {server_info}")
        print(f"  protocolVersion  : {client.protocol_version}")
        print(f"  Mcp-Session-Id   : {client.session_id}")
        print(f"  capabilities     : {pretty(caps)}")

        # 2) tools/list
        hr("TOOL LIST")
        tools_resp = client.list_tools()
        tools = (tools_resp.get("result") or {}).get("tools", [])
        print(f"{len(tools)} tool(s): {[t.get('name') for t in tools]}")
        geocode = next((t for t in tools if t.get("name") == "geocode"), None)
        if geocode is None:
            print(f"{C.RED}Tool 'geocode' not found!{C.RESET}")
            return 1
        analyze_tool_definition(geocode)

        # 3) Test battery
        hr("geocode TEST BATTERY")
        test_cases: list[tuple[str, dict[str, Any]]] = [
            ("ChatGPT bug case: Mairie de Loray / max=1", {"text": "Mairie de Loray", "maximumResponses": 1}),
            ("Multiple results: Mairie de Loray / max=3", {"text": "Mairie de Loray", "maximumResponses": 3}),
            ("Max results: Paris / max=10", {"text": "Paris", "maximumResponses": 10}),
            ("Without maximumResponses (default): Tour Eiffel", {"text": "Tour Eiffel"}),
            ("Precise address: 8 rue de la Paix Paris", {"text": "8 rue de la Paix Paris", "maximumResponses": 1}),
            ("Special characters: Besançon / max=1", {"text": "Besançon", "maximumResponses": 1}),
            ("Unlikely text (0 result?): xyzqwerty123", {"text": "xyzqwerty123zzz", "maximumResponses": 1}),
        ]

        diagnostics: list[dict[str, Any]] = []
        for label, params in test_cases:
            resp = client.call_tool("geocode", params)
            diagnostics.append(analyze_call_result(f"{label}  args={params}", resp))

        # 4) Summary
        hr("DIAGNOSTIC SUMMARY")
        out_schema = geocode.get("outputSchema")
        any_structured = any(d.get("has_structured_content") for d in diagnostics)
        any_text_json = any(d.get("text_is_json") for d in diagnostics)

        print(f"{C.BOLD}1) geocontext MCP side{C.RESET}")
        print(f"   - outputSchema declared       : {'YES' if out_schema else 'NO'}")
        print(f"   - structuredContent returned  : {'YES' if any_structured else 'NO'}")
        print(f"   - content[].text = raw JSON   : {'YES' if any_text_json else 'NO'}")

        if out_schema and not any_structured:
            print(
                f"   {C.RED}>> INCONSISTENCY: an outputSchema is declared but no "
                f"structuredContent is returned. Strict clients (including ChatGPT) "
                f"may reject/hide the result.{C.RESET}"
            )
        if any_text_json and not any_structured:
            print(
                f"   {C.YELLOW}>> The result is returned as TEXT containing JSON, "
                f"without structuredContent. ChatGPT then treats that text as a 'document' "
                f"to cite (hence the Citation Marker + empty display_url/display_title).{C.RESET}"
            )

        print(f"\n{C.BOLD}2) ChatGPT side{C.RESET}")
        print(
            "   The 'multimodal_text' + 'Citation Marker' + empty display_url/display_title "
            "block is generated by ChatGPT's MCP connector, NOT by the server. "
            "ChatGPT interprets the tool output as a citable resource and truncates "
            "the display ([L1]...[L3]) due to missing display metadata."
        )

        return 0
    finally:
        client.close()


if __name__ == "__main__":
    raise SystemExit(main())