PyFluentBackend#

class ansys.fluent.mcp.solve.backends.pyfluent.PyFluentBackend(*, label: str = 'PyFluent (solve)')#

Bases: ansys.fluent.mcp.common.backend.Backend

Persistent PyFluent Solver session.

Overview#

connect

Connect to the configured backend or service.

disconnect

Close resources for the PyFluentBackend object.

is_connected

Return whether connected.

close_sync

Close backend resources from synchronous code.

list_named_objects

Discover every named-object collection.

get_named_object_names

Return instance names for a single named-object collection using get_object_names().

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: {path: {exists, is_active, is_user_creatable, kind}}.

mesh_adjacency_probe

Return {cellzone -> [adjacent_face_zone_names]}.

get_command_arguments

Introspect a Command’s keyword-argument signature via PyFluent.

describe_named_object_template

Describe the field shape of a fresh child of a NamedObject collection.

list_fields

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

get_targeted_context

Return the targeted context.

run_code

Execute Python code through the backend runtime.

get_help

Return {path, doc, kind, child_names, allowed_values}.

mesh_counts

Probe live mesh element totals via Fluent Scheme.

mesh_quality

Return mesh quality information from the backend.

mesh_check

Run the backend mesh check operation.

solver_status

Best-effort summary of where the solver currently is.

validate_code

Validate code.

screenshot

Capture a screenshot from the backend runtime.

Import detail#

from ansys.fluent.mcp.solve.backends.pyfluent import PyFluentBackend

Attribute detail#

PyFluentBackend.kind = 'pyfluent'#
PyFluentBackend.label = 'PyFluent (solve)'#
PyFluentBackend.endpoint: str | None = None#

Method detail#

async PyFluentBackend.connect(*, ip: str | None = None, port: int | None = None, password: str | None = None, server_info_file: str | None = None, precision: str = 'double', processor_count: int = 1, ui_mode: str = 'gui', product_version: str | None = None, dimension: int | str | None = None, mode: str | None = None, gpu: bool | list[int] | None = None, journal_file_names: str | list[str] | None = None, case_file_name: str | None = None, case_data_file_name: str | None = None, cwd: str | None = None, fluent_path: str | None = None, env: dict[str, Any] | None = None, graphics_driver: str | None = None, scheduler_options: dict[str, Any] | None = None, start_timeout: int | None = None, cleanup_on_exit: bool | None = None, additional_arguments: str | None = None, **_: Any) ansys.fluent.mcp.common.models.ConnectResult#

Connect to the configured backend or service.

Parameters:
ipOptional[str]

IP address of the Fluent server to connect to.

portOptional[int]

Port number of the Fluent server to connect to.

passwordOptional[str]

Password used when connecting to a Fluent server.

server_info_fileOptional[str]

Server-info file used to connect to an existing Fluent session.

precisionstr

Solver precision requested for the Fluent session.

processor_countint

Number of processors requested for the Fluent session.

ui_modestr

Fluent UI mode requested for launch.

product_versionOptional[str]

Fluent product version requested for launch.

dimensionOptional[int | str]

Solver dimension requested for the Fluent session.

modeOptional[str]

Execution or launch mode requested by the caller.

gpuOptional[bool | list[int]]

GPU option forwarded to the Fluent launch API.

journal_file_namesOptional[str | list[str]]

Journal files passed to Fluent during startup.

case_file_nameOptional[str]

Case file path passed to Fluent at startup.

case_data_file_nameOptional[str]

Case-data file path passed to Fluent at startup.

cwdOptional[str]

Working directory used when launching Fluent.

fluent_pathOptional[str]

Path for the fluent.

envOptional[dict[str, Any]]

Environment mapping to read instead of the process environment.

graphics_driverOptional[str]

Graphics driver requested for the Fluent session.

scheduler_optionsOptional[dict[str, Any]]

Scheduler options forwarded to the Fluent launch API.

start_timeoutOptional[int]

Timeout in seconds used while starting Fluent.

cleanup_on_exitOptional[bool]

Whether Fluent resources should be cleaned up when the process exits.

additional_argumentsOptional[str]

Additional launch arguments forwarded to Fluent.

_Any

Ignored compatibility options accepted by the backend interface.

Returns:
ConnectResult

ConnectResult produced by the operation.

async PyFluentBackend.disconnect() None#

Close resources for the PyFluentBackend object.

Returns:
None

The function completes through its side effects.

PyFluentBackend.is_connected() bool#

Return whether connected.

Returns:
bool

Boolean result of the operation.

PyFluentBackend.close_sync() None#

Close backend resources from synchronous code.

Returns:
None

The function completes through its side effects.

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

Discover every named-object collection.

Strategy:

  1. Try the Scheme api-get-named-object-names (one call, covers every collection Fluent knows about).

  2. Fall back to a generic depth-limited tree walk.

No hardcoded path list.

Returns:
dict[str, Any]

Mapping containing the operation result.

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

Return instance names for a single named-object collection using get_object_names().

A single cheap Scheme call that does not need to read the full collection state.

Falls back to the full list_named_objects() snapshot if the collection node does not expose get_object_names.

Parameters:
collection_pathstr

Path to the named-object collection.

Returns:
list[str]

Collection containing the operation results.

async PyFluentBackend.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 PyFluentBackend.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 PyFluentBackend.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 PyFluentBackend.get_node_attrs(paths: list[str], attrs: list[str]) dict[str, dict[str, Any]]#

Batched per-node settings-attribute fetch.

For each path resolves the settings node and asks Fluent for ALL requested attrs in one Scheme RPC via PyFluent’s node.get_attrs([...]). Returns {path: {attr: value}}.

attrs use the Scheme spelling ("active?", "read-only?", "user-creatable?", "allowed-values", "min", "max", "default", "units-quantity", "file-purpose"). Attrs the node does not expose are simply omitted from its inner dict — Fluent never raises for missing attrs in a batch request.

Unresolvable paths and nodes that do not implement get_attrs map to {}. Empty paths raises InvalidArgumentsError; empty attrs raises the same.

This is the foundation for enrichment passes such as describe_named_object_template that would otherwise issue one RPC per attribute.

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 PyFluentBackend.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, True), a single Scheme RPC — returning {relative_child_path: {attr: value}} for every descendant that exposes those attributes. This replaces N per-field get_node_attrs round-trips when the caller needs metadata for a whole subtree (such as validating every leaf in a set_named value dict before apply).

Falls back to {} gracefully when the node is unreachable or does not support the recursive form.

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 PyFluentBackend.probe_path(paths: list[str]) dict[str, dict[str, Any]]#

Cheap pre-flight: {path: {exists, is_active, is_user_creatable, kind}}.

Designed to be called once before a batch of mutating writes so the executor can skip steps that would fail anyway (path missing, settings node currently inactive, NamedObject family that does not accept create). Each path costs ONE Scheme RPC via the leaf’s batched get_attrs([active?, user-creatable?]) plus a local class-name lookup.

kind is one of "missing", "command", "query", "named_object", "group", "leaf". Unknown / unusual nodes default to "leaf".

Parameters:
pathslist[str]

Fluent object paths supplied to the operation.

Returns:
dict[str, dict[str, Any]]

Mapping containing the operation result.

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

Return {cellzone -> [adjacent_face_zone_names]}.

Read-only. Implementation walks every BC family that exposes adjacent_cell_zone (wall, velocity_inlet, pressure_outlet, mass_flow_inlet, …) and inverts the mapping. Coupled-wall shadow_face_zone entries are added to BOTH sides so the cellzone-↔-cellzone neighbour query (set-intersection on shared face names) finds CHT solid-fluid pairs.

IMPORTANT — PyFluent Query objects: both adjacent_cell_zone and shadow_face_zone are PyFluent Query objects (not Parameters). They are never included in get_state() output and must be read by calling them: instance.adjacent_cell_zone(). The implementation enumerates instance names via a batched get_state() (cheap), then calls the Query per-instance.

bc_filter (optional) restricts the walk to the listed BC families (e.g. ("wall",) to answer “walls adjacent to X”; ("velocity_inlet","pressure_inlet") for inlets). Unknown families are silently dropped. The coupled-wall shadow pass runs only when "wall" is in scope (or the filter is None).

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 (cellzones.set_state raises attr value not defined). Callers that need interior-face coverage must supplement with run_code.

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 PyFluentBackend.get_command_arguments(path: str) dict[str, Any] | None#

Introspect a Command’s keyword-argument signature via PyFluent.

Parameters:
pathstr

Fluent object path or file-system path to inspect.

Returns:
dict[str, Any] | None

Mapping containing the operation result.

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

Describe the field shape of a fresh child of a NamedObject collection.

Uses PyFluent’s static settings classes (child_object_type) so we never have to create-and-delete a transient object on the live session. Returns a flat {field_name: {type_hint, allowed_values?, child_kind}} map plus the create-command signature when present.

Parameters:
pathstr

Fluent object path or file-system path to inspect.

Returns:
dict[str, Any] | None

Mapping containing the operation result.

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

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

scope is a hint passed to Fluent’s field-info API: "any" (default), "surface", "cell", "node". Returns {"fields": [...], "scope": scope} or 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 PyFluentBackend.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 PyFluentBackend.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 PyFluentBackend.get_help(path: str) dict[str, Any]#

Return {path, doc, kind, child_names, allowed_values}.

Parameters:
pathstr

Fluent object path or file-system path to inspect.

Returns:
dict[str, Any]

Mapping containing the operation result.

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

Probe live mesh element totals via Fluent Scheme.

Returns {"cell_count", "face_count", "node_count"}; each value is int when the Scheme call resolved or None when unavailable (no mesh loaded, partition pending, Scheme call failed). Never raises — surfaces failures as an all- None payload so callers can branch on is None without try/except.

Why Scheme: Fluent does not surface mesh-element totals on the settings tree. The canonical accessor is (inquire-grids) — it returns a list whose LAST element is the global grid summary in the shape ( cells faces nodes ...). cadr/caddr/ cadddr extract the three counts. This matches the pattern Fluent itself uses internally (e.g. in wb-is-data-file-compatible? for cross-file count checks) and is stable across versions back to 19.0. rpgetvar 'tinfo/n-cells does NOT exist as an rpvar.

Returns:
dict[str, int | None]

Mapping containing the operation result.

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

Return mesh quality information from the backend.

Returns:
dict[str, float | None]

Mapping containing the operation result.

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

Run the backend mesh check operation.

Returns:
dict[str, Any]

Mapping containing the operation result.

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

Best-effort summary of where the solver currently is.

Returns:
dict[str, Any]

Mapping containing the operation result.

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

Validate code.

Parameters:
codestr

Python code or command text to execute or validate.

Returns:
RunCodeResult

RunCodeResult produced by the operation.

async PyFluentBackend.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.