Python Guide

from typing import Optional
import grpc
import geyser_pb2
import geyser_pb2_grpc
import time
import queue

# Authentication Plugin
class TokenAuthPlugin(grpc.AuthMetadataPlugin):
    def __init__(self, token: str):
        self.token = token

    def __call__(self, context, callback):
        metadata = (("x-token", self.token),)
        callback(metadata, None)

# Setup connection
def setup_connection(endpoint, token):
    auth_plugin = TokenAuthPlugin(token)
    call_credentials = grpc.metadata_call_credentials(auth_plugin)
    ssl_credentials = grpc.ssl_channel_credentials()
    combined_credentials = grpc.composite_channel_credentials(
        ssl_credentials, call_credentials
    )
    return combined_credentials

# Subscribe to updates
def subscribe_to_slots(endpoint, token):
    credentials = setup_connection(endpoint, token)
    
    with grpc.secure_channel(endpoint, credentials=credentials) as channel:
        stub = geyser_pb2_grpc.GeyserStub(channel)
        
        # Create subscription request for slots
        subscribe_request = geyser_pb2.SubscribeRequest()
        slot_filter = geyser_pb2.SubscribeRequestFilterSlots()
        subscribe_request.slots["slot_updates"].CopyFrom(slot_filter)
        
        # Create request iterator
        request_queue = queue.Queue()
        request_queue.put(subscribe_request)
        
        def request_iterator():
            while True:
                try:
                    request = request_queue.get(timeout=30)
                    if request is None:
                        break
                    yield request
                except queue.Empty:
                    # Send ping to keep connection alive
                    ping_request = geyser_pb2.SubscribeRequest()
                    ping_request.ping.id = int(time.time())
                    yield ping_request
        
        # Subscribe and process updates
        response_stream = stub.Subscribe(request_iterator())
        
        print("✅ Connected! Listening for slot updates...")
        
        try:
            for update in response_stream:
                if update.slot:
                    print(f"🎰 Slot: {update.slot.slot}, Status: {update.slot.status}")
        except KeyboardInterrupt:
            print("\n🛑 Stopping subscription...")
            request_queue.put(None)

# Usage
if __name__ == "__main__":
    endpoint = "yellowstone.eu.fluxrpc.com"  # or yellowstone.us.fluxrpc.com
    token = "<your-API-key>"
    subscribe_to_slots(endpoint, setup_connection(endpoint, token))