solidworks_mcp.ui.routers.llm

solidworks_mcp.ui.routers.llm

LLM clarify / inspect / go-orchestration routes for the Prefab CAD dashboard.

Attributes

DEFAULT_SESSION_ID module-attribute

DEFAULT_SESSION_ID = 'prefab-dashboard'

DEFAULT_USER_GOAL module-attribute

DEFAULT_USER_GOAL = 'Design a printable mounting component with documented constraints and fastener strategy.'

router module-attribute

router = APIRouter()

Classes

ClarifyWithAnswerRequest

Bases: BaseModel

Request payload for clarify that includes the user's typed answers.

FamilyInspectRequest

Bases: BaseModel

Request payload for design family classification.

GoOrchestrationRequest

Bases: BaseModel

Request payload for global Go orchestration action.

Functions

_resolve_user_goal

_resolve_user_goal(session_id: str, user_goal: str) -> str

Prefer the approved session goal when the client sends the default placeholder.

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

def _resolve_user_goal(session_id: str, user_goal: str) -> str:
    """Prefer the approved session goal when the client sends the default placeholder."""

    if user_goal and user_goal != DEFAULT_USER_GOAL:
        return user_goal

    session_row = get_design_session(session_id) or {}
    return str(session_row.get("user_goal") or DEFAULT_USER_GOAL)

clarify async

clarify(payload: ClarifyWithAnswerRequest) -> dict[str, Any]

Call the LLM to generate clarifying questions for the design goal.

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

@router.post("/api/ui/clarify")
async def clarify(payload: ClarifyWithAnswerRequest) -> dict[str, Any]:
    """Call the LLM to generate clarifying questions for the design goal."""
    return await request_clarifications(
        payload.session_id,
        user_goal=_resolve_user_goal(payload.session_id, payload.user_goal),
        user_answer=payload.user_answer,
    )

family_inspect async

family_inspect(payload: FamilyInspectRequest) -> dict[str, Any]

Call the LLM to classify the design family from the current goal.

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

@router.post("/api/ui/family/inspect")
async def family_inspect(payload: FamilyInspectRequest) -> dict[str, Any]:
    """Call the LLM to classify the design family from the current goal."""
    return await inspect_family(
        payload.session_id,
        user_goal=_resolve_user_goal(payload.session_id, payload.user_goal),
    )

get_design_session

get_design_session(session_id: str, db_path: Path | None = None) -> dict[str, Any] | None

Return one session row as a dictionary.

Parameters:

Name	Type	Description	Default
session_id	str	The session id value.	required
db_path	Path | None	The db path value. Defaults to None.	None

Returns:

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

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

def get_design_session(
    session_id: str, db_path: Path | None = None
) -> dict[str, Any] | None:
    """Return one session row as a dictionary.

    Args:
        session_id (str): The session id value.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        dict[str, Any] | None: A dictionary containing the resulting values.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        row = session.exec(
            select(DesignSession).where(DesignSession.session_id == session_id)
        ).first()

    if row is None:
        return None
    return {
        "session_id": row.session_id,
        "user_goal": row.user_goal,
        "source_mode": row.source_mode,
        "accepted_family": row.accepted_family,
        "status": row.status,
        "current_checkpoint_index": row.current_checkpoint_index,
        "metadata_json": row.metadata_json,
        "created_at": row.created_at,
        "updated_at": row.updated_at,
    }

inspect_family async

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

Run LLM-backed family classification and suggested checkpoints.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
user_goal	str	Free-text description of the design intent.	required
db_path	Path | None	Optional SQLite path override.	None
model_name	str | None	Optional provider-qualified model override.	None

Returns:

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

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

async def inspect_family(
    session_id: str,
    user_goal: str,
    *,
    db_path: Path | None = None,
    model_name: str | None = None,
) -> dict[str, Any]:
    """Run LLM-backed family classification and suggested checkpoints.

    Args:
        session_id: Dashboard session identifier.
        user_goal: Free-text description of the design intent.
        db_path: Optional SQLite path override.
        model_name: Optional provider-qualified model override.

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

    ensure_dashboard_session(session_id, user_goal=user_goal, db_path=db_path)
    session_row = get_design_session(session_id, db_path=db_path) or {}
    meta = _parse_json_blob(session_row.get("metadata_json"))
    resolved_model = normalize_model_name_for_provider(
        model_name or meta.get("model_name"),
        provider=sanitize_ui_text(meta.get("model_provider"), "github"),
        profile=sanitize_ui_text(meta.get("model_profile"), "balanced"),
    )
    resolved_endpoint = sanitize_ui_text(
        meta.get("local_endpoint"),
        os.getenv("SOLIDWORKS_UI_LOCAL_ENDPOINT", "http://127.0.0.1:11434/v1"),
    )

    local_family = meta.get("proposed_family", "") or "<not yet classified>"
    local_evidence = " | ".join(meta.get("family_evidence", [])) or "<none>"
    prompt = (
        "## TASK\n"
        "Apply the Feature-Tree-Reconstruction skill to classify the SolidWorks feature family "
        "and produce a human-reviewable checkpoint plan.\n\n"
        "## DESIGN GOAL\n"
        f"{user_goal}\n\n"
        "## MODEL CONTEXT\n"
        f"Active model path  : {meta.get('active_model_path', '') or '<none>'}\n"
        f"Active model status: {meta.get('active_model_status', '') or '<none>'}\n"
        f"Feature target refs: {meta.get('feature_target_text', '') or '<none>'}\n"
        f"Feature target status: {meta.get('feature_target_status', '') or '<none>'}\n\n"
        "## LOCAL CLASSIFIER EVIDENCE (pre-computed)\n"
        f"Family  : {local_family}\n"
        f"Evidence: {local_evidence}\n\n"
        "## REFERENCE CORPUS\n"
        f"{meta.get('rag_provenance_text', '') or '<none>'}\n\n"
        "## FEATURE-TREE RECONSTRUCTION SKILL\n"
        "Inspection sequence when model is available (use your mcp tools):\n"
        "  open_model → get_model_info → list_features(include_suppressed=True) "
        "→ get_mass_properties → classify_feature_tree\n"
        "Feature families: revolve | extrude | sheet_metal | advanced_solid | assembly | drawing | unknown\n"
        "Delegation rules:\n"
        "  - sheet_metal or advanced_solid → VBA-aware reconstruction path\n"
        "  - simple part family → direct MCP checkpoint plan\n"
        "  - assembly → component-first decomposition, part-level plan per component\n"
        "Guardrail: never reconstruct from silhouette only. "
        "If confidence is low and contradictory evidence exists, propose more inspection steps. "
        "When no model is attached, treat explicit user-supplied dimensions, named sketch phases, feature ordering, and manufacturing constraints as valid user-confirmed evidence for a conservative plan.\n\n"
        "## OUTPUT CONTRACT\n"
        "Return: family, confidence (high/medium/low), evidence[], warnings[], checkpoints[3-6].\n"
        "You must always emit at least 3 checkpoints when the prompt contains enough explicit geometry to start from an empty part, even if confidence is only low or medium.\n"
        "Each checkpoint: title, allowed_tools[] (from MCP tool catalog), rationale."
    )

    result = await _run_structured_agent(
        system_prompt=(
            "## ROLE\n"
            "You are a SolidWorks routing assistant applying the Feature-Tree-Reconstruction skill.\n"
            "Classify the feature family with evidence and confidence, then produce a safe "
            "checkpoint plan for human review.\n\n"
            "## ORCHESTRATION NOTES\n"
            "  - Inspection before planning: never produce a build plan without at least one "
            "evidence item from model inspection or user-confirmed context.\n"
            "  - Explicit user-specified dimensions, feature names, and operation ordering count as user-confirmed context when no model is attached.\n"
            "  - Propose 3-6 conservative checkpoints. Require human confirmation before each "
            "irreversible step.\n"
            "  - For sheet metal or unsupported advanced features, route to VBA-aware planning.\n"
            "  - Surface warnings when evidence is contradictory or confidence is low.\n"
            "  - Prefer 'extrude' for prompt-only parts built from named sketches and base extrusions unless stronger contrary evidence exists."
        ),
        user_prompt=prompt,
        result_type=FamilyInspection,
        model_name=resolved_model,
        local_endpoint=resolved_endpoint,
    )

    if isinstance(result, RecoverableFailure):
        ordered_features = _extract_explicit_feature_order(user_goal)
        if ordered_features:
            result = _coerce_explicit_feature_order_plan(
                user_goal,
                FamilyInspection(
                    family="extrude",
                    confidence="medium",
                    evidence=[
                        "Deterministic fallback engaged because model routing failed.",
                        "Goal includes an explicit feature-order sequence.",
                    ],
                    warnings=[
                        result.explanation,
                        "Using explicit feature-order checkpoints until model routing is healthy.",
                    ],
                    checkpoints=[],
                ),
            )
        else:
            merge_metadata(
                session_id,
                db_path=db_path,
                latest_message=result.explanation,
                latest_error_text=result.explanation,
                remediation_hint=(
                    result.remediation_steps[0]
                    if result.remediation_steps
                    else "Adjust provider/model settings, then retry inspect."
                ),
            )
            insert_tool_call_record(
                session_id=session_id,
                tool_name="ui.inspect_family",
                input_json=json.dumps({"user_goal": user_goal}, ensure_ascii=True),
                output_json=result.model_dump_json(),
                success=False,
                db_path=db_path,
            )
            return build_dashboard_state(session_id, db_path=db_path)

    result = _coerce_explicit_feature_order_plan(user_goal, result)

    evidence_payload = []
    for index, line in enumerate(result.evidence, start=1):
        insert_evidence_link(
            session_id=session_id,
            source_type="llm",
            source_id=f"family_evidence_{index}",
            relevance_score=0.85,
            rationale=line,
            payload_json=json.dumps({"family": result.family}, ensure_ascii=True),
            db_path=db_path,
        )
        evidence_payload.append(line)

    if result.checkpoints:
        replacement_rows: list[dict[str, Any]] = []
        for index, checkpoint in enumerate(result.checkpoints, start=1):
            replacement_rows.append(
                {
                    "checkpoint_index": index,
                    "title": checkpoint.title,
                    "planned_action_json": json.dumps(
                        {
                            "title": checkpoint.title,
                            "goal": checkpoint.title,
                            "tools": checkpoint.allowed_tools,
                            "rationale": checkpoint.rationale,
                            **checkpoint.execution,
                        },
                        ensure_ascii=True,
                    ),
                    "approved_by_user": index == 1,
                }
            )

        replace_plan_checkpoints(
            session_id=session_id,
            checkpoints=replacement_rows,
            db_path=db_path,
        )

    metadata = merge_metadata(
        session_id,
        db_path=db_path,
        user_goal=user_goal,
        proposed_family=result.family,
        family_confidence=result.confidence,
        family_evidence=evidence_payload,
        family_warnings=result.warnings,
        latest_message=f"Updated family classification to '{result.family}' from GitHub Copilot.",
        latest_error_text="",
        remediation_hint="",
    )
    insert_tool_call_record(
        session_id=session_id,
        tool_name="ui.inspect_family",
        input_json=json.dumps({"user_goal": user_goal}, ensure_ascii=True),
        output_json=result.model_dump_json(),
        success=True,
        db_path=db_path,
    )
    insert_evidence_link(
        session_id=session_id,
        source_type="llm",
        source_id="family_inspection",
        relevance_score=0.93,
        rationale="LLM family classification and checkpoint suggestions from GitHub Copilot.",
        payload_json=json.dumps(metadata, ensure_ascii=True),
        db_path=db_path,
    )
    return build_dashboard_state(session_id, db_path=db_path)

orchestrate_go async

orchestrate_go(payload: GoOrchestrationRequest) -> dict[str, Any]

Run the full Go orchestration pipeline (clarify → inspect → plan checkpoints).

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

@router.post("/api/ui/orchestrate/go")
async def orchestrate_go(payload: GoOrchestrationRequest) -> dict[str, Any]:
    """Run the full Go orchestration pipeline (clarify → inspect → plan checkpoints)."""
    return await run_go_orchestration(
        payload.session_id,
        user_goal=_resolve_user_goal(payload.session_id, payload.user_goal),
        assumptions_text=payload.assumptions_text,
        user_answer=payload.user_answer,
    )

request_clarifications async

request_clarifications(session_id: str, user_goal: str, *, user_answer: str = '', db_path: Path | None = None, model_name: str | None = None) -> dict[str, Any]

Generate focused follow-up questions for the current design goal using LLM.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
user_goal	str	Free-text description of the design intent.	required
user_answer	str	Any previous answers the user has already provided.	''
db_path	Path | None	Optional SQLite path override.	None
model_name	str | None	Optional provider-qualified model override.	None

Returns:

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

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

async def request_clarifications(
    session_id: str,
    user_goal: str,
    *,
    user_answer: str = "",
    db_path: Path | None = None,
    model_name: str | None = None,
) -> dict[str, Any]:
    """Generate focused follow-up questions for the current design goal using LLM.

    Args:
        session_id: Dashboard session identifier.
        user_goal: Free-text description of the design intent.
        user_answer: Any previous answers the user has already provided.
        db_path: Optional SQLite path override.
        model_name: Optional provider-qualified model override.

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

    ensure_dashboard_session(session_id, user_goal=user_goal, db_path=db_path)
    session_row = get_design_session(session_id, db_path=db_path) or {}
    meta = _parse_json_blob(session_row.get("metadata_json"))
    resolved_model = normalize_model_name_for_provider(
        model_name or meta.get("model_name"),
        provider=sanitize_ui_text(meta.get("model_provider"), "github"),
        profile=sanitize_ui_text(meta.get("model_profile"), "balanced"),
    )
    resolved_endpoint = sanitize_ui_text(
        meta.get("local_endpoint"),
        os.getenv("SOLIDWORKS_UI_LOCAL_ENDPOINT", "http://127.0.0.1:11434/v1"),
    )

    answer_section = (
        f"\n## USER ANSWERS / CLARIFICATIONS\n{user_answer}" if user_answer else ""
    )
    assumptions_text = sanitize_ui_text(
        meta.get("assumptions_text"), "<none specified>"
    )
    prompt = (
        "## TASK\n"
        "Prepare a SolidWorks design brief using the Printer-Profile-Tolerancing skill.\n"
        "Return a normalized_brief and at most three clarifying_questions that unblock the next modeling step.\n\n"
        "## DESIGN GOAL\n"
        f"{user_goal}\n\n"
        "## MANUFACTURING ASSUMPTIONS\n"
        f"{assumptions_text}\n\n"
        "## MODEL CONTEXT\n"
        f"Active model path : {meta.get('active_model_path', '') or '<none>'}\n"
        f"Feature target refs: {meta.get('feature_target_text', '') or '<none>'}\n"
        f"Reference corpus   : {meta.get('rag_provenance_text', '') or '<none>'}"
        f"{answer_section}\n\n"
        "## OUTPUT CONTRACT\n"
        "normalized_brief: concise paragraph with explicit dimensions/tolerances where known (≥10 chars).\n"
        "questions       : list of up to 3 highest-leverage questions that unblock modeling.\n"
        "  - Return an empty question list when the goal, assumptions, and user answers already provide enough detail to sketch and dimension the next feature.\n"
        "  - Include material/layer-height/nozzle values if missing from assumptions.\n"
        "  - Include critical fit/clearance targets if unspecified.\n"
        "  - Do not ask questions already answered above."
    )

    result = await _run_structured_agent(
        system_prompt=(
            "## ROLE\n"
            "You are a CAD planning assistant applying the Printer-Profile-Tolerancing skill.\n"
            "Normalize goals into manufacturing-ready language with explicit tolerance/clearance "
            "targets (e.g. '0.30 mm mating clearance', '0.2 mm layer height'). "
            "Ask only the highest-leverage questions that unblock the SolidWorks modeling steps. "
            "Always surface material, nozzle size, and orientation constraints when present in the goal. "
            "If the user has already supplied explicit dimensions, wall thickness, fit targets, and feature placement, do not keep restating the same asks; return zero questions instead."
        ),
        user_prompt=prompt,
        result_type=ClarificationResponse,
        model_name=resolved_model,
        local_endpoint=resolved_endpoint,
    )

    if isinstance(result, RecoverableFailure):
        merge_metadata(
            session_id,
            db_path=db_path,
            latest_message=result.explanation,
            clarifying_questions=[],
            latest_error_text=result.explanation,
            remediation_hint=(
                result.remediation_steps[0]
                if result.remediation_steps
                else "Configure provider credentials and retry."
            ),
        )
        insert_tool_call_record(
            session_id=session_id,
            tool_name="ui.request_clarifications",
            input_json=json.dumps({"user_goal": user_goal}, ensure_ascii=True),
            output_json=result.model_dump_json(),
            success=False,
            db_path=db_path,
        )
        return build_dashboard_state(session_id, db_path=db_path)

    metadata = merge_metadata(
        session_id,
        db_path=db_path,
        user_goal=user_goal,
        normalized_brief=result.normalized_brief,
        clarifying_questions=result.questions,
        user_clarification_answer=user_answer,
        latest_message="Generated clarifying questions from GitHub Copilot.",
        latest_error_text="",
        remediation_hint="",
    )
    insert_tool_call_record(
        session_id=session_id,
        tool_name="ui.request_clarifications",
        input_json=json.dumps({"user_goal": user_goal}, ensure_ascii=True),
        output_json=result.model_dump_json(),
        success=True,
        db_path=db_path,
    )
    insert_evidence_link(
        session_id=session_id,
        source_type="llm",
        source_id="clarification_response",
        relevance_score=0.9,
        rationale="Normalized brief and follow-up questions from GitHub Copilot.",
        payload_json=json.dumps(metadata, ensure_ascii=True),
        db_path=db_path,
    )
    return build_dashboard_state(session_id, db_path=db_path)

run_go_orchestration async

run_go_orchestration(session_id: str, *, user_goal: str, assumptions_text: str | None = None, user_answer: str = '', db_path: Path | None = None, api_origin: str = DEFAULT_API_ORIGIN) -> dict[str, Any]

Run a single end-to-end pass: approve brief, update preferences, clarify, inspect.

Parameters:

Name	Type	Description	Default
session_id	str	Dashboard session identifier.	required
user_goal	str	Free-text description of the design intent.	required
assumptions_text	str | None	Optional manufacturing assumptions to persist.	None
user_answer	str	Any previous answers the user has already provided.	''
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/llm_service.py

async def run_go_orchestration(
    session_id: str,
    *,
    user_goal: str,
    assumptions_text: str | None = None,
    user_answer: str = "",
    db_path: Path | None = None,
    api_origin: str = DEFAULT_API_ORIGIN,
) -> dict[str, Any]:
    """Run a single end-to-end pass: approve brief, update preferences, clarify, inspect.

    Args:
        session_id: Dashboard session identifier.
        user_goal: Free-text description of the design intent.
        assumptions_text: Optional manufacturing assumptions to persist.
        user_answer: Any previous answers the user has already provided.
        db_path: Optional SQLite path override.
        api_origin: Base URL of the running FastAPI server.

    Returns:
        Full dashboard state payload.
    """
    from .session_service import (  # noqa: PLC0415
        approve_design_brief,
        build_dashboard_state,
        update_ui_preferences,
    )

    try:
        goal_text = sanitize_ui_text(user_goal, DEFAULT_USER_GOAL)
        approve_design_brief(session_id, goal_text, db_path=db_path)

        session_row = get_design_session(session_id, db_path=db_path) or {}
        meta = _parse_json_blob(session_row.get("metadata_json"))
        update_ui_preferences(
            session_id,
            assumptions_text=assumptions_text,
            model_provider=str(meta.get("model_provider") or "github"),
            model_profile=str(meta.get("model_profile") or "balanced"),
            model_name=meta.get("model_name"),
            local_endpoint=meta.get("local_endpoint"),
            db_path=db_path,
        )

        await request_clarifications(
            session_id,
            goal_text,
            user_answer=user_answer,
            db_path=db_path,
        )
        await inspect_family(session_id, goal_text, db_path=db_path)

        persist_ui_action(
            session_id,
            tool_name="ui.orchestrate_go",
            db_path=db_path,
            metadata_updates={
                "orchestration_status": (
                    "Go run completed: inputs saved, clarifications refreshed, engineering review updated."
                ),
                "latest_message": "Go run completed across workflow, review, and model output lanes.",
                "latest_error_text": "",
                "remediation_hint": "",
            },
            input_payload={
                "user_goal": goal_text,
                "assumptions_text": assumptions_text,
                "user_answer": user_answer,
            },
            output_payload={
                "status": "success",
                "message": "Go orchestration completed.",
            },
        )
    except Exception as exc:
        logger.exception("[ui.run_go_orchestration] failed session_id={}", session_id)
        merge_metadata(
            session_id,
            db_path=db_path,
            orchestration_status="Go run failed.",
            latest_error_text=str(exc),
            remediation_hint="Review provider credentials/model selection and retry Go.",
        )
    return build_dashboard_state(session_id, db_path=db_path, api_origin=api_origin)