gRPC client in Python#

Installation#

A ready-made package for Python is not provided. The client code must be generated locally with grpc_tools.protoc, as described in Generating a gRPC client.

python3 -m venv .venv
.venv/bin/pip install grpcio-tools
.venv/bin/python3 -m grpc_tools.protoc -I proto \
  --python_out=gen/python --grpc_python_out=gen/python \
  $(find proto -name "*.proto")

Connection and authorization#

Three parameters are required to connect:

GREENAPI_GRPC_HOST — the address of the GREEN-API gRPC server: grpc.green-api.com:443. The address is specified in the host:port format, without the https:// scheme;
GREENAPI_ID_INSTANCE — the idInstance value from your personal account;
GREENAPI_API_TOKEN_INSTANCE — the apiTokenInstance value from your personal account.

The connection to the gRPC server is established using TLS. The authorization data is passed separately in each RPC call. For details, see Authorization in the gRPC API.

Checking the connection#

To check the connection, you can call the GetStateInstance method, which returns the current state of the instance. If the method completes successfully and returns the INSTANCE_STATE_AUTHORIZED state, the connection to the gRPC server and authorization are configured correctly, and the instance is ready to work. The other instance states and possible errors are described in the Checking the connection section of the gRPC API overview.

import os

import grpc
from greenapi.instance.v1 import instance_pb2, instance_pb2_grpc
from greenapi.type.v1 import enums_pb2

channel = grpc.secure_channel(
    os.environ["GREENAPI_GRPC_HOST"],  # "grpc.green-api.com:443"
    grpc.ssl_channel_credentials(),
)
client = instance_pb2_grpc.InstanceServiceStub(channel)

metadata = (
    ("x-instance-id", os.environ["GREENAPI_ID_INSTANCE"]),
    ("authorization", "Bearer " + os.environ["GREENAPI_API_TOKEN_INSTANCE"]),
)

try:
    response = client.GetStateInstance(instance_pb2.GetStateInstanceRequest(), metadata=metadata)
except grpc.RpcError as err:
    raise SystemExit(f"{err.code().name}: {err.details()}")

print(enums_pb2.InstanceState.Name(response.state_instance))  # INSTANCE_STATE_AUTHORIZED

The first call#

import os

import grpc
from greenapi.message.v1 import message_pb2, message_pb2_grpc

channel = grpc.secure_channel(
    os.environ["GREENAPI_GRPC_HOST"],  # "grpc.green-api.com:443"
    grpc.ssl_channel_credentials(),
)
client = message_pb2_grpc.MessageServiceStub(channel)

metadata = (
    ("x-instance-id", os.environ["GREENAPI_ID_INSTANCE"]),
    ("authorization", "Bearer " + os.environ["GREENAPI_API_TOKEN_INSTANCE"]),
)

response = client.SendMessage(
    message_pb2.SendMessageRequest(
        chat_id="11001234567@c.us",
        message="hello from gRPC",
    ),
    metadata=metadata,
)
print(response.id_message)

A single gRPC connection can be used for all services. The client Stubs of the other services are created from the same channel.

How to upload a file#

The file is transferred using a client stream. The first UploadFileRequest message contains UploadFileMetadata with the file name and content type. The client then sequentially sends messages with chunk, which contain parts of the file.

When the request iterator is finished, gRPC completes sending the client stream (half-close) and waits for a single UploadFileResponse from the server.

The general contract of a streaming upload and the chunk size requirements are described in Streaming calls in the gRPC API.

In Python, the client stream is passed as an iterator. When the iterator is finished, gRPC stops sending messages to the server and waits for a single response.

def upload_file(client, metadata, path, content_type, chunk_size=64 * 1024):
    def requests():
        yield message_pb2.UploadFileRequest(
            metadata=message_pb2.UploadFileMetadata(
                file_name=os.path.basename(path),
                content_type=content_type,
            )
        )
        with open(path, "rb") as f:
            while chunk := f.read(chunk_size):
                yield message_pb2.UploadFileRequest(chunk=chunk)

    response = client.UploadFile(requests(), metadata=metadata)
    return response.url_file

url = upload_file(client, metadata, "/path/to/photo.jpg", "image/jpeg")

How to get a QR code#

The ScanQrCode method uses a server stream: the server sequentially sends responses with QR codes until the authorization attempt is complete. The statuses and the behaviour of the method are described in Getting a QR code over gRPC.

try:
    for resp in client.ScanQrCode(instance_pb2.ScanQrCodeRequest(), metadata=metadata):
        if resp.status == instance_pb2.QR_CODE_STATUS_QR_CODE:
            show_qr("data:image/png;base64," + base64.b64encode(resp.qr_png).decode())
        elif resp.status == instance_pb2.QR_CODE_STATUS_ALREADY_LOGGED:
            raise RuntimeError("the instance is already authorized, Logout is required")
except grpc.RpcError as err:
    # DEADLINE_EXCEEDED — nobody scanned it; FAILED_PRECONDITION — the instance is not ready
    print(err.code(), err.details())

When the server closes the stream, the iterator is exhausted and the for loop ends automatically.

Error handling#

You can get the gRPC status with err.code(), and the error description with err.details():

try:
    response = client.SendMessage(request, metadata=metadata)
except grpc.RpcError as err:
    print(err.code(), err.details())

The possible gRPC statuses and recommendations on retries are described in gRPC API error codes.

Before automatically retrying sending methods, read Resending messages: a repeated call may result in the same message being sent again.