Add robust external MCP testing: conformance + Claude/Codex client sims

Layers an out-of-process protocol-conformance and client-compatibility suite on
top of the integration scaffolding vendored from #1509:

- conftest.py: one session-scoped Echidna+MCP server shared by all modules
  (honors ECHIDNA_MCP_URL to target an already-running server)
- _mcp_client.py: dependency-light wire-level JSON-RPC/HTTP helpers
- test_mcp_conformance.py: asserts the transport guarantees strict clients need
  (notification -> 202 empty body, GET SSE -> 405, handshake, error codes)
- test_mcp_codex.py: replays Codex's strict rmcp handshake
- test_mcp_claude.py: drives the server with the official `mcp` SDK (Claude)
- test_mcp.py: refactored to share the conftest fixture (originally @datradito)
- mcp-tests.yml: CI runs the full suite (tools + conformance + Claude + Codex)

Verified end-to-end: 12/12 pass against the current (fixed) server; against the
pre-fix haskell-mcp-server (9fd60af) the conformance + Codex tests fail on the
`200 {}` notification (the bug that broke Codex) while the lenient Claude SDK
passes -- which is exactly why the strict conformance/Codex layer is needed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: gustavo-grieco

.github/workflows/mcp-tests.yml

@@ -1,12 +1,23 @@
 name: MCP integration tests
 
+# Out-of-process MCP testing: spins up a real Echidna campaign with the MCP
+# server and drives it the way real agent clients do. Besides the tool/semantic
+# tests it runs the client-compatibility suite:
+#   * test_mcp_conformance.py -- wire-protocol regression guard (202/405/handshake)
+#   * test_mcp_codex.py       -- replays Codex's strict rmcp handshake
+#   * test_mcp_claude.py      -- drives the server with the official `mcp` SDK
+# Both client checks are transport-level and need no API keys, so they run on
+# every push/PR. (A live-model smoke using examples/mcp_agent.py needs
+# ANTHROPIC_API_KEY / Codex auth and is intentionally not part of CI.)
+
 on:
   push:
     branches: ["**"]
   pull_request:
 
 jobs:
-  mcp-tests:
+  mcp-client-compat:
+    name: MCP client compatibility (Claude + Codex)
     runs-on: ubuntu-latest
 
     steps:
@@ -29,7 +40,7 @@ jobs:
           python-version: "3.11"
 
       - name: Install test dependencies
-        run: pip install pytest httpx
+        run: pip install -r tests/mcp/requirements-test.txt
 
-      - name: Run MCP integration tests
-        run: pytest tests/mcp/test_mcp.py -v
+      - name: Run MCP test suite (tools + conformance + Claude + Codex)
+        run: pytest tests/mcp -v

tests/mcp/_mcp_client.py

@@ -0,0 +1,61 @@
+"""
+Minimal, dependency-light MCP-over-HTTP client helpers shared by the
+conformance and Codex-replay test layers.
+
+These intentionally do NOT use a high-level MCP SDK: they speak the wire
+protocol directly so the tests can assert on exact status codes, headers and
+bodies — the things real clients (Codex's rmcp, Anthropic's MCP client) are
+strict about and that a lenient `httpx.post(...).json()` would silently hide.
+"""
+
+import httpx
+
+PROTOCOL_VERSION = "2025-06-18"
+SUPPORTED_VERSIONS = ("2025-06-18", "2025-03-26", "2024-11-05")
+
+
+def rpc(url, method, params=None, id=1, protocol_version=PROTOCOL_VERSION,
+        accept="application/json, text/event-stream", timeout=30):
+    """POST a single JSON-RPC message; return the raw httpx.Response (never parsed)."""
+    body = {"jsonrpc": "2.0", "method": method}
+    if id is not None:
+        body["id"] = id
+    if params is not None:
+        body["params"] = params
+    headers = {"Content-Type": "application/json", "Accept": accept}
+    # Per the spec the MCP-Protocol-Version header is sent on requests *after*
+    # initialization, not on `initialize` itself.
+    if protocol_version is not None and method != "initialize":
+        headers["MCP-Protocol-Version"] = protocol_version
+    return httpx.post(url, json=body, headers=headers, timeout=timeout)
+
+
+def http_get(url, accept="text/event-stream", timeout=10):
+    """GET the MCP endpoint (clients use this to open the server->client SSE stream)."""
+    return httpx.get(url, headers={"Accept": accept}, timeout=timeout)
+
+
+def handshake(url, protocol_version=PROTOCOL_VERSION):
+    """Run the client half of the MCP handshake; return the InitializeResult dict.
+
+    initialize -> (read result) -> notifications/initialized
+    """
+    resp = rpc(url, "initialize", {
+        "protocolVersion": protocol_version,
+        "capabilities": {},
+        "clientInfo": {"name": "echidna-mcp-tests", "version": "0"},
+    }, id=0, protocol_version=None)
+    resp.raise_for_status()
+    result = resp.json()["result"]
+    negotiated = result.get("protocolVersion", protocol_version)
+    rpc(url, "notifications/initialized", id=None, protocol_version=negotiated)
+    return result
+
+
+def call_tool(url, name, arguments=None, protocol_version=PROTOCOL_VERSION):
+    """tools/call helper that returns the parsed JSON-RPC `result`."""
+    resp = rpc(url, "tools/call",
+               {"name": name, "arguments": arguments or {}},
+               id=2, protocol_version=protocol_version)
+    resp.raise_for_status()
+    return resp.json().get("result", {})

tests/mcp/conftest.py

@@ -0,0 +1,94 @@
+"""
+Shared pytest fixtures for the out-of-process MCP test suite.
+
+A single session-scoped Echidna campaign (with the MCP server enabled) is
+started once and reused by every test module: the tool/semantic tests
+(test_mcp.py), the wire-protocol conformance tests (test_mcp_conformance.py),
+and the client-simulation tests (test_mcp_claude.py, test_mcp_codex.py).
+
+The Echidna launch/wait logic was extracted from the original test_mcp.py
+contributed by Dani Tradito (@datradito) in crytic/echidna#1509.
+"""
+
+import os
+import subprocess
+import time
+
+import pytest
+
+from _mcp_client import PROTOCOL_VERSION, rpc
+
+MCP_PORT = int(os.environ.get("ECHIDNA_MCP_PORT", "8080"))
+MCP_URL = f"http://127.0.0.1:{MCP_PORT}/mcp"
+ECHIDNA_BIN = os.environ.get("ECHIDNA_BIN", "echidna")
+CONTRACT = os.path.join(os.path.dirname(__file__), "contracts", "EchidnaMCPTest.sol")
+
+
+def _server_ready() -> bool:
+    """True once the MCP server answers a real `initialize` request."""
+    try:
+        resp = rpc(MCP_URL, "initialize", {
+            "protocolVersion": PROTOCOL_VERSION,
+            "capabilities": {},
+            "clientInfo": {"name": "readiness-probe", "version": "0"},
+        }, id=0, protocol_version=None, timeout=2)
+        return resp.status_code == 200 and "result" in resp.json()
+    except Exception:
+        return False
+
+
+@pytest.fixture(scope="session")
+def echidna_server(tmp_path_factory):
+    """Start an Echidna campaign with the MCP server for the whole test session.
+
+    If ECHIDNA_MCP_URL is set, use that already-running server instead of
+    starting one (handy for pointing the suite at a live campaign or, in CI, a
+    deliberately broken server to prove the conformance tests have teeth).
+    """
+    external = os.environ.get("ECHIDNA_MCP_URL")
+    if external:
+        yield {"url": external, "port": None, "protocol_version": PROTOCOL_VERSION}
+        return
+
+    corpus_dir = tmp_path_factory.mktemp("corpus")
+
+    env = os.environ.copy()
+    env["PATH"] = os.path.expanduser("~/.local/bin") + os.pathsep + env.get("PATH", "")
+    # Echidna shells out to solc/crytic-compile; an active VIRTUAL_ENV breaks that.
+    env.pop("VIRTUAL_ENV", None)
+
+    cmd = [
+        ECHIDNA_BIN, CONTRACT,
+        "--contract", "EchidnaMCPTest",
+        "--server", str(MCP_PORT),
+        "--format", "text",
+        "--test-limit", "1000000000",
+        "--corpus-dir", str(corpus_dir),
+    ]
+    proc = subprocess.Popen(
+        cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env
+    )
+
+    # Wait (compiling the contract can take a while) for the server to accept a
+    # real MCP initialize, not merely an open socket.
+    deadline = time.time() + 120
+    while time.time() < deadline:
+        if proc.poll() is not None:
+            out, err = proc.communicate()
+            raise RuntimeError(f"Echidna exited early.\nstdout:\n{out}\nstderr:\n{err}")
+        if _server_ready():
+            break
+        time.sleep(0.5)
+    else:
+        proc.terminate()
+        out, err = proc.communicate(timeout=5)
+        raise RuntimeError(f"MCP server did not become ready.\nstdout:\n{out}\nstderr:\n{err}")
+
+    yield {"url": MCP_URL, "port": MCP_PORT, "protocol_version": PROTOCOL_VERSION}
+
+    proc.terminate()
+    try:
+        proc.communicate(timeout=5)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait()

tests/mcp/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+# The Claude client test (test_mcp_claude.py) uses the async `mcp` SDK.
+asyncio_mode = auto

tests/mcp/requirements-test.txt

@@ -0,0 +1,12 @@
+# Dependencies for the out-of-process MCP test suite (tests/mcp/).
+#
+# Kept separate from requirements.txt (the heavier LangGraph/Claude agent demo
+# deps) so CI installs only what the tests need.
+
+httpx>=0.27,<0.29
+pytest>=8.0,<9.0
+pytest-asyncio>=0.24,<0.26
+
+# Official Model Context Protocol SDK -- used as the reference "Claude-family"
+# client in test_mcp_claude.py (transport only; no API key required).
+mcp>=1.2,<2

tests/mcp/test_mcp.py

@@ -1,37 +1,26 @@
 """
-Basic integration tests for the Echidna MCP server.
+Tool/semantic integration tests for the Echidna MCP server.
 
-Tests the core workflow: inject transactions, check coverage,
-reset priorities, and verify status.
+Verifies the core MCP tools return sensible results against a live Echidna
+campaign: status, inject_fuzz_transactions, show_coverage, clear_fuzz_priorities.
+
+Originally contributed by Dani Tradito (@datradito) in crytic/echidna#1509;
+adapted here to share the session-scoped `echidna_server` fixture in conftest.py.
 
 Run with:
     pytest tests/mcp/test_mcp.py
 """
 
-import pytest
-import socket
-import time
-import os
-import subprocess
 import httpx
 
 
-MCP_PORT = 8080
-MCP_URL = f"http://localhost:{MCP_PORT}/mcp"
-CONTRACT = "tests/mcp/contracts/EchidnaMCPTest.sol"
-
-
-# ---------------------------------------------------------------------------
-# Fixtures
-# ---------------------------------------------------------------------------
-
-def _call_tool(tool: str, args: dict = None) -> dict:
+def _call_tool(url: str, tool: str, args: dict = None) -> dict:
     payload = {
         "jsonrpc": "2.0", "id": 1,
         "method": "tools/call",
         "params": {"name": tool, "arguments": args or {}},
     }
-    resp = httpx.post(MCP_URL, json=payload, timeout=30)
+    resp = httpx.post(url, json=payload, timeout=30)
     resp.raise_for_status()
     return resp.json().get("result", {})
 
@@ -40,66 +29,19 @@ def _text(result: dict) -> str:
     return result.get("content", [{}])[0].get("text", "")
 
 
-@pytest.fixture(scope="module")
-def echidna(tmp_path_factory):
-    """Start an Echidna campaign with the MCP server for the whole test module."""
-    corpus_dir = tmp_path_factory.mktemp("corpus")
-    env = os.environ.copy()
-    env["PATH"] = os.path.expanduser("~/.local/bin") + ":" + env.get("PATH", "")
-
-    cmd = [
-        "echidna", CONTRACT,
-        "--contract", "EchidnaMCPTest",
-        "--server", str(MCP_PORT),
-        "--format", "text",
-        "--test-limit", "1000000000",
-        "--corpus-dir", str(corpus_dir),
-    ]
-    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-                            text=True, env=env)
-
-    # Wait up to 10 s for the server to accept connections
-    deadline = time.time() + 10
-    while time.time() < deadline:
-        try:
-            with socket.create_connection(("localhost", MCP_PORT), timeout=1):
-                break
-        except OSError:
-            time.sleep(0.5)
-    else:
-        proc.terminate()
-        out, err = proc.communicate(timeout=5)
-        raise RuntimeError(
-            f"Echidna MCP server did not start.\nstdout: {out}\nstderr: {err}"
-        )
-
-    yield proc
-
-    proc.terminate()
-    try:
-        proc.communicate(timeout=5)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        proc.wait()
-
-
-# ---------------------------------------------------------------------------
-# Tests
-# ---------------------------------------------------------------------------
-
-def test_status(echidna):
+def test_status(echidna_server):
     """status tool returns campaign metrics."""
-    result = _call_tool("status")
+    result = _call_tool(echidna_server["url"], "status")
     text = _text(result)
     assert text, "status returned empty response"
     assert "corpus" in text.lower() or "coverage" in text.lower(), (
         f"Unexpected status text: {text[:200]}"
     )
 
 
-def test_inject_transactions(echidna):
+def test_inject_transactions(echidna_server):
     """inject_fuzz_transactions accepts a call sequence and confirms injection."""
-    result = _call_tool("inject_fuzz_transactions", {
+    result = _call_tool(echidna_server["url"], "inject_fuzz_transactions", {
         "transactions": (
             "transfer(0x1111111111111111111111111111111111111111, 100);"
             "approve(0x2222222222222222222222222222222222222222, 50)"
@@ -112,23 +54,23 @@ def test_inject_transactions(echidna):
     )
 
 
-def test_show_coverage(echidna):
+def test_show_coverage(echidna_server):
     """show_coverage returns a non-empty coverage report."""
-    result = _call_tool("show_coverage")
+    result = _call_tool(echidna_server["url"], "show_coverage")
     text = _text(result)
     assert isinstance(text, str), "show_coverage response is not a string"
     assert len(text) > 0, "show_coverage returned empty report"
 
 
-def test_clear_priorities_and_status(echidna):
+def test_clear_priorities_and_status(echidna_server):
     """clear_fuzz_priorities succeeds, then status is still reachable."""
-    clear = _call_tool("clear_fuzz_priorities")
+    clear = _call_tool(echidna_server["url"], "clear_fuzz_priorities")
     clear_text = _text(clear)
     assert clear_text, "clear_fuzz_priorities returned empty response"
     assert "requested" in clear_text.lower() or "clear" in clear_text.lower(), (
         f"Unexpected clear response: {clear_text[:200]}"
     )
 
     # Status should still work after clearing
-    status = _call_tool("status")
+    status = _call_tool(echidna_server["url"], "status")
     assert _text(status), "status failed after clear_fuzz_priorities"