async def async_client_call(
    method: str,
    channel: Channel,
    data: dict[str, Message | pb.Part | bytes | str | dict[str, t.Any]],
    sanity: bool = True,
    timeout: int | None = 90,
    assert_data: pb.Response | t.Callable[[pb.Response], bool] | None = None,
    assert_code: grpc.StatusCode | None = None,
    assert_details: str | None = None,
    assert_trailing_metadata: aio.Metadata | None = None,
    protocol_version: str = LATEST_PROTOCOL_VERSION,
) -> pb.Response | None:
    """
    Invoke a given API method via a client.

    Args:
        method: The method name to call.
        channel: The given aio.Channel to use for invoking the RPC. Channels shouldn't include
                 any client interceptors. as we will handle interceptors' lifecycle separately.
        data: The data to send to the server.
        assert_data: The data to assert against the response.
        timeout: The timeout for the RPC.
        sanity: Whether to perform sanity check on the response.
        assert_code: The code to assert against the response.
        assert_details: The details to assert against the response.

    Returns:
        The response from the server.
    """
    pb, _ = import_generated_stubs(protocol_version)

    res: pb.Response | None = None
    try:
        Call = channel.unary_unary(
            f"/bentoml.grpc.{protocol_version}.BentoService/Call",
            request_serializer=pb.Request.SerializeToString,
            response_deserializer=pb.Response.FromString,
        )
        output: aio.UnaryUnaryCall[pb.Request, pb.Response] = Call(
            pb.Request(api_name=method, **data), timeout=timeout
        )
        res = await t.cast(t.Awaitable[pb.Response], output)
        return_code = await output.code()
        details = await output.details()
        trailing_metadata = await output.trailing_metadata()
        if sanity:
            assert isinstance(res, pb.Response)
        if assert_data:
            if callable(assert_data):
                assert assert_data(res), f"Failed while checking data: {output}"
            else:
                assert res == assert_data, f"Failed while checking data: {output}"
    except aio.AioRpcError as call:
        return_code = call.code()
        details = call.details()
        trailing_metadata = call.trailing_metadata()
    if assert_code is not None:
        assert return_code == assert_code, (
            f"Method '{method}' returns {return_code} while expecting {assert_code}."
        )
    if assert_details is not None:
        assert assert_details == details, (
            f"Details '{assert_details}' is not in '{details}'."
        )
    if assert_trailing_metadata is not None:
        assert trailing_metadata == assert_trailing_metadata, (
            f"Trailing metadata '{trailing_metadata}' while expecting '{assert_trailing_metadata}'."
        )
    return res
Invoke a given API method via a client.

Args:
method: The method name to call.
channel: The given aio.Channel to use for invoking the RPC. Channels shouldn't include
any client interceptors. as we will handle interceptors' lifecycle separately.
data: The data to send to the server.
assert_data: The data to assert against the response.
timeout: The timeout for the RPC.
sanity: Whether to perform sanity check on the response.
assert_code: The code to assert against the response.
assert_details: The details to assert against the response.

Returns:
The response from the server.
src/bentoml/testing/grpc/__init__.py