solidworks_mcp.agents.retrieval_index

solidworks_mcp.agents.retrieval_index

Local retrieval index builder for feature-tree audits, tool docs, and failures.

Attributes

DEFAULT_DB_PATH module-attribute

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

Functions

_chunk_text

_chunk_text(text: str, chunk_size: int = 1000, overlap: int = 150) -> list[str]

Split long text into overlapping chunks for simple local retrieval.

Parameters:

Name	Type	Description	Default
text	str	Input text processed by the operation.	required
chunk_size	int	Maximum number of characters to keep in each chunk. Defaults to 1000.	1000
overlap	int	Number of overlapping characters between chunks. Defaults to 150.	150

Returns:

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

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

def _chunk_text(text: str, chunk_size: int = 1000, overlap: int = 150) -> list[str]:
    """Split long text into overlapping chunks for simple local retrieval.

    Args:
        text (str): Input text processed by the operation.
        chunk_size (int): Maximum number of characters to keep in each chunk. Defaults to
                          1000.
        overlap (int): Number of overlapping characters between chunks. Defaults to 150.

    Returns:
        list[str]: A list containing the resulting items.
    """
    normalized = (text or "").strip()
    if not normalized:
        return []

    if len(normalized) <= chunk_size:
        return [normalized]

    chunks: list[str] = []
    start = 0
    while start < len(normalized):
        end = min(start + chunk_size, len(normalized))
        chunks.append(normalized[start:end])
        if end == len(normalized):
            break
        start = max(0, end - overlap)
    return chunks

_read_text

_read_text(path: Path) -> str

Build internal text.

Parameters:

Name	Type	Description	Default
path	Path	Filesystem path for the operation.	required

Returns:

Name	Type	Description
str	str	The resulting text value.

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

def _read_text(path: Path) -> str:
    """Build internal text.

    Args:
        path (Path): Filesystem path for the operation.

    Returns:
        str: The resulting text value.
    """

    try:
        return path.read_text(encoding="utf-8")
    except OSError:
        return ""

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_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
    ]