Source code for vibesketch.agent.mcp
"""The MCP server: vibe-sketch as tools for an agent (roadmap WP 2.6).
``vibe-sketch mcp`` speaks the Model Context Protocol over standard input and
output (newline-delimited JSON-RPC 2.0: ``initialize``, ``ping``,
``tools/list``, ``tools/call``). Every tool is a call to the REST API, so an
agent edits documents through the same revision counter as the browser: when
a vibe-sketch server is already running on this machine, the agent works on
the documents its user has open, and the user sees each edit within seconds.
When none is running, the MCP server starts one in the background and the
``status`` tool gives its address for the user to open.
The protocol layer is a few dozen lines of the standard library; nothing here
needs an MCP SDK.
"""
from __future__ import annotations
import json
import sys
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import IO, Any
from urllib.parse import urlencode
from vibesketch.agent.client import RestClient, ServerError
from vibesketch.document.io import write_output
__all__ = ["PROTOCOL_VERSIONS", "Tool", "handle", "make_tools", "serve"]
#: Protocol revisions this server speaks, newest first.
PROTOCOL_VERSIONS = ("2025-06-18", "2025-03-26", "2024-11-05")
INSTRUCTIONS = (
"vibe-sketch draws chemical structures and reactions and writes quantum-chemistry "
"inputs. Start with `status` to see the open documents and the address of the app the "
"user can open. Add structures from SMILES, a name, a peptide, DNA or RNA sequence, "
"HELM, MOL, SDF, RXN or CDXML with "
"`add_structures`; species and reactions are derived from the drawing. `embed_3d` makes "
"3D structures; `write_inputs` writes ORCA or vibe-qc inputs for chosen species; "
"`map_reaction` and `write_neb_inputs` turn a reaction into a nudged-elastic-band input. "
"Writers refuse, with reasons, what they cannot express; read the reasons rather than "
"guessing. "
"Edits are revisioned and shared with the user's browser."
)
class ToolError(Exception):
"""A tool could not do what was asked; the message goes back to the agent."""
[docs]
@dataclass(frozen=True)
class Tool:
name: str
description: str
schema: dict[str, Any]
run: Callable[[dict[str, Any]], Any]
[docs]
def listing(self) -> dict[str, Any]:
return {"name": self.name, "description": self.description, "inputSchema": self.schema}
def _schema(properties: dict[str, Any], required: tuple[str, ...] = ()) -> dict[str, Any]:
return {
"type": "object",
"properties": properties,
"required": list(required),
"additionalProperties": False,
}
_DOCUMENT = {"type": "string", "description": "Document id; the open document when left out."}
_SPEC_HELP = (
"Program-neutral job specification: task (sp, opt, freq, opt+freq, tddft, nmr), method "
"(B3LYP, PBE0, r2SCAN-3c, HF, MP2, DLPNO-CCSD(T) ...), basis (def2-SVP ...), aux_basis "
"(def2/J), approximation (RIJCOSX), dispersion (D3BJ, D4), solvent {model: CPCM|SMD, "
"name}, states (TDDFT), constraints [{kind, atoms, value}], resources {nprocs, "
"maxcore_mb}."
)
def _summary(state: dict[str, Any]) -> dict[str, Any]:
"""A document without its canvas: what an agent reasons about."""
document = state["document"]
return {
"id": state["id"],
"revision": state["revision"],
"file": state.get("file"),
"unsaved": state.get("dirty"),
"species": [
{
"id": s["id"],
**({"label": s["label"]} if s.get("label") else {}),
"smiles": s.get("smiles"),
"charge": s.get("charge", 0),
"multiplicity": s.get("multiplicity", 1),
**({"name": s["name"]["value"]} if s.get("name") else {}),
**(
{
"nanostructure": {
k: s["nanostructure"][k] for k in ("key", "description", "formula")
}
}
if s.get("nanostructure")
else {}
),
**({"capped": s["capped"]} if s.get("capped") else {}),
**(
{"solid": {k: s["solid"][k] for k in ("text", "key") if k in s["solid"]}}
if s.get("solid")
else {}
),
"has_3d": bool(s.get("conformers")),
}
for s in document.get("species", [])
],
"reactions": document.get("reactions", []),
"jobs": document.get("jobs", []),
# R-group cores, bracketed groups and R-group definitions (#90).
**({"generics": document["generics"]} if document.get("generics") else {}),
**({"rgroups": document["rgroups"]} if document.get("rgroups") else {}),
"warnings": state.get("warnings", []),
**({"result": state["result"]} if state.get("result") else {}),
}
[docs]
def make_tools(client: RestClient) -> list[Tool]:
"""The tools, each a call to the server ``client`` points at."""
def document_id(arguments: dict[str, Any]) -> str:
given = arguments.get("document")
if given:
return str(given)
session = client.call("GET", "api/session")
if session.get("default"):
return str(session["default"])
documents = session.get("documents") or []
if not documents:
raise ToolError("no document is open; use new_document or open_document first")
return str(documents[-1]["id"])
def state_of(identifier: str) -> dict[str, Any]:
state: dict[str, Any] = client.call("GET", f"api/documents/{identifier}")
return state
def edit(arguments: dict[str, Any], method: str, path: str, body: dict[str, Any]) -> Any:
"""An edit based on the current revision, as the agent."""
identifier = document_id(arguments)
revision = state_of(identifier)["revision"]
payload = {"revision": revision, "actor": "agent", "client": "mcp", **body}
return _summary(client.call(method, f"api/documents/{identifier}/{path}", payload))
def status(arguments: dict[str, Any]) -> Any:
session = client.call("GET", "api/session")
return {
"app": client.base_url,
"documents": session["documents"],
"open": session["default"],
"writers": session["writers"],
"naming": session["naming"],
"opsin": session.get("opsin", False),
}
def new_document(arguments: dict[str, Any]) -> Any:
body: dict[str, Any] = {"actor": "agent", "client": "mcp"}
if arguments.get("format"):
body |= {"format": arguments["format"], "data": arguments.get("data", "")}
return _summary(client.call("POST", "api/documents", body))
def open_document(arguments: dict[str, Any]) -> Any:
from vibesketch.document.formats import format_for_path
from vibesketch.limits import MAX_INPUT_BYTES, MAX_QVF_BYTES, read_file
path = Path(str(arguments["path"])).expanduser()
if not path.is_file():
raise ToolError(f"{path} is not a file")
fmt = format_for_path(str(path))
raw = read_file(path, path.name, limit=MAX_QVF_BYTES if fmt == "qvf" else MAX_INPUT_BYTES)
body: dict[str, Any] = {"actor": "agent", "client": "mcp"}
if fmt in ("qvf", "cdx"):
import base64
data = base64.b64encode(raw).decode("ascii")
body |= {"format": fmt, "data": data, "encoding": "base64"}
elif fmt == "vsk":
body["document"] = json.loads(raw.decode("utf-8"))
else:
body |= {"format": fmt, "data": raw.decode("utf-8")}
return _summary(client.call("POST", "api/documents", body))
def show_document(arguments: dict[str, Any]) -> Any:
return _summary(state_of(document_id(arguments)))
def report(arguments: dict[str, Any]) -> Any:
names = "true" if arguments.get("names", True) else "false"
return client.call("GET", f"api/documents/{document_id(arguments)}/report?names={names}")
def add_structures(arguments: dict[str, Any]) -> Any:
body = {
"format": arguments["format"],
"data": arguments["data"],
"mode": arguments.get("mode", "append"),
}
return edit(arguments, "POST", "import", body)
def set_species(arguments: dict[str, Any]) -> Any:
body: dict[str, Any] = {
k: arguments[k] for k in ("label", "multiplicity", "name") if k in arguments
}
if not body:
raise ToolError("give a label, a multiplicity or a name")
species = arguments["species"]
return edit(arguments, "PATCH", f"species/{species}", body)
def clean_up(arguments: dict[str, Any]) -> Any:
return edit(arguments, "POST", "clean", {})
def tidy_scheme(arguments: dict[str, Any]) -> Any:
body = {"width": arguments["width"]} if "width" in arguments else {}
return edit(arguments, "POST", "tidy", body)
def embed_3d(arguments: dict[str, Any]) -> Any:
body: dict[str, Any] = {
k: arguments[k] for k in ("species", "count", "replace") if k in arguments
}
return edit(arguments, "POST", "embed", body)
def relax_3d(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
keys = ("species", "constraints", "force_field", "fmax", "max_steps")
body: dict[str, Any] = {k: arguments[k] for k in keys if k in arguments}
if arguments.get("embed", True):
missing = [
s["id"]
for s in _summary(state_of(identifier))["species"]
if not s["has_3d"] and (body.get("species") is None or s["id"] in body["species"])
]
if missing:
edit({"document": identifier}, "POST", "embed", {"species": missing})
revision = state_of(identifier)["revision"]
payload = {"revision": revision, "actor": "agent", "client": "mcp", **body}
state = client.call("POST", f"api/documents/{identifier}/relax", payload)
return {**_summary(state), "relaxed": state.get("relaxed", {})}
def map_reaction(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
reaction = arguments["reaction"]
body = {"pins": list(arguments.get("pins", []))}
return client.call("POST", f"api/documents/{identifier}/reactions/{reaction}/mapping", body)
def write_neb_inputs(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
reaction = arguments["reaction"]
body: dict[str, Any] = {
"program": arguments.get("program", "orca"),
"spec": arguments["spec"],
"pins": list(arguments.get("pins", [])),
"format": "json",
}
written = client.call("POST", f"api/documents/{identifier}/reactions/{reaction}/neb", body)
directory = arguments.get("directory")
if not directory:
return written
target = Path(str(directory)).expanduser()
target.mkdir(parents=True, exist_ok=True)
paths = []
for file in written["files"]:
path = target / file["name"]
path.write_text(file["content"], encoding="utf-8")
paths.append(str(path))
return {"written": paths, "mapping": written["mapping"]}
def write_inputs(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
body: dict[str, Any] = {
"program": arguments["program"],
"spec": arguments["spec"],
"format": "json",
}
if "species" in arguments:
body["species"] = arguments["species"]
if arguments.get("embed", True):
missing = [
s["id"]
for s in _summary(state_of(identifier))["species"]
if not s["has_3d"] and (body.get("species") is None or s["id"] in body["species"])
]
if missing:
edit({"document": identifier}, "POST", "embed", {"species": missing})
written = client.call("POST", f"api/documents/{identifier}/inputs", body)
directory = arguments.get("directory")
if not directory:
return written
target = Path(str(directory)).expanduser()
target.mkdir(parents=True, exist_ok=True)
paths = []
for file in written["files"]:
write_output(target / file["name"], file["content"].encode("utf-8"))
paths.append(str(target / file["name"]))
return {"written": paths, "refused": written["refused"]}
def export(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
query = "?" + urlencode(
[(str(key), arguments[key]) for key in ("style", "dpi") if key in arguments]
+ ([("stereo_labels", "true")] if arguments.get("stereo_labels") else [])
+ ([("sequences", "true")] if arguments.get("sequences") else [])
)
content, media_type = client.raw(
"GET", f"api/documents/{identifier}/export/{arguments['format']}{query}"
)
path = Path(str(arguments["path"])).expanduser()
path.parent.mkdir(parents=True, exist_ok=True)
write_output(path, content)
return {"path": str(path), "bytes": len(content), "media_type": media_type}
def save_document(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
if not arguments.get("path"):
return _summary(client.call("POST", f"api/documents/{identifier}/save"))
path = Path(str(arguments["path"])).expanduser()
fmt = "vsk" if path.suffix.lower() == ".vsk" else "qvf"
if path.suffix.lower() not in (".vsk", ".qvf"):
path = path.with_suffix(".qvf")
content, _ = client.raw("GET", f"api/documents/{identifier}/export/{fmt}")
write_output(path, content)
return {"path": str(path), "revision": state_of(identifier)["revision"]}
def list_writers(arguments: dict[str, Any]) -> Any:
return client.call("GET", "api/writers")
def submit_jobs(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
body: dict[str, Any] = {"program": arguments["program"], "spec": arguments["spec"]}
if "species" in arguments:
body["species"] = arguments["species"]
if arguments.get("scheme"):
body["reactions"] = []
elif "reactions" in arguments:
body["reactions"] = arguments["reactions"]
if "container" in arguments:
body["container"] = bool(arguments["container"])
revision = state_of(identifier)["revision"]
answer = client.call(
"POST",
f"api/documents/{identifier}/submit",
{"revision": revision, "actor": "agent", "client": "mcp", **body},
)
return {
"group": answer["group"],
"submitted": answer["submitted"],
"refused": answer["refused"],
"fallbacks": answer["fallbacks"],
}
def energy_constants(arguments: dict[str, Any]) -> Any:
return client.call("GET", "api/energy-recipes/constants")
def add_energy_recipe(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
return client.call(
"POST",
f"api/documents/{identifier}/energy-recipes",
{
**{key: arguments[key] for key in ("id", "analysis", "spec", "program")},
"revision": state_of(identifier)["revision"],
"actor": "agent",
"client": "mcp",
},
)
def compile_energy_recipe(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
return client.call(
"POST",
f"api/documents/{identifier}/energy-recipes/{arguments['recipe']}/jobs",
{
"revision": state_of(identifier)["revision"],
"actor": "agent",
"client": "mcp",
},
)
def energy_recipe_results(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
path = f"api/documents/{identifier}/energy-recipes/{arguments['recipe']}/results"
body = {"runs": arguments["runs"], "figure": arguments.get("figure", False)}
if body["figure"]:
content, _ = client.raw("POST", path, body)
return {"svg": content.decode("utf-8")}
return client.call("POST", path, body)
def show_recipes(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
query = "?" + urlencode({"recipe": arguments["recipe"]}) if "recipe" in arguments else ""
return client.call("GET", f"api/documents/{identifier}/recipes{query}")
def set_recipe(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
revision = state_of(identifier)["revision"]
editor = {"revision": revision, "actor": "agent", "client": "mcp"}
if "recipe" in arguments:
if "reference_contract" in arguments:
raise ValueError("set the reference contract inside the complete recipe")
recipe = arguments["recipe"]
answer = client.call(
"PUT", f"api/documents/{identifier}/recipes/{recipe['id']}",
{**editor, "recipe": recipe},
) # fmt: skip
else:
body: dict[str, Any] = {
k: arguments[k]
for k in ("id", "name", "species", "reference_contract", "profile")
if k in arguments
}
if arguments.get("scheme"):
body["reactions"] = []
elif "reactions" in arguments:
body["reactions"] = arguments["reactions"]
answer = client.call(
"POST", f"api/documents/{identifier}/recipes/baseline", {**editor, **body}
)
return {"revision": answer["revision"], "report": answer["report"]}
def remove_recipe(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
revision = state_of(identifier)["revision"]
answer = client.call(
"POST", f"api/documents/{identifier}/recipes/{arguments['recipe']}/remove",
{"revision": revision, "actor": "agent", "client": "mcp"},
) # fmt: skip
return _summary(answer)
def compile_campaign(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
keys = ("directory", "id", "host", "program", "python", "runner", "cpus", "mem_mb",
"accept_model_license", "stage_runtimes", "wall_time_seconds",
"check_wall_time_seconds") # fmt: skip
body = {k: arguments[k] for k in keys if k in arguments}
revision = state_of(identifier)["revision"]
answer = client.call(
"POST", f"api/documents/{identifier}/recipes/{arguments['recipe']}/campaigns",
{"revision": revision, "actor": "agent", "client": "mcp", **body},
) # fmt: skip
return {
"campaign": answer["campaign"],
"run_command": answer["run_command"],
"warnings": answer["warnings"],
}
def review_campaign(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
keys = ("directory", "node", "reviewer", "reason", "criteria", "warnings", "evidence")
return client.call(
"POST",
f"api/documents/{identifier}/recipes/{arguments['recipe']}/campaigns/"
f"{arguments['campaign']}/review",
{
"revision": state_of(identifier)["revision"],
"actor": "agent",
"client": "mcp",
**{k: arguments[k] for k in keys if k in arguments},
},
)
def revalidate_campaign(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
keys = ("directory", "node", "reviewer", "reason")
return client.call(
"POST",
f"api/documents/{identifier}/recipes/{arguments['recipe']}/campaigns/"
f"{arguments['campaign']}/revalidate",
{
"revision": state_of(identifier)["revision"],
"actor": "agent",
"client": "mcp",
**{k: arguments[k] for k in keys if k in arguments},
},
)
def campaign_status(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
recipe, campaign = arguments["recipe"], arguments["campaign"]
return client.call(
"GET", f"api/documents/{identifier}/recipes/{recipe}/campaigns/{campaign}"
)
def render_campaign_stages(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
recipe, campaign = arguments["recipe"], arguments["campaign"]
keys = ("species", "directory", "width", "height", "publication", "images")
return client.call(
"POST", f"api/documents/{identifier}/recipes/{recipe}/campaigns/{campaign}/renders",
{k: arguments[k] for k in keys if k in arguments},
) # fmt: skip
def retry_campaign_node(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
recipe, campaign = arguments["recipe"], arguments["campaign"]
return client.call(
"POST", f"api/documents/{identifier}/recipes/{recipe}/campaigns/{campaign}/retry",
{"node": arguments["node"]},
) # fmt: skip
def show_runs(arguments: dict[str, Any]) -> Any:
return client.call("GET", f"api/documents/{document_id(arguments)}/runs")
def import_run_result(arguments: dict[str, Any]) -> Any:
import base64
from pathlib import Path
from vibesketch.document.results import MAX_RESULT_BYTES
from vibesketch.limits import read_file
path = Path(arguments["file"])
data = read_file(path, path.name, limit=MAX_RESULT_BYTES)
return edit(
arguments,
"POST",
"results",
{
"run": arguments["run"],
"format": arguments["format"],
"source_name": path.name,
"data": base64.b64encode(data).decode(),
"replace": arguments.get("replace", False),
},
)
def external_chain_report(arguments: dict[str, Any]) -> Any:
import base64
from pathlib import Path
from vibesketch.document.external_results import MAX_EVIDENCE_BYTES
from vibesketch.limits import read_file
path = Path(arguments["file"])
body = {
"data": base64.b64encode(read_file(path, path.name, limit=MAX_EVIDENCE_BYTES)).decode(),
"sha256": arguments["sha256"],
"reference_contract": arguments["reference_contract"],
"format": arguments.get("format", "json"),
}
endpoint = f"api/documents/{document_id(arguments)}/external-chain-report"
if body["format"] == "json":
return client.call("POST", endpoint, body)
data, _ = client.raw("POST", endpoint, body)
return data.decode()
def campaign_energy_report(arguments: dict[str, Any]) -> Any:
import base64
from urllib.parse import quote
recipe = quote(arguments["recipe"], safe="")
campaign = quote(arguments["campaign"], safe="")
body = {
k: arguments[k]
for k in ("stage", "reactions", "coefficients", "format")
if k in arguments
}
path = (
f"api/documents/{document_id(arguments)}/recipes/{recipe}/campaigns/{campaign}/energies"
)
if body.get("format", "json") == "json":
return client.call("POST", path, body)
data, media = client.raw("POST", path, body)
if body["format"] in {"qvf", "svg"}:
return {
"format": body["format"],
"media_type": media,
"data_base64": base64.b64encode(data).decode(),
}
return {"format": body["format"], "media_type": media, "text": data.decode("utf-8")}
def show_reaction_energies(arguments: dict[str, Any]) -> Any:
body = {
k: arguments[k]
for k in ("group", "runs", "reactions", "coefficients")
if k in arguments
}
return client.call("POST", f"api/documents/{document_id(arguments)}/results/report", body)
def export_energy_table(arguments: dict[str, Any]) -> Any:
from vibesketch.document.tables import MEDIA_TYPES
fmt = arguments["format"]
body = {
k: arguments[k]
for k in ("group", "runs", "reactions", "coefficients")
if k in arguments
}
data, _ = client.raw(
"POST",
f"api/documents/{document_id(arguments)}/results/report",
{**body, "format": fmt},
)
return {"format": fmt, "media_type": MEDIA_TYPES[fmt], "text": data.decode("utf-8")}
def export_spectrum(arguments: dict[str, Any]) -> Any:
from vibesketch.document.spectra import MEDIA_TYPES
body = {k: v for k, v in arguments.items() if k != "document"}
data, _ = client.raw(
"POST", f"api/documents/{document_id(arguments)}/results/spectrum", body
)
fmt = arguments.get("format", "csv")
return {"format": fmt, "media_type": MEDIA_TYPES[fmt], "text": data.decode("utf-8")}
def export_property_table(arguments: dict[str, Any]) -> Any:
from vibesketch.document.tables import MEDIA_TYPES
fmt = arguments.get("format", "csv")
data, _ = client.raw(
"GET", f"api/documents/{document_id(arguments)}/properties?format={fmt}"
)
return {"format": fmt, "media_type": MEDIA_TYPES[fmt], "text": data.decode("utf-8")}
def place_property_text(arguments: dict[str, Any]) -> Any:
body = {"fields": arguments["fields"]} if "fields" in arguments else {}
return edit(arguments, "POST", f"species/{arguments['species']}/property-text", body)
def set_amounts(arguments: dict[str, Any]) -> Any:
return edit(
arguments,
"PUT",
f"reactions/{arguments['reaction']}/amounts",
{"amounts": arguments["amounts"]},
)
def show_stoichiometry(arguments: dict[str, Any]) -> Any:
fmt = arguments.get("format", "json")
path = (
f"api/documents/{document_id(arguments)}/reactions/{arguments['reaction']}"
f"/stoichiometry?format={fmt}"
)
if fmt == "json":
return client.call("GET", path)
data, media = client.raw("GET", path)
return {"format": fmt, "media_type": media, "text": data.decode("utf-8")}
def show_stereo(arguments: dict[str, Any]) -> Any:
return client.call("GET", f"api/documents/{document_id(arguments)}/stereo")
def label_stereo(arguments: dict[str, Any]) -> Any:
return edit(arguments, "POST", "stereo-labels", {"clear": bool(arguments.get("clear"))})
def show_nmr(arguments: dict[str, Any]) -> Any:
fmt = arguments.get("format", "json")
query = [("run", arguments["run"]), ("format", fmt)]
if arguments.get("reference"):
query.append(("reference", arguments["reference"]))
query += [("nucleus", n) for n in arguments.get("nuclei", [])]
path = f"api/documents/{document_id(arguments)}/results/nmr?{urlencode(query)}"
if fmt == "json":
return client.call("GET", path)
data, media = client.raw("GET", path)
return {"format": fmt, "media_type": media, "text": data.decode("utf-8")}
def place_nmr_labels(arguments: dict[str, Any]) -> Any:
body: dict[str, Any] = {
k: arguments[k] for k in ("run", "reference", "nuclei") if k in arguments
}
return edit(arguments, "POST", "results/nmr-labels", body)
def add_nanostructure(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
body: dict[str, Any] = {
k: arguments[k]
for k in (
"species_id",
"label",
"charge",
"multiplicity",
"picture_style",
"picture_label",
)
if k in arguments
}
payload = {
"revision": state_of(identifier)["revision"],
"actor": "agent",
"client": "mcp",
"parameters": arguments["parameters"],
**body,
}
state = client.call("POST", f"api/documents/{identifier}/nanostructures", payload)
return {**_summary(state), "nanostructure": state["nanostructure"]}
def cap_generic(arguments: dict[str, Any]) -> Any:
return edit(
arguments, "POST", f"generics/{arguments['generic']}/cap", {"cap": arguments["cap"]}
)
def add_surface(arguments: dict[str, Any]) -> Any:
return edit(
arguments,
"POST",
"surfaces",
{k: arguments[k] for k in ("parameters", "at", "width", "height") if k in arguments},
)
def attach_to_surface(arguments: dict[str, Any]) -> Any:
return edit(
arguments,
"POST",
f"generics/{arguments['generic']}/attach",
{"species": arguments["surface"], "anchor": arguments.get("anchor", "top")},
)
def list_surfaces(arguments: dict[str, Any]) -> Any:
return client.call("GET", f"api/documents/{document_id(arguments)}/surfaces")
def depict_nanostructure(arguments: dict[str, Any]) -> Any:
keys = ("style", "label", "rear", "break", "width")
body: dict[str, Any] = {k: arguments[k] for k in keys if k in arguments}
return edit(arguments, "POST", f"species/{arguments['species']}/picture", body)
def list_pictures(arguments: dict[str, Any]) -> Any:
return client.call("GET", f"api/documents/{document_id(arguments)}/pictures")
def attach_to_nanostructure(arguments: dict[str, Any]) -> Any:
body = {"species": arguments["species"], "anchor": arguments.get("anchor", "upper-end")}
if "node" in arguments:
body["node"] = arguments["node"]
return edit(arguments, "POST", f"generics/{arguments['generic']}/attach", body)
def add_half_reaction(arguments: dict[str, Any]) -> Any:
keys = ("reactants", "products", "electrons", "conditions", "arrow", "over")
body: dict[str, Any] = {k: arguments[k] for k in keys if k in arguments}
return edit(arguments, "POST", "half-reactions", body)
def check_balance(arguments: dict[str, Any]) -> Any:
return client.call("GET", f"api/documents/{document_id(arguments)}/balance")
def add_tlc_plate(arguments: dict[str, Any]) -> Any:
return edit(arguments, "POST", "tlc", {"lanes": arguments["lanes"]})
def list_named_reactions(arguments: dict[str, Any]) -> Any:
return client.call(
"GET", "api/reaction-library?" + urlencode({"q": arguments.get("query", "")})
)
def add_named_reaction(arguments: dict[str, Any]) -> Any:
return edit(
arguments,
"POST",
"reaction-library",
{
"entry": arguments["entry"],
"kind": arguments.get("kind", "example"),
},
)
def list_templates(arguments: dict[str, Any]) -> Any:
return client.call(
"GET",
"api/template-library?"
+ urlencode(
{
"q": arguments.get("query", ""),
"category": arguments.get("category", ""),
"kind": arguments.get("kind", ""),
}
),
)
def add_template(arguments: dict[str, Any]) -> Any:
return edit(
arguments,
"POST",
"template-library",
{
"entry": arguments["entry"],
"scaffold": arguments.get("scaffold", False),
},
)
def add_apparatus(arguments: dict[str, Any]) -> Any:
body: dict[str, Any] = {
k: arguments[k] for k in ("kind", "at", "scale", "label") if k in arguments
}
return edit(arguments, "POST", "apparatus", body)
def highlight_atoms(arguments: dict[str, Any]) -> Any:
body: dict[str, Any] = {
k: arguments[k] for k in ("species", "atoms", "colour") if k in arguments
}
return edit(arguments, "POST", "highlights", body)
def clear_highlights(arguments: dict[str, Any]) -> Any:
body = {"species": arguments["species"]} if arguments.get("species") else {}
return edit(arguments, "POST", "highlights/clear", body)
def abbreviations(path: str) -> Any:
def call(arguments: dict[str, Any]) -> Any:
body = {k: arguments[k] for k in ("species", "labels") if arguments.get(k)}
return edit(arguments, "POST", path, body)
return call
def enumerate_rgroups(arguments: dict[str, Any]) -> Any:
body = {"core": arguments["core"]} if arguments.get("core") else {}
return edit(arguments, "POST", "enumerate-rgroups", body)
def list_abbreviations(arguments: dict[str, Any]) -> Any:
return client.call("GET", f"api/documents/{document_id(arguments)}/abbreviations")
def show_groups(arguments: dict[str, Any]) -> Any:
return client.call("GET", f"api/documents/{document_id(arguments)}/groups")
def cancel_run(arguments: dict[str, Any]) -> Any:
return client.call("POST", f"api/queue/runs/{arguments['run']}/cancel")
def cancel_group(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
return client.call("POST", f"api/documents/{identifier}/groups/{arguments['group']}/cancel")
def show_in_vibe_view(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
species = str(arguments["species"])
summary = _summary(state_of(identifier))
known = {s["id"]: s for s in summary["species"]}
if species in known and not known[species]["has_3d"]:
edit({"document": identifier}, "POST", "embed", {"species": [species]})
return client.call("POST", f"api/documents/{identifier}/view", {"species": species})
def result_views(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
return client.call("GET", f"api/documents/{identifier}/runs/{arguments['run']}/views")
def show_result_view(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
payload = {
k: arguments[k] for k in ("view", "artifact", "mode", "orbital") if k in arguments
}
return client.call(
"POST", f"api/documents/{identifier}/runs/{arguments['run']}/view", payload
)
def attach_view_file(arguments: dict[str, Any]) -> Any:
return edit(
arguments,
"POST",
f"runs/{arguments['run']}/view-files",
{"name": arguments["name"], "data": arguments["data"]},
)
def return_geometry(arguments: dict[str, Any]) -> Any:
identifier = document_id(arguments)
return client.call(
"POST",
f"api/documents/{identifier}/geometry-return",
{"data": arguments["data"], "actor": "agent", "client": "mcp"},
)
species_list = {"type": "array", "items": {"type": "string"}}
return [
Tool("status", "The app's address for the user, the open documents and the writers.",
_schema({}), status),
Tool("new_document", "A new document, empty or from structures: format smiles (a "
"reaction as A.B>>C), name, inchi, peptide, dna or rna (one-letter sequences, "
"one per line or FASTA), helm, mol, sdf, rxn, cdxml, ket, xyz, orca (an ORCA 6 "
"output) or orca-hess (an ORCA Hessian file), with data.",
_schema({"format": {"type": "string"}, "data": {"type": "string"}}), new_document),
Tool("open_document", "Open a .qvf (the vibe family's format; from vibe-sketch or any "
"other vibe program), .vsk, MOL, SDF, RXN, CDXML, CDX, KET, SMILES or XYZ file, or an "
"ORCA output (.out) or Hessian file (.hess), as a new document.",
_schema({"path": {"type": "string"}}, ("path",)),
open_document),
Tool("show_document", "Species (SMILES, charge, multiplicity, name, whether it has 3D), "
"reactions, jobs, generic structures (R-group cores, bracketed groups, R-group "
"definitions, as CXSMILES) and warnings of a document.",
_schema({"document": _DOCUMENT}),
show_document),
Tool("report", "Formula, masses, InChI and the checked name of every species.",
_schema({"document": _DOCUMENT, "names": {"type": "boolean"}}), report),
Tool("add_structures", "Draw structures from format/data onto the document, beside "
"what is there (mode append) or instead of it (mode replace); the formats are "
"new_document's.",
_schema({"document": _DOCUMENT, "format": {"type": "string"},
"data": {"type": "string"},
"mode": {"type": "string", "enum": ["append", "replace"]}},
("format", "data")), add_structures),
Tool("set_species", "Set what the drawing cannot say about a species: its compound "
"number (label), its multiplicity, or a name ('' removes it).",
_schema({"document": _DOCUMENT, "species": {"type": "string"},
"label": {"type": "string"}, "multiplicity": {"type": "integer"},
"name": {"type": "string"}}, ("species",)), set_species),
Tool("clean_up", "Standard bond lengths and angles for every molecule in the drawing, "
"its arrangement and stereochemistry kept.", _schema({"document": _DOCUMENT}),
clean_up),
Tool("tidy_scheme", "Lay the whole reaction scheme out as papers draw it: molecules "
"cleaned and on their arrows' line, reagents above, conditions below, compound "
"numbers under their structures, steps in order; rows longer than width (bond "
"lengths, default 24) turn down; a molecule used twice or three times "
"branches down and up, and two sequences that join form a Y. Refused, with "
"the reason, if the reactions would read differently or the scheme branches "
"in a way the layout cannot express.",
_schema({"document": _DOCUMENT, "width": {"type": "number"}}), tidy_scheme),
Tool("embed_3d", "3D structures (ETKDG, then MMFF94) for species, the drawn "
"stereochemistry checked; all species when none are named.",
_schema({"document": _DOCUMENT, "species": species_list,
"count": {"type": "integer"}, "replace": {"type": "boolean"}}),
embed_3d),
Tool("relax_3d", "Relax species' 3D structures with FIRE on MMFF94 (force_field "
"mmff94, or uff), the drawn stereochemistry checked, holding constraints exactly: "
"kind distance (2 atoms, value in Å), angle (3) or dihedral (4, degrees), or "
"position (1 atom, fixed); without a value, the starting value. Atoms are 0-based, "
"the drawing's first and then the hydrogens; constraints go with one species. "
"Species without 3D are embedded first. Returns energies, steps and the "
"constraint checks per species.",
_schema({"document": _DOCUMENT, "species": species_list,
"constraints": {"type": "array", "items": {
"type": "object",
"properties": {"kind": {"type": "string",
"enum": ["distance", "angle", "dihedral",
"position"]},
"atoms": {"type": "array", "items": {"type": "integer"}},
"value": {"type": "number"}},
"required": ["kind", "atoms"]}},
"force_field": {"type": "string", "enum": ["mmff94", "uff"]},
"fmax": {"type": "number"}, "max_steps": {"type": "integer"},
"embed": {"type": "boolean"}}), relax_3d),
Tool("write_inputs", "Write program inputs (program orca, vibeqc, qcschema, xyz) for "
"species with one specification; species without 3D are embedded first. "
+ _SPEC_HELP + " Returns the files, or writes them to directory, and lists "
"refused species with reasons.",
_schema({"document": _DOCUMENT, "program": {"type": "string"},
"spec": {"type": "object", "description": _SPEC_HELP},
"species": species_list, "directory": {"type": "string"},
"embed": {"type": "boolean"}}, ("program", "spec")), write_inputs),
Tool("map_reaction", "Which reactant atom becomes which product atom in a reaction, by "
"the fewest bonds formed and broken: show it before writing a reaction path. "
"Ambiguous pairings are listed; pins (reactant:atom=product:atom, 1-based, for "
"example m1:3=m3:5) correct the pairing.",
_schema({"document": _DOCUMENT, "reaction": {"type": "string"},
"pins": {"type": "array", "items": {"type": "string"}}}, ("reaction",)),
map_reaction),
Tool("write_neb_inputs", "Nudged-elastic-band inputs for a reaction: program orca "
"(spec.kind neb-ts, neb-ci or neb) or vibeqc (neb-ci or neb); spec has method, "
"basis and optionally aux_basis, approximation, dispersion, solvent, resources, "
"images (default 8) and preopt_ends. Returns the files and the mapping, or "
"writes them to directory; refused with reasons when the mapping is ambiguous.",
_schema({"document": _DOCUMENT, "reaction": {"type": "string"},
"program": {"type": "string", "enum": ["orca", "vibeqc"]},
"spec": {"type": "object"},
"pins": {"type": "array", "items": {"type": "string"}},
"directory": {"type": "string"}}, ("reaction", "spec")),
write_neb_inputs),
Tool("export", "Export the document to a file: format qvf, svg, png, pdf, eps, emf, "
"mol, sdf, rxn, smiles, inchi, helm or fasta (species imported from a sequence), "
"ket, vsk or chemfig (a LaTeX fragment of \\chemfig structures and \\schemestart "
"schemes).",
_schema({"document": _DOCUMENT, "format": {"type": "string"},
"path": {"type": "string"},
"style": {"type": "string",
"description": "Publication preset; defaults to document style."},
"dpi": {"type": "integer", "minimum": 36, "maximum": 2400,
"description": "PNG resolution; default 300 DPI."},
"stereo_labels": {"type": "boolean",
"description": "Print CIP descriptors in figures."},
"sequences": {"type": "boolean",
"description": "Print species built from a sequence as "
"it (H-Tyr-Gly-OH) in figures."}},
("format", "path")), export),
Tool("save_document", "Save the document: to path as QVF (the default; .vsk when the "
"path ends so), or back to the file the server opened it from.",
_schema({"document": _DOCUMENT, "path": {"type": "string"}}), save_document),
Tool("submit_jobs", "Submit ORCA (program orca) or vibe-qc (program vibeqc) "
"calculations to vibe-queue as one group, embedding what has no 3D structure; the "
"jobs and runs are recorded on the document. Name species, or reactions (their "
"reactants, reagents and products, each distinct species once), or scheme true "
"for every reaction; the answer names the group. A vibe-qc job goes as a QVF job "
"container that keeps the sketch with its result (container false: a run "
"script); fallbacks says why a script went instead. " + _SPEC_HELP,
_schema({"document": _DOCUMENT, "program": {"type": "string",
"enum": ["orca", "vibeqc"]},
"spec": {"type": "object", "description": _SPEC_HELP},
"species": species_list,
"reactions": {"type": "array", "items": {"type": "string"}},
"scheme": {"type": "boolean"},
"container": {"type": "boolean",
"description": "vibe-qc jobs as QVF job containers "
"(default true) or run scripts."}},
("program", "spec")), submit_jobs),
Tool("show_recipes", "The document's staged recipes (#42): every stage per species as "
"vibe-qc would run it, what the operator supplies at launch, and the refusals.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"}}), show_recipes),
Tool("energy_constants", "List cited thermodynamic choices and aqueous offsets.",
_schema({}), energy_constants),
Tool("add_energy_recipe", "Save a redox or binding recipe with explicit constants.",
_schema({"document": _DOCUMENT, "id": {"type": "string"},
"analysis": {"type": "object"}, "spec": {"type": "object"},
"program": {"type": "string", "enum": ["orca", "vibeqc"]}},
("id", "analysis", "spec", "program")), add_energy_recipe),
Tool("compile_energy_recipe", "Compile recorded jobs and input files; does not submit.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"}}, ("recipe",)),
compile_energy_recipe),
Tool("energy_recipe_results", "Evaluate explicit imported runs with cited constants.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"},
"runs": {"type": "array", "items": {"type": "string"}},
"figure": {"type": "boolean"}},
("recipe", "runs")), energy_recipe_results),
Tool("set_recipe", "Save a staged recipe on the document, replacing one with the same id: "
"the publication baseline (MACE-OFF23 optimisation, PBE/def2-SVP optimisation "
"without dispersion, same-surface Hessian, DLPNO-CCSD(T)/cc-pVDZ TightPNO) for "
"species, reactions or the whole scheme, or a full recipe object with stable "
"stage ids, parents and explicit geometry sources. Returns the review.",
_schema({"document": _DOCUMENT, "id": {"type": "string"}, "name": {"type": "string"},
"species": species_list,
"reactions": {"type": "array", "items": {"type": "string"}},
"scheme": {"type": "boolean"},
"profile": {"type": "string", "enum": ["baseline", "dd"]},
"reference_contract": {"type": "string",
"enum": ["all_tested_channels", "real_restricted_cc"],
"description": "CC contract; default all tested channels."},
"recipe": {"type": "object", "description": "A complete recipe."}}),
set_recipe),
Tool("compile_campaign", "Compile a saved recipe into a vibe-queue campaign in a new, "
"empty directory (absolute path): a compute and a validate node per species and "
"numerical stage, each stage starting from the geometry its predecessor's "
"validator accepted. A MACE stage needs accept_model_license, the acknowledgement "
"of an eligible operator. Returns the record and the vq campaign run command.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"},
"directory": {"type": "string"}, "id": {"type": "string"},
"host": {"type": "string"}, "program": {"type": "string"},
"python": {"type": "string"},
"runner": {"type": "array", "items": {"type": "string"}},
"cpus": {"type": "integer"}, "mem_mb": {"type": "integer"},
"wall_time_seconds": {"type": "integer", "minimum": 1, "maximum": 31536000},
"check_wall_time_seconds": {"type": "integer", "minimum": 1,
"maximum": 31536000},
"stage_runtimes": {"type": "object",
"additionalProperties": {"type": "object"}},
"accept_model_license": {"type": "boolean"}},
("recipe", "directory")), compile_campaign),
Tool("review_campaign", "Explicit operator review: compile a separate immutable "
"continuation of a failed terminal DLPNO validation. Requires retained evidence "
"and restricted-reference criteria; preserves the original failed campaign.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"},
"campaign": {"type": "string"}, "directory": {"type": "string"},
"node": {"type": "string"}, "reviewer": {"type": "string"},
"reason": {"type": "string"}, "criteria": {"type": "object"},
"warnings": {"type": "array", "items": {"type": "string"}},
"evidence": {"type": "array", "items": {"type": "string"}}},
("recipe", "campaign", "directory", "node", "reviewer", "reason",
"criteria", "warnings", "evidence")), review_campaign),
Tool("revalidate_campaign", "Compile an immutable validator-upgrade continuation for "
"a retained DLPNO result rejected by the older correlated-energy reader. "
"Requires strict current acceptance; preserves original failures "
"and never reruns chemistry.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"},
"campaign": {"type": "string"}, "directory": {"type": "string"},
"node": {"type": "string"}, "reviewer": {"type": "string"},
"reason": {"type": "string"}},
("recipe", "campaign", "directory", "node", "reviewer", "reason")),
revalidate_campaign),
Tool("campaign_status", "A compiled campaign's stages: each one's process state in "
"vibe-queue (with attempts and retries) and, apart from it, its scientific "
"acceptance with the validator's verdict.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"},
"campaign": {"type": "string"}}, ("recipe", "campaign")),
campaign_status),
Tool("render_campaign_stages", "Render a species' start and each accepted stage of a "
"campaign in one orientation, from its verified artifacts: one aligned QVF per "
"geometry with its trajectory, a renders.json record, and vibe-view's images when "
"vibe-view is installed. directory is a new, empty absolute path.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"},
"campaign": {"type": "string"}, "species": {"type": "string"},
"directory": {"type": "string"}, "width": {"type": "integer"},
"height": {"type": "integer"}, "publication": {"type": "boolean"},
"images": {"type": "boolean"}},
("recipe", "campaign", "species", "directory")), render_campaign_stages),
Tool("retry_campaign_node", "Authorise the next attempt of a failed campaign node; the "
"next vq campaign run sends it, recorded as a retry of the failed attempt.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"},
"campaign": {"type": "string"}, "node": {"type": "string"}},
("recipe", "campaign", "node")), retry_campaign_node),
Tool("remove_recipe", "Remove a recipe from the document.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"}}, ("recipe",)),
remove_recipe),
Tool("show_runs", "The document's runs in vibe-queue with their state, newest first.",
_schema({"document": _DOCUMENT}), show_runs),
Tool("import_run_result", "Import a verified ORCA output or finished QVF result for a "
"recorded run; an ORCA optimisation's converged geometry becomes a conformer of "
"the species. format orca-hess checks an ORCA Hessian file against the run's "
"stationary point. Input identity must match; replacement is explicit.",
_schema({"document": _DOCUMENT, "run": {"type": "string"},
"file": {"type": "string"},
"format": {"type": "string", "enum": ["orca", "qvf", "orca-hess"]},
"replace": {"type": "boolean"}},
("run", "file", "format")), import_run_result),
Tool("external_chain_report", "Verify a pinned external chain archive and report "
"electronic energies. Retains original jobs and reference/correlated "
"evidence; never creates campaign runs or waives reference stability.",
_schema({"document": _DOCUMENT, "file": {"type": "string"},
"sha256": {"type": "string"},
"reference_contract": {"type": "string", "enum": ["real_restricted_cc"]},
"format": {"type": "string", "enum": ["json", "csv", "markdown", "latex"]}},
("file", "sha256", "reference_contract")), external_chain_report),
Tool("campaign_energy_report", "Electronic reaction energies from an explicitly selected "
"campaign and correlated stage. Requires every campaign check accepted, verifies "
"artifact/parent/review provenance and balanced coefficients; never fabricates runs. "
"Export JSON, CSV, Markdown, LaTeX, SVG or QVF; binary exports are base64.",
_schema({"document": _DOCUMENT, "recipe": {"type": "string"},
"campaign": {"type": "string"}, "stage": {"type": "string"},
"reactions": {"type": "array", "items": {"type": "string"}},
"coefficients": {"type": "object"},
"format": {"type": "string",
"enum": ["json", "csv", "markdown", "latex", "svg", "qvf"]}},
("recipe", "campaign", "stage")), campaign_energy_report),
Tool("show_reaction_energies", "Products minus reactants from an explicit group or "
"source runs. Refuses missing, stale, incompatible or unbalanced results; "
"optional coefficients name both chemical sides.",
_schema({"document": _DOCUMENT, "group": {"type": "string"},
"runs": {"type": "array", "items": {"type": "string"}},
"reactions": {"type": "array", "items": {"type": "string"}},
"coefficients": {"type": "object"}}), show_reaction_energies),
Tool("export_spectrum", "Export a run's IR, Raman or UV-Vis spectrum from its imported "
"result: the line list (csv or text; a missing intensity is blank) or the curve "
"broadened with a gaussian or lorentzian of a given FWHM (cm-1 or eV) on a grid of "
"the axis unit (cm-1; eV, nm or cm-1 for uvvis).",
_schema({"document": _DOCUMENT, "run": {"type": "string"},
"spectrum": {"type": "string", "enum": ["ir", "raman", "uvvis"]},
"format": {"type": "string", "enum": ["csv", "text", "broadened"]},
"lineshape": {"type": "string", "enum": ["gaussian", "lorentzian"]},
"fwhm": {"type": "number"},
"fwhm_unit": {"type": "string", "enum": ["cm-1", "eV"]},
"axis": {"type": "string", "enum": ["cm-1", "eV", "nm"]},
"start": {"type": "number"}, "stop": {"type": "number"},
"step": {"type": "number"}}, ("run", "spectrum")), export_spectrum),
Tool("export_property_table", "Formula, masses, elemental analysis (Anal. Calcd "
"line) and calculated descriptors (Crippen logP and MR, TPSA, Lipinski donors and "
"acceptors, rotatable bonds) per species, as csv, markdown or latex.",
_schema({"document": _DOCUMENT,
"format": {"type": "string", "enum": ["csv", "markdown", "latex"]}}),
export_property_table),
Tool("place_property_text", "Place a species' properties as text under its "
"structure in the drawing; fields from formula, mass, analysis, logp, tpsa "
"(default formula, mass, analysis).",
_schema({"document": _DOCUMENT, "species": {"type": "string"},
"fields": {"type": "array", "items": {"type": "string", "enum": [
"formula", "mass", "analysis", "logp", "tpsa"]}}}, ("species",)),
place_property_text),
Tool("set_amounts", "Enter a reaction's amounts for its stoichiometry table, "
"replacing earlier ones: per species one of mmol, mass_mg, or volume_ml with "
"density_g_ml or concentration_mol_l, or equivalents alone; optional coefficient. "
"An empty list clears them.",
_schema({"document": _DOCUMENT, "reaction": {"type": "string"},
"amounts": {"type": "array", "items": {"type": "object"}}},
("reaction", "amounts")), set_amounts),
Tool("show_stoichiometry", "A reaction's stoichiometry table: molar masses, "
"equivalents, mmol, mass, volume, the limiting reactant and theoretical and "
"isolated yields; json, or a csv, markdown or latex table.",
_schema({"document": _DOCUMENT, "reaction": {"type": "string"},
"format": {"type": "string",
"enum": ["json", "csv", "markdown", "latex"]}},
("reaction",)), show_stoichiometry),
Tool("show_stereo", "Each species' CIP descriptors (R/S, r/s, E/Z, M/P from the "
"new CIP labeler), its undefined stereocentres and double bonds, and its "
"enhanced-stereo groups.",
_schema({"document": _DOCUMENT}), show_stereo),
Tool("label_stereo", "Write the drawn structures' CIP descriptors into the drawing as "
"labels, or remove them with clear.",
_schema({"document": _DOCUMENT, "clear": {"type": "boolean"}}), label_stereo),
Tool("show_nmr", "A run's NMR shieldings per set of equivalent nuclei and, against "
"a reference run at the same level of theory (TMS, CFCl3, ...), its shifts; json "
"or a csv, markdown or latex line list.",
_schema({"document": _DOCUMENT, "run": {"type": "string"},
"reference": {"type": "string"},
"nuclei": {"type": "array", "items": {"type": "string"}},
"format": {"type": "string",
"enum": ["json", "csv", "markdown", "latex"]}}, ("run",)),
show_nmr),
Tool("place_nmr_labels", "Put a run's NMR shifts against a reference run as text "
"beside the atoms of its species' drawn structure (default nuclei H and C).",
_schema({"document": _DOCUMENT, "run": {"type": "string"},
"reference": {"type": "string"},
"nuclei": {"type": "array", "items": {"type": "string"}}},
("run", "reference")), place_nmr_labels),
Tool("add_surface", "Add or replace a schematic surface, electrode, stack, nanoparticle "
"or nanotube-array. Parameters: id, kind, material, facet, layers (bottom to top), "
"diameter_nm, ligand_shell, nanotube species, "
"anchors {name: {u,v,dx,dy,site,repeat}}. "
"Not a physical 3D model.",
_schema({"document": _DOCUMENT, "parameters": {"type": "object"},
"at": {"type": "array", "items": {"type": "number"},
"minItems": 2, "maxItems": 2},
"width": {"type": "number"}, "height": {"type": "number"}},
("parameters",)), add_surface),
Tool("list_surfaces", "List surfaces, anchors, site types, repeats and attached generics.",
_schema({"document": _DOCUMENT}), list_surfaces),
Tool("attach_to_surface", "Join a labelled generic to a surface anchor (default top).",
_schema({"document": _DOCUMENT, "generic": {"type": "string"},
"surface": {"type": "string"}, "anchor": {"type": "string"}},
("generic", "surface")), attach_to_surface),
Tool("add_nanostructure", "Generate a carbon nanostructure from its parameters and add "
"it as a species with its 3D atoms (a cluster model; no SMILES, identified by a "
"NANO- key). parameters.kind is nanotube {walls: [{n, m}] innermost first, "
"length_nm or periods, ends: [lower, upper] each H|OH|COOH|capped (capped: "
"(5,5) only)}, graphene {width_nm, length_nm, edges: H|OH|COOH}, nanoribbon "
"{edge: armchair|zigzag, width (dimer lines or chains), periods, edges}, or "
"fullerene {size: 60|70}. Tubes, sheets and ribbons take sidewall [{group, site}] "
"and defects [{kind: vacancy|stone-wales, site}], a site being {along: 0-1, "
"around: degrees} on a tube or {along, across: 0-1} on a sheet.",
_schema({"document": _DOCUMENT,
"parameters": {"type": "object"},
"picture_style": {"type": "string",
"enum": ["schematic", "lattice", "end-on", "unrolled"]},
"picture_label": {"type": "boolean"},
"species_id": {"type": "string"},
"label": {"type": "string"},
"charge": {"type": "integer"},
"multiplicity": {"type": "integer", "minimum": 1}},
("parameters",)), add_nanostructure),
Tool("cap_generic", "Make a calculable species of a labelled generic structure, one "
"bonded to an attachment label such as CNT or Au, by replacing each label with H "
"or CH3. Nothing is capped unless asked; the species records the cap.",
_schema({"document": _DOCUMENT,
"generic": {"type": "string", "description": "The generic's id, e.g. g1."},
"cap": {"type": "string", "enum": ["H", "CH3"]}},
("generic", "cap")), cap_generic),
Tool("depict_nanostructure", "Put a picture of a generated nanostructure species on "
"the canvas as one object: lattice (side projection, rear wall light or omitted), "
"schematic (outline with hexagon texture), end-on (one ring per wall) or unrolled "
"(the sheet with Ch and T). Long tubes get a break mark unless break is off.",
_schema({"document": _DOCUMENT,
"species": {"type": "string"},
"style": {"type": "string",
"enum": ["lattice", "schematic", "end-on", "unrolled"]},
"label": {"type": "boolean"},
"rear": {"type": "string", "enum": ["light", "omit"]},
"break": {"type": "string", "enum": ["auto", "on", "off"]},
"width": {"type": "number", "minimum": 1, "maximum": 100}},
("species",)), depict_nanostructure),
Tool("list_pictures", "List picture nodes, species, boxes and anchors on the canvas.",
_schema({"document": _DOCUMENT}, ()), list_pictures),
Tool("attach_to_nanostructure", "Move a labelled structure (bonded to CNT, Au, ...) "
"onto an anchor of a nanostructure picture, so figures print them as one: "
"anchor lower-end, upper-end or sidewall.",
_schema({"document": _DOCUMENT,
"generic": {"type": "string"},
"species": {"type": "string"},
"anchor": {"type": "string"},
"node": {"type": "integer", "minimum": 0,
"description": "Canvas node from list_pictures."}},
("generic", "species")), attach_to_nanostructure),
Tool("add_half_reaction", "Draw an electrode half-reaction beside the drawing. "
"Participants are SMILES, {smiles, coefficient, state: s|l|g|aq|ads} or a solid "
"given by its formula {formula: 'Fe2O3', coefficient, state}; over are reagents "
"drawn above the arrow (a {formula} catalyst or a SMILES); electrons "
"{count, side: reactants|products, at_arrow}; conditions are lines below the "
"arrow, a potential written as '-1.2 V vs Ag/AgCl (3 M KCl)' (SHE, NHE, SCE, RHE), "
"'pH 7.4', '25 °C' or '2 h'. The reaction records electrons, states, coefficients "
"and typed quantities.",
_schema({"document": _DOCUMENT,
"reactants": {"type": "array"},
"products": {"type": "array"},
"electrons": {"type": "object"},
"conditions": {"type": "array", "items": {"type": "string"}},
"arrow": {"type": "string"},
"over": {"type": "array"}},
("reactants", "products")), add_half_reaction),
Tool("check_balance", "Each reaction's atom and charge balance with its electrons: "
"what each side holds and the imbalance, reported and never fixed.",
_schema({"document": _DOCUMENT}), check_balance),
Tool("add_tlc_plate", "Draw a TLC plate beside the drawing from shapes and text: "
"lanes, each with a label and the Rf (0 to 1) of up to six spots.",
_schema({"document": _DOCUMENT,
"lanes": {"type": "array", "items": {"type": "object", "properties": {
"label": {"type": "string"},
"spots": {"type": "array", "items": {"type": "number"}}}}}},
("lanes",)), add_tlc_plate),
Tool("list_named_reactions", "Search the offline reaction library by name or alias. "
"Returns stable IDs, sources and example/scaffold/reference status.",
_schema({"query": {"type": "string"}}), list_named_reactions),
Tool("add_named_reaction", "Append an editable teaching example or explicitly unfinished "
"scaffold. Scaffolds contain labels and a graphic arrow, no molecular structures. "
"Reference-only entries cannot be inserted.",
_schema({"document": _DOCUMENT, "entry": {"type": "string"},
"kind": {"type": "string", "enum": ["example", "scaffold"]}},
("entry",)), add_named_reaction),
Tool("list_templates", "Search offline compound, fragment and named-reaction templates. "
"Returns stable IDs, categories, sources and incomplete status.",
_schema({"query": {"type": "string"}, "category": {"type": "string"},
"kind": {"type": "string", "enum": [
"compound", "fragment", "polymer", "nanostructure", "example", "scaffold",
"reference"]}}),
list_templates),
Tool("add_template", "Append an editable compound, generic R-site fragment or reaction. "
"Unfinished reaction scaffolds require scaffold=true. Reference terms cannot insert.",
_schema({"document": _DOCUMENT, "entry": {"type": "string"},
"scaffold": {"type": "boolean"}}, ("entry",)), add_template),
Tool("add_apparatus", "Draw a laboratory apparatus pictogram from shapes: kind "
"flask, conical-flask, beaker, condenser, separatory-funnel, column, vial, "
"hotplate, three-electrode-cell, two-compartment-cell, potentiostat, "
"tube-furnace, gas-cylinder or uv-lamp; at [x, y] its bottom centre in drawing "
"coordinates (default beside the drawing), scale 0.25 to 4, an optional label "
"under it.",
_schema({"document": _DOCUMENT, "kind": {"type": "string"},
"at": {"type": "array", "items": {"type": "number"},
"minItems": 2, "maxItems": 2},
"scale": {"type": "number"}, "label": {"type": "string"}},
("kind",)), add_apparatus),
Tool("highlight_atoms", "Mark atoms of a species in colour for the canvas and figures: "
"atoms numbered from 1 as in the report (C2 is 2), colour orange, sky-blue, green, "
"blue, vermillion or purple; bonds between marked atoms are marked too.",
_schema({"document": _DOCUMENT, "species": {"type": "string"},
"atoms": {"type": "array", "items": {"type": "integer"}},
"colour": {"type": "string", "enum": ["orange", "sky-blue", "green",
"blue", "vermillion", "purple"]}},
("species", "atoms")), highlight_atoms),
Tool("clear_highlights", "Remove every highlight, or those of one species.",
_schema({"document": _DOCUMENT, "species": {"type": "string"}}), clear_highlights),
Tool("enumerate_rgroups", "Add all constraint-valid R-group combinations as species; "
"retain the core and definitions. Refuses unbounded or oversized enumeration.",
_schema({"document": _DOCUMENT, "core": {"type": "string"}}, required=("document",)),
enumerate_rgroups),
Tool("list_abbreviations", "The abbreviations (Ph, Boc, OTf ...) that can be "
"contracted, and those contracted in each species of the drawing.",
_schema({"document": _DOCUMENT}), list_abbreviations),
Tool("contract_abbreviations", "Draw groups such as Ph, Boc or OTf as their label, on "
"the canvas and in figures, for one species or all; without labels every known "
"group that leaves most of a molecule drawn. The structure is unchanged.",
_schema({"document": _DOCUMENT, "species": {"type": "string"},
"labels": {"type": "array", "items": {"type": "string"}}}),
abbreviations("abbreviations")),
Tool("expand_abbreviations", "Draw contracted abbreviations as their atoms again, for "
"one species or all, of the given labels or all.",
_schema({"document": _DOCUMENT, "species": {"type": "string"},
"labels": {"type": "array", "items": {"type": "string"}}}),
abbreviations("abbreviations/expand")),
Tool("export_energy_table", "Export an energy table with source runs and coefficients.",
_schema({"document": _DOCUMENT,
"format": {"type": "string", "enum": ["csv", "markdown", "latex"]},
"group": {"type": "string"},
"runs": {"type": "array", "items": {"type": "string"}},
"reactions": {"type": "array", "items": {"type": "string"}},
"coefficients": {"type": "object"}}, ("format",)), export_energy_table),
Tool("show_groups", "The document's run groups (one per submission) with the runs "
"in each state and whether each group has finished.",
_schema({"document": _DOCUMENT}), show_groups),
Tool("cancel_run", "Cancel a vibe-queue run that has not finished.",
_schema({"run": {"type": "string"}}, ("run",)), cancel_run),
Tool("cancel_group", "Cancel the runs of a group that have not finished.",
_schema({"document": _DOCUMENT, "group": {"type": "string"}}, ("group",)),
cancel_group),
Tool("result_views", "List a run's retained result data and missing-data reasons.",
_schema({"document":_DOCUMENT, "run":{"type":"string"}}, ("run",)), result_views),
Tool("show_result_view", "Open a run's selected result view in vibe-view.",
_schema({"document":_DOCUMENT, "run":{"type":"string"},
"view":{"type":"string", "enum":["orbitals","density","mode","spectrum"]},
"artifact":{"type":"string"}, "mode":{"type":"integer","minimum":1},
"orbital":{"type":"string"}}, ("run","view")), show_result_view),
Tool("attach_view_file", "Attach a companion with checked run geometry. "
"data is base64; name is a plain filename, never a server path.",
_schema({"document":_DOCUMENT, "run":{"type":"string"}, "name":{"type":"string"},
"data":{"type":"string"}}, ("run","name","data")), attach_view_file),
Tool("return_geometry", "Append a checked QVF geometry without moving the drawing. "
"data is the base64 archive; its origin must name this document.",
_schema({"document": _DOCUMENT, "data": {"type": "string"}}, ("data",)),
return_geometry),
Tool("show_in_vibe_view", "Show a species in 3D in vibe-view, the family's viewer, as a "
"QVF (embedded in 3D first when needed).",
_schema({"document": _DOCUMENT, "species": {"type": "string"}}, ("species",)),
show_in_vibe_view),
Tool("list_writers", "The input writers, their tasks and the fields they can express.",
_schema({}), list_writers),
] # fmt: skip
def _result(value: Any, *, error: bool = False) -> dict[str, Any]:
text = value if isinstance(value, str) else json.dumps(value, indent=1, ensure_ascii=False)
return {"content": [{"type": "text", "text": text}], "isError": error}
[docs]
def handle(message: Any, tools: dict[str, Tool]) -> dict[str, Any] | None:
"""The response to one JSON-RPC message; ``None`` for a notification."""
if not isinstance(message, dict) or message.get("jsonrpc") != "2.0":
return {
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32600, "message": "not JSON-RPC 2.0"},
}
if "id" not in message:
return None # notifications/initialized, cancellations: nothing to answer
identifier, method = message["id"], message.get("method")
params = message.get("params") or {}
def answer(result: dict[str, Any]) -> dict[str, Any]:
return {"jsonrpc": "2.0", "id": identifier, "result": result}
def failure(code: int, text: str) -> dict[str, Any]:
return {"jsonrpc": "2.0", "id": identifier, "error": {"code": code, "message": text}}
if method == "initialize":
from vibesketch import __version__
asked = params.get("protocolVersion")
version = asked if asked in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0]
return answer(
{
"protocolVersion": version,
"capabilities": {"tools": {"listChanged": False}},
"serverInfo": {"name": "vibe-sketch", "version": __version__},
"instructions": INSTRUCTIONS,
}
)
if method == "ping":
return answer({})
if method == "tools/list":
return answer({"tools": [tool.listing() for tool in tools.values()]})
if method == "tools/call":
tool = tools.get(str(params.get("name")))
if tool is None:
return failure(-32602, f"no tool named {params.get('name')!r}")
arguments = params.get("arguments") or {}
if not isinstance(arguments, dict):
return failure(-32602, "arguments must be an object")
missing = [k for k in tool.schema.get("required", []) if k not in arguments]
if missing:
return answer(_result(f"missing arguments: {', '.join(missing)}", error=True))
try:
return answer(_result(tool.run(arguments)))
except ServerError as error:
return answer(_result({"status": error.status, "detail": error.detail}, error=True))
except (ToolError, OSError, ValueError, KeyError) as error:
return answer(_result(str(error), error=True))
return failure(-32601, f"no method {method!r}")
[docs]
def serve(client: RestClient, reader: IO[str] | None = None, writer: IO[str] | None = None) -> None:
"""Answer MCP messages from ``reader`` (standard input) until it closes."""
source = reader if reader is not None else sys.stdin
sink = writer if writer is not None else sys.stdout
tools = {tool.name: tool for tool in make_tools(client)}
for line in source:
if not line.strip():
continue
try:
message = json.loads(line)
except ValueError:
response: Any = {"jsonrpc": "2.0", "id": None,
"error": {"code": -32700, "message": "not JSON"}} # fmt: skip
else:
if isinstance(message, list): # a batch, from protocol 2025-03-26
answers = [a for a in (handle(m, tools) for m in message) if a is not None]
response = answers or None
else:
response = handle(message, tools)
if response is not None:
sink.write(json.dumps(response, ensure_ascii=False) + "\n")
sink.flush()