# Connection
Establish a gRPC connection to NextBlock's API using Python and grpcio.
This page shows the connection pattern and authentication flow. Replace the placeholder client creation with the client generated from [`nextblock-proto`](https://github.com/nextblock-ag/nextblock-proto).
## Prerequisites
Install the required dependencies:
```bash
pip install grpcio grpcio-tools
pip install solders solana
pip install asyncio
```
Generate the Python gRPC client from the proto specs:
```bash
# Clone the proto repository
git clone https://github.com/nextblock-ag/nextblock-proto
cd nextblock-proto
# Generate Python gRPC client
python -m grpc_tools.protoc --python_out=. --grpc_python_out=. -I. *.proto
```
## Connection Setup
import asyncio
import grpc
from typing import Optional, Dict, Any
from dataclasses import dataclass
import ssl
# Authentication metadata interceptor
class AuthInterceptor(grpc.aio.ClientInterceptor):
def __init__(self, api_key: str):
self.api_key = api_key
async def intercept_unary_unary(self, continuation, client_call_details, request):
# Add authorization header to all requests
metadata = list(client_call_details.metadata or [])
metadata.append(('authorization', self.api_key))
new_details = client_call_details._replace(metadata=metadata)
return await continuation(new_details, request)
async def intercept_unary_stream(self, continuation, client_call_details, request):
# Add authorization header to streaming requests
metadata = list(client_call_details.metadata or [])
metadata.append(('authorization', self.api_key))
new_details = client_call_details._replace(metadata=metadata)
return await continuation(new_details, request)
# Connection configuration
@dataclass
class NextBlockConfig:
endpoint: str = "frankfurt.nextblock.io:443"
api_key: str = ""
use_tls: bool = True
timeout: float = 30.0
keepalive_time_ms: int = 60000 # 60 seconds
keepalive_timeout_ms: int = 15000 # 15 seconds
@classmethod
def from_env(cls) -> 'NextBlockConfig':
import os
return cls(
endpoint=os.getenv('NEXTBLOCK_ENDPOINT', 'frankfurt.nextblock.io:443'),
api_key=os.getenv('NEXTBLOCK_API_KEY', ''),
use_tls=os.getenv('NEXTBLOCK_USE_TLS', 'true').lower() == 'true',
timeout=float(os.getenv('NEXTBLOCK_TIMEOUT', '30')),
)
# Create secure channel with authentication
async def create_nextblock_channel(config: NextBlockConfig) -> grpc.aio.Channel:
# Configure channel options
options = [
('grpc.keepalive_time_ms', config.keepalive_time_ms),
('grpc.keepalive_timeout_ms', config.keepalive_timeout_ms),
('grpc.keepalive_permit_without_stream', True),
('grpc.http2.max_pings_without_data', 0),
('grpc.http2.min_ping_interval_without_data_ms', 300000), # 5 minutes
]
if config.use_tls:
# Create secure channel with TLS
credentials = grpc.ssl_channel_credentials()
channel = grpc.aio.secure_channel(
config.endpoint,
credentials,
options=options
)
else:
# Create insecure channel
channel = grpc.aio.insecure_channel(
config.endpoint,
options=options
)
return channel
# Create authenticated client
async def create_nextblock_client(config: NextBlockConfig):
# Create the channel
channel = await create_nextblock_channel(config)
# Add authentication interceptor
auth_interceptor = AuthInterceptor(config.api_key)
intercept_channel = grpc.aio.intercept_channel(channel, auth_interceptor)
# Create the client stub
# Import your generated gRPC client here
# from your_generated_proto import api_pb2_grpc
# client = api_pb2_grpc.ApiStub(intercept_channel)
return intercept_channel, None # Replace None with your generated client stub
# Connection manager with health checking
class NextBlockConnectionManager:
def __init__(self, config: NextBlockConfig):
self.config = config
self.channel: Optional[grpc.aio.Channel] = None
self.client = None
self.is_connected = False
async def connect(self) -> bool:
"""Establish connection to NextBlock"""
try:
self.channel, self.client = await create_nextblock_client(self.config)
# Test the connection
await self.health_check()
self.is_connected = True
print(f"Successfully connected to NextBlock at {self.config.endpoint}")
return True
except Exception as e:
print(f"Failed to connect to NextBlock: {e}")
self.is_connected = False
return False
async def health_check(self) -> bool:
"""Check if the connection is healthy"""
if not self.channel:
return False
try:
# Test connection with a simple call
# await self.client.ping(Empty()) # Uncomment with real client
print("Connection health check passed")
return True
except Exception as e:
print(f"Health check failed: {e}")
return False
async def disconnect(self):
"""Close the connection"""
if self.channel:
await self.channel.close()
self.is_connected = False
print("Disconnected from NextBlock")
async def __aenter__(self):
await self.connect()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.disconnect()
## Usage Example
async def main():
# Configure connection
config = NextBlockConfig(
endpoint="frankfurt.nextblock.io:443",
api_key="<your-api-key-here>",
use_tls=True,
timeout=30.0
)
# Method 1: Direct connection
channel, client = await create_nextblock_client(config)
try:
# Use the client for API calls
print("Connected to NextBlock!")
# Your API calls would go here
# response = await client.ping(Empty())
finally:
await channel.close()
# Method 2: Using connection manager (recommended)
async with NextBlockConnectionManager(config) as manager:
if manager.is_connected:
print("Connection manager established connection")
# Use manager.client for API calls
# response = await manager.client.ping(Empty())
else:
print("Failed to establish connection")
# Run the example
if __name__ == "__main__":
asyncio.run(main())
## Advanced Connection Features
import logging
from typing import Callable, Any
import backoff
# Enhanced connection manager with retry logic
class EnhancedConnectionManager(NextBlockConnectionManager):
def __init__(self, config: NextBlockConfig, max_retries: int = 3):
super().__init__(config)
self.max_retries = max_retries
self.retry_count = 0
@backoff.on_exception(
backoff.expo,
(grpc.RpcError, ConnectionError),
max_tries=3,
base=2,
max_value=60
)
async def connect_with_retry(self) -> bool:
"""Connect with exponential backoff retry"""
self.retry_count += 1
logging.info(f"Connection attempt {self.retry_count}")
success = await self.connect()
if not success:
raise ConnectionError(f"Failed to connect on attempt {self.retry_count}")
self.retry_count = 0 # Reset on success
return True
async def call_with_retry(self, func: Callable, *args, **kwargs) -> Any:
"""Execute gRPC call with automatic retry"""
@backoff.on_exception(
backoff.expo,
grpc.RpcError,
max_tries=3,
giveup=lambda e: e.code() == grpc.StatusCode.UNAUTHENTICATED
)
async def _call():
return await func(*args, **kwargs)
return await _call()
# Connection pool for high-throughput applications
class ConnectionPool:
def __init__(self, config: NextBlockConfig, pool_size: int = 5):
self.config = config
self.pool_size = pool_size
self.connections: List[NextBlockConnectionManager] = []
self.current_index = 0
async def initialize(self):
"""Initialize connection pool"""
for i in range(self.pool_size):
manager = NextBlockConnectionManager(self.config)
if await manager.connect():
self.connections.append(manager)
else:
logging.warning(f"Failed to create connection {i+1}/{self.pool_size}")
logging.info(f"Connection pool initialized with {len(self.connections)} connections")
def get_connection(self) -> NextBlockConnectionManager:
"""Get next available connection (round-robin)"""
if not self.connections:
raise RuntimeError("No available connections in pool")
connection = self.connections[self.current_index]
self.current_index = (self.current_index + 1) % len(self.connections)
return connection
async def close_all(self):
"""Close all connections in the pool"""
for connection in self.connections:
await connection.disconnect()
self.connections.clear()
## Environment Configuration
import os
from typing import Dict, Any
# Load configuration from environment variables
def load_config_from_env() -> NextBlockConfig:
return NextBlockConfig(
endpoint=os.getenv('NEXTBLOCK_ENDPOINT', 'frankfurt.nextblock.io:443'),
api_key=os.getenv('NEXTBLOCK_API_KEY', ''),
use_tls=os.getenv('NEXTBLOCK_USE_TLS', 'true').lower() == 'true',
timeout=float(os.getenv('NEXTBLOCK_TIMEOUT', '30.0')),
keepalive_time_ms=int(os.getenv('NEXTBLOCK_KEEPALIVE_TIME_MS', '60000')),
keepalive_timeout_ms=int(os.getenv('NEXTBLOCK_KEEPALIVE_TIMEOUT_MS', '15000')),
)
# Configuration validation
def validate_config(config: NextBlockConfig) -> bool:
if not config.api_key:
raise ValueError("API key is required")
if not config.endpoint:
raise ValueError("Endpoint is required")
if config.timeout <= 0:
raise ValueError("Timeout must be positive")
return True
# Example with configuration validation
async def main_with_validation():
try:
config = load_config_from_env()
validate_config(config)
async with EnhancedConnectionManager(config) as manager:
await manager.connect_with_retry()
print("Successfully connected with retry logic!")
except ValueError as e:
print(f"Configuration error: {e}")
except Exception as e:
print(f"Connection error: {e}")
if __name__ == "__main__":
asyncio.run(main_with_validation())
## Available Endpoints
* **Frankfurt**: `frankfurt.nextblock.io` (Europe)
* **Amsterdam**: `amsterdam.nextblock.io` (Europe)
* **London**: `london.nextblock.io` (Europe)
* **Singapore**: `singapore.nextblock.io` (Asia)
* **Tokyo**: `tokyo.nextblock.io` (Asia)
* **New York**: `ny.nextblock.io` (US East)
* **Salt Lake City**: `slc.nextblock.io` (US West)
* **Dublin**: `dublin.nextblock.io` (Europe)
* **Vilnius**: `vilnius.nextblock.io` (Europe)
## Best Practices
1. **Use async/await**: Leverage Python's asyncio for non-blocking operations
2. **Use TLS by default**: Prefer secure connections unless you explicitly operate in a trusted internal environment
3. **Implement keepalive**: Configure appropriate keepalive settings
4. **Handle errors gracefully**: Use retry logic with exponential backoff
5. **Validate configuration**: Check all required settings before connecting
6. **Use connection pooling**: For high-throughput applications
7. **Monitor connection health**: Implement regular health checks
8. **Close connections properly**: Always close connections when done
---
# Agent Instructions: Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.
Perform an HTTP GET request on the current page URL with the `ask` query parameter:
```
GET https://docs.nextblock.io/api/examples/python/connection.md?ask=
```
The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.
Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.