294 lines
8.9 KiB
Python
294 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Cursor MCP bridge for a remote Windows CODESYS agent."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from typing import Any, Dict
|
|
|
|
import httpx
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
|
|
AGENT_URL = os.environ.get("CODESYS_AGENT_URL", "http://127.0.0.1:8787").rstrip("/")
|
|
AGENT_TOKEN = os.environ.get("CODESYS_AGENT_TOKEN", "")
|
|
REQUEST_TIMEOUT_SEC = float(os.environ.get("CODESYS_AGENT_REQUEST_TIMEOUT_SEC", "30"))
|
|
|
|
|
|
if not AGENT_TOKEN:
|
|
raise RuntimeError("CODESYS_AGENT_TOKEN is required.")
|
|
|
|
|
|
mcp = FastMCP("codesys-bridge")
|
|
|
|
|
|
def _headers() -> Dict[str, str]:
|
|
return {"x-api-token": AGENT_TOKEN, "content-type": "application/json"}
|
|
|
|
|
|
def _client() -> httpx.Client:
|
|
return httpx.Client(timeout=REQUEST_TIMEOUT_SEC)
|
|
|
|
|
|
def _post(path: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
with _client() as client:
|
|
response = client.post(
|
|
f"{AGENT_URL}{path}",
|
|
headers=_headers(),
|
|
content=json.dumps(payload),
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def _get(path: str) -> Dict[str, Any]:
|
|
with _client() as client:
|
|
response = client.get(f"{AGENT_URL}{path}", headers=_headers())
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
@mcp.tool()
|
|
def codesys_health() -> Dict[str, Any]:
|
|
"""Check Windows VM CODESYS agent health."""
|
|
with _client() as client:
|
|
response = client.get(f"{AGENT_URL}/health")
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
@mcp.tool()
|
|
def codesys_submit_job(action: str, payload: Dict[str, Any], dry_run: bool = False) -> Dict[str, Any]:
|
|
"""Submit an asynchronous CODESYS action."""
|
|
return _post(
|
|
"/jobs",
|
|
{
|
|
"action": action,
|
|
"payload": payload,
|
|
"dry_run": dry_run,
|
|
},
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def codesys_get_job(job_id: str) -> Dict[str, Any]:
|
|
"""Get CODESYS job status/result by ID."""
|
|
return _get(f"/jobs/{job_id}")
|
|
|
|
|
|
@mcp.tool()
|
|
def create_project(project_name: str, output_dir: str, template: str = "standard", dry_run: bool = False) -> Dict[str, Any]:
|
|
"""Create a new CODESYS project in the Windows VM."""
|
|
return codesys_submit_job(
|
|
action="create_project",
|
|
payload={
|
|
"project_name": project_name,
|
|
"output_dir": output_dir,
|
|
"template": template,
|
|
},
|
|
dry_run=dry_run,
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def read_project_inventory(project_path: str, max_depth: int = 3, dry_run: bool = False) -> Dict[str, Any]:
|
|
"""Read an existing project tree/inventory."""
|
|
return codesys_submit_job(
|
|
action="read_project_inventory",
|
|
payload={"project_path": project_path, "max_depth": max_depth},
|
|
dry_run=dry_run,
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def stage_existing_project(
|
|
source_path: str,
|
|
projects_root_dir: str,
|
|
project_folder_name: str = "",
|
|
move: bool = False,
|
|
overwrite: bool = False,
|
|
dry_run: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Copy/move an existing project into a standardized projects root folder."""
|
|
return codesys_submit_job(
|
|
action="stage_existing_project",
|
|
payload={
|
|
"source_path": source_path,
|
|
"projects_root_dir": projects_root_dir,
|
|
"project_folder_name": project_folder_name,
|
|
"move": move,
|
|
"overwrite": overwrite,
|
|
},
|
|
dry_run=dry_run,
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def write_pou(
|
|
project_path: str,
|
|
object_path: str = "",
|
|
object_name: str = "",
|
|
declaration: str = "",
|
|
implementation: str = "",
|
|
updates: list = [],
|
|
dry_run: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Write/update textual objects (POUs, FBs, GVLs, NVLs, structs, programs).
|
|
|
|
For a single object: provide object_path (e.g. "Device/Plc Logic/Application/3.Function Blocks/fb_boiler")
|
|
or object_name (unique name), plus declaration and/or implementation text.
|
|
For batch: provide updates list of dicts with object_path/object_name + declaration/implementation.
|
|
"""
|
|
payload: Dict[str, Any] = {"project_path": project_path, "dry_run": dry_run}
|
|
if updates:
|
|
payload["updates"] = updates
|
|
else:
|
|
if object_path:
|
|
payload["object_path"] = object_path
|
|
if object_name:
|
|
payload["object_name"] = object_name
|
|
if declaration:
|
|
payload["declaration"] = declaration
|
|
if implementation:
|
|
payload["implementation"] = implementation
|
|
return codesys_submit_job(action="write_pou", payload=payload, dry_run=dry_run)
|
|
|
|
|
|
@mcp.tool()
|
|
def read_device_io(
|
|
project_path: str,
|
|
channels_only: bool = True,
|
|
max_depth: int = 10,
|
|
export_csv_path: str = "",
|
|
dry_run: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Read device tree with I/O channels, parameters, and variable mappings.
|
|
|
|
Returns device identification (type/model), I/O channel details (input/output,
|
|
bit size, IEC type), and current variable mappings for each device in the project.
|
|
Set channels_only=False to include non-IO parameters too.
|
|
Optionally export all IO mappings as CSV on the VM.
|
|
"""
|
|
payload: Dict[str, Any] = {
|
|
"project_path": project_path,
|
|
"channels_only": channels_only,
|
|
"max_depth": max_depth,
|
|
}
|
|
if export_csv_path:
|
|
payload["export_csv_path"] = export_csv_path
|
|
return codesys_submit_job(action="read_device_io", payload=payload, dry_run=dry_run)
|
|
|
|
|
|
@mcp.tool()
|
|
def write_io_mapping(
|
|
project_path: str,
|
|
mappings: list = [],
|
|
import_csv_path: str = "",
|
|
dry_run: bool = True,
|
|
) -> Dict[str, Any]:
|
|
"""Map IEC variables to device I/O channels.
|
|
|
|
Each mapping dict needs: device_path (slash-separated, e.g. "Device/EtherCAT_Master/EK1100/K1"),
|
|
param_id or param_name (from read_device_io output), and variable (IEC variable expression).
|
|
Alternatively provide import_csv_path to bulk-import mappings from a CSV file on the VM.
|
|
dry_run=True by default for safety -- set False to actually write.
|
|
"""
|
|
payload: Dict[str, Any] = {
|
|
"project_path": project_path,
|
|
"dry_run": dry_run,
|
|
}
|
|
if mappings:
|
|
payload["mappings"] = mappings
|
|
if import_csv_path:
|
|
payload["import_csv_path"] = import_csv_path
|
|
return codesys_submit_job(action="write_io_mapping", payload=payload, dry_run=dry_run)
|
|
|
|
|
|
@mcp.tool()
|
|
def manage_device(
|
|
project_path: str,
|
|
operations: list = [],
|
|
dry_run: bool = True,
|
|
) -> Dict[str, Any]:
|
|
"""Add, remove, update, rename, enable/disable devices, or set device parameters.
|
|
|
|
Each operation dict needs action_type plus action-specific fields:
|
|
- add: parent_path, name, device_type, device_id, device_version [, method="add"|"plug"|"insert"]
|
|
- remove: device_path
|
|
- update: device_path, device_type, device_id, device_version
|
|
- rename: device_path, new_name
|
|
- enable/disable: device_path, enabled (bool)
|
|
- set_param: device_path, param_id or param_name, value
|
|
|
|
Use device_type/device_id/device_version values from read_device_io output.
|
|
dry_run=True by default for safety.
|
|
"""
|
|
return codesys_submit_job(
|
|
action="manage_device",
|
|
payload={
|
|
"project_path": project_path,
|
|
"operations": operations,
|
|
"dry_run": dry_run,
|
|
},
|
|
dry_run=dry_run,
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def sync_project(
|
|
project_path: str,
|
|
textual_updates: list = [],
|
|
io_mappings: list = [],
|
|
device_params: list = [],
|
|
dry_run: bool = True,
|
|
) -> Dict[str, Any]:
|
|
"""Batch-push local changes into a CODESYS project in a single open/save cycle.
|
|
|
|
textual_updates: list of {object_path, declaration?, implementation?}
|
|
io_mappings: list of {device_path, param_id, variable}
|
|
device_params: list of {device_path, param_id or param_name, value}
|
|
|
|
This is the primary "push" tool -- reads all changes from Cursor and applies them at once.
|
|
dry_run=True by default for safety.
|
|
"""
|
|
return codesys_submit_job(
|
|
action="sync_project",
|
|
payload={
|
|
"project_path": project_path,
|
|
"textual_updates": textual_updates,
|
|
"io_mappings": io_mappings,
|
|
"device_params": device_params,
|
|
"dry_run": dry_run,
|
|
},
|
|
dry_run=dry_run,
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def build_project(project_path: str, dry_run: bool = False) -> Dict[str, Any]:
|
|
"""Build/compile a CODESYS project."""
|
|
return codesys_submit_job(
|
|
action="build_project",
|
|
payload={"project_path": project_path},
|
|
dry_run=dry_run,
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
def download_to_device(project_path: str, target_name: str, start_application: bool = False, dry_run: bool = False) -> Dict[str, Any]:
|
|
"""Download project to configured target device."""
|
|
return codesys_submit_job(
|
|
action="download_to_device",
|
|
payload={
|
|
"project_path": project_path,
|
|
"target_name": target_name,
|
|
"start_application": start_application,
|
|
},
|
|
dry_run=dry_run,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run()
|