Python Async gRPC Client Examples#

These snippets build on the Python Async gRPC Client Connect and Solve example. Start cuopt_grpc_server first, and pass the server host and port to Client (not CUOPT_REMOTE_*). Always call delete when finished, and pass variable_names to result() if you want named get_vars().

Log Streaming#

After submit, stream solver log lines until the job completes:

from cuopt.grpc.linear_programming import Client, JobStatus

client = Client("localhost", 5001)
job_id = client.submit(dm, settings)
try:
    client.start_log_stream(
        job_id, callback=lambda line, _done: print(line, flush=True)
    )
    if client.wait(job_id, timeout=120) != JobStatus.COMPLETED:
        raise RuntimeError("job did not complete")

    solution = client.result(job_id, variable_names=["x0", "x1"])
    print(solution.get_termination_reason(), solution.get_primal_objective())
finally:
    try:
        client.join_log_stream(job_id)
    finally:
        client.delete(job_id)

Incumbent Streaming (MIP)#

Register incumbent callbacks the same way as for a local solve: add a GetSolutionCallback (from cuopt.linear_programming.internals) on SolverSettings with set_mip_callback(). For gRPC, pass that settings to submit, then call start_incumbent_stream with the same settings so those callbacks receive incumbents while the job runs.

incumbent_stream_demo.py

 1# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 2# SPDX-License-Identifier: Apache-2.0
 3
 4"""MIP incumbent streaming via the Python async gRPC client.
 5
 6Same ``set_mip_callback`` registration as a local solve, plus
 7``start_incumbent_stream`` so those callbacks fire while the remote job runs.
 8
 9Start the server first::
10
11    cuopt_grpc_server --port 5001 --workers 1
12
13Then::
14
15    python incumbent_stream_demo.py
16"""
17
18from cuopt.grpc.linear_programming import Client, JobStatus
19from cuopt.linear_programming.internals import GetSolutionCallback
20from cuopt.linear_programming.problem import INTEGER, MAXIMIZE, Problem
21from cuopt.linear_programming.solver_settings import SolverSettings
22
23
24class IncumbentPrinter(GetSolutionCallback):
25    def get_solution(self, solution, solution_cost, solution_bound, user_data):
26        print(
27            f"incumbent cost={float(solution_cost[0]):.4f} "
28            f"values={solution.tolist()}",
29            flush=True,
30        )
31
32
33problem = Problem("incumbent_stream_demo")
34x = problem.addVariable(lb=0, ub=10, vtype=INTEGER, name="x")
35y = problem.addVariable(lb=0, ub=10, vtype=INTEGER, name="y")
36problem.addConstraint(x + y <= 10, name="c1")
37problem.addConstraint(x - y >= 0, name="c2")
38problem.setObjective(x + 2 * y, sense=MAXIMIZE)
39
40settings = SolverSettings()
41settings.set_mip_callback(IncumbentPrinter(), None)
42settings.set_parameter("time_limit", 30)
43
44client = Client("localhost", 5001)
45job_id = client.submit(problem, settings)
46try:
47    client.start_incumbent_stream(job_id, settings=settings)
48    if client.wait(job_id, timeout=120) != JobStatus.COMPLETED:
49        raise RuntimeError("job did not complete")
50    client.join_incumbent_stream(job_id)
51    names = [v.getVariableName() for v in problem.getVariables()]
52    solution = client.result(job_id, variable_names=names)
53    print(solution.get_termination_reason(), solution.get_primal_objective())
54finally:
55    client.delete(job_id)

See Also#