forked from 1b-agent/opencadstudio-upstream
Karim Jerbi
97afb57333
Expose creation fields, constraints and examples for the 11 supported types without expanding startup tool schemas. Share the creation contract and constructor with execution to keep accepted inputs consistent. Validation: MCP, contract/factory and entity creation regression tests.
233 lines
9.8 KiB
Python
233 lines
9.8 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"]
|
|
guides = request(process, {
|
|
"jsonrpc": "2.0", "id": "guides", "method": "resources/list",
|
|
})
|
|
guide_uris = {r["uri"] for r in guides["result"]["resources"]
|
|
if r["uri"].startswith("cad://guides/")}
|
|
assert guide_uris == {f"cad://guides/{name}" for name in
|
|
("workflow", "architecture", "capture", "batch-bindings")}
|
|
for uri in sorted(guide_uris):
|
|
resource = request(process, {
|
|
"jsonrpc": "2.0", "id": uri, "method": "resources/read", "params": {"uri": uri},
|
|
})
|
|
assert resource["result"]["contents"][0]["text"].startswith("# ")
|
|
guide_tool = request(process, {
|
|
"jsonrpc": "2.0", "id": "guide-tool", "method": "tools/call",
|
|
"params": {"name": "ocs_read", "arguments": {
|
|
"ocs_session_id": "offline", "op": "guide", "parameters": {"name": "capture"},
|
|
}},
|
|
})
|
|
assert not guide_tool["result"].get("isError"), guide_tool
|
|
assert guide_tool["result"]["structuredContent"]["ok"] is True
|
|
assert "resource_link" in guide_tool["result"]["structuredContent"]["contents"][0]["text"]
|
|
entity_schema = request(process, {
|
|
"jsonrpc": "2.0", "id": "entity-schema", "method": "tools/call",
|
|
"params": {"name": "ocs_read", "arguments": {
|
|
"op": "entity_schema", "parameters": {"type": "Circle"},
|
|
}},
|
|
})
|
|
assert not entity_schema["result"].get("isError"), entity_schema
|
|
contract = entity_schema["result"]["structuredContent"]["schema"]
|
|
assert contract["required"] == ["type", "center", "radius"]
|
|
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()
|