"""A small client for vibe-sketch's REST API, on the standard library alone.
The MCP server calls the same API the browser does, so an agent's edit and a
human's go through one revision counter and show up in each other's view.
"""
from __future__ import annotations
import json
import socket
import threading
import urllib.error
import urllib.request
from typing import Any
__all__ = ["RestClient", "ServerError", "find_server", "start_server"]
[docs]
class ServerError(RuntimeError):
"""The server answered with an error status; ``detail`` is its explanation."""
def __init__(self, status: int, detail: Any) -> None:
self.status = status
self.detail = detail
super().__init__(f"{status}: {detail}")
[docs]
class RestClient:
"""JSON calls against one vibe-sketch server."""
def __init__(self, base_url: str, *, timeout: float = 300.0) -> None:
self.base_url = base_url.rstrip("/") + "/"
self.timeout = timeout
[docs]
def raw(self, method: str, path: str, body: Any = None) -> tuple[bytes, str]:
"""The response body and its content type; raises :class:`ServerError`."""
data = None if body is None else json.dumps(body).encode("utf-8")
request = urllib.request.Request(
self.base_url + path.lstrip("/"),
data=data,
method=method,
headers={"Content-Type": "application/json"} if data is not None else {},
)
try:
with urllib.request.urlopen(request, timeout=self.timeout) as response:
return response.read(), response.headers.get("Content-Type", "")
except urllib.error.HTTPError as error:
text = error.read().decode("utf-8", "replace")
try:
detail = json.loads(text).get("detail", text)
except (ValueError, AttributeError):
detail = text
raise ServerError(error.code, detail) from None
[docs]
def call(self, method: str, path: str, body: Any = None) -> Any:
"""The decoded JSON response."""
content, _ = self.raw(method, path, body)
return json.loads(content) if content else None
[docs]
def find_server(host: str = "127.0.0.1", ports: range = range(8766, 8776)) -> str | None:
"""The address of a vibe-sketch server running on this machine, or ``None``.
``serve`` takes the first free port from 8766, so the search walks the
same range and recognises vibe-sketch by its session route.
"""
for port in ports:
url = f"http://{host}:{port}/"
try:
session = RestClient(url, timeout=1.0).call("GET", "api/session")
except (OSError, ServerError, ValueError):
continue
if isinstance(session, dict) and {"documents", "default", "writers"} <= set(session):
return url
return None
[docs]
def start_server(store: Any = None, host: str = "127.0.0.1") -> str:
"""Start a vibe-sketch server in a background thread; returns its address.
The port is chosen by the operating system and the socket is handed to
uvicorn already bound, so no other process can take it in between.
"""
import time
import uvicorn
from vibesketch.server import create_app
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((host, 0))
port = int(sock.getsockname()[1])
from vibesketch.server.guard import LOOPBACK, is_loopback
hosts = LOOPBACK | {host} if is_loopback(host) else None
config = uvicorn.Config(create_app(store, hosts=hosts), log_level="warning", lifespan="off")
server = uvicorn.Server(config)
thread = threading.Thread(
target=server.run, kwargs={"sockets": [sock]}, daemon=True, name="vibe-sketch-server"
)
thread.start()
deadline = time.monotonic() + 30
while not server.started:
if not thread.is_alive() or time.monotonic() > deadline:
raise RuntimeError("the vibe-sketch server did not start")
time.sleep(0.02)
return f"http://{host}:{port}/"