QUIC Transaction Submission

Use QUIC when you already have signed Solana transaction bytes and want a low-overhead submission path from Python.

The client below:

1. Connects to a regional QUIC endpoint

2. Authenticates with your API key on a bidirectional stream

3. Reuses the connection to send raw transaction bytes on unidirectional streams

Example

import asyncio
import os
import ssl

import certifi
from aioquic.asyncio.client import connect
from aioquic.quic.configuration import QuicConfiguration

AUTH_OK = b"\x00"
MAX_TX_SIZE = 1232


class NextblockQuicClient:
    def __init__(self, client_cm, protocol):
        self._client_cm = client_cm
        self._protocol = protocol

    @classmethod
    async def connect(cls, host: str, port: int, api_key: str) -> "NextblockQuicClient":
        configuration = QuicConfiguration(
            is_client=True,
            alpn_protocols=["nb-tx/1"],
        )
        configuration.verify_mode = ssl.CERT_REQUIRED
        configuration.load_verify_locations(cafile=certifi.where())

        client_cm = connect(
            host,
            port,
            configuration=configuration,
            server_name=host,
        )
        protocol = await client_cm.__aenter__()

        reader, writer = await protocol.create_stream()
        writer.write(api_key.encode("utf-8"))
        await writer.drain()
        writer.write_eof()

        response = await reader.readexactly(1)
        if response != AUTH_OK:
            await client_cm.__aexit__(None, None, None)
            raise RuntimeError("authentication rejected")

        return cls(client_cm, protocol)

    async def send_transaction(self, raw_tx: bytes) -> None:
        if len(raw_tx) > MAX_TX_SIZE:
            raise ValueError(f"transaction too large: {len(raw_tx)}")

        _, writer = await self._protocol.create_stream(is_unidirectional=True)
        writer.write(raw_tx)
        await writer.drain()
        writer.write_eof()

    async def close(self) -> None:
        await self._client_cm.__aexit__(None, None, None)


async def main() -> None:
    api_key = os.environ.get("NEXTBLOCK_API_KEY")
    if not api_key:
        raise RuntimeError("Set NEXTBLOCK_API_KEY before running")

    client = await NextblockQuicClient.connect(
        "london.nextblock.io",
        11100,
        api_key,
    )

    try:
        # Replace this with the serialized bytes of your signed transaction.
        # Example with solders: raw_tx = bytes(signed_transaction)
        raw_tx = b"\x00\x01\x02\x03"
        await client.send_transaction(raw_tx)
        print("transaction queued")
    finally:
        await client.close()


asyncio.run(main())

What To Replace

• Read the API key from NEXTBLOCK_API_KEY or your own config source.

• Pick the regional host closest to your deployment.

• Replace raw_tx with the serialized bytes of your signed transaction.

Notes

• Send raw transaction bytes, not base64.

• Keep the QUIC connection open and reuse it for multiple transactions. This example closes after one send only to keep the sample short.

• QUIC does not support the extra gRPC submission flags or atomic bundle submission.

• If you already build transactions with solders or solana-py, serialize the signed transaction first and pass the bytes to send_transaction().

See QUIC Transaction Submission for the full endpoint list and protocol summary.