solidworks_mcp.agents.harness

solidworks_mcp.agents.harness

PydanticAI harness for validating custom agent prompt responses.

Attributes

AGENTS_DIR module-attribute

AGENTS_DIR = Path('.github') / 'agents'

IMPORT_ERROR module-attribute

IMPORT_ERROR = None

TModel module-attribute

TModel = TypeVar('TModel', bound=BaseModel)

Classes

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.

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.

Functions

_extract_data

_extract_data(result: Any) -> Any

Build internal extract data.

Parameters:

Name	Type	Description	Default
result	Any	The result value.	required

Returns:

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

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

def _extract_data(result: Any) -> Any:
    """Build internal extract data.

    Args:
        result (Any): The result value.

    Returns:
        Any: The result produced by the operation.
    """

    if hasattr(result, "data"):
        return result.data
    if hasattr(result, "output"):
        return result.output
    return result

_load_agent_prompt

_load_agent_prompt(agent_file_name: str) -> str

Build internal agent prompt.

Parameters:

Name	Type	Description	Default
agent_file_name	str	The agent file name value.	required

Returns:

Name	Type	Description
str	str	The resulting text value.

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

def _load_agent_prompt(agent_file_name: str) -> str:
    """Build internal agent prompt.

    Args:
        agent_file_name (str): The agent file name value.

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

    path = AGENTS_DIR / agent_file_name
    raw = path.read_text(encoding="utf-8")

    # Strip YAML frontmatter and keep only instruction body.
    if raw.startswith("---"):
        parts = raw.split("---", 2)
        if len(parts) == 3:
            return parts[2].strip()
    return raw.strip()

insert_error

insert_error(record: ErrorRecord, run_id: str | None = None, db_path: Path | None = None) -> None

Persist an error with normalized root cause and remediation guidance.

Parameters:

Name	Type	Description	Default
record	ErrorRecord	The record value.	required
run_id	str | None	The run 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 insert_error(
    record: ErrorRecord, run_id: str | None = None, db_path: Path | None = None
) -> None:
    """Persist an error with normalized root cause and remediation guidance.

    Args:
        record (ErrorRecord): The record value.
        run_id (str | None): The run 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:
        session.add(
            ErrorCatalog(
                run_id=run_id,
                source=record.source,
                tool_name=record.tool_name,
                error_type=record.error_type,
                error_message=record.error_message,
                root_cause=record.root_cause,
                remediation=record.remediation,
                created_at=_utc_now_iso(),
            )
        )
        session.commit()

insert_run

insert_run(*, run_id: str, agent_name: str, prompt: str, status: str, output_json: str | None, model_name: str | None, db_path: Path | None = None) -> None

Record one prompt run and optionally the validated output payload.

Parameters:

Name	Type	Description	Default
run_id	str	The run id value.	required
agent_name	str	The agent name value.	required
prompt	str	The prompt value.	required
status	str	The status value.	required
output_json	str | None	The output json value.	required
model_name	str | None	Embedding model name to use.	required
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_run(
    *,
    run_id: str,
    agent_name: str,
    prompt: str,
    status: str,
    output_json: str | None,
    model_name: str | None,
    db_path: Path | None = None,
) -> None:
    """Record one prompt run and optionally the validated output payload.

    Args:
        run_id (str): The run id value.
        agent_name (str): The agent name value.
        prompt (str): The prompt value.
        status (str): The status value.
        output_json (str | None): The output json value.
        model_name (str | None): Embedding model name to use.
        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(
            AgentRun(
                run_id=run_id,
                agent_name=agent_name,
                prompt=prompt,
                model_name=model_name,
                status=status,
                output_json=output_json,
                created_at=_utc_now_iso(),
            )
        )
        session.commit()

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