`Backend`#

class ansys.fluent.mcp.common.backend.Backend#

Bases: abc.ABC

Common interface for all backends.

Concrete subclasses override only the methods their product supports. Default implementations raise BackendUnavailableError.

Overview#

Abstract methods

`connect`	Connect to the configured backend or service.
`is_connected`	Return whether connected.

Methods

`disconnect`	Override if the backend holds a persistent resource.
`status`	Return backend status information.
`list_named_objects`	List named objects entries.
`get_named_object_names`	Return the instance names for a single named-object collection.
`find_named_object`	Resolve a symbolic name across every named-object collection.
`get_state`	Return the state.
`get_active_status`	Return the active status.
`get_allowed_values`	Return the allowed values.
`get_node_attrs`	Batched per-node settings-attribute fetch.
`get_node_attrs_bulk`	Recursively fetch attributes for all children of `parent_path`.
`probe_path`	Cheap pre-flight probe for a batch of settings paths.
`get_command_arguments`	Return the keyword-argument signature of a command path.
`get_help`	Return the help.
`solver_status`	Return solver status information from the backend.
`describe_named_object_template`	Describe the field shape of a fresh child of a named-object collection.
`list_fields`	Enumerate solver field/variable names available for reports/post.
`get_targeted_context`	Return the targeted context.
`mesh_adjacency_probe`	Return `{cellzone -> [adjacent_face_zone_names]}`.
`find_api`	Retrieve candidate Fluent settings APIs for `query`.
`run_code`	Execute Python code through the backend runtime.
`validate_code`	Default validation: AST parse + forbidden-call scan.
`mesh_counts`	Return live mesh element totals.
`mesh_quality`	Return live mesh-quality summary statistics.
`mesh_check`	Return Fluent’s full `mesh.check` report as structured data.
`activate_component`	Start or resume the managed Fluids One component.
`deactivate_component`	Stop the managed Fluids One component.
`update_component`	Update the managed Fluids One component.
`refresh_component`	Refresh the managed Fluids One component.
`screenshot`	Capture a screenshot from the backend runtime.
`invalidate_mesh_cache`	Drop cached mesh-probe results.
`maybe_invalidate_mesh_cache`	Drop the mesh cache when `code` matches a mutation marker.
`invalidate_cache`	Clear cached backend state.
`invalidate_live_caches`	Drop caches that depend on solver state.

Attributes

`kind`
`label`
`MESH_MUTATION_MARKERS`

Import detail#

from ansys.fluent.mcp.common.backend import Backend

Attribute detail#

Backend.kind: str = 'unknown'#

Backend.label: str = 'Unknown backend'#

Backend.MESH_MUTATION_MARKERS: tuple[str, Ellipsis] = ('file.read_case', 'file.read_mesh', 'file.read_case_data', 'file.replace_mesh', 'mesh.replace',...#

Method detail#

abstractmethod Backend.connect(**kwargs: Any) → ansys.fluent.mcp.common.models.ConnectResult#

Async:

Connect to the configured backend or service.

Parameters:

kwargsAny: Keyword arguments forwarded to the callable.

Returns:

ConnectResult: ConnectResult produced by the operation.

async Backend.disconnect() → None#

Override if the backend holds a persistent resource.

Returns:

None: The function completes through its side effects.

abstractmethod Backend.is_connected() → bool#

Return whether connected.

Returns:

bool: Boolean result of the operation.

Backend.status(leaf: str) → ansys.fluent.mcp.common.models.SessionStatus#

Return backend status information.

Parameters:

leafstr: Leaf MCP server instance under test.

Returns:

SessionStatus: SessionStatus produced by the operation.

async Backend.list_named_objects() → dict[str, Any]#

List named objects entries.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.get_named_object_names(collection_path: str) → list[str]#

Return the instance names for a single named-object collection.

Default implementation calls list_named_objects() and filters. Backends with a direct get_object_names() accessor (e.g. PyFluent) should override this for a cheaper single-node round-trip.

Parameters:

collection_pathstr: Path to the named-object collection.

Returns:

list[str]: Collection containing the operation results.

async Backend.find_named_object(name: str) → list[dict[str, Any]]#

Resolve a symbolic name across every named-object collection.

Accepts:

Literal identifiers ("inlet-1") — exact and substring matches are returned, exact first.
Glob patterns ("wall-*", "*bar*|*tabzone*") — every live name matching the pattern is returned with "exact": True (the pattern itself stands in for an exact intent against the matched key) and an "is_pattern": True flag on each entry plus a top-level pattern_source payload so callers know the source query was a pattern.

Default implementation calls list_named_objects() and filters; backends with a faster index may override.

Parameters:

namestr: Name of the object, module, or setting being processed.

Returns:

list[dict[str, Any]]: Mapping containing the operation result.

async Backend.get_state(paths: list[str] | None = None) → dict[str, Any]#

Return the state.

Parameters:

pathslist[str] | None: Fluent object paths supplied to the operation.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.get_active_status(paths: list[str]) → dict[str, bool]#

Return the active status.

Parameters:

pathslist[str]: Fluent object paths supplied to the operation.

Returns:

dict[str, bool]: Mapping containing the operation result.

async Backend.get_allowed_values(paths: list[str]) → dict[str, list[Any]]#

Return the allowed values.

Parameters:

pathslist[str]: Fluent object paths supplied to the operation.

Returns:

dict[str, list[Any]]: Mapping containing the operation result.

async Backend.get_node_attrs(paths: list[str], attrs: list[str]) → dict[str, dict[str, Any]]#

Batched per-node settings-attribute fetch.

Returns {path: {attr: value}}. Backends with a live solver session should override; the default raises BackendUnavailableError so callers can fall back to the per-attr accessors (get_active_status, get_allowed_values).

Parameters:

pathslist[str]: Fluent object paths supplied to the operation.
attrslist[str]: Attribute names requested from the backend object.

Returns:

dict[str, dict[str, Any]]: Mapping containing the operation result.

async Backend.get_node_attrs_bulk(parent_path: str, attrs: list[str]) → dict[str, dict[str, Any]]#

Recursively fetch attributes for all children of parent_path.

Calls node.get_attrs(attrs, recursive=True) in a single Scheme RPC, replacing N per-field round-trips when the caller needs metadata for a whole subtree (e.g. validating every field in a set_named value dict).

Returns {relative_child_path: {attr: value}}. Attr spellings follow the Scheme convention ("min", "max", "units-quantity", "allowed-values", "active?").

The default returns {} so callers can fall back gracefully to per-path get_node_attrs calls.

Parameters:

parent_pathstr: Parent Fluent object path used for bulk lookup.
attrslist[str]: Attribute names requested from the backend object.

Returns:

dict[str, dict[str, Any]]: Mapping containing the operation result.

async Backend.probe_path(paths: list[str]) → dict[str, dict[str, Any]]#

Cheap pre-flight probe for a batch of settings paths.

Returns {path: {exists, is_active, is_user_creatable, kind}} in a single batched RPC. Live backends should override; the default raises BackendUnavailableError.

Parameters:

pathslist[str]: Fluent object paths supplied to the operation.

Returns:

dict[str, dict[str, Any]]: Mapping containing the operation result.

async Backend.get_command_arguments(path: str) → dict[str, Any] | None#

Return the keyword-argument signature of a command path.

Returns None if the backend cannot introspect commands or the path is not a command. The result shape is:

{"argument_names": ["type", "name"], "arguments": {"type": {...}, ...}}

Backends with a live solver session should override.

Parameters:

pathstr: Fluent object path or file-system path to inspect.

Returns:

dict[str, Any] | None: Mapping containing the operation result.

async Backend.get_help(path: str) → dict[str, Any]#

Return the help.

Parameters:

pathstr: Filesystem path or API path to process.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.solver_status() → dict[str, Any]#

Return solver status information from the backend.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.describe_named_object_template(path: str) → dict[str, Any] | None#

Describe the field shape of a fresh child of a named-object collection.

Returns None when the backend cannot introspect templates. Live backends should override.

Parameters:

pathstr: Fluent object path or file-system path to inspect.

Returns:

dict[str, Any] | None: Mapping containing the operation result.

async Backend.list_fields(*, scope: str = 'any') → dict[str, Any] | None#

Enumerate solver field/variable names available for reports/post.

Returns None when unavailable.

Parameters:

scopestr: Scope used to limit the field or API lookup.

Returns:

dict[str, Any] | None: Mapping containing the operation result.

async Backend.get_targeted_context(*, paths_to_check: list[str], named_object_types: list[str] | None = None, instance_state_fetch: list[str] | None = None) → dict[str, Any]#

Return the targeted context.

Parameters:

paths_to_checklist[str]: Fluent object paths to validate or inspect.
named_object_typeslist[str] | None: Named-object families that should be considered during lookup.
instance_state_fetchlist[str] | None: Whether named-object instance state should be fetched.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.mesh_adjacency_probe(cellzones: list[str], *, bc_filter: tuple[str, Ellipsis] | None = None) → dict[str, list[str]]#

Return {cellzone -> [adjacent_face_zone_names]}.

Implementations walk every BC family that exposes adjacent_cell_zone (wall, velocity_inlet, pressure_outlet, mass_flow_inlet, …) and invert the mapping. Coupled-wall shadow_face_zone entries are added to BOTH sides so cellzone↔cellzone neighbour queries (set-intersection on shared face names) find CHT solid–fluid pairs.

bc_filter restricts the walk to the listed BC families (e.g. ("wall",) to answer “walls adjacent to X”, ("velocity_inlet", "pressure_inlet") for inlets, …). None (default) walks every supported family. The shadow traversal only contributes when "wall" is in scope (or the filter is None) — interior families have no shadow concept.

Limitation: interior face zones are not enumerated — they are gated INACTIVE on boundary_conditions.interior[*] and the mesh.adjacency Command’s argument-binding side-channel is broken on Fluent ≥ 27.1. Callers needing interior coverage must supplement with run_code.

Backends without a live solver session must raise BackendUnavailableError.

Parameters:

cellzoneslist[str]: Cell zone names used for the mesh adjacency query.
bc_filtertuple[str, …] | None: Boundary-condition filter used to limit the probe.

Returns:

dict[str, list[str]]: Mapping containing the operation result.

async Backend.find_api(query: str, *, top_k: int = 10, kinds: list[str] | None = None, under: str | None = None) → list[dict[str, Any]]#

Retrieve candidate Fluent settings APIs for query.

Uses the configured lexical ApiRetriever, which performs BM25 ranking over the bundled Fluent API catalog and PyFluent class docstrings.

Returns a list of {path, kind, score, ...} hits. Backends may override to add live cross-checks (e.g. filter out paths whose immediate parent does not exist on the connected solver).

Parameters:

querystr: Search query supplied by the caller.
top_kint: Maximum number of search hits to return.
kindslist[str] | None: Optional API object kinds used to filter search results.
understr | None: Optional root path used to constrain API search results.

Returns:

list[dict[str, Any]]: Mapping containing the operation result.

async Backend.run_code(code: str, *, namespace: dict[str, Any] | None = None, filename: str = '') → ansys.fluent.mcp.common.models.RunCodeResult#

Execute Python code through the backend runtime.

Parameters:

codestr: Python code or command text to execute or validate.
namespacedict[str, Any] | None: Namespace used to resolve the backend object or route.
filenamestr: File name or path used by the backend operation.

Returns:

RunCodeResult: RunCodeResult produced by the operation.

async Backend.validate_code(code: str) → ansys.fluent.mcp.common.models.RunCodeResult#

Default validation: AST parse + forbidden-call scan.

Backends with a live solver session can override to add semantic checks against the connected model.

Parameters:

codestr: Python code snippet to validate or execute.

Returns:

RunCodeResult: Result produced by the function.

async Backend.mesh_counts() → dict[str, int | None]#

Return live mesh element totals.

Output shape: {"cell_count", "face_count", "node_count"} — each value is an int when the underlying solver exposes the count, or None when the count is unavailable (no mesh loaded, partition pending, or the backend has no introspection path to mesh totals). Callers MUST treat None as unknown — never as zero.

The default raises BackendUnavailableError so backends that have no live solver (Fluids One geometry / mesh / post leaves) decline the probe; the smart-defaults wrapper catches this and falls back to an all-None payload. Solve backends override this to query Fluent (e.g. via the Scheme variables tinfo/n-cells etc.).

Returns:

dict[str, int | None]: Mapping containing the operation result.

async Backend.mesh_quality() → dict[str, float | None]#

Return live mesh-quality summary statistics.

Output shape: {"min_orthogonal_quality", "max_ortho_skew", "max_aspect_ratio"} — each value is a float in its native Fluent range (orthogonal quality 0..1 where higher is better; ortho skew 0..1 where lower is better; aspect ratio ≥ 1) or None when the metric is not available (no mesh loaded, mesh.quality failed, or the report wording on the connected Fluent build doesn’t match the parser).

Callers MUST treat None as unknown — never as a passing score. Backends should cache the result (mesh quality is invariant between mutations) and invalidate on mesh-modifying operations (case load, scale, translate, rotate, adapt, repair).

The default raises BackendUnavailableError. Solve backends override this to invoke Fluent’s mesh.quality settings command and parse the captured transcript.

Returns:

dict[str, float | None]: Mapping containing the operation result.

async Backend.mesh_check() → dict[str, Any]#

Return Fluent’s full mesh.check report as structured data.

Output shape (all keys optional — a value is None when the corresponding line was not present in the captured transcript):

{
    "domain_extents": {"x": (min, max), "y": (min, max), "z": (min, max)},
    "volume_min": float | None,
    "volume_max": float | None,
    "volume_total": float | None,
    "face_area_min": float | None,
    "face_area_max": float | None,
    "errors": list[str],          # lines preceded by "Error:"
    "warnings": list[str],        # lines preceded by "Warning:"
    "raw": str,                   # the verbatim transcript chunk
}

mesh.check is the right pre-flight diagnostic — it covers topology (left-handed cells, non-positive volumes, face handedness, periodic / boundary-pair sanity) on top of the bulk statistics. It does NOT print quality numbers in modern Fluent builds (see mesh_quality() for those).

The default raises BackendUnavailableError.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.activate_component() → dict[str, Any]#

Start or resume the managed Fluids One component.

Sends POST /api/session/components//activate. Returns a status dict. Backends that do not talk to Fluids One raise BackendUnavailableError.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.deactivate_component() → dict[str, Any]#

Stop the managed Fluids One component.

Sends POST /api/session/components//deactivate. Returns a status dict. Backends that do not talk to Fluids One raise BackendUnavailableError.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.update_component() → dict[str, Any]#

Update the managed Fluids One component.

Sends POST /api/session/components//update. Returns a status dict. Backends that do not talk to Fluids One raise BackendUnavailableError.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.refresh_component() → dict[str, Any]#

Refresh the managed Fluids One component.

Sends POST /api/session/components//refresh. Returns a status dict. Backends that do not talk to Fluids One raise BackendUnavailableError.

Returns:

dict[str, Any]: Mapping containing the operation result.

async Backend.screenshot(*, view: str | None = None) → dict[str, Any]#

Capture a screenshot from the backend runtime.

Parameters:

viewOptional[str]: Graphics view or camera preset requested by the caller.

Returns:

dict[str, Any]: Mapping containing the operation result.

Backend.invalidate_mesh_cache() → None#

Drop cached mesh-probe results.

Returns:

None: The function completes through its side effects.

Backend.maybe_invalidate_mesh_cache(code: str) → bool#

Drop the mesh cache when code matches a mutation marker.

Parameters:

codestr: run_code snippet to inspect.

Returns:

bool: True when the cache was cleared.

Backend.invalidate_cache() → None#

Clear cached backend state.

Returns:

None: The function completes through its side effects.

Backend.invalidate_live_caches() → None#

Drop caches that depend on solver state.

Called by the framework after every run_code and after connect. The mesh-probe cache is preserved; use invalidate_mesh_cache() or maybe_invalidate_mesh_cache() to drop it explicitly.

Returns:

None: The function completes through its side effects.