Session setup — connection, health, and application runtime

Python client examples for the Connection, Health, and ApplicationRuntime gRPC services.

See Connection, Health, and ApplicationRuntime for these services’ complete reference material.

Python

import grpc
from ansys.api.fluent.v1 import (
    connection_pb2, connection_pb2_grpc,
    health_pb2, health_pb2_grpc,
    application_runtime_pb2, application_runtime_pb2_grpc,
)

channel = grpc.insecure_channel("<server-address>")
metadata = [("password", "<password>")]
connection_stub = connection_pb2_grpc.ConnectionStub(channel)
health_stub = health_pb2_grpc.HealthStub(channel)
application_runtime_stub = application_runtime_pb2_grpc.ApplicationRuntimeStub(channel)

Note

Calling Connection.Connect is optional for most deployments. Passing metadata=[("password", "...")] on each RPC call is sufficient. Use Connection only when your deployment requires explicit version negotiation or session lifecycle control.

Checking server health

Health.Check returns the serving status; assert SERVING_STATUS_SERVING before sending any other RPCs.

Python

# Check with an empty service name — covers the server as a whole.
health_response = health_stub.Check(
    health_pb2.HealthCheckRequest(service=""),
    metadata=metadata,
)
print(health_response.status)  # -> HealthCheckResponse.SERVING_STATUS_SERVING

# Check by fully-qualified service name.
health_response = health_stub.Check(
    health_pb2.HealthCheckRequest(
        service="ansys.api.fluent.v1.health.Health"
    ),
    metadata=metadata,
)
status_enum = health_pb2.HealthCheckResponse.DESCRIPTOR.fields_by_name["status"].enum_type
print(status_enum.values_by_number[health_response.status].name)
# -> SERVING_STATUS_SERVING

# Repeated calls must all return the same status.
statuses = [
    health_stub.Check(health_pb2.HealthCheckRequest(), metadata=metadata).status
    for _ in range(3)
]
print(len(set(statuses)) == 1)  # -> True

Connecting with version negotiation

Connection.Connect opens a server-streaming call; read the first response to confirm the server accepted the version and password.

Python

stream = connection_stub.Connect(
    connection_pb2.ConnectRequest(
        request_type=connection_pb2.ConnectRequest.REQUEST_TYPE_CONNECT,
        password="your-server-password",
        version="27.1.0",
    ),
    metadata=metadata,
)
print(stream is not None)  # -> True

first_response = next(iter(stream))
stream.cancel()
print(hasattr(first_response, "error_code"))  # -> True
print(first_response.error_code)              # -> CONNECTION_ERROR_NONE

# Connecting without specifying a version.
stream = connection_stub.Connect(
    connection_pb2.ConnectRequest(
        request_type=connection_pb2.ConnectRequest.REQUEST_TYPE_CONNECT,
        password="your-server-password",
    ),
    metadata=metadata,
)
first_response = next(iter(stream))
stream.cancel()
print(first_response.error_code == connection_pb2.CONNECTION_ERROR_UNKNOWN)  # -> True

Reading the product version

GetProductVersion returns the major, minor, and patch numbers of the running Fluent build.

Python

version_response = application_runtime_stub.GetProductVersion(
    application_runtime_pb2.GetProductVersionRequest(),
    metadata=metadata,
)
print(version_response.major)  # -> 27
print(version_response.minor)  # -> 1
print(version_response.patch)  # -> 0  (or higher for a patched build)

Reading build information

GetBuildInfo returns the build timestamp, numeric build ID, VCS revision, and branch of the running binary.

Python

build_info = application_runtime_stub.GetBuildInfo(
    application_runtime_pb2.GetBuildInfoRequest(),
    metadata=metadata,
)
print(len(build_info.build_time) > 0)     # -> True  (e.g. '2025-01-15T10:30:00')
print(build_info.build_id > 0)            # -> True
print(len(build_info.vcs_revision) > 0)   # -> True  (e.g. 'abc123def')
print(len(build_info.vcs_branch) > 0)     # -> True  (e.g. 'main')

Reading process information

GetControllerProcessInfo and GetSolverProcessInfo return hostname, PID, and working directory of the respective Fluent processes.

Python

controller_process_info = application_runtime_stub.GetControllerProcessInfo(
    application_runtime_pb2.GetControllerProcessInfoRequest(),
    metadata=metadata,
)
print(controller_process_info.hostname)           # -> 'compute-node-01'
print(controller_process_info.process_id)         # -> 12345  (an integer PID)
print(controller_process_info.working_directory)  # -> '/scratch/my_project'

solver_process_info = application_runtime_stub.GetSolverProcessInfo(
    application_runtime_pb2.GetSolverProcessInfoRequest(),
    metadata=metadata,
)
print(solver_process_info.process_id > 0)        # -> True
print(len(solver_process_info.hostname) > 0)     # -> True

Reading the application mode

GetMode identifies whether Fluent is running as a meshing session, solver session, or a specialised variant.

Python

app_mode_response = application_runtime_stub.GetMode(
    application_runtime_pb2.GetModeRequest(),
    metadata=metadata,
)
print(application_runtime_pb2.Mode.Name(app_mode_response.mode))
# -> MODE_SOLVER  (or MODE_MESHING, etc.)

Enabling beta features

IsBetaEnabled queries the current state; EnableBeta activates beta features for the session — the change persists until the server restarts.

Python

beta_status_response = application_runtime_stub.IsBetaEnabled(
    application_runtime_pb2.IsBetaEnabledRequest(),
    metadata=metadata,
)
print(isinstance(beta_status_response.is_beta_enabled, bool))  # -> True

application_runtime_stub.EnableBeta(
    application_runtime_pb2.EnableBetaRequest(),
    metadata=metadata,
)

beta_status_response = application_runtime_stub.IsBetaEnabled(
    application_runtime_pb2.IsBetaEnabledRequest(),
    metadata=metadata,
)
print(beta_status_response.is_beta_enabled)  # -> True

Recording a Python journal

StartPythonJournal begins recording all Fluent API calls; StopPythonJournal ends recording and returns the journal as a string (when no file name was given).

Python

# Start an in-memory journal (no file name).
start_response = application_runtime_stub.StartPythonJournal(
    application_runtime_pb2.StartPythonJournalRequest(),
    metadata=metadata,
)

# --- perform simulation work here ---

# Stop and retrieve the recorded journal string.
stop_response = application_runtime_stub.StopPythonJournal(
    application_runtime_pb2.StopPythonJournalRequest(
        journal_id=start_response.journal_id
    ),
    metadata=metadata,
)
print(isinstance(stop_response.journal_str, str))  # -> True

See Connection, Health, and ApplicationRuntime for the complete reference material.