solidworks_mcp.ui.routers.model

solidworks_mcp.ui.routers.model

Model open/connect routes for the Prefab CAD dashboard.

Attributes

DEFAULT_SESSION_ID module-attribute

DEFAULT_SESSION_ID = 'prefab-dashboard'

router module-attribute

router = APIRouter()

Classes

ConnectTargetModelRequest

Bases: BaseModel

Request payload for attaching an active SolidWorks target model.

OpenTargetModelRequest

Bases: BaseModel

Request payload for opening a target model by file path.

UploadedFilePayload

Bases: BaseModel

Browser-uploaded file payload returned by Prefab's OpenFilePicker action.

Functions

connect_model async

connect_model(payload: ConnectTargetModelRequest) -> dict[str, Any]

Attach a target SolidWorks document and derive grounded feature-tree context.

Source code in src/solidworks_mcp/ui/routers/model.py

@router.post("/api/ui/model/connect")
async def connect_model(payload: ConnectTargetModelRequest) -> dict[str, Any]:
    """Attach a target SolidWorks document and derive grounded feature-tree context."""
    return await connect_target_model(
        payload.session_id,
        model_path=payload.model_path,
        uploaded_files=(
            [f.model_dump() for f in payload.uploaded_files]
            if payload.uploaded_files
            else None
        ),
        feature_target_text=payload.feature_target_text,
    )

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)

open_model async

open_model(payload: OpenTargetModelRequest) -> dict[str, Any]

Open a SolidWorks model by path and reflect its feature tree into the session.

Source code in src/solidworks_mcp/ui/routers/model.py

@router.post("/api/ui/model/open")
async def open_model(payload: OpenTargetModelRequest) -> dict[str, Any]:
    """Open a SolidWorks model by path and reflect its feature tree into the session."""
    return await open_target_model(
        payload.session_id,
        model_path=payload.model_path,
        feature_target_text=payload.feature_target_text,
    )

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)