solidworks_mcp.agents.soc_exporter

solidworks_mcp.agents.soc_exporter

SolidWorks-as-Code exporter.

Reads ToolCallRecord rows from SQLite for a session and emits a clean, runnable Python script that mirrors the structure of build_u_bracket_artifact.py.

Usage::

from solidworks_mcp.agents.soc_exporter import export_session

export_session("my-session-id", output_path="my_part.py")

CLI::

python -m solidworks_mcp.agents.soc_exporter <session_id> <output_path>

Attributes

_CHECKPOINT_RULE module-attribute

_CHECKPOINT_RULE = '-' * 52

_SCRIPT_FOOTER module-attribute

_SCRIPT_FOOTER = '    finally:\n        await adapter.disconnect()\n\n\nif __name__ == "__main__":\n    asyncio.run(build_part())\n'

_SCRIPT_HEADER module-attribute

_SCRIPT_HEADER = 'from __future__ import annotations\n\nimport asyncio\nfrom pathlib import Path\nfrom typing import Any\n\nfrom solidworks_mcp.adapters import create_adapter\nfrom solidworks_mcp.adapters.base import ExtrusionParameters\nfrom solidworks_mcp.config import load_config\n\n\ndef require(result: Any, label: str) -> Any:\n    if not result.is_success:\n        raise RuntimeError(f"{label} failed: {result.error}")\n    return result\n\n\nasync def build_part() -> None:\n    config = load_config()\n    adapter = await create_adapter(config)\n    await adapter.connect()\n    try:\n'

Classes

_CodeGen

_CodeGen()

Stateful generator that tracks entity variables across one sketch.

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

def __init__(self) -> None:
    self._lines: list[str] = []
    # entity_id (str like "Line_1") → python variable name (str like "line_1")
    self._entity_vars: dict[str, str] = {}
    self._counters: dict[str, int] = {}
    self._in_sketch = False

Methods:

emit_generic

emit_generic(bare_name: str, inp: dict[str, Any]) -> None

Emit adapter.(**inp) for any tool not in _DISPATCH.

This gives every logged tool call valid, runnable Python code even if the exporter has no specialized emitter for it. The adapter method name and its keyword-argument names come directly from the logged input_json, so they match the actual adapter API.

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

def emit_generic(self, bare_name: str, inp: dict[str, Any]) -> None:
    """Emit adapter.<bare_name>(**inp) for any tool not in _DISPATCH.

    This gives every logged tool call valid, runnable Python code even if
    the exporter has no specialized emitter for it.  The adapter method
    name and its keyword-argument names come directly from the logged
    input_json, so they match the actual adapter API.
    """
    if not inp:
        self._blank()
        self._emit(f'require(await adapter.{bare_name}(), "{bare_name}")')
    else:
        kwargs = ", ".join(f"{k}={_r(v)}" for k, v in inp.items() if v is not None)
        self._blank()
        self._emit(f'require(await adapter.{bare_name}({kwargs}), "{bare_name}")')

Functions:

_checkpoint_comment

_checkpoint_comment(cp: dict[str, Any]) -> str

Render a SoCCheckpoint as a parseable comment block.

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

def _checkpoint_comment(cp: dict[str, Any]) -> str:
    """Render a SoCCheckpoint as a parseable comment block."""
    label = cp.get("label", "")
    file_path = cp.get("file_path", "")
    first_id = cp.get("first_record_id")
    last_id = cp.get("last_record_id")
    records_range = (
        f"{first_id}-{last_id}" if first_id and last_id else str(last_id or "")
    )
    lines = [
        f"        # -- checkpoint {_CHECKPOINT_RULE}",
        f"        # label:    {label}",
        f"        # file:     {file_path}",
    ]
    if records_range:
        lines.append(f"        # records:  {records_range}")
    lines.append(f"        # {_CHECKPOINT_RULE}")
    return "\n".join(lines)

_cli

_cli() -> None

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

def _cli() -> None:
    if len(sys.argv) < 3:
        print(
            "Usage: python -m solidworks_mcp.agents.soc_exporter <session_id> <output.py>"
        )
        sys.exit(1)
    session_id = sys.argv[1]
    output_path = Path(sys.argv[2])
    written = export_session(session_id, output_path)
    print(f"Exported {session_id!r} → {written}")

_coord

_coord(inp: dict[str, Any], *keys: str, default: float = 0.0) -> float

Return the first non-None value from inp for the given keys as a float.

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

def _coord(inp: dict[str, Any], *keys: str, default: float = 0.0) -> float:
    """Return the first non-None value from inp for the given keys as a float."""
    for k in keys:
        v = inp.get(k)
        if v is not None:
            return float(v)
    return default

_entity_id_from_output

_entity_id_from_output(output: dict[str, Any]) -> str | None

Extract entity_id from an AdapterResult output dict.

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

def _entity_id_from_output(output: dict[str, Any]) -> str | None:
    """Extract entity_id from an AdapterResult output dict."""
    data = output.get("data")
    if isinstance(data, dict):
        return data.get("entity_id") or data.get("id")
    if isinstance(data, str):
        return data
    return None

_fmt_num

_fmt_num(v: float) -> str

Format a float without trailing zeros.

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

def _fmt_num(v: float) -> str:
    """Format a float without trailing zeros."""
    s = f"{v:.6g}"
    return s

_parse_input

_parse_input(input_json: str | None) -> dict[str, Any]

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

def _parse_input(input_json: str | None) -> dict[str, Any]:
    if not input_json:
        return {}
    try:
        result = json.loads(input_json)
        return result if isinstance(result, dict) else {}
    except (json.JSONDecodeError, TypeError):
        return {}

_parse_output

_parse_output(output_json: str | None) -> dict[str, Any]

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

def _parse_output(output_json: str | None) -> dict[str, Any]:
    if not output_json:
        return {}
    try:
        result = json.loads(output_json)
        return result if isinstance(result, dict) else {}
    except (json.JSONDecodeError, TypeError):
        return {}

_r

_r(v: Any) -> str

Compact repr: omit None values, prefer bare strings for short strs.

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

def _r(v: Any) -> str:
    """Compact repr: omit None values, prefer bare strings for short strs."""
    return repr(v)

export_session

export_session(session_id: str, output_path: str | Path, *, checkpoint_id: int | None = None, db_path: Path | None = None, skip_failed: bool = True) -> Path

Export a session's ToolCallRecords as a runnable Python script.

Parameters:

Name	Type	Description	Default
session_id	str	The session ID to export.	required
output_path	str | Path	Destination .py file path.	required
checkpoint_id	int | None	If set, export only records from this checkpoint.	None
db_path	Path | None	Override default SQLite DB path.	None
skip_failed	bool	Skip failed tool call records (default True).	True

Returns:

Type	Description
Path	Path to the written file.

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

def export_session(
    session_id: str,
    output_path: str | Path,
    *,
    checkpoint_id: int | None = None,
    db_path: Path | None = None,
    skip_failed: bool = True,
) -> Path:
    """Export a session's ToolCallRecords as a runnable Python script.

    Args:
        session_id: The session ID to export.
        output_path: Destination .py file path.
        checkpoint_id: If set, export only records from this checkpoint.
        db_path: Override default SQLite DB path.
        skip_failed: Skip failed tool call records (default True).

    Returns:
        Path to the written file.
    """
    from .history_db import list_soc_checkpoints

    records = list_tool_call_records(
        session_id, checkpoint_id=checkpoint_id, db_path=db_path
    )
    checkpoints = list_soc_checkpoints(session_id, db_path=db_path)
    script = generate_script(
        records,
        session_id=session_id,
        checkpoints=checkpoints,
        skip_failed=skip_failed,
    )
    out = Path(output_path)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(script, encoding="utf-8")
    return out

generate_script

generate_script(records: list[dict[str, Any]], *, session_id: str | None = None, checkpoints: list[dict[str, Any]] | None = None, skip_failed: bool = True) -> str

Generate a Python script from a list of ToolCallRecord dicts.

Parameters:

Name	Type	Description	Default
records	list[dict[str, Any]]	Ordered list of ToolCallRecord dicts (from list_tool_call_records).	required
session_id	str | None	Optional session ID to embed in header comment.	None
checkpoints	list[dict[str, Any]] | None	Optional list of SoCCheckpoint dicts (from list_soc_checkpoints). When provided, checkpoint comment blocks are inserted between the corresponding record boundaries.	None
skip_failed	bool	Skip records where success=False (default True).	True

Returns:

Type	Description
str	Complete Python script as a string.

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

def generate_script(
    records: list[dict[str, Any]],
    *,
    session_id: str | None = None,
    checkpoints: list[dict[str, Any]] | None = None,
    skip_failed: bool = True,
) -> str:
    """Generate a Python script from a list of ToolCallRecord dicts.

    Args:
        records: Ordered list of ToolCallRecord dicts (from list_tool_call_records).
        session_id: Optional session ID to embed in header comment.
        checkpoints: Optional list of SoCCheckpoint dicts (from list_soc_checkpoints).
            When provided, checkpoint comment blocks are inserted between the
            corresponding record boundaries.
        skip_failed: Skip records where success=False (default True).

    Returns:
        Complete Python script as a string.
    """
    # Build a mapping from last_record_id → checkpoint for fast lookup
    cp_after: dict[int, dict[str, Any]] = {}
    if checkpoints:
        for cp in checkpoints:
            lid = cp.get("last_record_id")
            if lid is not None:
                cp_after[int(lid)] = cp

    gen = _CodeGen()

    for rec in records:
        if skip_failed and not rec.get("success", True):
            continue
        tool_name: str = rec.get("tool_name", "")
        inp = _parse_input(rec.get("input_json"))
        out = _parse_output(rec.get("output_json"))
        gen.process(tool_name, inp, out)

        # Emit checkpoint block if this record is the last in a checkpoint
        rec_id = rec.get("id")
        if rec_id is not None and int(rec_id) in cp_after:
            gen._lines.append("")
            gen._lines.append(_checkpoint_comment(cp_after[int(rec_id)]))
            gen._lines.append("")

    header_comment = ""
    if session_id:
        header_comment = f"# session_id: {session_id}\n"

    body = "\n".join(gen.body_lines())
    if not body.strip():
        body = "        pass  # no recorded tool calls"

    return f"{header_comment}{_SCRIPT_HEADER}{body}\n{_SCRIPT_FOOTER}"

list_tool_call_records

list_tool_call_records(session_id: str, checkpoint_id: int | None = None, db_path: Path | None = None, include_reverted: bool = False) -> 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
include_reverted	bool	Include records with status='reverted'. Defaults to False.	False

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,
    include_reverted: bool = False,
) -> 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.
        include_reverted (bool): Include records with status='reverted'. Defaults to False.

    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)
        if not include_reverted:
            query = query.where(
                (ToolCallRecord.status == None) | (ToolCallRecord.status != "reverted")  # noqa: E711
            )
        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,
            "status": row.status,
            "created_at": row.created_at,
        }
        for row in rows
    ]