Tools from MCP servers in the web app's chat (serve/mcp.py)

A small MCP client (JSON-RPC 2.0, stdio and Streamable HTTP, standard library
only): servers from the run config ("mcp_servers", or Claude Desktop's
"mcpServers" block) or --mcp-config start with the server; their tools are
offered as <server>__<tool>. A request that sends "strata_mcp": true (the web
app does) gets them: the server runs the calls, feeds the results back and lets
the model continue, up to mcp.max_rounds; the stream carries strata_mcp events.
Other clients are unchanged. Errors, crashes and timeouts become "error: ..."
results; long results are cut at mcp.max_result_chars; Stop cancels a running
tool. GET /mcp lists servers and tools; the web app shows each call as a
collapsible block, a switch in the Sampling drawer and a Monitor card.
The mock engine takes --script more than once (answers in turn).
This commit is contained in:
Niko1221 committed 2026-09-28 17:28:01 +02:00
1 parent de71982be9
commit 4c623f1d78
8 files changed
+1663 -36

No files matched your search

+45 -1
View File
@@ -4,7 +4,8 @@ The technical side of Strata: every measured number, the API, images, all settin
New here? Start with the [README](../README.md) - it has everything you need to install and use it.
> **On this page:** [Speed](#speed-measured) · [Other GPUs](#other-gpus-estimated) · [Which model?](#which-model) ·
> [Requirements](#before-you-start) · [Windows](#windows) · [Linux](#linux) · [API](#using-it) · [Images](#images-vision) ·
> [Requirements](#before-you-start) · [Windows](#windows) · [Linux](#linux) · [API](#using-it) ·
> [MCP tools](#tools-from-mcp-servers) · [Images](#images-vision) ·
> [Troubleshooting](#troubleshooting) · [How it works](#how-it-works)
---
@@ -244,6 +245,7 @@ The server listens on `http://127.0.0.1:8080` (change with `--port` in setup, or
| Model list / health | `GET /v1/models`, `GET /health` |
| What the model is doing right now | `GET /status` |
| Everything the Monitor tab shows (engine, live state, last requests, hardware) | `GET /metrics` |
| The MCP servers, their state and tools ([below](#tools-from-mcp-servers)) | `GET /mcp` |
```bash
curl http://127.0.0.1:8080/v1/chat/completions -H "Content-Type: application/json" -d '{
@@ -314,6 +316,48 @@ each batch got them). `top_k` keeps at most 64 candidates: `0` ("off") or anythi
---
## Tools from MCP servers
The chat page can give the model tools from [MCP](https://modelcontextprotocol.io) servers, as LM Studio and Claude
Desktop do: reading your files, fetching web pages, searching, anything an MCP server offers. List the servers in
`strata-<model>.json` under `"mcp_servers"` - the same shape as Claude Desktop's `mcpServers` block, which you can
also paste as it is (key `"mcpServers"`):
```json
"mcp_servers": {
"files": {"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "C:\\Users\\me\\Documents\\notes"]},
"search": {"url": "http://127.0.0.1:3000/mcp", "headers": {"Authorization": "Bearer ..."}}
},
"mcp": {"timeout_s": 60, "max_result_chars": 20000, "max_rounds": 8}
```
Or keep them in their own file and start the server with `--mcp-config path\to\claude_desktop_config.json` (a file
with an `mcpServers` block; add it to the `serve/server.py` line of your run script). Restart Strata after a change.
- **A program** (`command`, `args`, optional `env` and `cwd`) is started by Strata and spoken to over its
stdin/stdout; `npx`, `uvx`, `python` and friends are found on `PATH` as usual (Node.js is needed for `npx`
servers). **An address** (`url`, optional `headers`) uses MCP's Streamable HTTP transport (the older SSE-only
transport is not supported). `"disabled": true` leaves an entry out.
- The servers start with Strata, in the background; the server window says what each one offers
(`MCP server 'files': 14 tools (...)`), or why it did not start - its tools are then left out and the chat works
without them. The Monitor tab lists them, and the Sampling drawer has **Use tools from MCP servers** (on by
default). A server that stops later is started again at its next call.
- In the chat each call shows as a small block (tool, arguments, result); the model reads the result and goes on,
up to `max_rounds` calls in a row per answer. A tool that fails or takes longer than `timeout_s` (default 60 s)
gives the model an `error: ...` result instead of ending the chat. Results longer than `max_result_chars`
(default 20,000 characters) are cut, with a note, before the model reads them. Stop stops a running tool too.
- Only the chat page uses them. API clients (omp, Claude Code, OpenAI and Anthropic SDKs) see the API exactly as
before and keep their own tools; a request to `/v1/chat/completions` opts in with `"strata_mcp": true` (it then
gets `strata_mcp` tool events in the stream).
**Security.** MCP tools run on your PC with your user's rights, and **the model decides when to call them** - also
because of what it reads (a web page or a file can contain instructions). Give a filesystem server only the folders
it needs, prefer read-only tools, and don't add servers you don't trust. The tools can only be used from the chat
page itself (a request with another site's Origin or without a JSON content type is refused); if Strata is reachable
from other devices, set an API key.
---
## Images (vision)
The model has a vision encoder: [`mmproj-Qwen3.8-Flash-Next-BF16.gguf`](https://huggingface.co/ISTA-DASLab/Qwen3.8-Flash-Next-GSQ-RCO-GGUF)
+669
View File
@@ -0,0 +1,669 @@
"""serve/mcp.py - tools from MCP (Model Context Protocol) servers for the web app's chats.
The user lists servers in the run config (`"mcp_servers"`, or Claude Desktop's `"mcpServers"` block pasted as is) or
in a separate file (`--mcp-config`). Each one is either a program Strata starts (`command`, `args`, `env`, `cwd`:
the stdio transport, newline-delimited JSON-RPC 2.0 on its stdin/stdout) or an address (`url`, `headers`: the
Streamable HTTP transport, JSON-RPC POSTed, answered as JSON or as an event stream). Both are implemented here with
the standard library only; no MCP SDK is needed.
`McpHub` owns every configured server: it starts them in the background when the server starts, offers their tools to
the model as OpenAI function tools named `<server>__<tool>` (two servers may both have a `search`), and routes a call
back to the server that has it. A server that does not start is left out with one log line; the chat works without
it. A server that stops later (a crash) is started again on its next call.
Tool results reach the model as text: the MCP content blocks joined, capped at `max_result_chars` (a directory tree or
a web page can be megabytes, far more than a context holds) with a note saying so. A failed call - the tool's own
error, a crash, a timeout - becomes a result that starts with "error:", so the model can react instead of the chat
breaking.
"""
from __future__ import annotations
import collections
import json
import os
import re
import shutil
import signal
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
PROTOCOL = "2025-06-18" # the MCP revision Strata asks for; the server's answer is used as given
DEFAULTS = {"timeout_s": 60.0, "max_result_chars": 20000, "max_rounds": 8, "start_timeout_s": 120.0}
class McpError(RuntimeError):
"""A server answered with an error, stopped, or could not be reached."""
class McpTimeout(McpError):
pass
class McpCancelled(McpError):
"""The chat was stopped while the call ran."""
class _Slot:
"""One request waiting for its response."""
def __init__(self):
self.done = threading.Event()
self.msg = None
self.error = None
def _wait(slot: _Slot, timeout: float, cancel: threading.Event | None, what: str):
"""Wait for a response, checking the chat's stop button and the deadline a few times a second."""
deadline = time.monotonic() + timeout
while not slot.done.wait(0.05):
if cancel is not None and cancel.is_set():
raise McpCancelled("stopped")
if time.monotonic() > deadline:
raise McpTimeout(f"{what} timed out after {timeout:g} s")
if slot.error:
raise slot.error
msg = slot.msg
if "error" in msg:
err = msg["error"] if isinstance(msg["error"], dict) else {"message": str(msg["error"])}
raise McpError(f"{err.get('message') or 'error'} (code {err.get('code')})")
return msg.get("result") or {}
# ------------------------------------------------------------------------------------------------ transports
class StdioTransport:
"""A server Strata starts: JSON-RPC messages one per line on its stdin and stdout; its stderr is its log."""
kind = "stdio"
def __init__(self, name: str, cfg: dict):
self.name, self.cfg = name, cfg
self.proc = None
self.lock = threading.Lock() # the pending requests
self.write_lock = threading.Lock() # stdin; separate, so a full pipe can't block the reader
self.pending: dict[int, _Slot] = {}
self.next_id = 0
self.stderr = collections.deque(maxlen=40) # its last log lines, for the error message when it fails
self.protocol = PROTOCOL
def start(self):
env = dict(os.environ)
env.update({str(k): str(v) for k, v in (self.cfg.get("env") or {}).items()})
command = str(self.cfg["command"])
# On Windows `npx`, `uvx` and friends are .cmd files that CreateProcess finds only with their extension
exe = shutil.which(command, path=env.get("PATH")) or command
extra = {}
if os.name == "nt":
extra["creationflags"] = getattr(subprocess, "CREATE_NO_WINDOW", 0)
else:
extra["start_new_session"] = True # its own process group, so close() ends its children too
try:
self.proc = subprocess.Popen([exe, *[str(a) for a in self.cfg.get("args") or []]], stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=self.cfg.get("cwd") or None,
env=env, **extra)
except OSError as e:
raise McpError(f"could not start {command!r}: {e}") from None
threading.Thread(target=self._read, daemon=True).start()
threading.Thread(target=self._read_stderr, daemon=True).start()
def alive(self) -> bool:
return self.proc is not None and self.proc.poll() is None and not getattr(self, "ended", False)
@staticmethod
def _lines(f):
"""A pipe's lines until it closes (or close() closed it under us)."""
try:
yield from f
except (OSError, ValueError):
return
def _read_stderr(self):
for raw in self._lines(self.proc.stderr):
line = raw.decode("utf-8", "replace").rstrip()
if line:
self.stderr.append(line)
if os.environ.get("STRATA_DEBUG"):
print(f"[mcp {self.name}] {line}", flush=True)
def _read(self):
for raw in self._lines(self.proc.stdout):
line = raw.decode("utf-8", "replace").strip()
if not line:
continue
try:
msg = json.loads(line)
except ValueError: # a server that prints to stdout; the spec forbids it
self.stderr.append(line)
continue
for m in msg if isinstance(msg, list) else [msg]:
if isinstance(m, dict):
self._dispatch(m)
self.ended = True
code = self.proc.poll()
err = McpError(f"the server stopped{f' (exit code {code})' if code is not None else ''}{self._tail()}")
with self.lock:
slots, self.pending = list(self.pending.values()), {}
for s in slots:
s.error = err
s.done.set()
def _tail(self) -> str:
return f": {self.stderr[-1][:300]}" if self.stderr else ""
def _dispatch(self, m: dict):
if "method" in m:
if "id" in m: # a request from the server: we offer no client features
try:
if m["method"] == "ping":
self._send({"jsonrpc": "2.0", "id": m["id"], "result": {}})
else:
self._send({"jsonrpc": "2.0", "id": m["id"],
"error": {"code": -32601, "message": f"method not supported: {m['method']}"}})
except McpError:
pass
return # notifications (log messages, list changes): not used
with self.lock:
slot = self.pending.pop(m.get("id"), None)
if slot is not None: # an answer to a call that timed out is simply dropped
slot.msg = m
slot.done.set()
def _send(self, msg: dict):
data = json.dumps(msg, ensure_ascii=False).encode("utf-8") + b"\n"
try:
with self.write_lock:
self.proc.stdin.write(data)
self.proc.stdin.flush()
except (OSError, ValueError):
raise McpError(f"the server stopped{self._tail()}") from None
def request(self, method: str, params: dict, timeout: float, cancel=None):
if not self.alive():
raise McpError(f"the server is not running{self._tail()}")
slot = _Slot()
with self.lock:
self.next_id += 1
rid = self.next_id
self.pending[rid] = slot
try:
self._send({"jsonrpc": "2.0", "id": rid, "method": method, "params": params})
return _wait(slot, timeout, cancel, method)
except (McpTimeout, McpCancelled) as e:
with self.lock:
self.pending.pop(rid, None)
try: # tell the server to stop working on it
self.notify("notifications/cancelled", {"requestId": rid, "reason": str(e)})
except McpError:
pass
raise
def notify(self, method: str, params: dict | None = None):
self._send({"jsonrpc": "2.0", "method": method, **({"params": params} if params is not None else {})})
def close(self):
"""As the spec says: close its stdin, give it a moment to exit, then end it (and what it started)."""
p = self.proc
if p is None:
return
try:
p.stdin.close()
except OSError:
pass
try:
p.wait(timeout=2)
except subprocess.TimeoutExpired:
self._kill_tree(p)
for f in (p.stdout, p.stderr):
try:
f.close()
except OSError:
pass
@staticmethod
def _kill_tree(p):
try:
if os.name == "nt": # npx.cmd -> node: end the whole tree
subprocess.run(["taskkill", "/PID", str(p.pid), "/T", "/F"], stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, timeout=10)
else:
os.killpg(p.pid, signal.SIGTERM)
try:
p.wait(timeout=2)
except subprocess.TimeoutExpired:
os.killpg(p.pid, signal.SIGKILL)
except (OSError, subprocess.SubprocessError):
pass
try:
p.kill()
except OSError:
pass
class HttpTransport:
"""A server at an address (Streamable HTTP): every message is a POST; the answer is JSON or an event stream that
carries it. The session id the server hands out at `initialize` rides on every later request."""
kind = "http"
def __init__(self, name: str, cfg: dict):
self.name, self.url = name, str(cfg["url"])
self.headers = {str(k): str(v) for k, v in (cfg.get("headers") or {}).items()}
self.session = None
self.protocol = None # sent as MCP-Protocol-Version once negotiated
self.next_id = 0
self.lock = threading.Lock()
self.broken = None
def start(self):
pass
def alive(self) -> bool:
return self.broken is None
def _headers(self) -> dict:
h = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream",
"User-Agent": "strata", **self.headers}
if self.session:
h["Mcp-Session-Id"] = self.session
if self.protocol:
h["MCP-Protocol-Version"] = self.protocol
return h
def _post(self, msg: dict, rid, timeout: float, slot: _Slot | None):
try:
req = urllib.request.Request(self.url, data=json.dumps(msg, ensure_ascii=False).encode("utf-8"),
headers=self._headers(), method="POST")
with urllib.request.urlopen(req, timeout=timeout) as r:
if r.headers.get("Mcp-Session-Id"):
self.session = r.headers["Mcp-Session-Id"]
if slot is None: # a notification: 202, nothing to read
return
ctype = r.headers.get("Content-Type", "")
if "text/event-stream" in ctype:
found = self._from_events(r, rid)
else:
body = r.read()
found = None
parsed = json.loads(body) if body.strip() else None
for m in parsed if isinstance(parsed, list) else [parsed]:
if isinstance(m, dict) and m.get("id") == rid and "method" not in m:
found = m
if found is None:
raise McpError("the server's answer did not contain a response")
slot.msg = found
except urllib.error.HTTPError as e:
detail = ""
try:
detail = e.read().decode("utf-8", "replace")[:300]
except OSError:
pass
if e.code == 404 and self.session: # the session expired: a restart starts a new one
self.broken = McpError("the server ended the session")
err = McpError(f"HTTP {e.code} {e.reason}{': ' + detail if detail else ''}")
if slot is None:
raise err from None
slot.error = err
except (OSError, ValueError, McpError) as e:
err = e if isinstance(e, McpError) else McpError(f"could not reach {self.url}: {e}")
if slot is None:
raise err from None
slot.error = err
finally:
if slot is not None:
slot.done.set()
@staticmethod
def _from_events(r, rid):
"""Read the event stream until the event that answers `rid` (it may carry notifications first)."""
data = []
for raw in r:
line = raw.decode("utf-8", "replace").rstrip("\r\n")
if line.startswith("data:"):
data.append(line[5:].lstrip(" "))
elif not line and data:
try:
m = json.loads("\n".join(data))
except ValueError:
m = None
data = []
for x in m if isinstance(m, list) else [m]:
if isinstance(x, dict) and x.get("id") == rid and "method" not in x:
return x
return None
def request(self, method: str, params: dict, timeout: float, cancel=None):
if self.broken:
raise self.broken
with self.lock:
self.next_id += 1
rid = self.next_id
slot = _Slot()
# the POST runs on its own thread so the stop button and the deadline are noticed while it waits
threading.Thread(target=self._post, args=({"jsonrpc": "2.0", "id": rid, "method": method, "params": params},
rid, timeout + 5, slot), daemon=True).start()
try:
return _wait(slot, timeout, cancel, method)
except (McpTimeout, McpCancelled) as e:
threading.Thread(target=self._quiet_notify, args=("notifications/cancelled",
{"requestId": rid, "reason": str(e)}), daemon=True).start()
raise
def _quiet_notify(self, method, params):
try:
self.notify(method, params)
except McpError:
pass
def notify(self, method: str, params: dict | None = None):
self._post({"jsonrpc": "2.0", "method": method, **({"params": params} if params is not None else {})},
None, 10, None)
def close(self):
if not self.session:
return
try: # end the session on the server (it may not allow that)
req = urllib.request.Request(self.url, headers=self._headers(), method="DELETE")
urllib.request.urlopen(req, timeout=3).close()
except (OSError, ValueError):
pass
# ------------------------------------------------------------------------------------------------ one server
class McpServer:
"""One configured server: started once (and again after it stopped), with the tools it listed."""
def __init__(self, name: str, cfg: dict, settings: dict):
self.name, self.cfg, self.settings = name, cfg, settings
self.kind = "http" if cfg.get("url") else "stdio"
self.status = "idle" # idle -> starting -> ready | failed; ready -> stopped when it ends
self.error = None
self.tools: list[dict] = []
self.info: dict = {}
self.transport = None
self.lock = threading.Lock()
self.last_start = 0.0
def start(self) -> bool:
with self.lock:
return self._start()
def _start(self) -> bool:
if self.transport is not None:
self.transport.close()
self.status, self.error, self.last_start = "starting", None, time.monotonic()
t = HttpTransport(self.name, self.cfg) if self.kind == "http" else StdioTransport(self.name, self.cfg)
self.transport = t
timeout = float(self.settings["start_timeout_s"])
try:
t.start()
res = t.request("initialize", {"protocolVersion": PROTOCOL, "capabilities": {},
"clientInfo": {"name": "strata", "title": "Strata", "version": "1"}},
timeout)
t.protocol = str(res.get("protocolVersion") or PROTOCOL)
self.info = {k: v for k, v in (res.get("serverInfo") or {}).items() if k in ("name", "title", "version")}
self.info["protocol"] = t.protocol
t.notify("notifications/initialized")
tools, cursor = [], None
for _ in range(100): # pages; a server that never stops paging is cut off
try:
page = t.request("tools/list", {"cursor": cursor} if cursor else {}, timeout)
except McpError as e:
if "code -32601" in str(e) and not tools: # a server without tools (prompts/resources only)
break
raise
tools += [x for x in page.get("tools") or [] if isinstance(x, dict) and x.get("name")]
cursor = page.get("nextCursor")
if not cursor:
break
self.tools, self.status = tools, "ready"
names = ", ".join(x["name"] for x in tools[:8]) + (", ..." if len(tools) > 8 else "")
print(f"[strata] MCP server {self.name!r}: {len(tools)} tool{'s' * (len(tools) != 1)}"
f"{' (' + names + ')' if tools else ''}", flush=True)
return True
except McpError as e:
self.status, self.error, self.tools = "failed", str(e), []
t.close()
print(f"[strata] MCP server {self.name!r} did not start: {e}. Its tools are left out; the chat works "
"without them.", flush=True)
return False
def call(self, tool: str, arguments: dict, timeout: float, cancel=None) -> dict:
with self.lock:
if self.transport is None or not self.transport.alive():
# it stopped since its last call (a crash, an expired session): start it again; one that failed to
# start is tried again at most every 10 s, so a broken command does not run on every call
if self.status == "failed" and time.monotonic() - self.last_start < 10:
raise McpError(f"the server is not running ({self.error or 'it stopped'})")
print(f"[strata] MCP server {self.name!r} had stopped; starting it again", flush=True)
if not self._start():
raise McpError(f"the server could not be started again: {self.error}")
t = self.transport
try:
return t.request("tools/call", {"name": tool, "arguments": arguments}, timeout, cancel)
except McpError as e:
if not t.alive():
self.status, self.error = "stopped", str(e)
raise
def close(self):
with self.lock:
if self.transport is not None:
self.transport.close()
if self.status in ("ready", "starting"):
self.status = "stopped"
# ------------------------------------------------------------------------------------------------ the hub
def result_text(result: dict) -> str:
"""An MCP tool result as the text the model reads: the text blocks as they are, other blocks named."""
parts = []
for block in result.get("content") or []:
if not isinstance(block, dict):
continue
kind = block.get("type")
if kind == "text":
parts.append(str(block.get("text", "")))
elif kind == "resource":
res = block.get("resource") or {}
parts.append(str(res["text"]) if "text" in res else f"[resource {res.get('uri', '')} "
f"({res.get('mimeType', 'binary')}), not shown]")
elif kind == "resource_link":
parts.append(f"[resource link: {block.get('name') or ''} {block.get('uri', '')}]".replace(" ", " "))
elif kind in ("image", "audio"):
size = len(block.get("data") or "") * 3 // 4
parts.append(f"[{kind} ({block.get('mimeType', '?')}, {size:,} bytes): not shown to the model]")
else:
parts.append(json.dumps(block, ensure_ascii=False))
if not parts and result.get("structuredContent") is not None:
parts.append(json.dumps(result["structuredContent"], ensure_ascii=False))
return "\n".join(parts)
def _clean(name: str) -> str:
return re.sub(r"[^A-Za-z0-9_-]", "_", str(name)) or "x"
class McpHub:
"""Every configured server, and the merged, namespaced tool list the model sees."""
def __init__(self, servers: dict[str, dict], settings: dict | None = None):
self.settings = {**DEFAULTS, **(settings or {})}
self.servers = {name: McpServer(name, cfg, self.settings) for name, cfg in servers.items()}
self.threads: list[threading.Thread] = []
self._routes: dict[str, tuple[McpServer, str]] = {}
def start(self, wait: bool = False):
"""Start every server on its own thread (npx may download a package first: that must not delay the chat)."""
self.threads = [threading.Thread(target=s.start, daemon=True, name=f"mcp-{s.name}")
for s in self.servers.values()]
for t in self.threads:
t.start()
if wait:
self.wait(None)
def wait(self, timeout: float | None) -> None:
"""Wait (up to `timeout` s in all) for servers that are still starting."""
end = None if timeout is None else time.monotonic() + timeout
for t in self.threads:
t.join(None if end is None else max(0.0, end - time.monotonic()))
def routes(self) -> dict[str, tuple[McpServer, str]]:
"""tool name the model sees -> (server, the tool's own name). A server that stopped after it was ready keeps
its tools here: a call starts it again."""
out = {}
for s in self.servers.values():
if s.status not in ("ready", "stopped"):
continue
for t in s.tools:
name = f"{_clean(s.name)}__{_clean(t['name'])}"[:64]
n = 2
while name in out: # two names that differ only in cleaned characters
name = f"{name[:60]}_{n}"
n += 1
out[name] = (s, t["name"])
self._routes = out
return out
def template_tools(self, exclude=()) -> list[dict]:
"""The tools in the chat template's form ({name, description, parameters}); `exclude`: names the request
brought itself (those win)."""
out = []
for name, (s, tool) in self.routes().items():
if name in exclude:
continue
t = next(x for x in s.tools if x["name"] == tool)
schema = t.get("inputSchema") if isinstance(t.get("inputSchema"), dict) else {}
out.append({"name": name, "description": str(t.get("description") or t.get("title") or ""),
"parameters": schema or {"type": "object", "properties": {}}})
return out
def openai_tools(self) -> list[dict]:
return [{"type": "function", "function": t} for t in self.template_tools()]
def call(self, name: str, arguments: dict, cancel: threading.Event | None = None) -> dict:
"""Run one tool -> {"ok", "text" (what the model reads, capped), "chars" (its full length), "truncated",
"ms", "server", "tool"}. Errors become text starting with "error:"; only McpCancelled is raised."""
t0 = time.monotonic()
server, tool = self._routes.get(name) or self.routes().get(name, (None, name))
out = {"server": server.name if server else None, "tool": tool}
if server is None:
text, ok = f"error: there is no tool named {name!r}", False
else:
try:
res = server.call(tool, arguments if isinstance(arguments, dict) else {},
float(self.settings["timeout_s"]), cancel)
text, ok = result_text(res), not res.get("isError")
if not ok:
text = "error: " + (text or "the tool reported an error")
except McpCancelled:
raise
except McpError as e:
text, ok = f"error: {e}", False
cap = int(self.settings["max_result_chars"])
full = len(text)
if cap > 0 and full > cap:
text = text[:cap] + (f"\n\n[... truncated: the tool returned {full:,} characters; only the first "
f"{cap:,} are shown]")
out.update(ok=ok, text=text, chars=full, truncated=cap > 0 and full > cap,
ms=int((time.monotonic() - t0) * 1000))
return out
def status(self) -> dict:
"""GET /mcp: every server, its state and its tools (for the web app)."""
routes = self.routes()
servers = []
for s in self.servers.values():
names = {tool: n for n, (srv, tool) in routes.items() if srv is s}
servers.append({"name": s.name, "transport": s.kind, "status": s.status, "error": s.error,
"info": s.info,
"tools": [{"name": names.get(t["name"], t["name"]), "tool": t["name"],
"description": str(t.get("description") or "")[:300]} for t in s.tools]})
return {"servers": servers, "tools": len(routes),
"settings": {k: self.settings[k] for k in ("timeout_s", "max_result_chars", "max_rounds")}}
def close(self):
for s in self.servers.values():
try:
s.close()
except Exception: # noqa: BLE001 - closing is best effort
pass
# ------------------------------------------------------------------------------------------------ config
def _check_server(name, cfg, where) -> dict:
if not isinstance(cfg, dict):
raise SystemExit(f"[strata] {where}: MCP server {name!r} must be an object with \"command\" or \"url\"")
if cfg.get("url"):
if cfg.get("type") == "sse":
raise SystemExit(f"[strata] {where}: MCP server {name!r} uses the old SSE transport (\"type\": \"sse\"); "
"Strata speaks Streamable HTTP - most servers offer it at /mcp")
if not isinstance(cfg.get("headers") or {}, dict):
raise SystemExit(f"[strata] {where}: MCP server {name!r}: \"headers\" must be an object")
return cfg
if not isinstance(cfg.get("command"), str) or not cfg["command"].strip():
raise SystemExit(f"[strata] {where}: MCP server {name!r} needs \"command\" (a program to start) or \"url\"")
if not isinstance(cfg.get("args") or [], list):
raise SystemExit(f"[strata] {where}: MCP server {name!r}: \"args\" must be a list")
if not isinstance(cfg.get("env") or {}, dict):
raise SystemExit(f"[strata] {where}: MCP server {name!r}: \"env\" must be an object")
return cfg
def servers_from(block, where: str) -> dict[str, dict]:
"""{"name": {...}} with the entries checked; `"disabled": true` leaves one out."""
if block is None:
return {}
if not isinstance(block, dict):
raise SystemExit(f"[strata] {where}: the MCP servers must be an object {{\"name\": {{...}}}}")
return {str(n): _check_server(n, c, where) for n, c in block.items()
if not (isinstance(c, dict) and c.get("disabled") is True)}
def settings_from(cfg: dict) -> dict:
"""The run config's optional `"mcp"` block: timeout_s, max_result_chars, max_rounds, start_timeout_s."""
out = {}
for key, value in (cfg.get("mcp") or {}).items():
number = isinstance(value, (int, float)) and not isinstance(value, bool)
if key in ("timeout_s", "start_timeout_s"):
if not number or value <= 0:
raise SystemExit(f"[strata] config mcp.{key}={value!r}: expected a number of seconds > 0")
out[key] = float(value)
elif key in ("max_result_chars", "max_rounds"):
if not number or value != int(value) or value < 1:
raise SystemExit(f"[strata] config mcp.{key}={value!r}: expected a positive integer")
out[key] = int(value)
else:
print(f"[strata] config mcp.{key}={value!r}: unknown key, ignored", flush=True)
return out
def hub_from_config(cfg: dict, mcp_config_path: str | None = None) -> McpHub | None:
"""The run config's `"mcp_servers"` (or `"mcpServers"`) plus the servers in the --mcp-config file (Claude
Desktop's format: {"mcpServers": {...}}); a name in both takes the file's entry. None when there are none."""
servers = {}
servers.update(servers_from(cfg.get("mcp_servers"), "config mcp_servers"))
servers.update(servers_from(cfg.get("mcpServers"), "config mcpServers"))
if mcp_config_path:
try:
with open(mcp_config_path, encoding="utf-8-sig") as f:
data = json.load(f)
except (OSError, ValueError) as e:
raise SystemExit(f"[strata] --mcp-config {mcp_config_path}: {e}") from None
block = data.get("mcpServers", data.get("mcp_servers")) if isinstance(data, dict) else None
if block is None:
raise SystemExit(f"[strata] --mcp-config {mcp_config_path}: expected {{\"mcpServers\": {{...}}}}")
servers.update(servers_from(block, f"--mcp-config {mcp_config_path}"))
if not servers:
return None
return McpHub(servers, settings_from(cfg))
if __name__ == "__main__": # python -m serve.mcp config.json: list what a config offers
hub = hub_from_config(json.load(open(sys.argv[1], encoding="utf-8-sig")) if len(sys.argv) > 1 else {},
sys.argv[2] if len(sys.argv) > 2 else None)
if hub is None:
sys.exit("no MCP servers configured")
hub.start(wait=True)
print(json.dumps(hub.status(), indent=1))
hub.close()
+172
View File
@@ -0,0 +1,172 @@
"""serve/mcp_fake_server.py - a small MCP server for serve/test_mcp.py (no MCP SDK, no network beyond localhost).
python serve/mcp_fake_server.py stdio: JSON-RPC lines on stdin/stdout
python serve/mcp_fake_server.py --http 8765 Streamable HTTP on 127.0.0.1:8765/mcp
Tools: echo (returns its text), fail (a tool error, isError), sleep (answers after `seconds`), big (`n` characters),
die (the process exits mid-call), add (two integers). tools/list pages two tools at a time (`--page N`), so the
client's pagination is exercised. `--crash-at-start` exits before answering anything. With FAKE_MCP_LOG set, the
server appends what it saw (calls, cancellations) to that file, one JSON line each.
"""
from __future__ import annotations
import json
import os
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
TOOLS = [
{"name": "echo", "description": "Returns the text it is given.",
"inputSchema": {"type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"]}},
{"name": "fail", "description": "Always fails.", "inputSchema": {"type": "object", "properties": {}}},
{"name": "sleep", "description": "Waits, then answers.",
"inputSchema": {"type": "object", "properties": {"seconds": {"type": "number"}}}},
{"name": "big", "description": "Returns n characters.",
"inputSchema": {"type": "object", "properties": {"n": {"type": "integer"}}}},
{"name": "die", "description": "Ends the server.", "inputSchema": {"type": "object", "properties": {}}},
{"name": "add", "description": "Adds two integers.",
"inputSchema": {"type": "object", "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}}}},
]
PAGE = 2
cancelled: set = set()
def log(event: dict):
path = os.environ.get("FAKE_MCP_LOG")
if path:
with open(path, "a", encoding="utf-8") as f:
f.write(json.dumps(event) + "\n")
def text(t, error=False):
return {"content": [{"type": "text", "text": t}], "isError": error}
def handle(msg: dict):
"""-> the response for a request, None for a notification."""
method, rid, params = msg.get("method"), msg.get("id"), msg.get("params") or {}
if "id" not in msg:
if method == "notifications/cancelled":
cancelled.add(params.get("requestId"))
log({"cancelled": params.get("requestId")})
return None
if method == "initialize":
return {"jsonrpc": "2.0", "id": rid, "result": {
"protocolVersion": params.get("protocolVersion", "2025-06-18"), "capabilities": {"tools": {}},
"serverInfo": {"name": "fake", "version": "0.1"}}}
if method == "tools/list":
start = int(params.get("cursor") or 0)
page = {"tools": TOOLS[start:start + PAGE]}
if start + PAGE < len(TOOLS):
page["nextCursor"] = str(start + PAGE)
return {"jsonrpc": "2.0", "id": rid, "result": page}
if method == "tools/call":
name, args = params.get("name"), params.get("arguments") or {}
log({"call": name, "arguments": args})
if name == "echo":
result = text(str(args.get("text", "")))
elif name == "fail":
result = text("it failed on purpose", error=True)
elif name == "sleep":
time.sleep(float(args.get("seconds", 1)))
result = text("slept")
elif name == "big":
result = text("y" * int(args.get("n", 100000)))
elif name == "die":
sys.stderr.write("dying on purpose\n")
sys.stderr.flush()
os._exit(3)
elif name == "add":
result = text(str(int(args.get("a", 0)) + int(args.get("b", 0))))
else:
return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32602, "message": f"unknown tool {name}"}}
return {"jsonrpc": "2.0", "id": rid, "result": result}
return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": f"no method {method}"}}
def stdio():
for f in (sys.stdin, sys.stdout, sys.stderr): # MCP's stdio is UTF-8, whatever the Windows code page is
f.reconfigure(encoding="utf-8")
out = threading.Lock()
def answer(msg):
r = handle(msg)
if r is not None and msg.get("id") not in cancelled:
with out:
sys.stdout.write(json.dumps(r) + "\n")
sys.stdout.flush()
for line in sys.stdin:
if not line.strip():
continue
msg = json.loads(line)
if msg.get("method") == "tools/call": # calls on threads: a slow one must not block cancellation
threading.Thread(target=answer, args=(msg,), daemon=True).start()
else:
answer(msg)
def http_server(port: int = 0) -> ThreadingHTTPServer:
"""Streamable HTTP at /mcp: JSON answers, except tools/call, which answers as an event stream (a progress
notification first); a session id from initialize is required afterwards."""
sessions = set()
class H(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.0"
def log_message(self, *a):
pass
def do_POST(self):
msg = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))))
if msg.get("method") != "initialize" and self.headers.get("Mcp-Session-Id") not in sessions:
self.send_response(400)
self.end_headers()
return
r = handle(msg)
if r is None:
self.send_response(202)
self.end_headers()
return
if msg.get("method") == "tools/call":
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.end_headers()
note = {"jsonrpc": "2.0", "method": "notifications/progress", "params": {"progress": 1}}
self.wfile.write(b"event: message\ndata: " + json.dumps(note).encode() + b"\n\n")
self.wfile.write(b"event: message\ndata: " + json.dumps(r).encode() + b"\n\n")
return
body = json.dumps(r).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
if msg.get("method") == "initialize":
sid = f"s-{len(sessions) + 1}"
sessions.add(sid)
self.send_header("Mcp-Session-Id", sid)
self.end_headers()
self.wfile.write(body)
def do_DELETE(self):
sessions.discard(self.headers.get("Mcp-Session-Id"))
self.send_response(200)
self.end_headers()
httpd = ThreadingHTTPServer(("127.0.0.1", port), H)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
return httpd
if __name__ == "__main__":
if "--page" in sys.argv:
PAGE = int(sys.argv[sys.argv.index("--page") + 1])
if "--crash-at-start" in sys.argv:
sys.stderr.write("cannot start: missing configuration\n")
sys.exit(2)
if "--http" in sys.argv:
server = http_server(int(sys.argv[sys.argv.index("--http") + 1]))
print(f"listening on http://127.0.0.1:{server.server_address[1]}/mcp", flush=True)
threading.Event().wait()
stdio()
+167 -20
View File
@@ -4,7 +4,9 @@
python -m serve.server --engine strata --config strata.json --port 8080 (the real engine, resident)
Endpoints: POST /v1/chat/completions (OpenAI, stream and non-stream), POST /v1/messages (Anthropic, stream and
non-stream), GET /v1/models, GET /health. One sequence at a time behind a FIFO (plan: one resident sequence).
non-stream), GET /v1/models, GET /health, GET /mcp. One sequence at a time behind a FIFO (plan: one resident sequence).
Tools from MCP servers (serve/mcp.py, `"mcp_servers"` in the config or --mcp-config) are offered only to requests that
ask for them with `"strata_mcp": true` - the web app does; other clients see exactly the API they always saw.
Images (optional, when the config has a "vision" entry): OpenAI image_url parts and Anthropic image blocks (base64
data, http(s) URLs or local file paths) go through `strata-vision` (the model's mmproj file) and reach the engine as
embeddings (`GENI`). JPEG/PNG/BMP/GIF go straight in; WebP, TIFF, AVIF, ... (agents like omp send WebP) are
@@ -41,6 +43,7 @@ sys.path.insert(0, str(ROOT / "tools"))
sys.path.insert(0, str(ROOT)) # run as a script (run-<model>.bat) as well as a module
from serve.frontend import (ChatTemplate, Event, OutputParser, anthropic_to_messages, # noqa: E402
images_of, openai_to_messages)
from serve.mcp import McpCancelled, hub_from_config # noqa: E402
IM_END = "<|im_end|>"
IMAGE_PAD = "<|image_pad|>"
@@ -54,16 +57,23 @@ class Engine(Protocol):
class MockEngine:
"""Replays a scripted completion (text) as token ids, one per step, then the end-of-turn token."""
"""Replays a scripted completion (text) as token ids, one per step, then the end-of-turn token. Given a list of
scripts, each request gets the next one and the last one repeats (a tool call, then the answer after it)."""
def __init__(self, tokenizer, script: str, max_context: int = 32768, delay_s: float = 0.0):
def __init__(self, tokenizer, script: str | list[str], max_context: int = 32768, delay_s: float = 0.0):
self.tok, self.max_context, self.delay = tokenizer, max_context, delay_s
self.script = tokenizer.encode(script, parse_special=True) + tokenizer.encode(IM_END, parse_special=True)
end = tokenizer.encode(IM_END, parse_special=True)
self.scripts = [tokenizer.encode(x, parse_special=True) + end for x in ([script] if isinstance(script, str)
else script)]
self.script, self.turns = self.scripts[0], 0
self.last_prompt: list[int] = []
def generate(self, ids, max_new, sampling, cancel, embeddings=None):
self.last_prompt = list(ids)
self.last_embeddings = embeddings
if len(self.scripts) > 1:
self.script = self.scripts[min(self.turns, len(self.scripts) - 1)]
self.turns += 1
for t in self.script[:max_new]:
if cancel.is_set():
return
@@ -506,6 +516,7 @@ class Service:
self.totals = {"since": time.time(), "requests": 0, "prompt_tokens": 0, "reused": 0, "output_tokens": 0,
"prompt_ms": 0.0, "decode_ms": 0.0}
self.status_lock = threading.Lock()
self.mcp = None # serve/mcp.py's McpHub when MCP servers are configured
self.stop_ids = set(tokenizer.encode(IM_END, parse_special=True) +
tokenizer.encode("<|endoftext|>", parse_special=True))
@@ -788,8 +799,98 @@ def _debug_req(api, req, messages, tools, max_new, thinking, prompt_tokens):
f"prompt_tokens={prompt_tokens} last={last.get('role')!r}:{preview!r}", flush=True)
# ------------------------------------------------------------------------------------------------ MCP tool loop
def run_with_mcp(svc: Service, hub, messages, tools, kw, ids, thinking, max_new, max_req, sampling, cancel,
mcp_names):
"""Service.run with the MCP tools executed here: the model writes a call to an MCP tool, the server runs it, adds
the call and its result to the conversation and lets the model continue - up to `max_rounds` times. Yields what
Service.run yields (text, thinking, the request's own tool calls) plus ("mcp", {...}) for the tool activity, and
one ("done", ...) at the very end with the output tokens of every round.
`mcp_names`: the MCP tools this request offered; any other call is one of the request's own tools and ends the
turn as always (the client answers it). MCP calls written in the same answer are then not run (their results
could not reach the model before the client's)."""
max_rounds = int(hub.settings["max_rounds"])
total, rounds, done = 0, 0, None
messages = list(messages)
while True:
text, reasoning, calls, own_calls = [], [], [], 0
for kind, x in svc.run(ids, thinking, tools, max_new, sampling, cancel):
if kind == "done":
done = x
continue
if kind == "event":
ev: Event = x
if ev.call is not None and ev.call.name in mcp_names:
if ev.kind == "tool_start": # the model has started writing a call: say so at once
yield "mcp", {"event": "start", "id": ev.call.id, "name": ev.call.name}
elif ev.kind == "tool_call":
calls.append(ev.call)
continue # its argument pieces are not streamed to the client
if ev.kind == "tool_call":
own_calls += 1
elif ev.kind == "content":
text.append(ev.text)
elif ev.kind == "reasoning":
reasoning.append(ev.text)
yield kind, x
total += done["completion_tokens"]
run_them = calls and not own_calls and done["finish"] == "stop" and not cancel.is_set()
if run_them and rounds >= max_rounds:
yield "mcp", {"event": "limit", "max_rounds": max_rounds}
run_them = False
if not run_them:
for c in calls: # announced, never run: close them in the client's view
yield "mcp", {"event": "result", "id": c.id, "ok": False, "skipped": True, "text": "not run",
"chars": 0, "truncated": False, "ms": 0}
break
rounds += 1
results = []
for c in calls:
s, tool = hub.routes().get(c.name, (None, c.name))
yield "mcp", {"event": "call", "id": c.id, "name": c.name, "server": s.name if s else None,
"tool": tool, "arguments": c.arguments, "round": rounds}
# The call runs on a thread while this generator keeps yielding heartbeats: they reach the client as
# keep-alives, which is how a client that went away (the web app's Stop) is noticed during a slow tool.
box = {}
def work(c=c, box=box):
try:
box["r"] = hub.call(c.name, c.arguments, cancel)
except McpCancelled:
box["cancelled"] = True
worker = threading.Thread(target=work, daemon=True)
worker.start()
try:
while worker.is_alive():
worker.join(1.0)
if worker.is_alive():
yield "ping", None
except GeneratorExit:
cancel.set() # the client is gone: stop the tool too
raise
if "r" not in box:
break
r = box["r"]
print(f"[strata] tool {c.name}: {'ok' if r['ok'] else 'error'}, {r['chars']:,} characters in "
f"{r['ms'] / 1000:.1f} s{' (truncated for the model)' if r['truncated'] else ''}", flush=True)
results.append(r["text"])
yield "mcp", {"event": "result", "id": c.id, **{k: r[k] for k in ("ok", "text", "chars", "truncated", "ms")}}
if cancel.is_set() or len(results) < len(calls):
done = {**done, "finish": "cancel"}
break
messages.append({"role": "assistant", "content": "".join(text).strip(),
**({"reasoning_content": "".join(reasoning).strip()} if reasoning else {}),
"tool_calls": [{"function": {"name": c.name, "arguments": c.arguments}} for c in calls]})
messages += [{"role": "tool", "content": r} for r in results]
ids, thinking, max_new = svc.prepare(messages, tools, kw, max_req)
yield "done", {**done, "completion_tokens": total, "prompt_tokens": len(ids)}
# ------------------------------------------------------------------------------------------------ OpenAI
def openai_chunks(svc: Service, req: dict, ids, thinking, tools, max_new, cancel):
def openai_chunks(svc: Service, req: dict, ids, thinking, tools, max_new, cancel, run=None):
"""`run`: the events to send instead of Service.run's (run_with_mcp); its ("mcp", {...}) items become chunks with
an empty delta and a `strata_mcp` field, which only the web app reads."""
cid, created = "chatcmpl-" + uuid.uuid4().hex[:24], int(time.time())
def chunk(delta, finish=None):
@@ -799,9 +900,13 @@ def openai_chunks(svc: Service, req: dict, ids, thinking, tools, max_new, cancel
yield chunk({"role": "assistant", "content": ""})
calls = 0
streamed = {} # tool call id -> index, for calls sent piece by piece
for kind, x in svc.run(ids, thinking, tools, max_new, req, cancel):
for kind, x in run if run is not None else svc.run(ids, thinking, tools, max_new, req, cancel):
if kind == "ping":
yield None
elif kind == "mcp":
c = chunk({})
c["strata_mcp"] = x
yield c
elif kind == "event":
ev: Event = x
if ev.kind == "reasoning" and ev.text:
@@ -825,16 +930,19 @@ def openai_chunks(svc: Service, req: dict, ids, thinking, tools, max_new, cancel
else:
finish = "tool_calls" if calls and x["finish"] == "stop" else {"cancel": "stop"}.get(x["finish"], x["finish"])
last = chunk({}, finish)
last["usage"] = {"prompt_tokens": len(ids), "completion_tokens": x["completion_tokens"],
"total_tokens": len(ids) + x["completion_tokens"]}
pt = x.get("prompt_tokens", len(ids)) # after MCP rounds: the last round's prompt
last["usage"] = {"prompt_tokens": pt, "completion_tokens": x["completion_tokens"],
"total_tokens": pt + x["completion_tokens"]}
yield last
def openai_collect(chunks) -> dict:
content, reasoning, by_index, last = [], [], {}, None
content, reasoning, by_index, last, mcp = [], [], {}, None, []
for c in chunks:
if c is None: # a heartbeat
continue
if c.get("strata_mcp"):
mcp.append(c["strata_mcp"])
d = c["choices"][0]["delta"]
content.append(d.get("content") or "")
reasoning.append(d.get("reasoning_content") or "")
@@ -852,6 +960,8 @@ def openai_collect(chunks) -> dict:
msg["reasoning_content"] = "".join(reasoning)
if calls:
msg["tool_calls"] = calls
if mcp:
msg["strata_mcp"] = mcp
return {"id": last["id"], "object": "chat.completion", "created": last["created"], "model": last["model"],
"choices": [{"index": 0, "message": msg, "finish_reason": last["choices"][0]["finish_reason"]}],
"usage": last["usage"]}
@@ -1017,6 +1127,11 @@ def make_handler(svc: Service):
if self._authorized():
self._json(200, {"shared": bool(svc.shared), "defaults": svc.shared})
return
if path == "/mcp":
# the MCP servers, their state and tools (the web app's switch and Monitor card)
if self._authorized():
self._json(200, svc.mcp.status() if svc.mcp else {"servers": [], "tools": 0})
return
if path == "":
body = (ROOT / "serve" / "web" / "index.html").read_bytes()
self.send_response(200)
@@ -1064,17 +1179,22 @@ def make_handler(svc: Service):
except EngineDied as e: # before the answer started (not streamed)
self._json(503, {"error": {"type": "server_error", "message": f"{e}; the next request restarts it"}})
def _settings(self):
# They change what every client gets, so only the app's own page may set them: JSON only (a form or a
# "simple" cross-site request can't send it without a CORS preflight, which this server never grants),
# and no foreign Origin
body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
def _own_page(self, what) -> bool:
"""Only JSON (a form or a "simple" cross-site request can't send it without a CORS preflight, which this
server never grants) and no foreign Origin: a web page elsewhere must not change settings or run tools."""
if not self.headers.get("Content-Type", "").startswith("application/json"):
self._json(415, {"error": {"message": "send application/json"}})
return
return False
origin = self.headers.get("Origin")
if origin and origin.split("://", 1)[-1] != self.headers.get("Host", ""):
self._json(403, {"error": {"message": "settings can only be changed from Strata's own page"}})
self._json(403, {"error": {"message": f"{what} only from Strata's own page"}})
return False
return True
def _settings(self):
# They change what every client gets, so only the app's own page may set them
body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
if not self._own_page("settings can be changed"):
return
try:
req = json.loads(body or b"{}")
@@ -1095,11 +1215,22 @@ def make_handler(svc: Service):
def _openai(self, req):
req = svc.with_shared(req, "openai")
messages, tools, kw = openai_to_messages(req)
max_new = int(req.get("max_completion_tokens") or req.get("max_tokens") or 0) # 0/-1: the rest of the context
max_req = max_new = int(req.get("max_completion_tokens") or req.get("max_tokens") or 0) # 0/-1: the rest
use_mcp = req.get("strata_mcp") is True and svc.mcp is not None # the web app's opt-in (serve/mcp.py)
own = {t.get("name") for t in tools or []}
if use_mcp:
if not self._own_page("MCP tools can be used"): # tools run with the user's rights on this PC
return
svc.mcp.wait(10) # servers still starting (only right after start)
extra = svc.mcp.template_tools(exclude=own) # the request's own tools win a name clash
use_mcp = bool(extra)
tools = (tools or []) + extra or None
ids, thinking, max_new = svc.prepare(messages, tools, kw, max_new)
_debug_req("openai", req, messages, tools, max_new, thinking, len(ids))
cancel = threading.Event()
chunks = openai_chunks(svc, req, ids, thinking, tools, max_new, cancel)
run = run_with_mcp(svc, svc.mcp, messages, tools, kw, ids, thinking, max_new, max_req, req, cancel,
{t["name"] for t in extra}) if use_mcp else None
chunks = openai_chunks(svc, req, ids, thinking, tools, max_new, cancel, run=run)
if not req.get("stream"):
return self._json(200, openai_collect(chunks))
self._sse()
@@ -1312,7 +1443,9 @@ def main() -> int:
ap.add_argument("--host", default=None,
help="the address to listen on: 127.0.0.1 = this PC only (the default), 0.0.0.0 = also other devices "
"on your network (set an API key); also \"host\" in the config")
ap.add_argument("--script", default="Thinking about it.</think>\n\nHello from the mock engine.")
ap.add_argument("--script", action="append",
help="the mock engine's answer (default: a short greeting); given more than once, requests get "
"them in turn and the last one repeats")
ap.add_argument("--port", type=int, default=8095)
ap.add_argument("--gpu", type=int, help="the GPU to run on, as nvidia-smi numbers them (also \"gpu\" in the config)")
ap.add_argument("--tokenizer", default=str(ROOT / "pack/full/tokenizer"),
@@ -1323,6 +1456,9 @@ def main() -> int:
"(default: reject with 400, like llama.cpp; also \"fit_max_tokens\": true in the config)")
ap.add_argument("--api-key", default=os.environ.get("STRATA_API_KEY", ""),
help="require this key on /v1/* (Authorization: Bearer ... or x-api-key); also $STRATA_API_KEY")
ap.add_argument("--mcp-config", help="a JSON file with MCP servers in Claude Desktop's format ({\"mcpServers\": "
"{...}}); the web app's chat can use their tools (also \"mcp_servers\" in "
"the config)")
a = ap.parse_args()
cfg = json.loads(Path(a.config).read_text(encoding="utf-8-sig")) if a.config else {} # Notepad adds a BOM
if a.gpu is not None:
@@ -1348,6 +1484,7 @@ def main() -> int:
merges = (tpath / "merges.txt").read_text(encoding="utf-8").split("\n")
types = json.loads((tpath / "token_type.json").read_text())
tok = ST.Tokenizer(tokens, merges, types)
hub = hub_from_config(cfg, a.mcp_config) # before the minutes of loading: a bad entry stops here
if a.engine == "strata":
if not cfg:
ap.error("--engine strata needs --config")
@@ -1365,7 +1502,8 @@ def main() -> int:
engine = StrataEngine(cfg["exe"], cfg["args"], cwd=cfg.get("cwd"), log=cfg.get("log"), env=env)
warn_tight_ram(engine.info.get("arena_mib"))
else:
engine, vision, sampling_defaults = MockEngine(tok, a.script), None, {}
engine, vision, sampling_defaults = MockEngine(tok, a.script or [
"Thinking about it.</think>\n\nHello from the mock engine."]), None, {}
# the model's own chat template (exported with its tokenizer), else the original model's
tpl = tpath / "chat_template.jinja"
svc = Service(engine, tok, ChatTemplate(tpl if tpl.exists() else ROOT / "serve/chat_template.jinja"),
@@ -1383,6 +1521,13 @@ def main() -> int:
", ".join(f"{k}={v}" for k, v in svc.shared.items()), flush=True)
except (OSError, ValueError):
svc.shared = {}
if hub is not None:
import atexit
svc.mcp = hub
print(f"[strata] starting {len(hub.servers)} MCP server{'s' * (len(hub.servers) != 1)} for the web app's "
f"chat: {', '.join(hub.servers)}", flush=True)
hub.start()
atexit.register(hub.close) # the servers Strata started end with it
httpd = serve(svc, host=a.host, port=a.port)
here = "127.0.0.1" if a.host in ("0.0.0.0", "", "::") else a.host
print(f"ready: http://{here}:{a.port}/v1 (OpenAI: /v1/chat/completions, Anthropic: /v1/messages, "
@@ -1415,6 +1560,8 @@ def main() -> int:
engine.close()
if vision:
vision.close()
if hub is not None:
hub.close()
return 0
+434
View File
@@ -0,0 +1,434 @@
"""serve/test_mcp.py - tools from MCP servers (serve/mcp.py) and the web app's tool loop, against the mock engine and
the fake MCP server in serve/mcp_fake_server.py (no GPU, no pack, no MCP SDK).
python -m unittest serve.test_mcp -v
"""
from __future__ import annotations
import json
import os
import sys
import tempfile
import threading
import time
import unittest
import urllib.error
import urllib.request
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from serve import mcp_fake_server as fake # noqa: E402
from serve.frontend import ChatTemplate # noqa: E402
from serve.mcp import McpCancelled, McpHub, hub_from_config # noqa: E402
from serve.server import ByteTokenizer, MockEngine, Service, serve # noqa: E402
ROOT = Path(__file__).resolve().parents[1]
FAKE = str(ROOT / "serve" / "mcp_fake_server.py")
CTX = 8192
def stdio(*args, **extra):
return {"command": sys.executable, "args": [FAKE, *args], **extra}
def call_script(name, **params):
"""The model's text for one tool call, as Qwen's template asks for it."""
body = "".join(f"<parameter={k}>\n{v}\n</parameter>\n" for k, v in params.items())
return f"</think>\n\nLet me check.\n\n<tool_call>\n<function={name}>\n{body}</function>\n</tool_call>"
class ScriptedEngine(MockEngine):
"""A mock engine with one script per call (the last one repeats): a tool call first, then the answer."""
def __init__(self, tok, scripts, max_context=CTX, delay_s=0.0):
super().__init__(tok, list(scripts), max_context=max_context, delay_s=delay_s)
self.prompts = []
def generate(self, ids, max_new, sampling, cancel, embeddings=None):
self.prompts.append(list(ids))
yield from super().generate(ids, max_new, sampling, cancel, embeddings)
def prompt_text(self, i):
return bytes(t for t in self.prompts[i] if t < 256).decode("utf-8", "replace")
# ------------------------------------------------------------------------------------------------ the client
class Client(unittest.TestCase):
"""The stdio client against the fake server: start, pagination, calls, errors, crashes, timeouts, the cap."""
@classmethod
def setUpClass(cls):
cls.log = tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl")
cls.log.close()
os.environ["FAKE_MCP_LOG"] = cls.log.name
cls.hub = McpHub({"fake": stdio(), "paged": stdio("--page", "4"), "broken": stdio("--crash-at-start"),
"missing": {"command": "strata-no-such-program"}},
{"timeout_s": 1.5, "max_result_chars": 1000})
cls.hub.start(wait=True)
@classmethod
def tearDownClass(cls):
cls.hub.close()
del os.environ["FAKE_MCP_LOG"]
os.unlink(cls.log.name)
def test_initialize_and_list_every_page(self):
s = self.hub.servers["fake"]
self.assertEqual(s.status, "ready")
self.assertEqual(s.info["name"], "fake")
self.assertEqual(s.info["protocol"], "2025-06-18")
want = [t["name"] for t in fake.TOOLS]
self.assertEqual([t["name"] for t in s.tools], want) # 3 pages of 2
self.assertEqual([t["name"] for t in self.hub.servers["paged"].tools], want) # 2 pages of 4
def test_namespaced_openai_tools(self):
tools = self.hub.openai_tools()
names = [t["function"]["name"] for t in tools]
self.assertIn("fake__echo", names)
self.assertIn("paged__echo", names)
echo = next(t for t in tools if t["function"]["name"] == "fake__echo")
self.assertEqual(echo["type"], "function")
self.assertEqual(echo["function"]["parameters"]["properties"]["text"]["type"], "string")
def test_a_server_that_fails_is_left_out(self):
for name, why in (("broken", "missing configuration"), ("missing", "could not start")):
with self.subTest(name=name):
s = self.hub.servers[name]
self.assertEqual(s.status, "failed")
self.assertIn(why, s.error)
self.assertFalse([t for t in self.hub.template_tools() if t["name"].startswith(name + "__")])
st = {s["name"]: s for s in self.hub.status()["servers"]}
self.assertEqual(st["broken"]["status"], "failed")
self.assertEqual(st["fake"]["status"], "ready")
def test_call(self):
self.hub.routes()
r = self.hub.call("fake__echo", {"text": "hello wörld"})
self.assertEqual((r["ok"], r["text"], r["server"], r["tool"]), (True, "hello wörld", "fake", "echo"))
self.assertEqual(self.hub.call("fake__add", {"a": 2, "b": 40})["text"], "42")
def test_errors_become_text(self):
self.hub.routes()
r = self.hub.call("fake__fail", {}) # the tool's own error (isError)
self.assertFalse(r["ok"])
self.assertEqual(r["text"], "error: it failed on purpose")
r = self.hub.call("nobody__nothing", {}) # not a tool at all
self.assertTrue(r["text"].startswith("error: there is no tool"))
# a JSON-RPC error from the server
s = self.hub.servers["fake"]
with self.assertRaises(Exception) as ctx:
s.call("no-such-tool", {}, 5)
self.assertIn("unknown tool", str(ctx.exception))
def test_timeout(self):
self.hub.routes()
t0 = time.monotonic()
r = self.hub.call("fake__sleep", {"seconds": 5})
self.assertLess(time.monotonic() - t0, 4)
self.assertFalse(r["ok"])
self.assertIn("timed out after 1.5 s", r["text"])
self.assertEqual(self.hub.call("fake__echo", {"text": "still here"})["text"], "still here")
def test_truncation(self):
self.hub.routes()
r = self.hub.call("fake__big", {"n": 50000})
self.assertTrue(r["ok"])
self.assertTrue(r["truncated"])
self.assertEqual(r["chars"], 50000)
self.assertTrue(r["text"].startswith("y" * 1000 + "\n\n[... truncated"))
self.assertIn("50,000 characters", r["text"])
self.assertLess(len(r["text"]), 1200)
def test_crash_mid_call_then_restart(self):
hub = McpHub({"solo": stdio()}, {"timeout_s": 5})
hub.start(wait=True)
try:
hub.routes()
r = hub.call("solo__die", {})
self.assertFalse(r["ok"])
self.assertIn("the server stopped", r["text"])
self.assertIn("dying on purpose", r["text"])
self.assertEqual(hub.servers["solo"].status, "stopped")
self.assertIn("solo__echo", hub.routes()) # its tools stay: the next call restarts it
self.assertEqual(hub.call("solo__echo", {"text": "back"})["text"], "back")
self.assertEqual(hub.servers["solo"].status, "ready")
finally:
hub.close()
def test_cancel(self):
self.hub.routes()
cancel = threading.Event()
threading.Timer(0.3, cancel.set).start()
t0 = time.monotonic()
with self.assertRaises(McpCancelled):
self.hub.call("fake__sleep", {"seconds": 1.2}, cancel)
self.assertLess(time.monotonic() - t0, 1.0)
time.sleep(0.3)
seen = [json.loads(line) for line in Path(self.log.name).read_text().splitlines()]
self.assertTrue(any("cancelled" in e for e in seen), seen) # notifications/cancelled reached the server
class HttpClient(unittest.TestCase):
"""Streamable HTTP: JSON answers, an event-stream answer, the session id."""
@classmethod
def setUpClass(cls):
cls.httpd = fake.http_server(0)
url = f"http://127.0.0.1:{cls.httpd.server_address[1]}/mcp"
cls.hub = McpHub({"web": {"url": url}, "down": {"url": "http://127.0.0.1:9/mcp"}}, {"timeout_s": 1.5})
cls.hub.start(wait=True)
@classmethod
def tearDownClass(cls):
cls.hub.close()
cls.httpd.shutdown()
cls.httpd.server_close()
def test_list_and_call(self):
s = self.hub.servers["web"]
self.assertEqual(s.status, "ready", s.error)
self.assertEqual(len(s.tools), len(fake.TOOLS)) # pages, with the session id on each request
self.assertEqual(s.transport.session, "s-1")
self.hub.routes()
self.assertEqual(self.hub.call("web__echo", {"text": "over http"})["text"], "over http") # event stream
self.assertEqual(self.hub.call("web__fail", {})["text"], "error: it failed on purpose")
def test_timeout_and_unreachable(self):
self.hub.routes()
r = self.hub.call("web__sleep", {"seconds": 4})
self.assertIn("timed out", r["text"])
self.assertEqual(self.hub.servers["down"].status, "failed")
self.assertIn("could not reach", self.hub.servers["down"].error)
class Config(unittest.TestCase):
def test_both_spellings_and_the_file(self):
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, "claude_desktop_config.json")
Path(path).write_text(json.dumps({"mcpServers": {"b": {"url": "http://x/mcp"}, "a": stdio()}}),
encoding="utf-8")
hub = hub_from_config({"mcp_servers": {"a": {"command": "old"}},
"mcpServers": {"c": stdio(), "off": {"command": "x", "disabled": True}},
"mcp": {"timeout_s": 5, "max_result_chars": 100, "max_rounds": 3}}, path)
self.assertEqual(sorted(hub.servers), ["a", "b", "c"])
self.assertEqual(hub.servers["a"].cfg["command"], sys.executable) # the file wins a name clash
self.assertEqual(hub.servers["b"].kind, "http")
self.assertEqual((hub.settings["timeout_s"], hub.settings["max_result_chars"], hub.settings["max_rounds"]),
(5.0, 100, 3))
self.assertIsNone(hub_from_config({}))
self.assertEqual(McpHub({}).settings["max_result_chars"], 20000)
def test_bad_entries_stop_the_start(self):
for cfg in ({"mcp_servers": {"x": {}}}, {"mcp_servers": {"x": {"command": "a", "args": "b"}}},
{"mcp_servers": ["x"]}, {"mcpServers": {"x": {"url": "http://a", "type": "sse"}}},
{"mcp_servers": {"x": {"command": "a"}}, "mcp": {"timeout_s": 0}},
{"mcp_servers": {"x": {"command": "a"}}, "mcp": {"max_rounds": 1.5}}):
with self.subTest(cfg=cfg), self.assertRaises(SystemExit):
hub_from_config(cfg)
with self.assertRaises(SystemExit):
hub_from_config({}, os.path.join(tempfile.gettempdir(), "strata-no-such-mcp.json"))
# ------------------------------------------------------------------------------------------------ the tool loop
class ToolLoop(unittest.TestCase):
"""The web app's chat (`"strata_mcp": true`): the model calls an MCP tool, the server runs it and the model
answers with its result; plain API requests never see the MCP tools."""
def setUp(self):
self.log = tempfile.NamedTemporaryFile(delete=False, suffix=".jsonl")
self.log.close()
os.environ["FAKE_MCP_LOG"] = self.log.name
self.hub = McpHub({"fake": stdio(), "broken": stdio("--crash-at-start")},
{"timeout_s": 10, "max_result_chars": 500, "max_rounds": 2})
self.hub.start(wait=True)
self.httpd = None
def tearDown(self):
if self.httpd:
self.httpd.shutdown()
self.httpd.server_close()
self.hub.close()
del os.environ["FAKE_MCP_LOG"]
os.unlink(self.log.name)
def start(self, *scripts, delay_s=0.0):
tok = ByteTokenizer()
self.engine = ScriptedEngine(tok, list(scripts), delay_s=delay_s)
self.svc = Service(self.engine, tok, ChatTemplate(ROOT / "serve/chat_template.jinja"))
self.svc.mcp = self.hub
self.httpd = serve(self.svc, port=0)
self.base = f"http://127.0.0.1:{self.httpd.server_address[1]}"
def post(self, body, headers=None, stream=True):
body = {"model": "m", "messages": [{"role": "user", "content": "check it"}], "stream": stream, **body}
req = urllib.request.Request(self.base + "/v1/chat/completions", data=json.dumps(body).encode(),
headers={"Content-Type": "application/json", **(headers or {})})
try:
with urllib.request.urlopen(req, timeout=30) as r:
text = r.read().decode()
return r.status, text
except urllib.error.HTTPError as e:
with e:
return e.code, e.read().decode()
@staticmethod
def chunks(text):
return [json.loads(line[6:]) for line in text.splitlines() if line.startswith("data: {")]
def test_tool_call_then_answer(self):
self.start(call_script("fake__echo", text="hello from the tool"), "</think>\n\nThe tool said hello.")
code, text = self.post({"strata_mcp": True})
self.assertEqual(code, 200, text)
cs = self.chunks(text)
mcp = [c["strata_mcp"] for c in cs if "strata_mcp" in c]
self.assertEqual([m["event"] for m in mcp], ["start", "call", "result"])
self.assertEqual(mcp[1]["arguments"], {"text": "hello from the tool"})
self.assertEqual((mcp[1]["server"], mcp[1]["tool"]), ("fake", "echo"))
self.assertEqual((mcp[2]["ok"], mcp[2]["text"]), (True, "hello from the tool"))
self.assertEqual(len({m["id"] for m in mcp}), 1)
content = "".join((c["choices"][0]["delta"].get("content") or "") for c in cs)
self.assertEqual(content, "Let me check.The tool said hello.")
self.assertFalse([c for c in cs if c["choices"][0]["delta"].get("tool_calls")]) # nothing for the client to run
self.assertEqual(cs[-1]["choices"][0]["finish_reason"], "stop")
self.assertEqual(len(self.engine.prompts), 2)
first, second = self.engine.prompt_text(0), self.engine.prompt_text(1)
self.assertIn('"name": "fake__echo"', first) # the MCP tools are in the prompt
self.assertNotIn("broken__", first) # the server that failed is left out
self.assertIn("<function=fake__echo>\n<parameter=text>\nhello from the tool\n</parameter>", second)
self.assertIn("<tool_response>\nhello from the tool\n</tool_response>", second)
self.assertEqual(cs[-1]["usage"]["prompt_tokens"], len(self.engine.prompts[1]))
def test_non_stream(self):
self.start(call_script("fake__add", a=2, b=3), "</think>\n\n5.")
code, text = self.post({"strata_mcp": True}, stream=False)
self.assertEqual(code, 200, text)
msg = json.loads(text)["choices"][0]["message"]
self.assertEqual(msg["content"], "Let me check.5.")
self.assertEqual(msg["strata_mcp"][-1]["text"], "5")
def test_plain_requests_get_no_mcp_tools(self):
self.start(call_script("fake__echo", text="x"), "</think>\n\nnever")
code, text = self.post({})
self.assertEqual(code, 200, text)
self.assertNotIn("fake__echo", self.engine.prompt_text(0))
cs = self.chunks(text)
self.assertFalse([c for c in cs if "strata_mcp" in c])
self.assertEqual(cs[-1]["choices"][0]["finish_reason"], "tool_calls") # returned to the client as always
self.assertEqual(len(self.engine.prompts), 1)
self.assertNotIn('"call"', Path(self.log.name).read_text()) # nothing ran
for req in ({"strata_mcp": "yes"}, {"strata_mcp": 1}): # only a real true opts in
self.post(req)
self.assertNotIn("fake__echo", self.engine.prompt_text(-1))
def test_own_tools_still_go_to_the_client(self):
own = [{"type": "function", "function": {"name": "lookup", "parameters": {"type": "object", "properties": {
"q": {"type": "string"}}}}}]
self.start(call_script("lookup", q="x"), "</think>\n\nnever")
code, text = self.post({"strata_mcp": True, "tools": own})
cs = self.chunks(text)
calls = [tc for c in cs for tc in c["choices"][0]["delta"].get("tool_calls") or []]
self.assertEqual(calls[0]["function"]["name"], "lookup")
self.assertEqual(cs[-1]["choices"][0]["finish_reason"], "tool_calls")
self.assertIn('"name": "lookup"', self.engine.prompt_text(0))
self.assertIn('"name": "fake__echo"', self.engine.prompt_text(0))
self.assertEqual(len(self.engine.prompts), 1)
def test_tool_error_is_a_result(self):
self.start(call_script("fake__fail"), "</think>\n\nIt failed.")
code, text = self.post({"strata_mcp": True})
mcp = [c["strata_mcp"] for c in self.chunks(text) if "strata_mcp" in c]
self.assertEqual((mcp[-1]["ok"], mcp[-1]["text"]), (False, "error: it failed on purpose"))
self.assertIn("<tool_response>\nerror: it failed on purpose\n</tool_response>", self.engine.prompt_text(1))
self.assertEqual(self.chunks(text)[-1]["choices"][0]["finish_reason"], "stop")
def test_truncated_for_the_model(self):
self.start(call_script("fake__big", n=3000), "</think>\n\nLong.")
self.post({"strata_mcp": True})
second = self.engine.prompt_text(1)
self.assertIn("y" * 500 + "\n\n[... truncated: the tool returned 3,000 characters", second)
self.assertNotIn("y" * 501, second)
def test_max_rounds(self):
self.start(call_script("fake__echo", text="again")) # the model never stops calling
code, text = self.post({"strata_mcp": True})
mcp = [c["strata_mcp"] for c in self.chunks(text) if "strata_mcp" in c]
self.assertEqual(len(self.engine.prompts), 3) # 2 rounds of tools, then the limit
self.assertEqual(sum(m["event"] == "call" for m in mcp), 2)
self.assertIn({"event": "limit", "max_rounds": 2}, mcp)
self.assertTrue(mcp[-1].get("skipped"))
self.assertEqual(self.chunks(text)[-1]["choices"][0]["finish_reason"], "stop")
def test_stop_during_a_tool(self):
"""Closing the connection while a slow tool runs stops the tool (notifications/cancelled) and the loop."""
self.start(call_script("fake__sleep", seconds=8), "</think>\n\nnever")
body = {"model": "m", "messages": [{"role": "user", "content": "x"}], "stream": True, "strata_mcp": True}
req = urllib.request.Request(self.base + "/v1/chat/completions", data=json.dumps(body).encode(),
headers={"Content-Type": "application/json"})
r = urllib.request.urlopen(req, timeout=30)
for raw in r:
if b'"event": "call"' in raw:
break
r.close() # the web app's Stop aborts the fetch
t0 = time.monotonic()
while time.monotonic() - t0 < 6 and "cancelled" not in Path(self.log.name).read_text():
time.sleep(0.1)
self.assertIn("cancelled", Path(self.log.name).read_text())
self.assertLess(time.monotonic() - t0, 6)
time.sleep(0.5)
self.assertEqual(len(self.engine.prompts), 1) # no second round
self.assertFalse(self.svc.status["busy"])
def test_only_from_the_app_s_own_page(self):
self.start("</think>\n\nhi")
req = urllib.request.Request(self.base + "/v1/chat/completions", data=json.dumps(
{"model": "m", "messages": [{"role": "user", "content": "x"}], "strata_mcp": True}).encode(),
headers={"Content-Type": "text/plain"}) # a cross-site "simple" request
with self.assertRaises(urllib.error.HTTPError) as ctx:
urllib.request.urlopen(req, timeout=10)
self.assertEqual(ctx.exception.code, 415)
code, _ = self.post({"strata_mcp": True}, {"Origin": "http://evil.example"})
self.assertEqual(code, 403)
self.assertEqual(self.engine.prompts, [])
host = self.base.split("://", 1)[1]
code, _ = self.post({"strata_mcp": True}, {"Origin": "http://" + host})
self.assertEqual(code, 200)
def test_get_mcp(self):
self.start("</think>\n\nhi")
with urllib.request.urlopen(self.base + "/mcp", timeout=10) as r:
st = json.loads(r.read())
servers = {s["name"]: s for s in st["servers"]}
self.assertEqual(servers["fake"]["status"], "ready")
self.assertEqual(servers["fake"]["transport"], "stdio")
self.assertIn("fake__echo", [t["name"] for t in servers["fake"]["tools"]])
self.assertEqual(servers["broken"]["status"], "failed")
self.assertIn("missing configuration", servers["broken"]["error"])
self.assertEqual(st["tools"], len(fake.TOOLS))
self.assertEqual(st["settings"]["max_rounds"], 2)
self.svc.api_key = "k"
with self.assertRaises(urllib.error.HTTPError):
urllib.request.urlopen(self.base + "/mcp", timeout=10)
self.svc.api_key = ""
class NoServers(unittest.TestCase):
def test_opt_in_without_servers_is_a_plain_chat(self):
tok = ByteTokenizer()
svc = Service(ScriptedEngine(tok, ["</think>\n\nhi"]), tok, ChatTemplate(ROOT / "serve/chat_template.jinja"))
httpd = serve(svc, port=0)
try:
base = f"http://127.0.0.1:{httpd.server_address[1]}"
with urllib.request.urlopen(base + "/mcp", timeout=10) as r:
self.assertEqual(json.loads(r.read()), {"servers": [], "tools": 0})
req = urllib.request.Request(base + "/v1/chat/completions", data=json.dumps(
{"model": "m", "messages": [{"role": "user", "content": "x"}], "strata_mcp": True}).encode(),
headers={"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=10) as r:
self.assertEqual(json.loads(r.read())["choices"][0]["message"]["content"], "hi")
finally:
httpd.shutdown()
httpd.server_close()
if __name__ == "__main__":
unittest.main()
+19
View File
@@ -51,6 +51,18 @@ main { flex: 1; min-height: 0; display: flex; }
@keyframes blink { 50% { opacity: 0; } }
.msg-error { color: var(--st-danger-text); background: var(--st-danger-tint); border-radius: var(--st-r-md); padding: 10px 14px;
font-size: var(--st-fs-md); }
/* MCP tool calls inside an answer: one compact collapsible block per call */
.st-bubble .tool-call { margin: var(--st-s-3) 0; }
.tool-call > summary { min-width: 0; padding: 8px 12px; }
.tool-call__name { font: 500 var(--st-fs-sm)/1.2 var(--st-font-mono); color: var(--st-ink); white-space: nowrap; }
.tool-call__preview { flex: 1; min-width: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; font-size: var(--st-fs-xs); }
.tool-call .st-badge, .tool-call summary .small { flex: none; }
.tool-call[data-state="error"] { border-color: color-mix(in srgb, var(--st-danger) 35%, var(--st-line)); }
.tool-call[data-state="running"] .st-badge::before, .tool-call[data-state="writing"] .st-badge::before { animation: blink 1s steps(2) infinite; }
.tool-call__label { font-size: var(--st-fs-xs); font-weight: var(--st-fw-medium); color: var(--st-ink-muted); margin: var(--st-s-2) 0 var(--st-s-1); }
.tool-call__pre { margin: 0; padding: 10px 12px; max-height: 320px; overflow: auto; white-space: pre-wrap; overflow-wrap: anywhere;
font: 400 12.5px/1.5 var(--st-font-mono); color: var(--st-ink-soft); background: var(--st-surface);
border: 1px solid var(--st-line-soft); border-radius: var(--st-r-sm); }
.composer-wrap { flex: none; padding: 0 var(--st-s-6) var(--st-s-6); }
.st-composer { max-width: 860px; margin: 0 auto; }
@@ -98,6 +110,13 @@ main { flex: 1; min-height: 0; display: flex; }
.table-wrap.all { max-height: 420px; overflow-y: auto; }
.req-totals { margin: 0; font-size: var(--st-fs-sm); }
.req-totals:empty { display: none; }
.mcp-card { display: flex; flex-direction: column; gap: var(--st-s-3); }
.mcp-server { display: flex; flex-direction: column; gap: var(--st-s-2); padding-top: var(--st-s-3); border-top: 1px solid var(--st-line-soft); }
.mcp-server:first-child { padding-top: 0; border-top: 0; }
.mcp-server__head { display: flex; align-items: center; gap: var(--st-s-2); flex-wrap: wrap; font-size: var(--st-fs-sm); }
.mcp-server__tools { display: flex; flex-wrap: wrap; gap: 6px; }
.mcp-server__tools .chip { height: 26px; padding: 0 10px; font-family: var(--st-font-mono); }
.mcp-note { margin: 0; }
/* ---------------------------------------------------------------- about */
.about { max-width: 860px; }
+147 -15
View File
@@ -78,6 +78,7 @@ function showTab(name) {
for (const v of ["chat", "monitor", "about"]) $(`view-${v}`).hidden = v !== tab;
if (location.hash.slice(1) !== tab) history.replaceState(null, "", tab === "chat" ? location.pathname : `#${tab}`);
if (tab === "chat") $("input").focus();
if (tab === "monitor") loadMcp();
if (lastMetrics) render(lastMetrics);
}
for (const b of document.querySelectorAll(".st-tab")) b.onclick = () => showTab(b.dataset.tab);
@@ -142,7 +143,7 @@ function setMetric(key, value, unit, sub) {
$(`ms-${key}`).textContent = sub || "";
}
let lastMetrics = null, metricsFailures = 0, keyWarned = false;
let lastMetrics = null, metricsFailures = 0, keyWarned = false, mcpTick = 0;
let reqShowAll = false; // the Monitor's request table: the last 12, or every one the server keeps (issue #35)
async function poll() {
try {
@@ -160,6 +161,7 @@ async function poll() {
} catch (e) {
if (++metricsFailures === 3) setPill("error", "Server not reachable");
}
if (tab === "monitor" && ++mcpTick % 10 === 0) loadMcp(); // server states change rarely: every 10 s
setTimeout(poll, 1000);
}
@@ -338,6 +340,41 @@ document.addEventListener("click", (e) => {
});
$("req-all").addEventListener("click", () => { reqShowAll = !reqShowAll; if (lastMetrics) render(lastMetrics); });
// ------------------------------------------------------------------ MCP servers (GET /mcp)
// Tools from the MCP servers in the run config: the chat offers them to the model (opt-in per request,
// "strata_mcp": true, which only this page sends); the Monitor lists the servers and what they offer.
let mcpInfo = {servers: [], tools: 0}, mcpRetry = null;
async function loadMcp() {
try {
const r = await fetch("/mcp", {headers: headers()});
if (!r.ok) return;
mcpInfo = await r.json();
} catch (e) { return; /* an older server: no MCP */ }
renderMcp();
clearTimeout(mcpRetry); // right after the start, servers may still be starting (npx downloads)
if ((mcpInfo.servers || []).some((s) => s.status === "starting")) mcpRetry = setTimeout(loadMcp, 3000);
}
const MCP_STATE = {ready: ["st-badge--generating", "Connected"], starting: ["st-badge--reading", "Starting"],
failed: ["st-badge--error", "Failed"], stopped: ["st-badge--queued", "Stopped"], idle: ["", "Waiting"]};
function renderMcp() {
const servers = mcpInfo.servers || [];
$("mcp-card").hidden = !servers.length;
$("mcp-row").hidden = !servers.length;
const ready = servers.filter((s) => s.status === "ready" || s.status === "stopped");
$("mcp-sum").textContent = servers.length ? `${fmt(mcpInfo.tools)} tools · ${ready.length} of ${servers.length} servers connected` : "";
$("mcp-row-sub").textContent = mcpInfo.tools ? `${fmt(mcpInfo.tools)} tools from ${ready.map((s) => s.name).join(", ")}; the model calls them when it decides to`
: "no server is connected yet (see the Monitor)";
$("mcp-list").innerHTML = servers.map((s) => {
const [cls, text] = MCP_STATE[s.status] || ["", s.status];
const info = s.info && s.info.name ? ` · ${s.info.name}${s.info.version ? ` ${s.info.version}` : ""}` : "";
return `<div class="mcp-server"><div class="mcp-server__head"><span class="st-badge ${cls}">${esc(text)}</span>` +
`<strong>${esc(s.name)}</strong><span class="muted small">${esc(s.transport)} · ${fmt(s.tools.length)} tools${esc(info)}</span></div>` +
(s.error ? `<div class="msg-error">${esc(s.error)}</div>` : "") +
(s.tools.length ? `<div class="mcp-server__tools">${s.tools.map((t) => `<span class="chip" title="${esc(t.description || "")}">${esc(t.tool)}</span>`).join("")}</div>` : "") +
`</div>`;
}).join("");
}
// ------------------------------------------------------------------ Markdown (escaped first, then formatted)
function inline(s) {
const codes = [];
@@ -405,7 +442,7 @@ function markdown(text) {
}
// ------------------------------------------------------------------ Chat
const DEFAULTS = {thinking: "high", temperature: 0.6, top_p: 0.95, top_k: 20, max: "", seed: "", show: true, esp: true};
const DEFAULTS = {thinking: "high", temperature: 0.6, top_p: 0.95, top_k: 20, max: "", seed: "", show: true, esp: true, mcp: true};
let settings = {...DEFAULTS, ...store.get("sampling", {})};
let messages = store.get("chat", []);
let attachments = []; // {name, url}
@@ -452,7 +489,7 @@ function msgEl(m, i) {
meta.textContent = `You · ${timeStr(m.time)}`;
el.appendChild(meta);
} else {
el.innerHTML = `<details class="st-collapse" hidden><summary>${icon("thinking", "st-icon st-icon--sm")}<span class="think-title"></span>` +
el.innerHTML = `<details class="st-collapse think" hidden><summary>${icon("thinking", "st-icon st-icon--sm")}<span class="think-title"></span>` +
`${icon("chevron", "st-icon st-icon--sm st-chev")}</summary><div class="st-collapse__body thinking"></div></details>` +
`<div class="st-bubble"></div><div class="st-msg__meta"><span class="meta-text"></span>` +
`<button class="st-btn st-btn--icon" data-msg-copy aria-label="Copy the answer" title="Copy">${icon("copy")}</button></div>`;
@@ -460,8 +497,56 @@ function msgEl(m, i) {
}
return el;
}
// One MCP tool call in the answer: a compact block (name, state, a one-line preview) that opens to the arguments and
// the result as the model read it. Its body is built only while open: a result can be 20,000 characters.
const TOOL_STATE = {writing: ["st-badge--reading", "Writing"], running: ["st-badge--generating", "Running"], done: ["", "Done"],
error: ["st-badge--error", "Error"], skipped: ["st-badge--queued", "Not run"]};
function toolHtml(t, k) {
const [cls, label] = TOOL_STATE[t.state] || ["", t.state];
const args = t.arguments == null ? "" : JSON.stringify(t.arguments, null, 2);
const preview = t.result != null ? t.result : args.replace(/\s+/g, " ");
let body = "";
if (t.open) {
body = `<div class="tool-call__label">Arguments</div><pre class="tool-call__pre">${esc(args || "(being written)")}</pre>`;
if (t.result != null) {
body += `<div class="tool-call__label">${t.ok ? "Result" : "Error"}${t.chars ? ` · ${fmt(t.chars)} characters` : ""}` +
`${t.truncated ? ", cut for the model" : ""}</div><pre class="tool-call__pre">${esc(t.result)}</pre>`;
}
}
return `<details class="st-collapse tool-call" data-tool="${k}" data-state="${esc(t.state)}"${t.open ? " open" : ""}>` +
`<summary>${icon("tool", "st-icon st-icon--sm")}<span class="tool-call__name" title="${esc(t.name || "")}">${esc(t.tool || t.name || "tool")}</span>` +
(t.server ? `<span class="muted small">${esc(t.server)}</span>` : "") +
`<span class="tool-call__preview muted">${esc(preview.slice(0, 200))}</span>` +
`<span class="st-badge ${cls}">${esc(label)}</span>${t.ms != null && t.state !== "skipped" ? `<span class="muted small">${fmt(t.ms / 1000, 1)} s</span>` : ""}` +
`${icon("chevron", "st-icon st-icon--sm st-chev")}</summary><div class="st-collapse__body">${body}</div></details>`;
}
// the answer's text with the tool blocks where the model called them
function answerHtml(m) {
if (!m.tools || !m.tools.length) return markdown(m.text || "");
let html = "", pos = 0;
m.tools.forEach((t, k) => {
const at = Math.min(Math.max(t.at || 0, pos), m.text.length);
if (at > pos) html += markdown(m.text.slice(pos, at));
pos = at;
html += toolHtml(t, k);
});
return html + markdown(m.text.slice(pos));
}
// a tool event from the stream (the `strata_mcp` field of a chunk)
function onTool(m, x) {
if (x.event === "limit") { m.limit = x.max_rounds; return; }
m.tools = m.tools || [];
let t = m.tools.find((y) => y.id === x.id);
if (!t) { t = {id: x.id, name: x.name, at: m.text.length, rat: m.reasoning.length, state: "writing"}; m.tools.push(t); }
if (x.event === "call") {
Object.assign(t, {name: x.name, server: x.server, tool: x.tool, arguments: x.arguments, round: x.round, state: "running"});
} else if (x.event === "result") {
Object.assign(t, {result: x.text, ok: x.ok, chars: x.chars, truncated: x.truncated, ms: x.ms,
state: x.skipped ? "skipped" : x.ok ? "done" : "error"});
}
}
function updateAssistant(el, m, streaming) {
const det = el.querySelector("details");
const det = el.querySelector("details.think");
if (m.reasoning) {
det.hidden = false;
const thinkingNow = streaming && !m.text;
@@ -478,10 +563,10 @@ function updateAssistant(el, m, streaming) {
if (m.error) {
bubble.innerHTML = `<div class="msg-error"></div>`;
bubble.firstChild.textContent = m.error;
} else if (!m.text && streaming) {
} else if (!m.text && streaming && !(m.tools && m.tools.length)) {
bubble.innerHTML = m.reasoning ? `<span class="muted cursor">Writing</span>` : `<span class="cursor"></span>`;
} else {
bubble.innerHTML = markdown(m.text || "");
bubble.innerHTML = answerHtml(m);
if (streaming) bubble.classList.add("cursor"); else bubble.classList.remove("cursor");
}
el.querySelector(".meta-text").textContent = m.meta || (streaming ? "" : m.stopped ? "Stopped" : "");
@@ -501,11 +586,20 @@ $("chat").addEventListener("click", (e) => {
const cc = e.target.closest("[data-code-copy]");
if (cc) { copyText(cc.closest(".st-code").querySelector("pre").textContent, cc); return; }
const mc = e.target.closest("[data-msg-copy]");
if (mc) { const i = +mc.closest(".st-msg").dataset.i; copyText(messages[i].text, mc); }
if (mc) { const i = +mc.closest(".st-msg").dataset.i; copyText(messages[i].text, mc); return; }
// a tool block: its open state lives in the message (the answer is rebuilt while it streams), so the click sets it
const sum = e.target.closest(".tool-call > summary");
if (sum) {
e.preventDefault();
const el = sum.closest(".st-msg"), m = messages[+el.dataset.i], t = m && m.tools && m.tools[+sum.parentElement.dataset.tool];
if (!t) return;
t.open = !t.open;
updateAssistant(el, m, !!busy && busy.msg === m);
}
});
$("chat").addEventListener("toggle", (e) => {
const d = e.target;
if (d.tagName !== "DETAILS") return;
if (d.tagName !== "DETAILS" || !d.classList.contains("think")) return;
if (d._auto) { d._auto = false; return; } // our own open/close, not the user's
d.dataset.touched = "1";
const body = d.querySelector(".thinking");
@@ -520,12 +614,31 @@ function apiMessages() {
const text = userText(m);
out.push({role: "user", content: imgs.length ? [{type: "text", text},
...imgs.map((i) => ({type: "image_url", image_url: {url: i.url}}))] : text});
} else if (m.text && !m.error) {
out.push({role: "assistant", content: m.text});
} else if (!m.error) {
out.push(...assistantMessages(m));
}
}
return out;
}
// An answer that used MCP tools goes back as the model wrote it: per round the text before the calls, the calls and
// their results (as the model read them), then the rest - so the next question can build on what the tools found.
function assistantMessages(m) {
const ran = (m.tools || []).filter((t) => t.round != null && t.result != null && t.state !== "skipped");
if (!ran.length) return m.text ? [{role: "assistant", content: m.text}] : [];
const out = [];
let pos = 0;
for (const r of [...new Set(ran.map((t) => t.round))]) {
const calls = ran.filter((t) => t.round === r);
const at = Math.min(Math.max(pos, calls[0].at || 0), m.text.length);
out.push({role: "assistant", content: m.text.slice(pos, at).trim(),
tool_calls: calls.map((t) => ({id: t.id, type: "function", function: {name: t.name, arguments: JSON.stringify(t.arguments || {})}}))});
for (const t of calls) out.push({role: "tool", tool_call_id: t.id, content: t.result});
pos = at;
}
const rest = m.text.slice(pos).trim();
if (rest) out.push({role: "assistant", content: rest});
return out;
}
function setBusy(on) {
$("stop-btn").hidden = !on;
@@ -560,6 +673,7 @@ async function send() {
if (settings.seed) body.seed = +settings.seed;
if (settings.max) body.max_tokens = +settings.max;
if (projectionLoaded()) body.experimental_speed_projection = !!settings.esp;
if (settings.mcp !== false && mcpInfo.tools > 0) body.strata_mcp = true; // this server may run MCP tools for it
let firstAt = null, thinkStart = null, usage = null, frame = 0;
const paint = () => { frame = 0; updateAssistant(el, m, true); scrollDown(); };
@@ -589,11 +703,19 @@ async function send() {
try { j = JSON.parse(data); } catch (e) { continue; }
if (j.error) throw new Error(j.error.message || "the engine reported an error");
if (j.usage) usage = j.usage;
if (j.strata_mcp) onTool(m, j.strata_mcp);
const d = (j.choices && j.choices[0] && j.choices[0].delta) || {};
if (d.reasoning_content) { if (!firstAt) firstAt = performance.now(); if (!thinkStart) thinkStart = performance.now(); m.reasoning += d.reasoning_content; }
const lastTool = m.tools && m.tools.length ? m.tools[m.tools.length - 1] : null; // a new round after a tool
if (d.reasoning_content) {
if (!firstAt) firstAt = performance.now();
if (!thinkStart) thinkStart = performance.now();
if (lastTool && m.reasoning && lastTool.rat === m.reasoning.length) m.reasoning += "\n\n";
m.reasoning += d.reasoning_content;
}
if (d.content) {
if (!firstAt) firstAt = performance.now();
if (thinkStart && m.thinkSecs == null) m.thinkSecs = (performance.now() - thinkStart) / 1000;
if (lastTool && m.text && lastTool.at === m.text.length) m.text += "\n\n";
m.text += d.content;
}
if (!frame) frame = requestAnimationFrame(paint);
@@ -612,6 +734,10 @@ async function send() {
} else if (m.stopped) {
m.meta = "Stopped";
}
for (const t of m.tools || []) if (t.state === "writing" || t.state === "running") { t.state = "skipped"; t.ms = null; }
const ran = (m.tools || []).filter((t) => t.state === "done" || t.state === "error").length;
if (ran) m.meta = `${m.meta ? `${m.meta} · ` : ""}${ran} tool call${ran > 1 ? "s" : ""}`;
if (m.limit) m.meta = `${m.meta || ""} · stopped at the limit of ${m.limit} tool rounds (mcp.max_rounds)`;
busy = null;
setBusy(false);
if (frame) cancelAnimationFrame(frame);
@@ -639,8 +765,11 @@ $("new-btn").onclick = () => {
};
$("export-btn").onclick = () => {
if (!messages.length) { toast("info", "Nothing to save yet"); return; }
const tools = (m) => (m.tools || []).filter((t) => t.result != null).map((t) =>
`<details><summary>Tool ${t.server ? `${t.server} / ` : ""}${t.tool || t.name}${t.ok ? "" : " (error)"}</summary>\n\n` +
`\`\`\`json\n${JSON.stringify(t.arguments || {}, null, 2)}\n\`\`\`\n\n\`\`\`\n${t.result}\n\`\`\`\n\n</details>\n\n`).join("");
const md = messages.map((m) => m.role === "user" ? `## You\n\n${m.text}\n` :
`## ${health.model}\n\n${m.reasoning ? `<details><summary>Thinking</summary>\n\n${m.reasoning}\n\n</details>\n\n` : ""}${m.text || m.error || ""}\n`).join("\n");
`## ${health.model}\n\n${m.reasoning ? `<details><summary>Thinking</summary>\n\n${m.reasoning}\n\n</details>\n\n` : ""}${tools(m)}${m.text || m.error || ""}\n`).join("\n");
const a = document.createElement("a");
a.href = URL.createObjectURL(new Blob([md], {type: "text/markdown"}));
a.download = `strata-chat-${new Date().toISOString().slice(0, 16).replace(/[:T]/g, "-")}.md`;
@@ -734,7 +863,7 @@ function openDrawer(open) {
$("drawer").dataset.open = String(open);
$("drawer").setAttribute("aria-hidden", String(!open));
$("scrim").hidden = !open;
if (open) { loadDrawer(); loadShared(); }
if (open) { loadDrawer(); loadShared(); loadMcp(); }
}
function loadDrawer(s = settings) {
for (const b of $("s-thinking").children) b.setAttribute("aria-checked", String(b.dataset.v === s.thinking));
@@ -743,6 +872,7 @@ function loadDrawer(s = settings) {
$("s-show").setAttribute("aria-checked", String(!!s.show));
$("s-esp").setAttribute("aria-checked", String(s.esp !== false));
$("esp-row").hidden = !projectionLoaded();
$("s-mcp").setAttribute("aria-checked", String(s.mcp !== false));
$("s-share").setAttribute("aria-checked", String(sharedOn));
outputs();
}
@@ -791,6 +921,7 @@ for (const b of $("s-thinking").children) b.onclick = () => { for (const x of $(
for (const id of ["s-temp", "s-topp", "s-topk"]) $(id).oninput = outputs;
$("s-show").onclick = () => $("s-show").setAttribute("aria-checked", String($("s-show").getAttribute("aria-checked") !== "true"));
$("s-esp").onclick = () => $("s-esp").setAttribute("aria-checked", String($("s-esp").getAttribute("aria-checked") !== "true"));
$("s-mcp").onclick = () => $("s-mcp").setAttribute("aria-checked", String($("s-mcp").getAttribute("aria-checked") !== "true"));
$("s-share").onclick = () => $("s-share").setAttribute("aria-checked", String($("s-share").getAttribute("aria-checked") !== "true"));
$("s-reset").onclick = () => loadDrawer(DEFAULTS);
$("s-apply").onclick = async () => {
@@ -798,7 +929,8 @@ $("s-apply").onclick = async () => {
settings = {thinking: sel ? sel.dataset.v : "high", temperature: +$("s-temp").value, top_p: +$("s-topp").value,
top_k: +$("s-topk").value, max: $("s-max").value.trim(), seed: $("s-seed").value.trim(),
show: $("s-show").getAttribute("aria-checked") === "true",
esp: $("s-esp").getAttribute("aria-checked") === "true"};
esp: $("s-esp").getAttribute("aria-checked") === "true",
mcp: $("s-mcp").getAttribute("aria-checked") === "true"};
store.set("sampling", settings);
const share = $("s-share").getAttribute("aria-checked") === "true";
openDrawer(false);
@@ -824,6 +956,6 @@ setBusy(false);
renderChat();
const startQuestion = new URLSearchParams(location.search).get("q"); // /?q=... starts a chat (a shortcut)
if (startQuestion) history.replaceState(null, "", location.pathname + location.hash);
loadHealth().then(() => { if (startQuestion) { $("input").value = startQuestion; send(); } });
loadHealth().then(loadMcp).then(() => { if (startQuestion) { $("input").value = startQuestion; send(); } });
showTab(location.hash.slice(1) || "chat");
poll();
+10
View File
@@ -124,6 +124,13 @@
<p class="muted req-totals" id="req-totals"></p>
</div>
</div>
<div class="st-card mcp-card" id="mcp-card" hidden>
<div class="req-head"><span class="card-title">MCP servers</span><span class="muted small" id="mcp-sum"></span></div>
<div class="mcp-list" id="mcp-list"></div>
<p class="muted small mcp-note">Their tools run on this PC with your rights, when the model decides to call them
(in this page's chat only; switch it off in Sampling).</p>
</div>
</div>
</section>
@@ -194,6 +201,9 @@
</div>
<div class="toggle-row"><span>Show thinking<br><span class="muted small">expanded while it streams</span></span>
<button class="st-toggle" id="s-show" role="switch" aria-checked="true" aria-label="Show thinking"></button></div>
<div class="toggle-row" id="mcp-row" hidden><span>Use tools from MCP servers<br><span class="muted small"
id="mcp-row-sub">the model may call them while it answers</span></span>
<button class="st-toggle" id="s-mcp" role="switch" aria-checked="true" aria-label="Use tools from MCP servers"></button></div>
<div class="toggle-row" id="esp-row" hidden><span>Experimental speed projection<br><span class="muted small">the
engine's control vector; off = the stock model. Switching reads the chat again once</span></span>
<button class="st-toggle" id="s-esp" role="switch" aria-checked="true" aria-label="Experimental speed projection"></button></div>