solidworks_mcp.agents

solidworks_mcp.agents

Agent-testing utilities for SolidWorks MCP custom agents.

Attributes

DEFAULT_DB_PATH module-attribute

DEFAULT_DB_PATH = Path('.solidworks_mcp') / 'agent_memory.sqlite3'

__all__ module-attribute

__all__ = ['DEFAULT_DB_PATH', 'DocsPlan', 'ErrorRecord', 'ManufacturabilityReview', 'RecoverableFailure', 'ToolRoutingDecision', 'build_local_retrieval_index', 'find_conversation_events', 'get_design_session', 'find_recent_errors', 'find_run_timeline', 'init_db', 'insert_evidence_link', 'insert_model_state_snapshot', 'insert_plan_checkpoint', 'insert_sketch_graph_snapshot', 'insert_tool_call_record', 'insert_conversation_event', 'list_evidence_links', 'list_model_state_snapshots', 'list_plan_checkpoints', 'list_sketch_graph_snapshots', 'list_tool_call_records', 'pretty_json', 'run_validated_prompt', 'update_plan_checkpoint', 'upsert_design_session']

Classes

DocsPlan

Bases: BaseModel

Validation shape for docs-engineering responses.

Attributes:

Name	Type	Description
audience	str	The audience value.
decisions	list[ToolRoutingDecision]	The decisions value.
demo_steps	list[str]	The demo steps value.
objective	str	The objective value.
sections	list[str]	The sections value.

ErrorRecord

Bases: BaseModel

A normalized error record from an MCP call or planning step.

Attributes:

Name	Type	Description
error_message	str	The error message value.
error_type	str	The error type value.
remediation	str	The remediation value.
root_cause	str	The root cause value.
source	str	The source value.
tool_name	str	The tool name value.

ManufacturabilityReview

Bases: BaseModel

Validation shape for printability-focused agent responses.

Attributes:

Name	Type	Description
assumptions	list[Assumption]	The assumptions value.
build_volume_check	str	The build volume check value.
orientation_guidance	str	The orientation guidance value.
recommendations	list[Recommendation]	The recommendations value.
summary	str	The summary value.
tolerance_clearance_notes	list[str]	The tolerance clearance notes value.

RecoverableFailure

Bases: BaseModel

Typed failure output used when agent needs user-guided retry.

Attributes:

Name	Type	Description
explanation	str	The explanation value.
remediation_steps	list[str]	The remediation steps value.
retry_focus	str | None	The retry focus value.
should_retry	bool	The should retry value.

ToolRoutingDecision

Bases: BaseModel

Validation shape for tool-selection/skills documentation prompts.

Attributes:

Name	Type	Description
fallback_strategy	list[str]	The fallback strategy value.
intent	str	The intent value.
selected_tool_group	str	The selected tool group value.
why	str	The why value.

Functions

build_local_retrieval_index

build_local_retrieval_index(*, output_path: Path | None = None, worked_examples_path: Path | None = None, tool_catalog_dir: Path | None = None, db_path: Path | None = None, max_recent_errors: int = 200) -> dict[str, Any]

Build a JSON retrieval index from local audits, tool docs, and failures.

Parameters:

Name	Type	Description	Default
output_path	Path | None	The output path value. Defaults to None.	None
worked_examples_path	Path | None	The worked examples path value. Defaults to None.	None
tool_catalog_dir	Path | None	The tool catalog dir value. Defaults to None.	None
db_path	Path | None	The db path value. Defaults to None.	None
max_recent_errors	int	The max recent errors value. Defaults to 200.	200

Returns:

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

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

def build_local_retrieval_index(
    *,
    output_path: Path | None = None,
    worked_examples_path: Path | None = None,
    tool_catalog_dir: Path | None = None,
    db_path: Path | None = None,
    max_recent_errors: int = 200,
) -> dict[str, Any]:
    """Build a JSON retrieval index from local audits, tool docs, and failures.

    Args:
        output_path (Path | None): The output path value. Defaults to None.
        worked_examples_path (Path | None): The worked examples path value. Defaults to
                                            None.
        tool_catalog_dir (Path | None): The tool catalog dir value. Defaults to None.
        db_path (Path | None): The db path value. Defaults to None.
        max_recent_errors (int): The max recent errors value. Defaults to 200.

    Returns:
        dict[str, Any]: A dictionary containing the resulting values.
    """
    if output_path is None:
        output_path = Path(".solidworks_mcp") / "retrieval" / "local_index.json"
    if worked_examples_path is None:
        worked_examples_path = Path("docs") / "user-guide" / "worked-examples.md"
    if tool_catalog_dir is None:
        tool_catalog_dir = Path("docs") / "user-guide" / "tool-catalog"
    if db_path is None:
        db_path = DEFAULT_DB_PATH

    chunks: list[dict[str, Any]] = []
    chunk_id = 0

    worked_text = _read_text(worked_examples_path)
    for idx, chunk in enumerate(_chunk_text(worked_text), start=1):
        chunk_id += 1
        chunks.append(
            {
                "id": f"audit-worked-examples-{idx}",
                "chunk_id": chunk_id,
                "source_type": "feature_tree_audit",
                "source": str(worked_examples_path),
                "text": chunk,
                "tags": ["audit", "worked_examples", "inspect_classify_delegate"],
            }
        )

    if tool_catalog_dir.exists():
        for doc_path in sorted(tool_catalog_dir.glob("*.md")):
            if doc_path.name.lower() == "index.md":
                continue
            doc_text = _read_text(doc_path)
            for idx, chunk in enumerate(_chunk_text(doc_text), start=1):
                chunk_id += 1
                chunks.append(
                    {
                        "id": f"tool-doc-{doc_path.stem}-{idx}",
                        "chunk_id": chunk_id,
                        "source_type": "tool_doc",
                        "source": str(doc_path),
                        "text": chunk,
                        "tags": ["tool_catalog", doc_path.stem],
                    }
                )

    for idx, error in enumerate(
        find_recent_errors(limit=max_recent_errors, db_path=db_path),
        start=1,
    ):
        chunk_id += 1
        error_text = (
            f"source={error.get('source')} | tool={error.get('tool_name')} | "
            f"type={error.get('error_type')} | message={error.get('error_message')} | "
            f"root_cause={error.get('root_cause')} | remediation={error.get('remediation')}"
        )
        chunks.append(
            {
                "id": f"failure-memory-{idx}",
                "chunk_id": chunk_id,
                "source_type": "failure_memory",
                "source": str(db_path),
                "text": error_text,
                "tags": ["failure_memory", error.get("tool_name", "unknown")],
                "created_at": error.get("created_at"),
            }
        )

    output_path.parent.mkdir(parents=True, exist_ok=True)

    payload: dict[str, Any] = {
        "version": "1.0",
        "generated_at": datetime.now(UTC).isoformat(),
        "stats": {
            "chunk_count": len(chunks),
            "worked_examples_source": str(worked_examples_path),
            "tool_catalog_source": str(tool_catalog_dir),
            "failure_memory_source": str(db_path),
        },
        "chunks": chunks,
    }

    output_path.write_text(
        json.dumps(payload, indent=2, ensure_ascii=True),
        encoding="utf-8",
    )
    return payload

find_conversation_events

find_conversation_events(conversation_id: str, db_path: Path | None = None) -> list[dict[str, Any]]

Retrieve all events for a conversation, ordered by creation time.

Parameters:

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

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: A list containing the resulting items.

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

def find_conversation_events(
    conversation_id: str, db_path: Path | None = None
) -> list[dict[str, Any]]:
    """Retrieve all events for a conversation, ordered by creation time.

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

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        rows = session.exec(
            select(ConversationEvent)
            .where(ConversationEvent.conversation_id == conversation_id)
            .order_by(ConversationEvent.id.asc())
        ).all()

    return [
        {
            "id": row.id,
            "conversation_id": row.conversation_id,
            "run_id": row.run_id,
            "event_type": row.event_type,
            "role": row.role,
            "content_snippet": row.content_snippet,
            "metadata_json": row.metadata_json,
            "created_at": row.created_at,
        }
        for row in rows
    ]

find_recent_errors

find_recent_errors(limit: int = 20, db_path: Path | None = None) -> list[dict[str, Any]]

Return recent errors so agents can avoid repeated failing states.

Parameters:

Name	Type	Description	Default
limit	int	The limit value. Defaults to 20.	20
db_path	Path | None	The db path value. Defaults to None.	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: A list containing the resulting items.

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

def find_recent_errors(
    limit: int = 20, db_path: Path | None = None
) -> list[dict[str, Any]]:
    """Return recent errors so agents can avoid repeated failing states.

    Args:
        limit (int): The limit value. Defaults to 20.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        rows = session.exec(
            select(ErrorCatalog).order_by(ErrorCatalog.id.desc()).limit(limit)  # type: ignore[union-attr]
        ).all()

    return [
        {
            "run_id": row.run_id,
            "source": row.source,
            "tool_name": row.tool_name,
            "error_type": row.error_type,
            "error_message": row.error_message,
            "root_cause": row.root_cause,
            "remediation": row.remediation,
            "created_at": row.created_at,
        }
        for row in rows
    ]

find_run_timeline

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

Reconstruct a complete timeline for one run, joining runs, tool events, and conversation events.

Parameters:

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

Returns:

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

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

def find_run_timeline(run_id: str, db_path: Path | None = None) -> dict[str, Any]:
    """Reconstruct a complete timeline for one run, joining runs, tool events, and conversation events.

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

    Returns:
        dict[str, Any]: A dictionary containing the resulting values.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)

    timeline: dict[str, Any] = {
        "run_id": run_id,
        "run_info": None,
        "events": [],
    }

    with Session(engine) as session:
        run_row = session.exec(
            select(AgentRun).where(AgentRun.run_id == run_id)
        ).first()

        if run_row:
            timeline["run_info"] = {
                "agent_name": run_row.agent_name,
                "prompt_preview": run_row.prompt[:200] if run_row.prompt else None,
                "model_name": run_row.model_name,
                "status": run_row.status,
                "created_at": run_row.created_at,
            }

        tool_events = session.exec(
            select(ToolEvent)
            .where(ToolEvent.run_id == run_id)
            .order_by(ToolEvent.id.asc())  # type: ignore[union-attr]
        ).all()

        convo_events = session.exec(
            select(ConversationEvent)
            .where(ConversationEvent.run_id == run_id)
            .order_by(ConversationEvent.id.asc())  # type: ignore[union-attr]
        ).all()

        events = []
        for evt in tool_events:
            events.append(
                {
                    "timestamp": evt.created_at,
                    "event_type": "tool",
                    "tool_name": evt.tool_name,
                    "phase": evt.phase,
                    "payload_preview": evt.payload_json[:100]
                    if evt.payload_json
                    else None,
                }
            )

        for evt in convo_events:
            events.append(
                {
                    "timestamp": evt.created_at,
                    "event_type": "message",
                    "role": evt.role,
                    "content_preview": evt.content_snippet[:100],
                    "metadata": evt.metadata_json,
                }
            )

        events.sort(key=lambda e: e["timestamp"] or "")  # type: ignore[return-value]
        timeline["events"] = events

    return timeline

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,
    }

init_db

init_db(db_path: Path | None = None) -> Path

Create SQLModel tables used by the lightweight agent memory system.

Parameters:

Name	Type	Description	Default
db_path	Path | None	The db path value. Defaults to None.	None

Returns:

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

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

def init_db(db_path: Path | None = None) -> Path:
    """Create SQLModel tables used by the lightweight agent memory system.

    Args:
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        Path: The result produced by the operation.
    """
    resolved = db_path or DEFAULT_DB_PATH
    engine = _build_engine(resolved)
    try:
        SQLModel.metadata.create_all(engine)
    finally:
        engine.dispose()
    return resolved

insert_conversation_event

insert_conversation_event(*, conversation_id: str, event_type: str, content_snippet: str, role: str | None = None, run_id: str | None = None, metadata_json: str | None = None, db_path: Path | None = None) -> None

Record a conversation event (message, system event, or tool call) linked to a run.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation id value.	required
event_type	str	The event type value.	required
content_snippet	str	The content snippet value.	required
role	str | None	The role value. Defaults to None.	None
run_id	str | None	The run id value. Defaults to None.	None
metadata_json	str | None	The metadata 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_conversation_event(
    *,
    conversation_id: str,
    event_type: str,
    content_snippet: str,
    role: str | None = None,
    run_id: str | None = None,
    metadata_json: str | None = None,
    db_path: Path | None = None,
) -> None:
    """Record a conversation event (message, system event, or tool call) linked to a run.

    Args:
        conversation_id (str): The conversation id value.
        event_type (str): The event type value.
        content_snippet (str): The content snippet value.
        role (str | None): The role value. Defaults to None.
        run_id (str | None): The run id value. Defaults to None.
        metadata_json (str | None): The metadata 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(
            ConversationEvent(
                conversation_id=conversation_id,
                run_id=run_id,
                event_type=event_type,
                role=role,
                content_snippet=content_snippet,
                metadata_json=metadata_json,
                created_at=_utc_now_iso(),
            )
        )
        session.commit()

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_plan_checkpoint

insert_plan_checkpoint(*, session_id: str, checkpoint_index: int, title: str, planned_action_json: str, approved_by_user: bool = False, executed: bool = False, result_json: str | None = None, rollback_snapshot_id: int | None = None, db_path: Path | None = None) -> int

Insert a new checkpoint and return its ID.

Parameters:

Name	Type	Description	Default
session_id	str	The session id value.	required
checkpoint_index	int	The checkpoint index value.	required
title	str	The title value.	required
planned_action_json	str	The planned action json value.	required
approved_by_user	bool	The approved by user value. Defaults to False.	False
executed	bool	The executed value. Defaults to False.	False
result_json	str | None	The result json value. Defaults to None.	None
rollback_snapshot_id	int | None	The rollback snapshot id 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_plan_checkpoint(
    *,
    session_id: str,
    checkpoint_index: int,
    title: str,
    planned_action_json: str,
    approved_by_user: bool = False,
    executed: bool = False,
    result_json: str | None = None,
    rollback_snapshot_id: int | None = None,
    db_path: Path | None = None,
) -> int:
    """Insert a new checkpoint and return its ID.

    Args:
        session_id (str): The session id value.
        checkpoint_index (int): The checkpoint index value.
        title (str): The title value.
        planned_action_json (str): The planned action json value.
        approved_by_user (bool): The approved by user value. Defaults to False.
        executed (bool): The executed value. Defaults to False.
        result_json (str | None): The result json value. Defaults to None.
        rollback_snapshot_id (int | None): The rollback snapshot id 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)
    now = _utc_now_iso()
    with Session(engine) as session:
        row = PlanCheckpoint(
            session_id=session_id,
            checkpoint_index=checkpoint_index,
            title=title,
            planned_action_json=planned_action_json,
            approved_by_user=approved_by_user,
            executed=executed,
            result_json=result_json,
            rollback_snapshot_id=rollback_snapshot_id,
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        return int(row.id or 0)

insert_sketch_graph_snapshot

insert_sketch_graph_snapshot(*, session_id: str, nodes_json: str, edges_json: str, model_path: str | None = None, graph_format: str = 'json', metadata_json: str | None = None, db_path: Path | None = None) -> None

Store lightweight sketch graph snapshots in SQLite (Section F).

Parameters:

Name	Type	Description	Default
session_id	str	The session id value.	required
nodes_json	str	The nodes json value.	required
edges_json	str	The edges json value.	required
model_path	str | None	The model path value. Defaults to None.	None
graph_format	str	The graph format value. Defaults to "json".	'json'
metadata_json	str | None	The metadata 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_sketch_graph_snapshot(
    *,
    session_id: str,
    nodes_json: str,
    edges_json: str,
    model_path: str | None = None,
    graph_format: str = "json",
    metadata_json: str | None = None,
    db_path: Path | None = None,
) -> None:
    """Store lightweight sketch graph snapshots in SQLite (Section F).

    Args:
        session_id (str): The session id value.
        nodes_json (str): The nodes json value.
        edges_json (str): The edges json value.
        model_path (str | None): The model path value. Defaults to None.
        graph_format (str): The graph format value. Defaults to "json".
        metadata_json (str | None): The metadata 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(
            SketchGraphSnapshot(
                session_id=session_id,
                model_path=model_path,
                graph_format=graph_format,
                nodes_json=nodes_json,
                edges_json=edges_json,
                metadata_json=metadata_json,
                created_at=_utc_now_iso(),
            )
        )
        session.commit()

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()

list_evidence_links

list_evidence_links(session_id: str, checkpoint_id: int | None = None, db_path: Path | None = None) -> list[dict[str, Any]]

List evidence rows for a session and optional checkpoint.

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
db_path	Path | None	The db path value. Defaults to None.	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: A list containing the resulting items.

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

def list_evidence_links(
    session_id: str,
    checkpoint_id: int | None = None,
    db_path: Path | None = None,
) -> list[dict[str, Any]]:
    """List evidence rows for a session and optional checkpoint.

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

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        query = select(EvidenceLink).where(EvidenceLink.session_id == session_id)
        if checkpoint_id is not None:
            query = query.where(EvidenceLink.checkpoint_id == checkpoint_id)
        rows = session.exec(query.order_by(EvidenceLink.id.asc())).all()  # type: ignore[union-attr]

    return [
        {
            "id": row.id,
            "session_id": row.session_id,
            "checkpoint_id": row.checkpoint_id,
            "source_type": row.source_type,
            "source_id": row.source_id,
            "relevance_score": row.relevance_score,
            "rationale": row.rationale,
            "payload_json": row.payload_json,
            "created_at": row.created_at,
        }
        for row in rows
    ]

list_model_state_snapshots

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

List snapshots for a session newest first for diff/rollback flows.

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
list[dict[str, Any]]	list[dict[str, Any]]: A list containing the resulting items.

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

def list_model_state_snapshots(
    session_id: str,
    db_path: Path | None = None,
) -> list[dict[str, Any]]:
    """List snapshots for a session newest first for diff/rollback flows.

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

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        rows = session.exec(
            select(ModelStateSnapshot)
            .where(ModelStateSnapshot.session_id == session_id)
            .order_by(ModelStateSnapshot.id.desc())  # type: ignore[union-attr]
        ).all()

    return [
        {
            "id": row.id,
            "session_id": row.session_id,
            "checkpoint_id": row.checkpoint_id,
            "model_path": row.model_path,
            "feature_tree_json": row.feature_tree_json,
            "mass_properties_json": row.mass_properties_json,
            "screenshot_path": row.screenshot_path,
            "state_fingerprint": row.state_fingerprint,
            "created_at": row.created_at,
        }
        for row in rows
    ]

list_plan_checkpoints

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

List all checkpoints for a session.

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
list[dict[str, Any]]	list[dict[str, Any]]: A list containing the resulting items.

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

def list_plan_checkpoints(
    session_id: str,
    db_path: Path | None = None,
) -> list[dict[str, Any]]:
    """List all checkpoints for a session.

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

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        rows = session.exec(
            select(PlanCheckpoint)
            .where(PlanCheckpoint.session_id == session_id)
            .order_by(PlanCheckpoint.checkpoint_index.asc())  # type: ignore[union-attr]
        ).all()

    return [
        {
            "id": row.id,
            "session_id": row.session_id,
            "checkpoint_index": row.checkpoint_index,
            "title": row.title,
            "planned_action_json": row.planned_action_json,
            "approved_by_user": row.approved_by_user,
            "executed": row.executed,
            "result_json": row.result_json,
            "rollback_snapshot_id": row.rollback_snapshot_id,
            "created_at": row.created_at,
            "updated_at": row.updated_at,
        }
        for row in rows
    ]

list_sketch_graph_snapshots

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

List sketch graph snapshots for a session.

Parameters:

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

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: A list containing the resulting items.

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

def list_sketch_graph_snapshots(
    session_id: str,
    model_path: str | None = None,
    db_path: Path | None = None,
) -> list[dict[str, Any]]:
    """List sketch graph snapshots for a session.

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

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        query = select(SketchGraphSnapshot).where(
            SketchGraphSnapshot.session_id == session_id
        )
        if model_path is not None:
            query = query.where(SketchGraphSnapshot.model_path == model_path)
        rows = session.exec(query.order_by(SketchGraphSnapshot.id.desc())).all()  # type: ignore[union-attr]

    return [
        {
            "id": row.id,
            "session_id": row.session_id,
            "model_path": row.model_path,
            "graph_format": row.graph_format,
            "nodes_json": row.nodes_json,
            "edges_json": row.edges_json,
            "metadata_json": row.metadata_json,
            "created_at": row.created_at,
        }
        for row in rows
    ]

list_tool_call_records

list_tool_call_records(session_id: str, checkpoint_id: int | None = None, db_path: Path | None = None) -> list[dict[str, Any]]

List tool call records for a session and optional checkpoint.

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
db_path	Path | None	The db path value. Defaults to None.	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: A list containing the resulting items.

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

def list_tool_call_records(
    session_id: str,
    checkpoint_id: int | None = None,
    db_path: Path | None = None,
) -> list[dict[str, Any]]:
    """List tool call records for a session and optional checkpoint.

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

    Returns:
        list[dict[str, Any]]: A list containing the resulting items.
    """
    resolved = init_db(db_path)
    engine = _build_engine(resolved)
    with Session(engine) as session:
        query = select(ToolCallRecord).where(ToolCallRecord.session_id == session_id)
        if checkpoint_id is not None:
            query = query.where(ToolCallRecord.checkpoint_id == checkpoint_id)
        rows = session.exec(query.order_by(ToolCallRecord.id.asc())).all()  # type: ignore[union-attr]

    return [
        {
            "id": row.id,
            "session_id": row.session_id,
            "checkpoint_id": row.checkpoint_id,
            "run_id": row.run_id,
            "tool_name": row.tool_name,
            "input_json": row.input_json,
            "output_json": row.output_json,
            "success": row.success,
            "latency_ms": row.latency_ms,
            "created_at": row.created_at,
        }
        for row in rows
    ]

pretty_json

pretty_json(model: BaseModel) -> str

Return pretty JSON for test output snapshots.

Parameters:

Name	Type	Description	Default
model	BaseModel	The model value.	required

Returns:

Name	Type	Description
str	str	The resulting text value.

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

def pretty_json(model: BaseModel) -> str:
    """Return pretty JSON for test output snapshots.

    Args:
        model (BaseModel): The model value.

    Returns:
        str: The resulting text value.
    """
    return json.dumps(model.model_dump(mode="json"), indent=2)

run_validated_prompt async

run_validated_prompt(*, agent_file_name: str, model_name: str, user_prompt: str, result_type: type[TModel], max_retries_on_recoverable: int = 1, db_path: Path | None = None) -> TModel | RecoverableFailure

Run one prompt through PydanticAI and validate the output schema.

Parameters:

Name	Type	Description	Default
agent_file_name	str	The agent file name value.	required
model_name	str	Embedding model name to use.	required
user_prompt	str	The user prompt value.	required
result_type	type[TModel]	The result type value.	required
max_retries_on_recoverable	int	The max retries on recoverable value. Defaults to 1.	1
db_path	Path | None	The db path value. Defaults to None.	None

Returns:

Type	Description
TModel | RecoverableFailure	TModel | RecoverableFailure: The result produced by the operation.

Raises:

Type	Description
RuntimeError	Pydantic_ai is not importable. Install dependencies and retry.

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

async def run_validated_prompt(
    *,
    agent_file_name: str,
    model_name: str,
    user_prompt: str,
    result_type: type[TModel],
    max_retries_on_recoverable: int = 1,
    db_path: Path | None = None,
) -> TModel | RecoverableFailure:
    """Run one prompt through PydanticAI and validate the output schema.

    Args:
        agent_file_name (str): The agent file name value.
        model_name (str): Embedding model name to use.
        user_prompt (str): The user prompt value.
        result_type (type[TModel]): The result type value.
        max_retries_on_recoverable (int): The max retries on recoverable value. Defaults to
                                          1.
        db_path (Path | None): The db path value. Defaults to None.

    Returns:
        TModel | RecoverableFailure: The result produced by the operation.

    Raises:
        RuntimeError: Pydantic_ai is not importable. Install dependencies and retry.
    """
    if Agent is None:  # pragma: no cover
        raise RuntimeError(
            "pydantic_ai is not importable. Install dependencies and retry."
        ) from IMPORT_ERROR

    base_run_id = str(uuid.uuid4())
    instructions = _load_agent_prompt(agent_file_name)
    system_prompt = (
        f"{instructions}\n\n"
        "If you cannot safely produce the requested structured output, return "
        "a RecoverableFailure with concrete remediation steps and a focused retry hint."
    )

    attempt = 0
    current_prompt = user_prompt

    while True:
        attempt += 1
        run_id = f"{base_run_id}-a{attempt}"

        prompt_snapshot = current_prompt

        agent = Agent(
            model_name,
            system_prompt=system_prompt,
            output_type=[result_type, RecoverableFailure],
        )

        try:
            result = await agent.run(current_prompt)
            payload = _extract_data(result)

            if isinstance(payload, RecoverableFailure):
                insert_run(
                    run_id=run_id,
                    agent_name=agent_file_name,
                    prompt=prompt_snapshot,
                    status="recoverable_failure",
                    output_json=payload.model_dump_json(indent=2),
                    model_name=model_name,
                    db_path=db_path,
                )
                insert_error(
                    ErrorRecord(
                        source="pydantic_ai",
                        tool_name="run_validated_prompt",
                        error_type="RecoverableFailure",
                        error_message=payload.explanation,
                        root_cause=payload.explanation,
                        remediation="; ".join(payload.remediation_steps)
                        if payload.remediation_steps
                        else "Follow retry_focus guidance and retry with narrower scope.",
                    ),
                    run_id=run_id,
                    db_path=db_path,
                )
                if payload.should_retry and attempt <= max_retries_on_recoverable:
                    retry_hint = (
                        payload.retry_focus or "Narrow scope and clarify assumptions."
                    )
                    remediation = "\n".join(
                        f"- {step}" for step in payload.remediation_steps
                    )
                    current_prompt = (
                        f"{user_prompt}\n\n"
                        "Retry context from previous recoverable failure:\n"
                        f"Explanation: {payload.explanation}\n"
                        f"Retry focus: {retry_hint}\n"
                        f"Remediation steps:\n{remediation}"
                    )
                    continue
                return payload

            validated = (
                payload
                if isinstance(payload, result_type)
                else result_type.model_validate(payload)
            )

            insert_run(
                run_id=run_id,
                agent_name=agent_file_name,
                prompt=prompt_snapshot,
                status="success",
                output_json=validated.model_dump_json(indent=2),
                model_name=model_name,
                db_path=db_path,
            )
            return validated
        except Exception as exc:
            insert_run(
                run_id=run_id,
                agent_name=agent_file_name,
                prompt=current_prompt,
                status="error",
                output_json=None,
                model_name=model_name,
                db_path=db_path,
            )
            insert_error(
                ErrorRecord(
                    source="pydantic_ai",
                    tool_name="run_validated_prompt",
                    error_type=exc.__class__.__name__,
                    error_message=str(exc),
                    root_cause="Agent output failed schema validation or model invocation",
                    remediation=(
                        "Review response schema, verify model provider credentials, and re-run with a narrower prompt."
                    ),
                ),
                run_id=run_id,
                db_path=db_path,
            )
            raise

update_plan_checkpoint

update_plan_checkpoint(checkpoint_id: int, *, approved_by_user: bool | None = None, executed: bool | None = None, result_json: str | None = None, rollback_snapshot_id: int | None = None, db_path: Path | None = None) -> None

Patch checkpoint approval/execution fields.

Parameters:

Name	Type	Description	Default
checkpoint_id	int	The checkpoint id value.	required
approved_by_user	bool | None	The approved by user value. Defaults to None.	None
executed	bool | None	The executed value. Defaults to None.	None
result_json	str | None	The result json value. Defaults to None.	None
rollback_snapshot_id	int | None	The rollback snapshot id 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 update_plan_checkpoint(
    checkpoint_id: int,
    *,
    approved_by_user: bool | None = None,
    executed: bool | None = None,
    result_json: str | None = None,
    rollback_snapshot_id: int | None = None,
    db_path: Path | None = None,
) -> None:
    """Patch checkpoint approval/execution fields.

    Args:
        checkpoint_id (int): The checkpoint id value.
        approved_by_user (bool | None): The approved by user value. Defaults to None.
        executed (bool | None): The executed value. Defaults to None.
        result_json (str | None): The result json value. Defaults to None.
        rollback_snapshot_id (int | None): The rollback snapshot id 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:
        row = session.exec(
            select(PlanCheckpoint).where(PlanCheckpoint.id == checkpoint_id)
        ).first()
        if row is None:
            return
        if approved_by_user is not None:
            row.approved_by_user = approved_by_user
        if executed is not None:
            row.executed = executed
        if result_json is not None:
            row.result_json = result_json
        if rollback_snapshot_id is not None:
            row.rollback_snapshot_id = rollback_snapshot_id
        row.updated_at = _utc_now_iso()
        session.add(row)
        session.commit()

upsert_design_session

upsert_design_session(*, session_id: str, user_goal: str, source_mode: str = 'model', accepted_family: str | None = None, status: str = 'active', current_checkpoint_index: int = 0, metadata_json: str | None = None, db_path: Path | None = None) -> None

Create or update one interactive design session row.

Parameters:

Name	Type	Description	Default
session_id	str	The session id value.	required
user_goal	str	The user goal value.	required
source_mode	str	The source mode value. Defaults to "model".	'model'
accepted_family	str | None	The accepted family value. Defaults to None.	None
status	str	The status value. Defaults to "active".	'active'
current_checkpoint_index	int	The current checkpoint index value. Defaults to 0.	0
metadata_json	str | None	The metadata 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 upsert_design_session(
    *,
    session_id: str,
    user_goal: str,
    source_mode: str = "model",
    accepted_family: str | None = None,
    status: str = "active",
    current_checkpoint_index: int = 0,
    metadata_json: str | None = None,
    db_path: Path | None = None,
) -> None:
    """Create or update one interactive design session row.

    Args:
        session_id (str): The session id value.
        user_goal (str): The user goal value.
        source_mode (str): The source mode value. Defaults to "model".
        accepted_family (str | None): The accepted family value. Defaults to None.
        status (str): The status value. Defaults to "active".
        current_checkpoint_index (int): The current checkpoint index value. Defaults to 0.
        metadata_json (str | None): The metadata 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)
    now = _utc_now_iso()
    with Session(engine) as session:
        row = session.exec(
            select(DesignSession).where(DesignSession.session_id == session_id)
        ).first()
        if row is None:
            session.add(
                DesignSession(
                    session_id=session_id,
                    user_goal=user_goal,
                    source_mode=source_mode,
                    accepted_family=accepted_family,
                    status=status,
                    current_checkpoint_index=current_checkpoint_index,
                    metadata_json=metadata_json,
                    created_at=now,
                    updated_at=now,
                )
            )
        else:
            row.user_goal = user_goal
            row.source_mode = source_mode
            row.accepted_family = accepted_family
            row.status = status
            row.current_checkpoint_index = current_checkpoint_index
            row.metadata_json = metadata_json
            row.updated_at = now
            session.add(row)
        session.commit()