Overview
Python is a versatile and widely-used programming language, known for its simplicity and strong community support. This document provides a step-by-step process for setting up a Python environment to interact with Solana gRPC (Yellowstone-compatible Geyser gRPC), including project setup, dependency management, and implementing authentication mechanisms.
Solana gRPC is available on Mainnet and Devnet for Scale and Business customers. The example below uses Mainnet. For Devnet, use your Solana Devnet endpoint hostname and token; see Endpoint and Token Configuration.
Authentication Required for Python
To establish a connection with the Solana gRPC server, initialize a secure gRPC channel with your server endpoint on port 443 and an authentication token. Configure compression on the channel to optimize the data delivered and metered.
def create_grpc_channel(endpoint: str, token: str) -> grpc.Channel:
# Remove http:// or https:// from the endpoint if present
endpoint = endpoint.replace('http://', '').replace('https://', '')
# Create authentication credentials
auth_creds = grpc.metadata_call_credentials(
lambda context, callback: callback((("x-token", token),), None)
)
# Create SSL credentials
ssl_creds = grpc.ssl_channel_credentials()
# Combine authentication and SSL credentials
combined_creds = grpc.composite_channel_credentials(ssl_creds, auth_creds)
# Create and return the secure channel
return grpc.secure_channel(
endpoint,
credentials=combined_creds,
compression=grpc.Compression.Gzip,
)
grpcio supports Gzip and Deflate only. zstd is not available in the Python client. See Compression for per-language configuration.
You can call the above function like this:
# Create channel
channel = create_grpc_channel(
"your-endpoint.example.com:443", # Use port 443
"your-auth-token"
)
# Create stub for making RPC calls
stub = geyser_pb2_grpc.GeyserStub(channel)
To learn how to split your Solana gRPC-enabled URL into endpoint and token, refer to this section - Endpoint and Token Configuration
Environment Setup and gRPC Preparation
Step 1: Create and Activate a Python Virtual Environment
Create a virtual environment to isolate dependencies:
python -m venv venv
Activate the environment by the following commands:
For macOS/Linux:
source venv/bin/activate
For Windows:
venv\Scripts\activate.bat
Step 2: Define Dependencies
Create a requirements.txt file to specify the necessary dependencies for gRPC and related functionalities:
echo "click==8.1.7
grpcio==1.63.0
grpcio-tools==1.63.0
protobuf==5.26.1
base58==2.1.1" > requirements.txt
Install the dependencies:
python -m pip install -r requirements.txt
Step 3: Prepare for gRPC Data Type Generation
Organize your project directory to handle .proto files and their generated stubs:
-
Create a directory for
.protofiles:mkdir proto # Add your .proto files here
You will need to save geyser.proto and solana-storage.proto in this directory. Take them from the yellowstone-grpc-proto definitions at tag v16.1.4+solana.4.3.2 or later. These define the Geyser interfaces that we will need to interact with Solana gRPC.
Version 1 transactions carry their compute and fee settings in Message.config, which proto v12.6.0 added. Stubs generated from older .proto files have no config field, so a version 1 message parses without error and those settings are silently dropped. Bank IDs and block footers need proto v13.0.0 or later, and block footer certificates need v14.0.1 or later.
-
Create a directory for the generated Python stubs:
mkdir generated
Step 4: Generate Python Stubs Using grpc_tools.protoc
Run the following command to generate Python files from the .proto definitions:
python -m grpc_tools.protoc \
-I./proto/ \
--python_out=./generated \
--pyi_out=./generated \
--grpc_python_out=./generated \
./proto/*
Verify that the generated files are available in the generated directory. Typical file names include:
geyser_pb2.pygeyser_pb2_grpc.pysolana_storage_pb2.pysolana_storage_pb2_grpc.py
Step 5: Enable Module Imports
To ensure Python treats the generated directory as a module, create an __init__.py file:
touch generated/__init__.py
You can use our reference file unless you wish to make your own changes.
Sample Python Application for Solana gRPC
A sample Python application demonstrating gRPC interaction with Solana gRPC is available in the following GitHub repository: Solana gRPC Python Sample App
This example includes:
- Establishing a connection to Solana gRPC.
- Sending requests and handling responses using gRPC.
- Managing Solana data via Solana gRPC.
Decode Block Footer Certificates
Set include_certificates on the block_footer filter to receive certificates carried by each footer. block_final_cert proves that its own slot is finalized; this can be earlier than the footer's slot. Use the certificate's slot when attributing finalization, not the footer's slot or bank ID. It always carries final_aggregate, and it also carries notar_aggregate when the block finalized on the slow path. skip_reward_cert and notar_reward_cert record the validators that voted skip or notarize on the slot.
Each aggregate holds a BLS signature and a signer_bitmap that records which validators signed. The example below decodes the bitmap and prints how many validators signed each aggregate. It uses the create_grpc_channel function from Authentication Required for Python.
import grpc
import generated.geyser_pb2 as geyser_pb2
import generated.geyser_pb2_grpc as geyser_pb2_grpc
def decode_signer_bitmap(signer_bitmap: bytes) -> list[int]:
# Header: a version byte, then the bit count as a little-endian u16
version = signer_bitmap[0]
bit_count = int.from_bytes(signer_bitmap[1:3], "little")
payload = signer_bitmap[3:]
if version == 0:
# One bit per validator
return [i for i in range(bit_count) if payload[i // 8] >> (i % 8) & 1]
if version == 1:
# Five base-3 digits per byte, where a nonzero digit marks a signer
return [i for i in range(bit_count) if payload[i // 5] // 3 ** (i % 5) % 3]
raise ValueError(f"Unknown signer bitmap version: {version}")
def main():
channel = create_grpc_channel(
"your-endpoint.example.com:443",
"your-auth-token"
)
stub = geyser_pb2_grpc.GeyserStub(channel)
request = geyser_pb2.SubscribeRequest(
block_footer={
"footers": geyser_pb2.SubscribeRequestFilterBlockFooter(include_certificates=True)
},
commitment=geyser_pb2.CommitmentLevel.FINALIZED,
)
for update in stub.Subscribe(iter([request])):
if not update.HasField("block_footer"):
continue
footer = update.block_footer
if not footer.HasField("block_final_cert"):
continue
cert = footer.block_final_cert
for name in ("final_aggregate", "notar_aggregate"):
if cert.HasField(name):
signers = decode_signer_bitmap(getattr(cert, name).signer_bitmap)
print(f"Slot {cert.slot} {name}: {len(signers)} signers")
if __name__ == "__main__":
main()
Additional Notes
- Ensure that your
.protofiles are correctly defined and added to theprotodirectory. - For detailed usage of the generated stubs, refer to the sample application linked above.
- If you encounter any issues, consider updating your dependencies or checking for compatibility with your Python version.