forked from 1b-agent/opencadstudio-upstream
Karim Jerbi
7b68d274ad
Add opt-in response_detail=summary with step and change counts, compact state, warnings, and actionable errors or prompts. Return an ocs_read batch_result request for full cached details without resuming mutations. Keep compact/full behavior and idempotent batch retries intact. Preserve active-step context and report unavailable cached results explicitly. Document cache scope and retention; update schemas and digest. Validation: 47 MCP unit tests, live stdio smoke test, rustfmt, and diff checks passed.
203 lines
8.3 KiB
Python
203 lines
8.3 KiB
Python
"""Protocol smoke test for the native client-neutral MCP endpoint."""
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
TOOLS = {"ocs_sessions", "ocs_read", "ocs_execute", "ocs_capture"}
|
|
MODERN = "2026-07-28"
|
|
|
|
|
|
def start(server: Path) -> subprocess.Popen[str]:
|
|
process = subprocess.Popen(
|
|
[str(server), "--mcp"],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
assert process.stdin and process.stdout
|
|
return process
|
|
|
|
|
|
def request(process: subprocess.Popen[str], payload: dict) -> dict:
|
|
assert process.stdin and process.stdout
|
|
process.stdin.write(json.dumps(payload) + "\n")
|
|
process.stdin.flush()
|
|
return json.loads(process.stdout.readline())
|
|
|
|
|
|
def close(process: subprocess.Popen[str]) -> None:
|
|
assert process.stdin
|
|
process.stdin.close()
|
|
assert process.wait(timeout=5) == 0
|
|
|
|
|
|
def legacy(server: Path) -> None:
|
|
process = start(server)
|
|
|
|
initialized = request(process, {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "initialize",
|
|
"params": {"protocolVersion": "2025-11-25", "capabilities": {}, "clientInfo": {"name": "smoke", "version": "1"}},
|
|
})
|
|
assert initialized["result"]["serverInfo"]["name"] == "OpenCADStudio"
|
|
assert "state.command.accepts" in initialized["result"]["instructions"]
|
|
process.stdin.write(json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}) + "\n")
|
|
process.stdin.flush()
|
|
tools = request(process, {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}})
|
|
definitions = {tool["name"]: tool for tool in tools["result"]["tools"]}
|
|
names = set(definitions)
|
|
assert names == TOOLS, names
|
|
assert "summary" in definitions["ocs_execute"]["inputSchema"]["properties"]["response_detail"]["enum"]
|
|
assert "batch_result" in definitions["ocs_read"]["inputSchema"]["properties"]["op"]["enum"]
|
|
execute_request = definitions["ocs_execute"]["inputSchema"]["properties"]["request"]
|
|
variants = execute_request.get("anyOf") or execute_request.get("oneOf")
|
|
# Resolve local references from the complete tool input schema.
|
|
root = definitions["ocs_execute"]["inputSchema"]
|
|
def resolve(variant):
|
|
if "$ref" not in variant:
|
|
return variant
|
|
ref = variant["$ref"]
|
|
assert ref.startswith("#/")
|
|
resolved = root
|
|
for part in ref[2:].split("/"):
|
|
resolved = resolved[part.replace("~1", "/").replace("~0", "~")]
|
|
return resolved
|
|
variants = [resolve(variant) for variant in variants]
|
|
op_enum = execute_request["properties"]["op"]["enum"]
|
|
assert len(variants) == len(op_enum)
|
|
for shipped in ("entities_create", "entities_delete", "entities_transform",
|
|
"block_define", "block_delete", "file_identity", "xdata_set",
|
|
"view_focus", "wblock", "plot",
|
|
"entities_copy_to", "group_create", "selection_set_save",
|
|
"selection_set_load", "close", "sysvar", "layout_create",
|
|
"page_setup_set", "text_replace"):
|
|
assert shipped in op_enum, shipped
|
|
assert "xdata_get" in definitions["ocs_read"]["inputSchema"]["properties"]["op"]["enum"]
|
|
assert "text_search" in definitions["ocs_read"]["inputSchema"]["properties"]["op"]["enum"]
|
|
assert "text_audit" in definitions["ocs_read"]["inputSchema"]["properties"]["op"]["enum"]
|
|
assert execute_request["properties"]["steps"]["maxItems"] == 64
|
|
variant_map = {
|
|
v["properties"]["op"].get("const") or v["properties"]["op"]["enum"][0]: v
|
|
for v in variants
|
|
}
|
|
assert variant_map["run"]["properties"]["cmd"]["type"] == "string"
|
|
assert "set_properties" in execute_request["properties"]["op"]["enum"]
|
|
assert "save_verified" in execute_request["properties"]["op"]["enum"]
|
|
assert "record_schema" in definitions["ocs_read"]["inputSchema"]["properties"]["op"]["enum"]
|
|
assert "audit" in definitions["ocs_read"]["inputSchema"]["properties"]["op"]["enum"]
|
|
read_parameters = definitions["ocs_read"]["inputSchema"]["properties"]["parameters"]["properties"]
|
|
assert {"collection", "where", "paths", "target_format", "target_version"} <= set(read_parameters)
|
|
assert variant_map["save_verified"]["properties"]["target_version"]["enum"][0] == "R14"
|
|
assert variant_map["input"]["properties"]["kind"]["enum"] == [
|
|
"text", "token", "point", "entity", "structure", "selection", "enter"
|
|
]
|
|
for name in {"ocs_sessions", "ocs_read", "ocs_execute"}:
|
|
assert definitions[name]["outputSchema"]["type"] == "object"
|
|
assert "resultType" not in tools["result"]
|
|
sessions = request(process, {
|
|
"jsonrpc": "2.0",
|
|
"id": 3,
|
|
"method": "tools/call",
|
|
"params": {"name": "ocs_sessions", "arguments": {"launch_if_none": False}},
|
|
})
|
|
assert not sessions["result"].get("isError"), sessions
|
|
missing_batch = request(process, {
|
|
"jsonrpc": "2.0", "id": 4, "method": "tools/call",
|
|
"params": {"name": "ocs_read", "arguments": {
|
|
"ocs_session_id": "missing", "op": "batch_result",
|
|
"parameters": {"request_id": "expired-batch"},
|
|
}},
|
|
})
|
|
assert missing_batch["result"]["isError"] is True
|
|
assert "absent or expired" in missing_batch["result"]["content"][0]["text"]
|
|
close(process)
|
|
|
|
|
|
def modern(server: Path) -> None:
|
|
process = start(server)
|
|
meta = {
|
|
"io.modelcontextprotocol/protocolVersion": MODERN,
|
|
"io.modelcontextprotocol/clientInfo": {"name": "smoke", "version": "1"},
|
|
"io.modelcontextprotocol/clientCapabilities": {},
|
|
}
|
|
discovered = request(process, {
|
|
"jsonrpc": "2.0",
|
|
"id": "discover",
|
|
"method": "server/discover",
|
|
"params": {"_meta": meta},
|
|
})
|
|
assert discovered["result"]["resultType"] == "complete"
|
|
assert discovered["result"]["ttlMs"] >= 0
|
|
assert discovered["result"]["cacheScope"] in {"public", "private"}
|
|
assert "io.modelcontextprotocol/tasks" in discovered["result"]["capabilities"]["extensions"]
|
|
|
|
tools = request(process, {
|
|
"jsonrpc": "2.0",
|
|
"id": "tools",
|
|
"method": "tools/list",
|
|
"params": {"_meta": meta},
|
|
})
|
|
assert {tool["name"] for tool in tools["result"]["tools"]} == TOOLS
|
|
assert tools["result"]["resultType"] == "complete"
|
|
assert tools["result"]["ttlMs"] >= 0
|
|
assert tools["result"]["cacheScope"] in {"public", "private"}
|
|
definitions = {tool["name"]: tool for tool in tools["result"]["tools"]}
|
|
assert definitions["ocs_execute"]["inputSchema"]["properties"]["response_detail"]["default"] == "compact"
|
|
assert definitions["ocs_capture"]["inputSchema"]["properties"]["scope"]["default"] == "viewport"
|
|
|
|
missing_cmd = request(process, {
|
|
"jsonrpc": "2.0",
|
|
"id": "missing-cmd",
|
|
"method": "tools/call",
|
|
"params": {
|
|
"name": "ocs_execute",
|
|
"arguments": {
|
|
"ocs_session_id": "missing",
|
|
"request": {"op": "run", "request_id": "run-1"},
|
|
},
|
|
"_meta": meta,
|
|
},
|
|
})
|
|
# Tool errors are text content; the output schema covers successes only.
|
|
assert missing_cmd["result"]["isError"] is True
|
|
assert "structuredContent" not in missing_cmd["result"]
|
|
error = missing_cmd["result"]["content"][0]["text"]
|
|
assert "Missing cmd" in error
|
|
assert "LINE 0,0 10,10" in error
|
|
|
|
invalid_mutation = request(process, {
|
|
"jsonrpc": "2.0",
|
|
"id": "invalid-mutation",
|
|
"method": "tools/call",
|
|
"params": {
|
|
"name": "ocs_execute",
|
|
"arguments": {"ocs_session_id": "missing", "request": {"op": "undo"}},
|
|
"_meta": meta,
|
|
},
|
|
})
|
|
assert invalid_mutation["result"]["isError"] is True
|
|
assert "structuredContent" not in invalid_mutation["result"]
|
|
assert "request_id" in invalid_mutation["result"]["content"][0]["text"]
|
|
|
|
rejected = request(process, {
|
|
"jsonrpc": "2.0",
|
|
"id": "unsupported",
|
|
"method": "tools/list",
|
|
"params": {"_meta": {"io.modelcontextprotocol/protocolVersion": "2099-01-01"}},
|
|
})
|
|
assert rejected["error"]["code"] == -32022
|
|
close(process)
|
|
|
|
|
|
def main() -> None:
|
|
server = Path(sys.argv[1] if len(sys.argv) > 1 else "target/debug/OpenCADStudio").resolve()
|
|
legacy(server)
|
|
modern(server)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|