Add OSA creation and file import MCP tools

README.md

@@ -138,6 +138,32 @@ The following example prompts show how you can interact with the VEOS MCP server
 
 </details>
 
+<details>
+<summary><b>OSA Creation</b></summary>
+
+- **veos_create_osa**
+  - Title: Create a new VEOS OSA file
+  - Description: Creates and saves a new, empty VEOS offline simulation application using the headless VEOS Model Console.
+  - Parameters:
+    - `osa_path` (string): Destination path of the OSA file. The parent directory must already exist.
+  - Read-only: **false**
+
+</details>
+
+<details>
+<summary><b>File Import</b></summary>
+
+- **veos_import_file**
+  - Title: Import a file into a VEOS simulation system
+  - Description: Imports a VEOS-supported file into an existing OSA and saves it using the headless VEOS Model or Build Console. Supported types are OSA, JSON, FMU, SIC, BSC, SMC, classic VECU, and adaptive VECU.
+  - Parameters:
+    - `osa_path` (string): Path to the OSA simulation system to modify.
+    - `import_file_path` (string): Path to the file to import using the default VEOS import and build settings.
+    - `file_type` (string, optional): Explicit import type. Required for VECU (`classic-vecu` or `adaptive-vecu`) and optional for other supported types.
+  - Read-only: **false**
+
+</details>
+
 <details>
 <summary><b>Simulator Control</b></summary>
 

src/veos_mcp/runtime.py

@@ -9,10 +9,9 @@
 from veos_mcp.models.errors import ErrorType, VeosError
 from veos_mcp.veos_cli import VeosCli
 from veos_mcp.veos_path_resolver import (
-    check_veos_installation_exists,
     get_linux_installations,
     get_windows_installations,
-    resolve_veos_path_for_version,
+    resolve_veos_installation,
 )
 
 mcp = FastMCP(
@@ -25,17 +24,17 @@
 
 
 def configure_cli(*, veos_version: str | None, veos_bin_path: str | None) -> None:
-    """Configure the server with the VEOS CLI installation directory."""
+    """Configure the server with the selected VEOS installation."""
 
     veos_installations = get_windows_installations() if sys.platform.startswith("win32") else get_linux_installations()
-
-    if veos_bin_path is not None:
-        veos_path = check_veos_installation_exists(veos_installations, veos_bin_path)
-    else:
-        veos_path = resolve_veos_path_for_version(veos_installations, veos_version)
+    installation = resolve_veos_installation(
+        veos_installations,
+        veos_version=veos_version,
+        veos_bin_path=veos_bin_path,
+    )
 
     global _veos_cli
-    _veos_cli = VeosCli(veos_path=veos_path)
+    _veos_cli = VeosCli(veos_path=installation.executable_path)
 
 
 def get_cli() -> VeosCli:

src/veos_mcp/tools/__init__.py

@@ -1,12 +1,16 @@
 """MCP tools exposed by the VEOS MCP server."""
 
+from veos_mcp.tools import file_import as file_import
 from veos_mcp.tools import log_file_access as log_file_access
+from veos_mcp.tools import osa_creation as osa_creation
 from veos_mcp.tools import sim_control as sim_control
 from veos_mcp.tools import system_extraction as system_extraction
 from veos_mcp.tools import system_modification as system_modification
 
 __all__ = [
+    "file_import",
     "log_file_access",
+    "osa_creation",
     "sim_control",
     "system_extraction",
     "system_modification",

src/veos_mcp/tools/file_import.py

@@ -0,0 +1,97 @@
+"""MCP tool for importing files with the VEOS Model and Build Consoles."""
+
+from pathlib import Path
+from typing import Literal
+
+from mcp.types import CallToolResult, ToolAnnotations
+
+from veos_mcp.runtime import (
+    create_command_result_response_error,
+    create_command_result_response_success,
+    create_error_response,
+    get_cli,
+    mcp,
+)
+
+ImportFileType = Literal[
+    "osa",
+    "json",
+    "fmu",
+    "sic",
+    "bsc",
+    "smc",
+    "classic-vecu",
+    "adaptive-vecu",
+]
+
+_MODEL_IMPORT_TYPES = {"osa", "json"}
+_BUILD_TYPES_BY_EXTENSION = {
+    ".fmu": "fmu",
+    ".sic": "sic",
+    ".bsc": "bsc",
+    ".smc": "smc",
+}
+
+
+def _resolve_file_type(import_file_path: str, file_type: ImportFileType | None) -> str | None:
+    if file_type is not None:
+        return file_type
+
+    extension = Path(import_file_path).suffix.lower()
+    if extension in {".osa", ".json"}:
+        return extension.removeprefix(".")
+    return _BUILD_TYPES_BY_EXTENSION.get(extension)
+
+
+@mcp.tool(
+    name="veos_import_file",
+    title="Import a file into a VEOS simulation system",
+    description=(
+        "Imports a VEOS-supported file into an existing OSA and saves the updated OSA using the headless VEOS "
+        "Model or Build Console. File types OSA, JSON, FMU, SIC, BSC, and SMC are inferred from their extension. "
+        "For VECU files, file_type must be classic-vecu or adaptive-vecu. An explicit file_type can also be used "
+        "for a supported container with a nonstandard extension. VEOS uses the default build settings."
+    ),
+    annotations=ToolAnnotations(
+        readOnlyHint=False,
+        destructiveHint=False,
+        idempotentHint=False,
+        openWorldHint=False,
+    ),
+)
+def veos_import_file(
+    osa_path: str,
+    import_file_path: str,
+    file_type: ImportFileType | None = None,
+) -> CallToolResult:
+    """Import a supported file into an OSA simulation system."""
+    resolved_file_type = _resolve_file_type(import_file_path, file_type)
+    if resolved_file_type is None:
+        return create_error_response(
+            f"Could not determine the VEOS import type for {import_file_path}. "
+            "Specify file_type; VECU files require classic-vecu or adaptive-vecu."
+        )
+
+    if resolved_file_type in _MODEL_IMPORT_TYPES:
+        command_result = get_cli().run_model(
+            "import",
+            osa_path,
+            "--path",
+            import_file_path,
+            "--save-only-on-success",
+        )
+    else:
+        command_result = get_cli().run_build(
+            resolved_file_type,
+            import_file_path,
+            "--output-file",
+            osa_path,
+        )
+
+    if not command_result.success:
+        return create_command_result_response_error(
+            command_result,
+            f"Failed to import file {import_file_path} into VEOS OSA {osa_path}.",
+        )
+
+    return create_command_result_response_success(command_result)

src/veos_mcp/tools/osa_creation.py

@@ -0,0 +1,42 @@
+"""MCP tool for creating VEOS simulation systems with the Model Console."""
+
+from mcp.types import CallToolResult, ToolAnnotations
+
+from veos_mcp.runtime import (
+    create_command_result_response_error,
+    create_command_result_response_success,
+    get_cli,
+    mcp,
+)
+
+
+@mcp.tool(
+    name="veos_create_osa",
+    title="Create a new VEOS OSA file",
+    description=(
+        "Creates and saves a new, empty VEOS offline simulation application (OSA) at the specified path using the "
+        "headless VEOS Model Console. The parent directory must already exist."
+    ),
+    annotations=ToolAnnotations(
+        readOnlyHint=False,
+        destructiveHint=False,
+        idempotentHint=False,
+        openWorldHint=False,
+    ),
+)
+def veos_create_osa(osa_path: str) -> CallToolResult:
+    """Create a new OSA simulation system."""
+    command_result = get_cli().run_model(
+        "modify",
+        osa_path,
+        "--modifications",
+        "[]",
+        "--create-new",
+        "--save-only-on-success",
+        "--verbosity",
+        "Error",
+    )
+    if not command_result.success:
+        return create_command_result_response_error(command_result, f"Failed to create VEOS OSA file {osa_path}.")
+
+    return create_command_result_response_success(command_result)

src/veos_mcp/veos_cli.py

@@ -8,6 +8,7 @@
 from veos_mcp.models.cli_command_result import CliCommandResult, CommandResultCode
 
 DEFAULT_COMMAND_TIMEOUT_SECONDS = 40.0
+DEFAULT_BUILD_COMMAND_TIMEOUT_SECONDS = 600.0
 _COMMAND_GATE = Lock()
 
 
@@ -130,9 +131,11 @@ def __init__(
         *,
         veos_path: Path,
         command_timeout_seconds: float = DEFAULT_COMMAND_TIMEOUT_SECONDS,
+        build_command_timeout_seconds: float = DEFAULT_BUILD_COMMAND_TIMEOUT_SECONDS,
     ) -> None:
         self.veos_path = veos_path
         self.command_timeout_seconds = command_timeout_seconds
+        self.build_command_timeout_seconds = build_command_timeout_seconds
 
     def run_sim(self, *arguments: str) -> CliCommandResult:
         """Run the VEOS simulator CLI with serialized access."""
@@ -142,14 +145,24 @@ def run_model(self, *arguments: str) -> CliCommandResult:
         """Run the VEOS model CLI with serialized access."""
         return self._run_locked(self.veos_path, ("model", *arguments))
 
+    def run_build(self, *arguments: str) -> CliCommandResult:
+        """Run the VEOS build CLI with serialized access."""
+        return self._run_locked(
+            self.veos_path,
+            ("build", *arguments),
+            command_timeout_seconds=self.build_command_timeout_seconds,
+        )
+
     def _run_locked(
         self,
         executable_path: Path,
         arguments: tuple[str, ...],
+        *,
+        command_timeout_seconds: float | None = None,
     ) -> CliCommandResult:
         with _COMMAND_GATE:
             return run_process_command(
                 executable_path,
                 arguments,
-                command_timeout_seconds=self.command_timeout_seconds,
+                command_timeout_seconds=(self.command_timeout_seconds if command_timeout_seconds is None else command_timeout_seconds),
             )

src/veos_mcp/veos_path_resolver.py

@@ -25,6 +25,11 @@ class _VeosInstallation:
     bin_path: Path
     source_name: str
 
+    @property
+    def executable_path(self) -> Path:
+        executable_name = "veos.exe" if sys.platform.startswith("win32") else "veos"
+        return self.bin_path / executable_name
+
 
 def _normalize_release(release: str | None) -> str | None:
     if release is None:
@@ -163,16 +168,33 @@ def get_linux_installations(
 
 
 def resolve_veos_path_for_version(installations: list[_VeosInstallation], veos_version: str | None) -> Path:
-    candidate_installation = _select_installation(installations, veos_version)
-    return candidate_installation.bin_path / "veos.exe" if sys.platform.startswith("win32") else candidate_installation.bin_path / "veos"
+    return _select_installation(installations, veos_version).executable_path
 
 
 def check_veos_installation_exists(installations: list[_VeosInstallation], veos_bin_path: str) -> Path:
     for installation in installations:
         if installation.bin_path == Path(veos_bin_path):
-            return installation.bin_path / "veos.exe" if sys.platform.startswith("win32") else installation.bin_path / "veos"
+            return installation.executable_path
     raise ValueError(
         f"Provided VEOS path is invalid, make sure that the provided path '{veos_bin_path}' "
         f"points to the bin directory of an existing VEOS installation. "
         f"Following installations were found: {', '.join(str(installation.bin_path) for installation in installations)}."
     )
+
+
+def resolve_veos_installation(
+    installations: list[_VeosInstallation],
+    *,
+    veos_version: str | None,
+    veos_bin_path: str | None,
+) -> _VeosInstallation:
+    if veos_bin_path is None:
+        return _select_installation(installations, veos_version)
+
+    requested_bin_path = Path(veos_bin_path)
+    for installation in installations:
+        if installation.bin_path == requested_bin_path:
+            return installation
+
+    check_veos_installation_exists(installations, veos_bin_path)
+    raise AssertionError("Unreachable")

tests/test_mcp_surface_smoketest.py

@@ -3,6 +3,7 @@
 import asyncio
 import os
 import sys
+from typing import Any
 
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
@@ -22,8 +23,10 @@ def test_list_all_tools() -> None:
     expected_tools = [
         "veos_add_signal_connections",
         "veos_apply_config",
+        "veos_create_osa",
         "veos_get_all_signals_and_ports",
         "veos_get_log_file",
+        "veos_import_file",
         "veos_list_all_available_log_files",
         "veos_load",
         "veos_remove_signal_connections",
@@ -46,6 +49,33 @@ async def list_tool_names() -> list[str]:
     assert actual_tools == expected_tools
 
 
+def test_import_tool_exposes_all_cli_file_types() -> None:
+    """Test that clients can select every supported CLI import route."""
+
+    async def get_file_type_schema() -> dict[str, Any]:
+        server_params = create_server_params()
+
+        async with stdio_client(server_params) as (read, write):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                tools = await session.list_tools()
+                import_tool = next(tool for tool in tools.tools if tool.name == "veos_import_file")
+                return import_tool.inputSchema["properties"]["file_type"]
+
+    file_type_schema = asyncio.run(get_file_type_schema())
+
+    assert file_type_schema["anyOf"][0]["enum"] == [
+        "osa",
+        "json",
+        "fmu",
+        "sic",
+        "bsc",
+        "smc",
+        "classic-vecu",
+        "adaptive-vecu",
+    ]
+
+
 def test_list_all_resource_templates() -> None:
     """Test that all expected resource templates can be listed from the stdio server."""