solidworks_mcp.adapters.intelligent_router¶

solidworks_mcp.adapters.intelligent_router ¶

Intelligent router for COM/VBA execution with optional response caching.

Attributes¶

OperationCallable `module-attribute` ¶

OperationCallable = Callable[..., Awaitable[AdapterResult[Any]]]

Classes¶

AdapterResult `dataclass` ¶

AdapterResult(status: AdapterResultStatus, data: T | None = None, error: str | None = None, execution_time: float | None = None, metadata: dict[str, Any] | None = None)

Bases: Generic[T]

Result wrapper for adapter operations.

Attributes:

Name	Type	Description
`data`	`T \| None`	The data value.
`error`	`str \| None`	The error value.
`execution_time`	`float \| None`	The execution time value.
`metadata`	`dict[str, Any] \| None`	The metadata value.
`status`	`AdapterResultStatus`	The status value.

Attributes¶

is_error `property` ¶

is_error: bool

Check if operation had an error.

Returns:

Name	Type	Description
`bool`	`bool`	True if error, otherwise False.

is_success `property` ¶

is_success: bool

Check if operation was successful.

Returns:

Name	Type	Description
`bool`	`bool`	True if success, otherwise False.

AdapterResultStatus ¶

Bases: StrEnum

Result status for adapter operations.

Attributes:

Name	Type	Description
`ERROR`	`Any`	The error value.
`SUCCESS`	`Any`	The success value.
`TIMEOUT`	`Any`	The timeout value.
`WARNING`	`Any`	The warning value.

ComplexityAnalyzer ¶

ComplexityAnalyzer(parameter_threshold: int = 12, score_threshold: float = 0.6)

Analyze operation complexity and recommend COM or VBA execution path.

Parameters:

Name	Type	Description	Default
`parameter_threshold`	`int`	The parameter threshold value. Defaults to 12.	`12`
`score_threshold`	`float`	The score threshold value. Defaults to 0.6.	`0.6`

Attributes:

Name	Type	Description
`_parameter_threshold`	`Any`	The parameter threshold value.
`_score_threshold`	`Any`	The score threshold value.

Initialize analyzer state.

Parameters:

Name	Type	Description	Default
`parameter_threshold`	`int`	The parameter threshold value. Defaults to 12.	`12`
`score_threshold`	`float`	The score threshold value. Defaults to 0.6.	`0.6`

Returns:

Name	Type	Description
`None`	`None`	None.

Source code in src/solidworks_mcp/adapters/complexity_analyzer.py

def __init__(
    self,
    parameter_threshold: int = 12,
    score_threshold: float = 0.6,
) -> None:
    """Initialize analyzer state.

    Args:
        parameter_threshold (int): The parameter threshold value. Defaults to 12.
        score_threshold (float): The score threshold value. Defaults to 0.6.

    Returns:
        None: None.
    """
    self._parameter_threshold = max(parameter_threshold, 1)
    self._score_threshold = max(min(score_threshold, 1.0), 0.1)
    self._profiles: dict[str, OperationProfile] = self._default_profiles()
    self._history: dict[str, dict[str, int]] = {}

Functions¶

analyze ¶

analyze(operation: str, payload: object) -> RoutingDecision

Produce a routing recommendation for an operation call.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Callable object executed by the helper.	required
`payload`	`object`	The payload value.	required

Returns:

Name	Type	Description
`RoutingDecision`	`RoutingDecision`	The result produced by the operation.

Source code in src/solidworks_mcp/adapters/complexity_analyzer.py

def analyze(self, operation: str, payload: object) -> RoutingDecision:
    """Produce a routing recommendation for an operation call.

    Args:
        operation (str): Callable object executed by the helper.
        payload (object): The payload value.

    Returns:
        RoutingDecision: The result produced by the operation.
    """
    profile = self._profiles.get(
        operation,
        OperationProfile(name=operation, base_complexity=0.2, vba_preferred=False),
    )
    parameter_count = self._count_parameters(payload)

    parameter_component = min(
        parameter_count / float(self._parameter_threshold), 1.0
    )
    history_component = self._history_bias(operation)
    complexity_score = min(
        1.0,
        (parameter_component * 0.45)
        + (profile.base_complexity * 0.40)
        + (history_component * 0.15),
    )

    prefer_vba = (
        parameter_count > self._parameter_threshold
        or profile.vba_preferred
        or complexity_score >= self._score_threshold
    )

    reason = (
        "parameter threshold exceeded"
        if parameter_count > self._parameter_threshold
        else "profile prefers VBA"
        if profile.vba_preferred
        else "complexity score exceeded threshold"
        if complexity_score >= self._score_threshold
        else "COM path preferred"
    )

    return RoutingDecision(
        operation=operation,
        parameter_count=parameter_count,
        complexity_score=complexity_score,
        prefer_vba=prefer_vba,
        reason=reason,
    )

record_result ¶

record_result(operation: str, route: str, success: bool) -> None

Record operation outcome for future routing influence.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Callable object executed by the helper.	required
`route`	`str`	The route value.	required
`success`	`bool`	The success value.	required

Returns:

Name	Type	Description
`None`	`None`	None.

Source code in src/solidworks_mcp/adapters/complexity_analyzer.py

def record_result(self, operation: str, route: str, success: bool) -> None:
    """Record operation outcome for future routing influence.

    Args:
        operation (str): Callable object executed by the helper.
        route (str): The route value.
        success (bool): The success value.

    Returns:
        None: None.
    """
    history = self._history.setdefault(
        operation,
        {
            "com_success": 0,
            "com_failure": 0,
            "vba_success": 0,
            "vba_failure": 0,
        },
    )
    key = f"{route}_{'success' if success else 'failure'}"
    if key in history:
        history[key] += 1

IntelligentRouter ¶

IntelligentRouter(analyzer: ComplexityAnalyzer, cache: ResponseCache, cacheable_operations: set[str] | None = None)

Route operations between COM and VBA paths using complexity analysis.

Parameters:

Name	Type	Description	Default
`analyzer`	`ComplexityAnalyzer`	The analyzer value.	required
`cache`	`ResponseCache`	The cache value.	required
`cacheable_operations`	`set[str] \| None`	The cacheable operations value. Defaults to None.	`None`

Attributes:

Name	Type	Description
`_analyzer`	`Any`	The analyzer value.
`_cache`	`Any`	The cache value.
`_cacheable_operations`	`Any`	The cacheable operations value.

Initialize router dependencies.

Parameters:

Name	Type	Description	Default
`analyzer`	`ComplexityAnalyzer`	The analyzer value.	required
`cache`	`ResponseCache`	The cache value.	required
`cacheable_operations`	`set[str] \| None`	The cacheable operations value. Defaults to None.	`None`

Returns:

Name	Type	Description
`None`	`None`	None.

Source code in src/solidworks_mcp/adapters/intelligent_router.py

def __init__(
    self,
    analyzer: ComplexityAnalyzer,
    cache: ResponseCache,
    cacheable_operations: set[str] | None = None,
) -> None:
    """Initialize router dependencies.

    Args:
        analyzer (ComplexityAnalyzer): The analyzer value.
        cache (ResponseCache): The cache value.
        cacheable_operations (set[str] | None): The cacheable operations value. Defaults to
                                                None.

    Returns:
        None: None.
    """
    self._analyzer = analyzer
    self._cache = cache
    self._cacheable_operations = cacheable_operations or {
        # Lightweight metadata operations
        "get_model_info",
        "list_features",
        "list_configurations",
        "get_file_properties",
        "get_dimension",
        # Analysis operations — NOT cached: model state changes with
        # every feature-creation tool call and the router has no
        # automatic invalidation. Caching these produced stale
        # volumes mid-build.
        # "get_mass_properties",
        # "calculate_mass_properties",
        "get_material_properties",
        "analyze_geometry",
        "check_interference",
        # Drawing analysis operations
        "analyze_drawing_comprehensive",
        "analyze_drawing_dimensions",
        "analyze_drawing_views",
        "analyze_drawing_annotations",
        "check_drawing_compliance",
        "check_drawing_standards",
        "compare_drawing_versions",
        # Classification and indexing
        "classify_feature_tree",
        "discover_solidworks_docs",
    }

Functions¶

execute `async` ¶

execute(operation: str, payload: object, call_args: tuple[Any, ...], call_kwargs: dict[str, Any], com_operation: OperationCallable, vba_operation: OperationCallable | None, cache_ttl_seconds: int | None = None) -> tuple[AdapterResult[Any], RouteResult]

Provide execute support for the intelligent router.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Callable object executed by the helper.	required
`payload`	`object`	The payload value.	required
`call_args`	`tuple[Any, ...]`	The call args value.	required
`call_kwargs`	`dict[str, Any]`	The call kwargs value.	required
`com_operation`	`OperationCallable`	The com operation value.	required
`vba_operation`	`OperationCallable \| None`	The vba operation value.	required
`cache_ttl_seconds`	`int \| None`	The cache ttl seconds value. Defaults to None.	`None`

Returns:

Type	Description
`tuple[AdapterResult[Any], RouteResult]`	tuple[AdapterResult[Any], RouteResult]: A tuple containing the resulting values.

Source code in src/solidworks_mcp/adapters/intelligent_router.py

async def execute(
    self,
    operation: str,
    payload: object,
    call_args: tuple[Any, ...],
    call_kwargs: dict[str, Any],
    com_operation: OperationCallable,
    vba_operation: OperationCallable | None,
    cache_ttl_seconds: int | None = None,
) -> tuple[AdapterResult[Any], RouteResult]:
    """Provide execute support for the intelligent router.

    Args:
        operation (str): Callable object executed by the helper.
        payload (object): The payload value.
        call_args (tuple[Any, ...]): The call args value.
        call_kwargs (dict[str, Any]): The call kwargs value.
        com_operation (OperationCallable): The com operation value.
        vba_operation (OperationCallable | None): The vba operation value.
        cache_ttl_seconds (int | None): The cache ttl seconds value. Defaults to None.

    Returns:
        tuple[AdapterResult[Any], RouteResult]: A tuple containing the resulting values.
    """
    if operation in self._cacheable_operations and self._cache.enabled:
        key = self._cache.make_key(operation, payload)
        cached = self._cache.get(key)
        if isinstance(cached, AdapterResult):
            return cached, RouteResult(route="cache", used_cache=True)

    decision = self._analyzer.analyze(operation=operation, payload=payload)

    if decision.prefer_vba and vba_operation is not None:
        vba_result = await self._safe_call(vba_operation, call_args, call_kwargs)
        self._analyzer.record_result(
            operation=operation,
            route="vba",
            success=vba_result.is_success,
        )
        if vba_result.is_success:
            self._cache_result(operation, payload, vba_result, cache_ttl_seconds)
            return vba_result, RouteResult(route="vba", used_cache=False)

    com_result = await self._safe_call(com_operation, call_args, call_kwargs)
    self._analyzer.record_result(
        operation=operation,
        route="com",
        success=com_result.is_success,
    )

    if com_result.is_success:
        self._cache_result(operation, payload, com_result, cache_ttl_seconds)
        return com_result, RouteResult(route="com", used_cache=False)

    if not decision.prefer_vba and vba_operation is not None:
        fallback_result = await self._safe_call(
            vba_operation,
            call_args,
            call_kwargs,
        )
        self._analyzer.record_result(
            operation=operation,
            route="vba",
            success=fallback_result.is_success,
        )
        if fallback_result.is_success:
            self._cache_result(
                operation,
                payload,
                fallback_result,
                cache_ttl_seconds,
            )
            return fallback_result, RouteResult(
                route="vba-fallback", used_cache=False
            )

    return com_result, RouteResult(route="com-error", used_cache=False)

ResponseCache ¶

ResponseCache(policy: CachePolicy)

Thread-safe in-memory cache for adapter response objects.

Parameters:

Name	Type	Description	Default
`policy`	`CachePolicy`	The policy value.	required

Attributes:

Name	Type	Description
`_lock`	`Any`	The lock value.
`_policy`	`Any`	The policy value.

Initialize this cache.

Parameters:

Name	Type	Description	Default
`policy`	`CachePolicy`	The policy value.	required

Returns:

Name	Type	Description
`None`	`None`	None.

Source code in src/solidworks_mcp/cache/response_cache.py

def __init__(self, policy: CachePolicy) -> None:
    """Initialize this cache.

    Args:
        policy (CachePolicy): The policy value.

    Returns:
        None: None.
    """
    self._policy = policy
    self._entries: dict[str, _CacheEntry[object]] = {}
    self._lock = RLock()

Attributes¶

enabled `property` ¶

enabled: bool

Return whether caching is enabled.

Returns:

Name	Type	Description
`bool`	`bool`	True if enabled, otherwise False.

Functions¶

get ¶

get(key: str) -> object | None

Fetch a cached value when present and unexpired.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key value.	required

Returns:

Type	Description
`object \| None`	object \| None: The result produced by the operation.

Source code in src/solidworks_mcp/cache/response_cache.py

def get(self, key: str) -> object | None:
    """Fetch a cached value when present and unexpired.

    Args:
        key (str): The key value.

    Returns:
        object | None: The result produced by the operation.
    """
    if not self._policy.enabled:
        return None

    now = time.time()
    with self._lock:
        entry = self._entries.get(key)
        if entry is None:
            return None
        if entry.expires_at <= now:
            self._entries.pop(key, None)
            return None
        return entry.value

make_key ¶

make_key(operation: str, payload: object) -> str

Build a deterministic cache key from an operation payload.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Callable object executed by the helper.	required
`payload`	`object`	The payload value.	required

Returns:

Name	Type	Description
`str`	`str`	The resulting text value.

Source code in src/solidworks_mcp/cache/response_cache.py

def make_key(self, operation: str, payload: object) -> str:
    """Build a deterministic cache key from an operation payload.

    Args:
        operation (str): Callable object executed by the helper.
        payload (object): The payload value.

    Returns:
        str: The resulting text value.
    """
    normalized = self._normalize_payload(payload)
    raw = f"{operation}:{normalized}".encode()
    return hashlib.sha256(raw).hexdigest()

set ¶

set(key: str, value: object, ttl_seconds: int | None = None) -> None

Store a cache value with expiration.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key value.	required
`value`	`object`	The value value.	required
`ttl_seconds`	`int \| None`	The ttl seconds value. Defaults to None.	`None`

Returns:

Name	Type	Description
`None`	`None`	None.

Source code in src/solidworks_mcp/cache/response_cache.py

def set(self, key: str, value: object, ttl_seconds: int | None = None) -> None:
    """Store a cache value with expiration.

    Args:
        key (str): The key value.
        value (object): The value value.
        ttl_seconds (int | None): The ttl seconds value. Defaults to None.

    Returns:
        None: None.
    """
    if not self._policy.enabled:
        return

    ttl = (
        ttl_seconds if ttl_seconds is not None else self._policy.default_ttl_seconds
    )
    now = time.time()
    entry = _CacheEntry(value=value, created_at=now, expires_at=now + max(ttl, 1))

    with self._lock:
        if len(self._entries) >= self._policy.max_entries:
            self._evict_oldest_unlocked()
        self._entries[key] = entry

RouteResult `dataclass` ¶

RouteResult(route: str, used_cache: bool)

Route execution metadata.

Attributes:

Name	Type	Description
`route`	`str`	The route value.
`used_cache`	`bool`	The used cache value.

solidworks_mcp.adapters.intelligent_router¶