Web search MCP servers compared

# The same 12 searches through four web search MCP servers, called the way an agent calls them (MCP tools).
# Each query has a page a good result list should contain near the top; we count how often it is in the first 5.
# pip install mcp tiktoken    export SKRYP_API_KEY=... FIRECRAWL_API_KEY=...    (needs Node.js and uv)
# timeout: 900
import asyncio
import os
import re
import statistics
import time
from urllib.parse import urlsplit

import tiktoken
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client

API = os.environ.get("SKRYP_API_URL", "https://api.skryp.dev")
ENC = tiktoken.get_encoding("cl100k_base")

# (query, domains that hold the answer)
QUERIES = [
    ("python requests timeout", ("requests.readthedocs.io", "docs.python-requests.org")),
    ("playwright python installation", ("playwright.dev",)),
    ("model context protocol specification", ("modelcontextprotocol.io",)),
    ("beautifulsoup find_all", ("crummy.com",)),
    ("tokio spawn", ("docs.rs", "tokio.rs")),
    ("South African Reserve Bank repo rate", ("resbank.co.za",)),
    ("next.js app router documentation", ("nextjs.org",)),
    ("IANA time zone database", ("iana.org",)),
    ("firecrawl pricing", ("firecrawl.dev",)),
    ("portal 2 steam", ("store.steampowered.com",)),
    ("postgres jsonb index", ("postgresql.org",)),
    ("kubernetes horizontal pod autoscaler", ("kubernetes.io",)),
]

# name: (how to start it, search tool, arguments)
SERVERS = {
    "Skryp": ("http", "search", lambda q: {"query": q, "limit": 10}),
    "Firecrawl": (StdioServerParameters(command="npx", args=["-y", "firecrawl-mcp"],
                                        env={**os.environ, "FIRECRAWL_API_KEY": os.environ.get("FIRECRAWL_API_KEY", "")}),
                  "firecrawl_search", lambda q: {"query": q, "limit": 10}),
    "DuckDuckGo": (StdioServerParameters(command="uvx", args=["duckduckgo-mcp-server"]), "search",
                   lambda q: {"query": q, "max_results": 10}),
    "open-websearch": (StdioServerParameters(command="npx", args=["-y", "open-websearch@latest"], env={**os.environ, "MODE": "stdio"}),
                       "search", lambda q: {"query": q, "limit": 10}),
}

URL = re.compile(r"https?://[^\s\"'<>)\]]+")


def urls_in(text: str) -> list[str]:
    """Distinct pages in a reply, in order (a link to a section of a page counts as that page)."""
    out = []
    for u in URL.findall(text):
        u = u.rstrip(".,;:").split("#", 1)[0]
        if u not in out and "duckduckgo.com/l/" not in u:
            out.append(u)
    return out


def hit(urls: list[str], domains: tuple[str, ...]) -> bool:
    return any((urlsplit(u).hostname or "").removeprefix("www.").endswith(domains) for u in urls[:5])


async def run(session: ClientSession, tool: str, args) -> list[dict]:
    rows = []
    for q, domains in QUERIES:
        t0 = time.monotonic()
        try:
            r = await session.call_tool(tool, args(q))
            text = "\n".join(getattr(c, "text", "") for c in r.content)
            err = text[:80] if r.is_error else None
        except Exception as e:  # a failed call is a result too
            text, err = "", f"{type(e).__name__}"
        urls = urls_in(text)
        rows.append({"q": q, "ok": hit(urls, domains), "ms": (time.monotonic() - t0) * 1000, "n": len(urls),
                     "tokens": len(ENC.encode(text, disallowed_special=())), "err": err})
    return rows


async def one(name: str) -> list[dict]:
    how, tool, args = SERVERS[name]
    if how == "http":
        client = create_mcp_http_client(headers={"Authorization": f"Bearer {os.environ['SKRYP_API_KEY']}"})
        async with client, streamable_http_client(f"{API}/mcp", http_client=client) as (r, w):
            async with ClientSession(r, w) as s:
                await s.initialize()
                return await run(s, tool, args)
    async with stdio_client(how, errlog=open(os.devnull, "w")) as (r, w):
        async with ClientSession(r, w) as s:
            await s.initialize()
            return await run(s, tool, args)


async def main() -> None:
    results = {name: await one(name) for name in SERVERS}
    print(f"{'server':<16}{'answer in top 5':>16}{'median s':>10}{'pages':>7}{'tokens':>8}  errors")
    for name, rows in results.items():
        errs = sum(1 for r in rows if r["err"])
        print(f"{name:<16}{sum(r['ok'] for r in rows):>10} of {len(rows)}{statistics.median(r['ms'] for r in rows) / 1000:>10.1f}"
              f"{statistics.median(r['n'] for r in rows):>7.0f}{statistics.median(r['tokens'] for r in rows):>8.0f}  {errs}")
    print("\nper query (x = the answer's site was in the top 5)")
    names = list(results)
    print(f"{'':<38}" + "".join(f"{n[:10]:>11}" for n in names))
    for i, (q, _) in enumerate(QUERIES):
        print(f"{q:<38}" + "".join(f"{('x' if results[n][i]['ok'] else '-'):>11}" for n in names))


asyncio.run(main())