MCP servers, over streamable HTTP

A server is a row with a URL; its tools are discovered by a button and
cached, then offered beside the built-in ones. Written by hand rather than
taken from the reference SDK, because that SDK's transport does its own
connecting -- and the one thing that must not be bypassed is check_url on
every hop. Owning the transport is the point; the framing beside it is the
small part.

Sessions are per call: initialize, initialized, the call, a best-effort
DELETE. Caching one wants an owner, a TTL, eviction, a lock and a shutdown
hook, and the server may expire it under all of that anyway -- ToolContext
is a session-free snapshot precisely so nothing in a tool holds live state.

A server's names and descriptions reach the model as instructions and are
bounded before they do; what it returns is escaped preformatted text, never
markdown. Tools are namespaced per server, so two servers exposing "search"
do not collide and neither shadows a built-in.

Also: a round's calls now run together under a semaphore, results indexed
so each tool turn stays paired with its call, and generation.status names
what is running -- a remote tool is latency-bound, and a silent pause is
what a hang looks like.

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

src/lembas/services/mcp/__init__.py

@@ -0,0 +1,26 @@
+"""An MCP client: remote servers over streamable HTTP.
+
+Three parts. `protocol` is the wire format and nothing else -- it knows no
+database and no HTTP. `client` owns the transport, which is where the SSRF guard
+lives and the reason none of this comes from the reference SDK. `registry` is
+where a row becomes a tool the chat loop can be offered.
+
+Local stdio servers are deliberately absent. Spawning a subprocess is the
+agentic-execution feature, which wants a confirmation model before it does
+anything; a URL is a different act with a different blast radius.
+"""
+
+from lembas.services.mcp.client import McpSpec, call_tool, list_tools, spec_from
+from lembas.services.mcp.protocol import McpError
+from lembas.services.mcp.registry import offer_name, refresh, tool_defs
+
+__all__ = [
+    "McpError",
+    "McpSpec",
+    "call_tool",
+    "list_tools",
+    "offer_name",
+    "refresh",
+    "spec_from",
+    "tool_defs",
+]

src/lembas/services/mcp/client.py

@@ -0,0 +1,273 @@
+"""Talking to a remote MCP server over streamable HTTP.
+
+One session per call, deliberately. A cached session would need an owner, a
+lifetime, eviction, a lock -- a round runs its tools concurrently -- and a
+shutdown hook, and the server may expire it underneath all of that anyway.
+`ToolContext` is a session-free snapshot precisely so that nothing inside a tool
+holds live state. The cost is one extra POST in front of a call that is already
+a network round trip inside a reply taking seconds; the upgrade, if it is ever
+worth it, is a dict in this module and invisible to everything else.
+
+Redirects are followed by hand and every hop is re-checked, for the reason
+`services/fetch.py` gives: an administrator can point this at any URL, and a
+name resolving to 127.0.0.1 walks past any check that only reads the text.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager, suppress
+from dataclasses import dataclass, field
+from typing import Any
+from urllib.parse import quote, urlparse
+
+import httpx
+
+from lembas.db.models import SECRET_BEARER, SECRET_HEADER, SECRET_QUERY, McpServer
+from lembas.services import fetch as fetch_service
+from lembas.services.crypto import decrypt
+from lembas.services.mcp import protocol
+from lembas.services.mcp.protocol import McpError
+
+log = logging.getLogger(__name__)
+
+# A JSON-RPC message is small; anything this large is a server misbehaving, and
+# reading it into memory before parsing is the failure to avoid.
+MAX_MESSAGE_BYTES = 4 * 1024 * 1024
+
+# tools/list is paged. Both bounds exist because the whole list goes into every
+# request as schema.
+MAX_PAGES = 10
+MAX_TOOLS = 100
+
+MIN_TIMEOUT, MAX_TIMEOUT = 1, 120
+MIN_CHARS, MAX_CHARS = 200, 40_000
+
+
+@dataclass(frozen=True)
+class McpSpec:
+    """One server, read while the session was open. See `custom_tools.HttpSpec`."""
+
+    slug: str
+    name: str
+    url: str
+    headers: dict[str, str] = field(default_factory=dict)
+    secret: str = ""
+    secret_placement: str = SECRET_BEARER
+    secret_name: str = "Authorization"
+    timeout: int = 30
+    max_chars: int = 8000
+    allow_private: bool = False
+
+
+def spec_from(row: McpServer) -> McpSpec:
+    return McpSpec(
+        slug=row.slug,
+        name=row.name or row.slug,
+        url=row.url or "",
+        headers=dict(row.headers_json or {}),
+        secret=decrypt(row.secret_encrypted),
+        secret_placement=row.secret_placement,
+        secret_name=row.secret_name or "Authorization",
+        timeout=min(max(int(row.timeout or 0), MIN_TIMEOUT), MAX_TIMEOUT),
+        max_chars=min(max(int(row.max_chars or 0), MIN_CHARS), MAX_CHARS),
+        allow_private=bool(row.allow_private),
+    )
+
+
+def _headers(spec: McpSpec) -> dict[str, str]:
+    headers = {
+        "User-Agent": fetch_service.USER_AGENT,
+        "Content-Type": "application/json",
+        # Both, because a server may answer either for the same request.
+        "Accept": "application/json, text/event-stream",
+        **{str(k): str(v) for k, v in spec.headers.items()},
+    }
+    if spec.secret:
+        if spec.secret_placement == SECRET_BEARER:
+            headers[spec.secret_name or "Authorization"] = f"Bearer {spec.secret}"
+        elif spec.secret_placement == SECRET_HEADER:
+            headers[spec.secret_name or "Authorization"] = spec.secret
+    return headers
+
+
+def _endpoint(spec: McpSpec) -> str:
+    url = fetch_service.check_url(spec.url, allow_private=spec.allow_private)
+    if spec.secret and spec.secret_placement == SECRET_QUERY:
+        joiner = "&" if urlparse(url).query else "?"
+        url = f"{url}{joiner}{quote(spec.secret_name)}={quote(spec.secret, safe='')}"
+    return url
+
+
+class Session:
+    """One initialised conversation with a server."""
+
+    def __init__(self, spec: McpSpec, client: httpx.AsyncClient) -> None:
+        self.spec = spec
+        self._client = client
+        self._url = ""
+        self._headers = _headers(spec)
+        self._session_id = ""
+        self._next_id = 0
+        self.protocol_version = ""
+        self.server_info: dict[str, Any] = {}
+
+    # --- Transport -----------------------------------------------------------
+    async def _post(self, message: dict[str, Any]) -> httpx.Response:
+        current = self._url
+        headers = dict(self._headers)
+        if self._session_id:
+            headers["Mcp-Session-Id"] = self._session_id
+        if self.protocol_version:
+            headers["MCP-Protocol-Version"] = self.protocol_version
+
+        origin = (urlparse(current).scheme, urlparse(current).netloc)
+        for _ in range(fetch_service.MAX_REDIRECTS + 1):
+            try:
+                response = await self._client.post(current, json=message, headers=headers)
+            except httpx.RequestError as exc:
+                raise McpError(f"Could not reach {self.spec.name}: {exc}") from exc
+
+            if not response.is_redirect:
+                return response
+
+            location = response.headers.get("location", "")
+            if not location:
+                raise McpError(f"{self.spec.name} redirected to nowhere.")
+            if response.status_code not in (307, 308):
+                # 301, 302 and 303 turn a POST into a GET, which means nothing
+                # to a JSON-RPC endpoint. Refused rather than guessed at.
+                raise McpError(
+                    f"{self.spec.name} answered {response.status_code}, which would "
+                    "turn the request into a GET. Point the URL at the endpoint itself."
+                )
+
+            current = fetch_service.check_url(
+                str(response.url.join(location)), allow_private=self.spec.allow_private
+            )
+            if (urlparse(current).scheme, urlparse(current).netloc) != origin:
+                headers.pop(self.spec.secret_name or "Authorization", None)
+                origin = (urlparse(current).scheme, urlparse(current).netloc)
+
+        raise McpError(f"{self.spec.name} redirected too many times.")
+
+    def _read(self, response: httpx.Response, *, request_id: int) -> dict[str, Any]:
+        if response.status_code >= 400:
+            raise McpError(f"{self.spec.name} returned HTTP {response.status_code}.")
+        return protocol.result_of(
+            response.content[:MAX_MESSAGE_BYTES],
+            response.headers.get("content-type", ""),
+            request_id=request_id,
+        )
+
+    # --- Lifecycle -----------------------------------------------------------
+    async def open(self) -> None:
+        self._url = _endpoint(self.spec)
+        self._next_id += 1
+        request_id = self._next_id
+
+        response = await self._post(
+            protocol.request(
+                "initialize",
+                {
+                    "protocolVersion": protocol.PROTOCOL_VERSION,
+                    "capabilities": {},
+                    "clientInfo": protocol.CLIENT_INFO,
+                },
+                request_id=request_id,
+            )
+        )
+        # Captured before the body is read: a server that issues one expects it
+        # on everything after this, including the initialized notification.
+        self._session_id = response.headers.get("mcp-session-id", "")
+        result = self._read(response, request_id=request_id)
+
+        self.protocol_version = str(result.get("protocolVersion") or protocol.PROTOCOL_VERSION)
+        info = result.get("serverInfo")
+        self.server_info = info if isinstance(info, dict) else {}
+        if self.protocol_version != protocol.PROTOCOL_VERSION:
+            log.info(
+                "%s speaks MCP %s, we asked for %s",
+                self.spec.name,
+                self.protocol_version,
+                protocol.PROTOCOL_VERSION,
+            )
+
+        await self._post(protocol.notification("notifications/initialized"))
+
+    async def call(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
+        """One request, re-initialising once if the session has expired."""
+        self._next_id += 1
+        request_id = self._next_id
+        response = await self._post(protocol.request(method, params, request_id=request_id))
+
+        if response.status_code == 404 and self._session_id:
+            # The server dropped the session. One retry, then it is a failure
+            # like any other -- a loop here would be a loop against a server
+            # that has decided to forget us.
+            log.info("%s expired its session; re-initialising", self.spec.name)
+            self._session_id = ""
+            await self.open()
+            self._next_id += 1
+            request_id = self._next_id
+            response = await self._post(protocol.request(method, params, request_id=request_id))
+
+        return self._read(response, request_id=request_id)
+
+    async def close(self) -> None:
+        """Best effort. A server with nothing to clean up answers 405."""
+        if not self._session_id:
+            return
+        headers = {**self._headers, "Mcp-Session-Id": self._session_id}
+        if self.protocol_version:
+            headers["MCP-Protocol-Version"] = self.protocol_version
+        with suppress(httpx.RequestError, McpError):
+            await self._client.delete(self._url, headers=headers)
+
+
+@asynccontextmanager
+async def session_for(spec: McpSpec) -> AsyncIterator[Session]:
+    """An initialised session, closed afterwards whatever happened."""
+    client = httpx.AsyncClient(timeout=spec.timeout, follow_redirects=False)
+    session = Session(spec, client)
+    try:
+        await session.open()
+        yield session
+    finally:
+        await session.close()
+        await client.aclose()
+
+
+async def list_tools(spec: McpSpec) -> tuple[list[dict[str, Any]], dict[str, Any], str]:
+    """Every tool a server advertises, plus what it said about itself.
+
+    Returns (tools, serverInfo, protocolVersion). Bounded at MAX_TOOLS: the list
+    is sent as schema on every request, so a server offering two hundred is a
+    server that would fill the window before anything was asked.
+    """
+    tools: list[dict[str, Any]] = []
+    async with session_for(spec) as session:
+        cursor = ""
+        for _ in range(MAX_PAGES):
+            result = await session.call("tools/list", {"cursor": cursor} if cursor else {})
+            for entry in result.get("tools") or []:
+                cleaned = protocol.clean_tool(entry)
+                if cleaned is None:
+                    log.info("%s advertised an unusable tool entry", spec.name)
+                elif len(tools) < MAX_TOOLS:
+                    tools.append(cleaned)
+            cursor = str(result.get("nextCursor") or "")
+            if not cursor or len(tools) >= MAX_TOOLS:
+                break
+        return tools, session.server_info, session.protocol_version
+
+
+async def call_tool(spec: McpSpec, name: str, arguments: dict[str, Any]) -> tuple[str, bool]:
+    """Run one tool. Returns (text, is_error)."""
+    async with session_for(spec) as session:
+        result = await session.call("tools/call", {"name": name, "arguments": arguments})
+        return protocol.content_to_text(result), bool(result.get("isError"))
+
+
+__all__ = ["McpError", "McpSpec", "Session", "call_tool", "list_tools", "spec_from"]

src/lembas/services/mcp/protocol.py

@@ -0,0 +1,195 @@
+"""The MCP wire format: JSON-RPC 2.0, and what comes back from a tool call.
+
+Written out rather than taken from the reference SDK. The client is a few
+hundred lines of framing, and the SDK's transport does its own connecting --
+which would mean the one thing that must not be bypassed, `fetch.check_url` on
+every hop, being bypassed. Owning the transport is the point; owning the framing
+beside it is the small part.
+
+A response arrives either as one JSON object or as an event stream carrying
+several messages, and a server may choose either for the same request. Both are
+read here so `client.py` does not have to care which it got.
+"""
+
+from __future__ import annotations
+
+import json
+from contextlib import suppress
+from typing import Any
+
+from lembas import __version__
+
+# What we tell a server we speak. A server answering an older version is not
+# refused: several in the wild still answer 2024-11-05 and work perfectly.
+PROTOCOL_VERSION = "2025-06-18"
+
+CLIENT_INFO = {"name": "LLeMbas", "version": __version__}
+
+# A tool's metadata is sent to the model as instructions, so it is bounded here
+# rather than trusted. A server advertising a 40 KB description would spend the
+# context window before the conversation started.
+MAX_DESCRIPTION = 1000
+MAX_SCHEMA_BYTES = 8192
+MAX_NAME = 64
+
+# Content that is not text is described rather than forwarded. A tool turn is a
+# string, images only reach models marked as having vision, and base64 in a tool
+# result is the fastest way to fill a window with nothing.
+UNSUPPORTED = "[{kind}: {detail} — not shown to the model]"
+
+
+class McpError(Exception):
+    """A failed exchange, with a message fit to show an administrator.
+
+    Same contract as `LLMError`, `SearchError` and `FetchError`: the message is
+    the whole error, and is safe to render.
+    """
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+        self.message = message
+
+
+def request(method: str, params: dict[str, Any] | None, *, request_id: int) -> dict[str, Any]:
+    message: dict[str, Any] = {"jsonrpc": "2.0", "id": request_id, "method": method}
+    if params is not None:
+        message["params"] = params
+    return message
+
+
+def notification(method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
+    message: dict[str, Any] = {"jsonrpc": "2.0", "method": method}
+    if params is not None:
+        message["params"] = params
+    return message
+
+
+def _messages(body: bytes, content_type: str) -> list[dict[str, Any]]:
+    """Every JSON-RPC message in a response body, whichever framing was used."""
+    text = body.decode("utf-8", "replace").strip()
+    if not text:
+        return []
+
+    if "text/event-stream" not in content_type.lower():
+        try:
+            document = json.loads(text)
+        except json.JSONDecodeError as exc:
+            raise McpError(f"That server did not answer with JSON: {exc}") from exc
+        return document if isinstance(document, list) else [document]
+
+    # Frames are `data:` lines gathered until a blank line -- the same framing
+    # services/sse.py writes. Read here rather than shared, because that module
+    # is a writer and this is a reader with a size cap.
+    out: list[dict[str, Any]] = []
+    data: list[str] = []
+    for line in text.splitlines() + [""]:
+        if line.startswith("data:"):
+            data.append(line[5:].lstrip())
+        elif not line.strip() and data:
+            # A frame that is not JSON is a comment or a keep-alive, not a
+            # message; the stream carries both.
+            with suppress(json.JSONDecodeError):
+                out.append(json.loads("\n".join(data)))
+            data = []
+    return out
+
+
+def result_of(body: bytes, content_type: str, *, request_id: int) -> dict[str, Any]:
+    """The result for one request, out of whatever the server sent back.
+
+    Raises `McpError` on a JSON-RPC error member, because that is a failure the
+    administrator needs the words of -- "Unknown tool" and "invalid params" are
+    the two that actually happen.
+    """
+    for message in _messages(body, content_type):
+        if not isinstance(message, dict) or message.get("id") != request_id:
+            continue
+        if "error" in message:
+            error = message["error"] or {}
+            code = error.get("code", "")
+            text = str(error.get("message") or "The server reported an error.")
+            raise McpError(f"{text}{f' (code {code})' if code != '' else ''}")
+        result = message.get("result")
+        return result if isinstance(result, dict) else {}
+
+    raise McpError("That server did not answer the request.")
+
+
+# --- Tool metadata -----------------------------------------------------------
+def clean_tool(entry: Any) -> dict[str, Any] | None:
+    """One advertised tool, bounded. None if there is nothing usable here."""
+    if not isinstance(entry, dict):
+        return None
+    name = str(entry.get("name") or "").strip()
+    if not name or len(name) > MAX_NAME:
+        return None
+
+    schema = entry.get("inputSchema")
+    if not isinstance(schema, dict) or schema.get("type") != "object":
+        schema = {"type": "object", "properties": {}}
+    elif len(json.dumps(schema)) > MAX_SCHEMA_BYTES:
+        # Kept callable rather than dropped: a model can still be told the tool
+        # exists, and an argument it guesses is no worse than not offering it.
+        schema = {"type": "object", "properties": {}}
+
+    return {
+        "name": name,
+        "title": str(entry.get("title") or "")[:MAX_NAME],
+        "description": str(entry.get("description") or "")[:MAX_DESCRIPTION],
+        "schema": schema,
+    }
+
+
+# --- Tool results ------------------------------------------------------------
+def _block_text(block: Any) -> str:
+    if not isinstance(block, dict):
+        return ""
+    kind = str(block.get("type") or "")
+
+    if kind == "text":
+        return str(block.get("text") or "")
+
+    if kind in ("image", "audio"):
+        size = len(str(block.get("data") or ""))
+        detail = f"{block.get('mimeType') or 'unknown type'}, about {size * 3 // 4} bytes"
+        return UNSUPPORTED.format(kind=kind, detail=detail)
+
+    if kind == "resource":
+        resource = block.get("resource")
+        if not isinstance(resource, dict):
+            return ""
+        uri = str(resource.get("uri") or "")
+        if isinstance(resource.get("text"), str):
+            return f"{uri}\n{resource['text']}" if uri else str(resource["text"])
+        detail = f"{uri or 'unnamed'}, {resource.get('mimeType') or 'unknown type'}"
+        return UNSUPPORTED.format(kind="resource", detail=detail)
+
+    if kind == "resource_link":
+        return f"{block.get('name') or 'resource'} ({block.get('uri') or ''})".strip()
+
+    # An addition to the protocol degrades to a note rather than to silence:
+    # a model told nothing came back will say nothing came back.
+    return UNSUPPORTED.format(kind="content", detail=kind or "no type")
+
+
+def content_to_text(result: dict[str, Any]) -> str:
+    """A `tools/call` result as the flat text a tool turn carries."""
+    blocks = result.get("content")
+    parts = [text for block in (blocks or []) if (text := _block_text(block).strip())]
+
+    if not parts and isinstance(result.get("structuredContent"), dict | list):
+        return json.dumps(result["structuredContent"], indent=2, ensure_ascii=False)
+
+    return "\n\n".join(parts)
+
+
+__all__ = [
+    "CLIENT_INFO",
+    "PROTOCOL_VERSION",
+    "McpError",
+    "clean_tool",
+    "content_to_text",
+    "notification",
+    "request",
+    "result_of",
+]

src/lembas/services/mcp/registry.py

@@ -0,0 +1,185 @@
+"""Turning MCP servers into tools a chat can be offered.
+
+Two names per tool. The server has its own, which is what `tools/call` must be
+given; we have an *offered* name, which is what goes in the schema the endpoint
+sees. They differ because two servers both exposing `search` would collide, a
+server exposing `notes_delete` would shadow a built-in, and endpoints accept a
+narrower character set than MCP does. The rename never leaves this module: the
+runner closes over the server's own name.
+
+The advertised list is cached on the row and refreshed by a button, the same
+shape as discovering a connection's models. A server is contacted when an
+administrator asks, not when a chat starts -- a slow server must not be able to
+delay every reply.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy.orm import Session as DBSession
+
+from lembas.db.models import McpServer, User
+from lembas.services import tool_access
+from lembas.services.fetch import FetchError
+from lembas.services.mcp import client
+from lembas.services.mcp.protocol import McpError
+from lembas.services.tools import FAMILY_MCP, ToolContext, ToolDef, ToolOutcome
+
+log = logging.getLogger(__name__)
+
+# What an endpoint will accept as a function name.
+FUNCTION_NAME = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")
+MAX_NAME = 64
+
+# How much of a reply is kept on the message row for the transcript.
+MAX_EVENT_CHARS = 2000
+MAX_SUMMARY_CHARS = 200
+
+
+def offer_name(server_slug: str, tool_name: str, *, taken: set[str]) -> str:
+    """A name the endpoint will accept, unique across everything offered.
+
+    Truncation can collide where the full names would not, so a numeric suffix
+    is appended until it does not. Deterministic given a stable iteration order,
+    which is why rows are walked in (position, slug) order everywhere.
+    """
+    combined = f"{server_slug}_{tool_name}".lower()
+    cleaned = re.sub(r"_+", "_", re.sub(r"[^a-z0-9_-]", "_", combined)).strip("_")
+    candidate = (cleaned or "tool")[:MAX_NAME]
+
+    suffix = 2
+    while candidate in taken:
+        tail = f"_{suffix}"
+        candidate = f"{(cleaned or 'tool')[: MAX_NAME - len(tail)]}{tail}"
+        suffix += 1
+
+    taken.add(candidate)
+    return candidate
+
+
+def _summary(arguments: dict[str, Any]) -> str:
+    parts = [f"{name}={value!r}" for name, value in arguments.items()]
+    text = ", ".join(parts)
+    return text[:MAX_SUMMARY_CHARS]
+
+
+def _runner(spec: client.McpSpec, tool_name: str, offered: str):
+    async def run(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
+        event = {
+            "name": offered,
+            "kind": "mcp",
+            "label": f"{spec.name} · {tool_name}",
+            "query": _summary(args),
+            "detail": spec.name,
+            "results": [],
+        }
+        try:
+            text, failed = await client.call_tool(spec, tool_name, args)
+        except (McpError, FetchError) as exc:
+            message = exc.message
+            log.info("mcp %s/%s failed: %s", spec.slug, tool_name, message)
+            return ToolOutcome(
+                f"The {tool_name} tool on {spec.name} failed: {message}",
+                {**event, "status": "error", "error": message[:200]},
+            )
+
+        text = text.strip()[: spec.max_chars]
+        if failed:
+            return ToolOutcome(
+                f"The tool reported an error:\n{text}" if text else "The tool reported an error.",
+                {**event, "status": "error", "error": text[:200] or "The tool reported an error."},
+            )
+        if not text:
+            return ToolOutcome(
+                f"{tool_name} returned nothing.", {**event, "status": "ok", "text": ""}
+            )
+        return ToolOutcome(text, {**event, "status": "ok", "text": text[:MAX_EVENT_CHARS]})
+
+    return run
+
+
+def _offered_tools(server: McpServer) -> list[dict[str, Any]]:
+    """The cached entries this server is currently willing to offer."""
+    overrides = server.tool_overrides_json or {}
+    # Absent means on, the rule the model capability flags follow: a tool that
+    # appeared in the last refresh should work rather than silently do nothing.
+    return [
+        entry
+        for entry in (server.tools_json or [])
+        if isinstance(entry, dict) and overrides.get(entry.get("name"), True)
+    ]
+
+
+def tool_defs(
+    db: DBSession, user: User | None, *, everything: bool = False, taken: set[str] | None = None
+) -> list[ToolDef]:
+    """One `ToolDef` per offerable tool across every visible server."""
+    claimed = taken if taken is not None else set()
+    out: list[ToolDef] = []
+
+    for server in tool_access.visible_mcp_servers(db, user, everything=everything):
+        spec = client.spec_from(server)
+        for entry in _offered_tools(server):
+            name = str(entry.get("name") or "")
+            offered = str(entry.get("offer_name") or "") or offer_name(
+                server.slug, name, taken=claimed
+            )
+            claimed.add(offered)
+            if not FUNCTION_NAME.match(offered):
+                continue
+            out.append(
+                ToolDef(
+                    name=offered,
+                    family=f"{FAMILY_MCP}:{server.slug}",
+                    description=entry.get("description") or f"{name}, from {server.name}.",
+                    parameters=entry.get("schema") or {"type": "object", "properties": {}},
+                    run=_runner(spec, name, offered),
+                )
+            )
+
+    return out
+
+
+async def refresh(db: DBSession, server: McpServer) -> tuple[int, str]:
+    """Contact a server and cache what it advertises. Returns (count, error).
+
+    Shaped like `api/admin.py:_refresh_models`, including writing `last_error`
+    and `last_checked_at` on both paths so the row says what happened rather
+    than only whether it worked.
+    """
+    spec = client.spec_from(server)
+    try:
+        advertised, info, version = await client.list_tools(spec)
+    except (McpError, FetchError) as exc:
+        server.last_error = exc.message
+        server.last_checked_at = datetime.now(UTC)
+        db.commit()
+        return 0, exc.message
+
+    taken: set[str] = set()
+    for entry in advertised:
+        entry["offer_name"] = offer_name(server.slug, entry["name"], taken=taken)
+
+    # Choices about tools that are still advertised survive; ones about tools
+    # that have gone are dropped rather than left to accumulate.
+    names = {entry["name"] for entry in advertised}
+    server.tool_overrides_json = {
+        name: on for name, on in (server.tool_overrides_json or {}).items() if name in names
+    }
+
+    server.tools_json = advertised
+    server.server_info = info
+    server.protocol_version = version
+    server.last_error = ""
+    server.last_checked_at = datetime.now(UTC)
+    db.commit()
+
+    log.info("mcp %s advertised %d tool(s)", server.slug, len(advertised))
+    return len(advertised), ""
+
+
+__all__ = ["offer_name", "refresh", "tool_defs"]