solidworks_mcp.ui.services.model_service

solidworks_mcp.ui.services.model_service

Model connection and upload service for the Prefab CAD assistant dashboard.

Handles opening a target SolidWorks model (open-only) and connecting to it (open + inspect feature tree + trigger preview refresh).

Attributes

DEFAULT_API_ORIGIN module-attribute

DEFAULT_API_ORIGIN = getenv('SOLIDWORKS_UI_API_ORIGIN', 'http://127.0.0.1:8766')

DEFAULT_PREVIEW_ORIENTATION module-attribute

DEFAULT_PREVIEW_ORIENTATION = 'current'

Functions

_resolve_model_path

_resolve_model_path(session_id: str, *, model_path: str | None, uploaded_files: list[dict[str, Any]] | None, feature_target_text: str | None, db_path: Path | None, api_origin: str) -> Path | None

Resolve and validate the model path from either upload payload or explicit path.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
model_path	str | None	Explicit absolute path string.	required
uploaded_files	list[dict[str, Any]] | None	Optional list of upload dicts.	required
feature_target_text	str | None	Optional feature target annotation.	required
db_path	Path | None	Optional SQLite path override.	required
api_origin	str	Base URL of the running FastAPI server.	required

Returns:

Type	Description
Path | None	Resolved Path on success, None when validation fails (error state
Path | None	already persisted to the session).

Source code in src/solidworks_mcp/ui/services/model_service.py

def _resolve_model_path(
    session_id: str,
    *,
    model_path: str | None,
    uploaded_files: list[dict[str, Any]] | None,
    feature_target_text: str | None,
    db_path: Path | None,
    api_origin: str,
) -> Path | None:
    """Resolve and validate the model path from either upload payload or explicit path.

    Args:
        session_id: Dashboard session identifier.
        model_path: Explicit absolute path string.
        uploaded_files: Optional list of upload dicts.
        feature_target_text: Optional feature target annotation.
        db_path: Optional SQLite path override.
        api_origin: Base URL of the running FastAPI server.

    Returns:
        Resolved ``Path`` on success, ``None`` when validation fails (error state
        already persisted to the session).
    """
    if uploaded_files:
        try:
            return materialize_uploaded_model(session_id, uploaded_files)
        except RuntimeError as exc:
            merge_metadata(
                session_id,
                db_path=db_path,
                latest_message="Uploaded model could not be prepared.",
                latest_error_text=str(exc),
                remediation_hint="Choose a valid .sldprt or .sldasm file and retry.",
                feature_target_text=feature_target_text or "",
                workflow_mode="edit_existing",
            )
            return None

    if model_path:
        normalized = sanitize_model_path_text(model_path)
        if not normalized:
            merge_metadata(
                session_id,
                db_path=db_path,
                latest_message="No target model was provided.",
                latest_error_text="Missing model path or uploaded model file.",
                remediation_hint="Choose a local SolidWorks file or provide an absolute model path.",
                feature_target_text=feature_target_text or "",
                workflow_mode="edit_existing",
            )
            return None
        resolved = Path(normalized).expanduser()
        if not resolved.exists():
            merge_metadata(
                session_id,
                db_path=db_path,
                latest_message="Target model path was not found.",
                latest_error_text=f"Missing file: {resolved}",
                remediation_hint="Provide an absolute path to an existing .sldprt or .sldasm file.",
                active_model_path=str(resolved),
                feature_target_text=feature_target_text or "",
                workflow_mode="edit_existing",
            )
            return None
        return resolved

    merge_metadata(
        session_id,
        db_path=db_path,
        latest_message="No target model was provided.",
        latest_error_text="Missing model path or uploaded model file.",
        remediation_hint="Choose a local SolidWorks file or provide an absolute model path.",
        feature_target_text=feature_target_text or "",
        workflow_mode="edit_existing",
    )
    return None

classify_feature_tree_snapshot

classify_feature_tree_snapshot(model_info: Mapping[str, Any] | None, features: list[Mapping[str, Any]] | None) -> dict[str, Any]

Classify a model family from model-info and feature-tree snapshots.

The output is intentionally simple and explainable so agents can use it as a planning primitive rather than as a black-box prediction.

Parameters:

Name	Type	Description	Default
model_info	Mapping[str, Any] | None	The model info value.	required
features	list[Mapping[str, Any]] | None	The features value.	required

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: A dictionary containing the resulting values.

Source code in src/solidworks_mcp/utils/feature_tree_classifier.py

def classify_feature_tree_snapshot(
    model_info: Mapping[str, Any] | None,
    features: list[Mapping[str, Any]] | None,
) -> dict[str, Any]:
    """Classify a model family from model-info and feature-tree snapshots.

    The output is intentionally simple and explainable so agents can use it as a planning
    primitive rather than as a black-box prediction.

    Args:
        model_info (Mapping[str, Any] | None): The model info value.
        features (list[Mapping[str, Any]] | None): The features value.

    Returns:
        dict[str, Any]: A dictionary containing the resulting values.
    """

    feature_list = list(features or [])
    feature_texts = [_feature_text(feature) for feature in feature_list]
    document_type = _as_lower_text((model_info or {}).get("type")) or "unknown"

    evidence: list[str] = []
    warnings: list[str] = []
    next_actions: list[str] = []

    family = "unknown"
    workflow = "inspect-more"
    confidence = "low"
    needs_vba = False

    if document_type == "assembly" or _has_any(feature_texts, _ASSEMBLY_TOKENS):
        family = "assembly"
        workflow = "assembly-planning"
        confidence = "high" if document_type == "assembly" else "medium"
        evidence.append("Assembly document or mate/component evidence detected")
        next_actions.extend(
            [
                "List components and mates before planning inserts or edits",
                "Delegate part-level reconstruction for each component separately",
            ]
        )
    elif document_type == "drawing" or _has_any(feature_texts, _DRAWING_TOKENS):
        family = "drawing"
        workflow = "drawing-review"
        confidence = "high" if document_type == "drawing" else "medium"
        evidence.append("Drawing document or drawing-view evidence detected")
        next_actions.append("Use drawing tools instead of part-modeling tools")
    elif _has_any(feature_texts, _SHEET_METAL_TOKENS):
        family = "sheet_metal"
        workflow = "vba-sheet-metal"
        confidence = "high"
        needs_vba = True
        evidence.extend(
            _match_examples(feature_texts, _SHEET_METAL_TOKENS)
            or ["Sheet metal features detected in the tree"]
        )
        next_actions.extend(
            [
                "Preserve base-flange and bend dependency order",
                "If cuts appear between Unfold and Fold, keep them in flat-pattern state",
                "Prefer a VBA-aware reconstruction plan until direct sheet metal tools exist",
            ]
        )
    elif _has_any(feature_texts, _ADVANCED_SOLID_TOKENS):
        family = "advanced_solid"
        workflow = "vba-advanced-solid"
        confidence = "medium"
        needs_vba = True
        evidence.extend(
            _match_examples(feature_texts, _ADVANCED_SOLID_TOKENS)
            or ["Advanced solid or surface features detected"]
        )
        next_actions.extend(
            [
                "Plan around loft/sweep/shell boundaries before issuing build commands",
                "Prefer macro-backed execution for unsupported direct-MCP features",
            ]
        )
    elif _has_any(feature_texts, _REVOLVE_TOKENS):
        family = "revolve"
        workflow = "direct-mcp-revolve"
        confidence = "high"
        evidence.extend(
            _match_examples(feature_texts, _REVOLVE_TOKENS)
            or ["Revolve features detected"]
        )
        next_actions.extend(
            [
                "Locate the axis sketch or centerline before recreating the profile",
                "Verify the half-profile closes on the revolve axis",
            ]
        )
    elif _has_any(feature_texts, _EXTRUDE_TOKENS):
        family = "extrude"
        workflow = "direct-mcp-extrude"
        confidence = "high"
        evidence.extend(
            _match_examples(feature_texts, _EXTRUDE_TOKENS)
            or ["Extrude features detected"]
        )
        next_actions.extend(
            [
                "Read the driving sketch before recreating downstream cuts or fillets",
                "Verify closed-loop profiles before extrusion",
            ]
        )
    else:
        non_reference_count = 0
        sketch_like_count = 0
        for feature in feature_list:
            feature_type = _as_lower_text(feature.get("type"))
            if feature_type and feature_type not in {
                "refplane",
                "originprofilefeature",
            }:
                non_reference_count += 1
            if feature_type in {"profilefeature", "3dprofilefeature", "sketch"}:
                sketch_like_count += 1

        if sketch_like_count > 0 and sketch_like_count == non_reference_count:
            family = "sketch_only"
            workflow = "inspect-more"
            confidence = "low"
            evidence.append("Only sketch-like features were found in the tree snapshot")
            warnings.append(
                "Feature tree may be incomplete for this adapter/runtime; do not infer the 3D family from sketches alone"
            )
            next_actions.extend(
                [
                    "Combine feature-tree output with mass properties and exported images",
                    "Prefer reading the original file in SolidWorks before planning a rebuild",
                ]
            )
        else:
            warnings.append(
                "No strong feature-family evidence found; keep planning provisional"
            )
            next_actions.extend(
                [
                    "Inspect more of the model state before building",
                    "Avoid committing to direct-MCP or VBA-only paths until evidence improves",
                ]
            )

    return {
        "document_type": document_type or "unknown",
        "family": family,
        "recommended_workflow": workflow,
        "confidence": confidence,
        "needs_vba": needs_vba,
        "evidence": evidence,
        "warnings": warnings,
        "next_actions": next_actions,
        "feature_count": len(feature_list),
    }

connect_target_model async

connect_target_model(session_id: str, *, model_path: str | None = None, uploaded_files: list[dict[str, Any]] | None = None, feature_target_text: str | None = None, db_path: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Open a target model, inspect its feature tree, and trigger a preview refresh.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
model_path	str | None	Absolute path to the SolidWorks model file.	None
uploaded_files	list[dict[str, Any]] | None	List of base64-encoded file upload dicts.	None
feature_target_text	str | None	Optional comma-separated feature target references.	None
db_path	Path | None	Optional SQLite path override.	None
api_origin	str	Base URL of the running FastAPI server.	DEFAULT_API_ORIGIN

Returns:

Type	Description
dict[str, Any]	Full dashboard state payload (includes preview state after the refresh).

Source code in src/solidworks_mcp/ui/services/model_service.py

async def connect_target_model(
    session_id: str,
    *,
    model_path: str | None = None,
    uploaded_files: list[dict[str, Any]] | None = None,
    feature_target_text: str | None = None,
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Open a target model, inspect its feature tree, and trigger a preview refresh.

    Args:
        session_id: Dashboard session identifier.
        model_path: Absolute path to the SolidWorks model file.
        uploaded_files: List of base64-encoded file upload dicts.
        feature_target_text: Optional comma-separated feature target references.
        db_path: Optional SQLite path override.
        api_origin: Base URL of the running FastAPI server.

    Returns:
        Full dashboard state payload (includes preview state after the refresh).
    """
    from .session_service import build_dashboard_state, ensure_dashboard_session  # noqa: PLC0415
    from .preview_service import refresh_preview  # noqa: PLC0415

    ensure_dashboard_session(session_id, db_path=db_path)
    adapter = None
    resolved_path: Path | None = None

    logger.info(
        "[ui.connect_target_model] session_id={} model_path={} uploaded_files={} feature_targets={}",
        session_id,
        model_path,
        len(uploaded_files) if uploaded_files else 0,
        feature_target_text or "",
    )

    resolved_path = _resolve_model_path(
        session_id,
        model_path=model_path,
        uploaded_files=uploaded_files,
        feature_target_text=feature_target_text,
        db_path=db_path,
        api_origin=api_origin,
    )
    if resolved_path is None:
        return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

    config = load_config()
    adapter = await create_adapter(config)
    model_info: dict[str, Any] = {}
    features: list[dict[str, Any]] = []
    attach_succeeded = False
    tool_input = {
        "model_path": str(resolved_path.resolve()),
        "uploaded_file_name": uploaded_files[0].get("name") if uploaded_files else None,
        "feature_target_text": feature_target_text or "",
    }
    try:
        await adapter.connect()
        open_result = await adapter.open_model(str(resolved_path.resolve()))
        if not open_result.is_success:
            raise RuntimeError(open_result.error or "Failed to open target model.")

        if hasattr(adapter, "get_model_info"):
            info_result = await adapter.get_model_info()
            if info_result.is_success and isinstance(info_result.data, dict):
                model_info = info_result.data

        if hasattr(adapter, "list_features"):
            feature_result = await adapter.list_features(include_suppressed=True)
            if feature_result.is_success and isinstance(feature_result.data, list):
                features = [
                    item for item in feature_result.data if isinstance(item, dict)
                ]

        from ._utils import feature_target_status  # noqa: PLC0415

        classification = classify_feature_tree_snapshot(model_info, features)
        target_status, matched_targets, missing_targets = feature_target_status(
            features, feature_target_text
        )

        snapshot_id = insert_model_state_snapshot(
            session_id=session_id,
            model_path=str(resolved_path.resolve()),
            feature_tree_json=json.dumps(features, ensure_ascii=True),
            state_fingerprint=f"{resolved_path.resolve()}::{resolved_path.stat().st_mtime_ns}",
            db_path=db_path,
        )
        for evidence_line in classification.get("evidence", []):
            insert_evidence_link(
                session_id=session_id,
                source_type="active_model",
                source_id=str(resolved_path.resolve()),
                relevance_score=0.96,
                rationale=str(evidence_line),
                payload_json=json.dumps(classification, ensure_ascii=True),
                db_path=db_path,
            )

        if matched_targets or missing_targets:
            insert_evidence_link(
                session_id=session_id,
                source_type="feature_target",
                source_id=str(resolved_path.resolve()),
                relevance_score=0.9 if matched_targets else 0.4,
                rationale=target_status,
                payload_json=json.dumps(
                    {"matched": matched_targets, "missing": missing_targets},
                    ensure_ascii=True,
                ),
                db_path=db_path,
            )

        metadata = merge_metadata(
            session_id,
            db_path=db_path,
            workflow_mode="edit_existing",
            active_model_path=str(resolved_path.resolve()),
            active_model_status=(
                f"Attached model: {resolved_path.name}"
                f" | type={model_info.get('type', 'unknown')}"
                f" | features={len(features)}"
            ),
            active_model_type=str(model_info.get("type") or ""),
            active_model_configuration=str(
                model_info.get("configuration") or "Default"
            ),
            feature_target_text=feature_target_text or "",
            feature_target_status=target_status,
            proposed_family=classification.get("family") or "unknown",
            family_confidence=classification.get("confidence") or "low",
            family_evidence=classification.get("evidence") or [],
            family_warnings=classification.get("warnings") or [],
            latest_message=(
                f"Opened target model {resolved_path.name}. Generating preview views..."
            ),
            preview_status="Generating preview views from attached model...",
            preview_stl_ready=False,
            preview_png_ready=False,
            preview_viewer_url="",
            latest_error_text="",
            remediation_hint="",
            latest_snapshot_id=snapshot_id,
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.connect_target_model",
            input_json=json.dumps(tool_input, ensure_ascii=True),
            output_json=json.dumps(metadata, ensure_ascii=True),
            success=True,
            db_path=db_path,
        )
        attach_succeeded = True
    except Exception as exc:
        logger.exception(
            "[ui.connect_target_model] failed session_id={} path={} error={}",
            session_id,
            str(resolved_path.resolve()) if resolved_path else "<none>",
            exc,
        )
        merge_metadata(
            session_id,
            db_path=db_path,
            workflow_mode="edit_existing",
            active_model_path=str(resolved_path.resolve()),
            feature_target_text=feature_target_text or "",
            latest_message="Failed to attach target model.",
            preview_status="Preview generation failed while attaching model.",
            latest_error_text=str(exc),
            remediation_hint="Open SolidWorks, verify COM access, and retry with a valid .sldprt/.sldasm path.",
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.connect_target_model",
            input_json=json.dumps(tool_input, ensure_ascii=True),
            output_json=json.dumps({"error": str(exc)}, ensure_ascii=True),
            success=False,
            db_path=db_path,
        )

    if attach_succeeded:
        try:
            return await refresh_preview(
                session_id,
                orientation=DEFAULT_PREVIEW_ORIENTATION,
                db_path=db_path,
                preview_dir=ensure_preview_dir(),
                api_origin=api_origin,
                adapter_override=adapter,
                active_model_path_override=str(resolved_path.resolve()),
                reopen_active_model=False,
            )
        except Exception as refresh_exc:
            logger.warning(
                "[ui.connect_target_model] post-attach preview refresh failed: {}",
                str(refresh_exc),
            )

    if adapter is not None:
        try:
            await adapter.disconnect()
        except Exception:
            logger.debug("Adapter disconnect failed during target-model cleanup")

    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

create_adapter async

create_adapter(config: SolidWorksMCPConfig) -> SolidWorksAdapter

Async factory function for creating SolidWorks adapters.

Parameters:

Name	Type	Description	Default
config	SolidWorksMCPConfig	Configuration values for the operation.	required

Returns:

Name	Type	Description
SolidWorksAdapter	SolidWorksAdapter	The result produced by the operation.

Source code in src/solidworks_mcp/adapters/factory.py

async def create_adapter(config: SolidWorksMCPConfig) -> SolidWorksAdapter:
    """Async factory function for creating SolidWorks adapters.

    Args:
        config (SolidWorksMCPConfig): Configuration values for the operation.

    Returns:
        SolidWorksAdapter: The result produced by the operation.
    """
    # Register adapters if not already done
    _register_default_adapters()

    # Create adapter using factory
    adapter = AdapterFactory.create_adapter(config)

    return adapter

ensure_preview_dir

ensure_preview_dir(preview_dir: Path | None = None) -> Path

Create and return the preview image directory.

Parameters:

Name	Type	Description	Default
preview_dir	Path | None	Override directory; defaults to .solidworks_mcp/ui_previews.	None

Returns:

Type	Description
Path	Resolved Path that is guaranteed to exist.

Source code in src/solidworks_mcp/ui/services/_utils.py

def ensure_preview_dir(preview_dir: Path | None = None) -> Path:
    """Create and return the preview image directory.

    Args:
        preview_dir: Override directory; defaults to ``.solidworks_mcp/ui_previews``.

    Returns:
        Resolved ``Path`` that is guaranteed to exist.
    """
    resolved = preview_dir or _DEFAULT_PREVIEW_DIR
    resolved.mkdir(parents=True, exist_ok=True)
    return resolved

insert_evidence_link

insert_evidence_link(*, session_id: str, source_type: str, source_id: str, checkpoint_id: int | None = None, relevance_score: float | None = None, rationale: str | None = None, payload_json: str | None = None, db_path: Path | None = None) -> None

Insert one evidence row used by planning/classification.

Parameters:

Name	Type	Description	Default
session_id	str	The session id value.	required
source_type	str	The source type value.	required
source_id	str	The source id value.	required
checkpoint_id	int | None	The checkpoint id value. Defaults to None.	None
relevance_score	float | None	The relevance score value. Defaults to None.	None
rationale	str | None	The rationale value. Defaults to None.	None
payload_json	str | None	The payload json value. Defaults to None.	None
db_path	Path | None	The db path value. Defaults to None.	None

Returns:

Name	Type	Description
None	None	None.

Source code in src/solidworks_mcp/agents/history_db.py

def insert_evidence_link(
    *,
    session_id: str,
    source_type: str,
    source_id: str,
    checkpoint_id: int | None = None,
    relevance_score: float | None = None,
    rationale: str | None = None,
    payload_json: str | None = None,
    db_path: Path | None = None,
) -> None:
    """Insert one evidence row used by planning/classification.

    Args:
        session_id (str): The session id value.
        source_type (str): The source type value.
        source_id (str): The source id value.
        checkpoint_id (int | None): The checkpoint id value. Defaults to None.
        relevance_score (float | None): The relevance score value. Defaults to None.
        rationale (str | None): The rationale value. Defaults to None.
        payload_json (str | None): The payload json value. Defaults to None.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        None: None.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        session.add(
            EvidenceLink(
                session_id=session_id,
                checkpoint_id=checkpoint_id,
                source_type=source_type,
                source_id=source_id,
                relevance_score=relevance_score,
                rationale=rationale,
                payload_json=payload_json,
                created_at=_utc_now_iso(),
            )
        )
        session.commit()

insert_model_state_snapshot

insert_model_state_snapshot(*, session_id: str, checkpoint_id: int | None = None, model_path: str | None = None, feature_tree_json: str | None = None, mass_properties_json: str | None = None, screenshot_path: str | None = None, state_fingerprint: str | None = None, db_path: Path | None = None) -> int

Insert model snapshot row and return snapshot ID for rollback tracking.

Parameters:

Name	Type	Description	Default
session_id	str	The session id value.	required
checkpoint_id	int | None	The checkpoint id value. Defaults to None.	None
model_path	str | None	The model path value. Defaults to None.	None
feature_tree_json	str | None	The feature tree json value. Defaults to None.	None
mass_properties_json	str | None	The mass properties json value. Defaults to None.	None
screenshot_path	str | None	The screenshot path value. Defaults to None.	None
state_fingerprint	str | None	The state fingerprint value. Defaults to None.	None
db_path	Path | None	The db path value. Defaults to None.	None

Returns:

Name	Type	Description
int	int	The computed numeric result.

Source code in src/solidworks_mcp/agents/history_db.py

def insert_model_state_snapshot(
    *,
    session_id: str,
    checkpoint_id: int | None = None,
    model_path: str | None = None,
    feature_tree_json: str | None = None,
    mass_properties_json: str | None = None,
    screenshot_path: str | None = None,
    state_fingerprint: str | None = None,
    db_path: Path | None = None,
) -> int:
    """Insert model snapshot row and return snapshot ID for rollback tracking.

    Args:
        session_id (str): The session id value.
        checkpoint_id (int | None): The checkpoint id value. Defaults to None.
        model_path (str | None): The model path value. Defaults to None.
        feature_tree_json (str | None): The feature tree json value. Defaults to None.
        mass_properties_json (str | None): The mass properties json value. Defaults to None.
        screenshot_path (str | None): The screenshot path value. Defaults to None.
        state_fingerprint (str | None): The state fingerprint value. Defaults to None.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        int: The computed numeric result.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        row = ModelStateSnapshot(
            session_id=session_id,
            checkpoint_id=checkpoint_id,
            model_path=model_path,
            feature_tree_json=feature_tree_json,
            mass_properties_json=mass_properties_json,
            screenshot_path=screenshot_path,
            state_fingerprint=state_fingerprint,
            created_at=_utc_now_iso(),
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        return int(row.id or 0)

insert_tool_call_record

insert_tool_call_record(*, session_id: str, tool_name: str, checkpoint_id: int | None = None, run_id: str | None = None, input_json: str | None = None, output_json: str | None = None, success: bool = True, latency_ms: float | None = None, db_path: Path | None = None) -> None

Insert one tool call execution record.

Parameters:

Name	Type	Description	Default
session_id	str	The session id value.	required
tool_name	str	The tool name value.	required
checkpoint_id	int | None	The checkpoint id value. Defaults to None.	None
run_id	str | None	The run id value. Defaults to None.	None
input_json	str | None	The input json value. Defaults to None.	None
output_json	str | None	The output json value. Defaults to None.	None
success	bool	The success value. Defaults to True.	True
latency_ms	float | None	The latency ms value. Defaults to None.	None
db_path	Path | None	The db path value. Defaults to None.	None

Returns:

Name	Type	Description
None	None	None.

Source code in src/solidworks_mcp/agents/history_db.py

def insert_tool_call_record(
    *,
    session_id: str,
    tool_name: str,
    checkpoint_id: int | None = None,
    run_id: str | None = None,
    input_json: str | None = None,
    output_json: str | None = None,
    success: bool = True,
    latency_ms: float | None = None,
    db_path: Path | None = None,
) -> None:
    """Insert one tool call execution record.

    Args:
        session_id (str): The session id value.
        tool_name (str): The tool name value.
        checkpoint_id (int | None): The checkpoint id value. Defaults to None.
        run_id (str | None): The run id value. Defaults to None.
        input_json (str | None): The input json value. Defaults to None.
        output_json (str | None): The output json value. Defaults to None.
        success (bool): The success value. Defaults to True.
        latency_ms (float | None): The latency ms value. Defaults to None.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        None: None.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        session.add(
            ToolCallRecord(
                session_id=session_id,
                checkpoint_id=checkpoint_id,
                run_id=run_id,
                tool_name=tool_name,
                input_json=input_json,
                output_json=output_json,
                success=success,
                latency_ms=latency_ms,
                created_at=_utc_now_iso(),
            )
        )
        session.commit()

load_config

load_config(config_file: str | None = None) -> SolidWorksMCPConfig

Load configuration from file and environment variables.

Parameters:

Name	Type	Description	Default
config_file	str | None	The config file value. Defaults to None.	None

Returns:

Name	Type	Description
SolidWorksMCPConfig	SolidWorksMCPConfig	The result produced by the operation.

Source code in src/solidworks_mcp/config.py

def load_config(config_file: str | None = None) -> SolidWorksMCPConfig:
    """Load configuration from file and environment variables.

    Args:
        config_file (str | None): The config file value. Defaults to None.

    Returns:
        SolidWorksMCPConfig: The result produced by the operation.
    """
    if config_file:
        config_path = Path(config_file)
        if config_path.exists() and config_path.suffix.lower() == ".json":
            import json

            with config_path.open("r", encoding="utf-8") as f:
                data = json.load(f)
            return SolidWorksMCPConfig(**data)
        return SolidWorksMCPConfig.from_env(str(config_path))

    return SolidWorksMCPConfig.from_env()

materialize_uploaded_model

materialize_uploaded_model(session_id: str, uploaded_files: list[dict[str, Any]] | None, *, upload_dir: Path | None = None) -> Path

Decode a base64-encoded uploaded model file and write it to the staging directory.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier used as a staging subdirectory.	required
uploaded_files	list[dict[str, Any]] | None	List of file payload dicts, each with name and data fields.	required
upload_dir	Path | None	Override for the upload staging directory.	None

Returns:

Type	Description
Path	Path pointing at the decoded file on disk.

Raises:

Type	Description
RuntimeError	When no file is provided, the name is missing, the suffix is unsupported, or the data field is not valid base64.

Source code in src/solidworks_mcp/ui/services/_utils.py

def materialize_uploaded_model(
    session_id: str,
    uploaded_files: list[dict[str, Any]] | None,
    *,
    upload_dir: Path | None = None,
) -> Path:
    """Decode a base64-encoded uploaded model file and write it to the staging directory.

    Args:
        session_id: Dashboard session identifier used as a staging subdirectory.
        uploaded_files: List of file payload dicts, each with ``name`` and ``data`` fields.
        upload_dir: Override for the upload staging directory.

    Returns:
        Path pointing at the decoded file on disk.

    Raises:
        RuntimeError: When no file is provided, the name is missing, the suffix is
            unsupported, or the data field is not valid base64.
    """
    if not uploaded_files:
        raise RuntimeError("No uploaded model file was provided.")

    upload = uploaded_files[0]
    file_name = Path(str(upload.get("name") or "")).name
    if not file_name:
        raise RuntimeError("Uploaded model is missing a filename.")

    suffix = Path(file_name).suffix.lower()
    if suffix not in SUPPORTED_MODEL_UPLOAD_SUFFIXES:
        allowed = ", ".join(sorted(SUPPORTED_MODEL_UPLOAD_SUFFIXES))
        raise RuntimeError(
            f"Unsupported uploaded model type: {suffix or 'unknown'}. Expected one of {allowed}."
        )

    encoded_data = upload.get("data")
    if not isinstance(encoded_data, str) or not encoded_data.strip():
        raise RuntimeError("Uploaded model is missing file data.")

    try:
        file_bytes = base64.b64decode(encoded_data, validate=True)
    except (binascii.Error, ValueError) as exc:
        raise RuntimeError("Uploaded model payload is not valid base64 data.") from exc

    target_dir = ensure_uploaded_model_dir(upload_dir) / session_id
    target_dir.mkdir(parents=True, exist_ok=True)
    target_path = target_dir / file_name
    target_path.write_bytes(file_bytes)
    return target_path

merge_metadata

merge_metadata(session_id: str, *, db_path: Path | None = None, user_goal: str | None = None, **updates: Any) -> dict[str, Any]

Read session metadata, merge updates into it, and write it back.

Implements the optimistic read-modify-write pattern used across all service functions that need to update one or more metadata keys without overwriting unrelated keys.

Parameters:

Name	Type	Description	Default
session_id	str	Target session identifier.	required
db_path	Path | None	Optional override for the SQLite database path.	None
user_goal	str | None	When provided, also updates the user_goal column.	None
**updates	Any	Arbitrary key-value pairs to merge into metadata.	{}

Returns:

Type	Description
dict[str, Any]	The merged metadata dict after the write.

Source code in src/solidworks_mcp/ui/services/_utils.py

def merge_metadata(
    session_id: str,
    *,
    db_path: Path | None = None,
    user_goal: str | None = None,
    **updates: Any,
) -> dict[str, Any]:
    """Read session metadata, merge *updates* into it, and write it back.

    Implements the optimistic read-modify-write pattern used across all
    service functions that need to update one or more metadata keys without
    overwriting unrelated keys.

    Args:
        session_id: Target session identifier.
        db_path: Optional override for the SQLite database path.
        user_goal: When provided, also updates the ``user_goal`` column.
        **updates: Arbitrary key-value pairs to merge into metadata.

    Returns:
        The merged metadata dict after the write.
    """
    session_row = get_design_session(session_id, db_path=db_path)
    metadata = parse_json_blob(session_row["metadata_json"]) if session_row else {}
    metadata.update(updates)

    effective_goal = user_goal or (
        session_row["user_goal"] if session_row else DEFAULT_USER_GOAL
    )
    effective_source = (
        session_row["source_mode"] if session_row else DEFAULT_SOURCE_MODE
    )
    effective_family = session_row["accepted_family"] if session_row else None
    effective_status = session_row["status"] if session_row else "active"
    effective_index = session_row["current_checkpoint_index"] if session_row else 0

    upsert_design_session(
        session_id=session_id,
        user_goal=effective_goal,
        source_mode=effective_source,
        accepted_family=effective_family,
        status=effective_status,
        current_checkpoint_index=effective_index,
        metadata_json=json.dumps(metadata, ensure_ascii=True),
        db_path=db_path,
    )
    return metadata

open_target_model async

open_target_model(session_id: str, *, model_path: str | None = None, uploaded_files: list[dict[str, Any]] | None = None, feature_target_text: str | None = None, db_path: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Open a target model in SolidWorks and persist session state.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
model_path	str | None	Absolute path to the SolidWorks model file.	None
uploaded_files	list[dict[str, Any]] | None	List of base64-encoded file upload dicts.	None
feature_target_text	str | None	Optional comma-separated feature target references.	None
db_path	Path | None	Optional SQLite path override.	None
api_origin	str	Base URL of the running FastAPI server.	DEFAULT_API_ORIGIN

Returns:

Type	Description
dict[str, Any]	Full dashboard state payload.

Source code in src/solidworks_mcp/ui/services/model_service.py

async def open_target_model(
    session_id: str,
    *,
    model_path: str | None = None,
    uploaded_files: list[dict[str, Any]] | None = None,
    feature_target_text: str | None = None,
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Open a target model in SolidWorks and persist session state.

    Args:
        session_id: Dashboard session identifier.
        model_path: Absolute path to the SolidWorks model file.
        uploaded_files: List of base64-encoded file upload dicts.
        feature_target_text: Optional comma-separated feature target references.
        db_path: Optional SQLite path override.
        api_origin: Base URL of the running FastAPI server.

    Returns:
        Full dashboard state payload.
    """
    from .session_service import build_dashboard_state, ensure_dashboard_session  # noqa: PLC0415

    ensure_dashboard_session(session_id, db_path=db_path)
    adapter = None
    resolved_path: Path | None = None

    logger.info(
        "[ui.open_target_model] session_id={} model_path={} uploaded_files={} feature_targets={}",
        session_id,
        model_path,
        len(uploaded_files) if uploaded_files else 0,
        feature_target_text or "",
    )

    resolved_path = _resolve_model_path(
        session_id,
        model_path=model_path,
        uploaded_files=uploaded_files,
        feature_target_text=feature_target_text,
        db_path=db_path,
        api_origin=api_origin,
    )
    if resolved_path is None:
        return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

    config = load_config()
    adapter = await create_adapter(config)
    model_info: dict[str, Any] = {}
    tool_input = {
        "model_path": str(resolved_path.resolve()),
        "uploaded_file_name": uploaded_files[0].get("name") if uploaded_files else None,
        "feature_target_text": feature_target_text or "",
    }
    try:
        await adapter.connect()
        open_result = await adapter.open_model(str(resolved_path.resolve()))
        if not open_result.is_success:
            raise RuntimeError(open_result.error or "Failed to open target model.")

        if hasattr(adapter, "get_model_info"):
            info_result = await adapter.get_model_info()
            if info_result.is_success and isinstance(info_result.data, dict):
                model_info = info_result.data

        metadata = merge_metadata(
            session_id,
            db_path=db_path,
            workflow_mode="edit_existing",
            active_model_path=str(resolved_path.resolve()),
            active_model_status=(
                f"Opened model: {resolved_path.name}"
                f" | type={model_info.get('type', 'unknown')}"
            ),
            active_model_type=str(model_info.get("type") or ""),
            active_model_configuration=str(
                model_info.get("configuration") or "Default"
            ),
            feature_target_text=feature_target_text or "",
            latest_message=f"Opened target model {resolved_path.name} in SolidWorks.",
            latest_error_text="",
            remediation_hint="",
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.open_target_model",
            input_json=json.dumps(tool_input, ensure_ascii=True),
            output_json=json.dumps(metadata, ensure_ascii=True),
            success=True,
            db_path=db_path,
        )
    except Exception as exc:
        logger.exception(
            "[ui.open_target_model] failed session_id={} path={} error={}",
            session_id,
            str(resolved_path.resolve()) if resolved_path else "<none>",
            exc,
        )
        merge_metadata(
            session_id,
            db_path=db_path,
            workflow_mode="edit_existing",
            active_model_path=str(resolved_path.resolve()),
            feature_target_text=feature_target_text or "",
            latest_message="Failed to open target model.",
            latest_error_text=str(exc),
            remediation_hint="Open SolidWorks, verify COM access, and retry with a valid .sldprt/.sldasm path.",
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.open_target_model",
            input_json=json.dumps(tool_input, ensure_ascii=True),
            output_json=json.dumps({"error": str(exc)}, ensure_ascii=True),
            success=False,
            db_path=db_path,
        )
    finally:
        if adapter is not None:
            try:
                await adapter.disconnect()
            except Exception:
                logger.debug("Adapter disconnect failed during open-model cleanup")

    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)

sanitize_model_path_text

sanitize_model_path_text(value: Any) -> str

Strip surrounding quotes from a model path string.

Parameters:

Name	Type	Description	Default
value	Any	Raw model path value from UI state.	required

Returns:

Type	Description
str	Cleaned path string, or "" if empty.

Source code in src/solidworks_mcp/ui/services/_utils.py

def sanitize_model_path_text(value: Any) -> str:
    """Strip surrounding quotes from a model path string.

    Args:
        value: Raw model path value from UI state.

    Returns:
        Cleaned path string, or ``""`` if empty.
    """
    text = sanitize_ui_text(value, "")
    if len(text) >= 2 and text[0] == text[-1] and text[0] in {'"', "'"}:
        text = text[1:-1].strip()
    return text