pynetty 🐍🛠️

Asynchronous networking library for Python, closely inspired by Netty (Java). Built on top of asyncio, with no external dependencies.

Coverage relative to Netty

Netty Py 🐍
├── Bootstrap                    ✅
├── ServerBootstrap              ✅
├── Channel                      ✅ (TCP)
├── ChannelPipeline              ✅
├── ChannelHandler               ✅ (inbound + outbound)
├── ChannelHandlerContext        ✅
├── EventLoop                    ✅
├── EventLoopGroup               ✅
├── ChannelFuture                ✅ (+ ChannelPromise)
├── ByteBuf                      ✅
├── Allocator                    ✅ (Unpooled + Pooled)
├── TCP                          ✅
├── UDP                          ✅ (DatagramChannel)
├── Socket options                ✅ (ChannelOption)
├── Backpressure                  ✅ (WriteBufferWaterMark)
├── Concurrency                   ✅ (ChannelExecutor)
├── TLS/SSL                       ✅ (SslContextBuilder)
├── Encoders / Decoders           ✅ (length-based framing)
├── Idle handlers                 ✅ (IdleStateHandler)
├── Timeouts                      ✅ (Read/WriteTimeoutHandler)
├── Exception handling            ✅ (exception_caught via the pipeline,
│                                      io.netty.*-style exception classes)
└── Declarative packet codec      ✅ (ProtolibCodec, via the protolib
                                       dependency)

Installation

pip install socket-netty

(The PyPI distribution is named socket-netty, but the importable package is still pynetty — your code stays import pynetty.)

Or, for local development from a clone of this repo:

pip install -e .

pynetty declares protolib as an install dependency, so either install method pulls it in automatically — you don't need a separate install step to use ProtolibCodec below.

Structure

pynetty/
  exceptions.py -> Netty-style exception hierarchy (DecoderException,
                    EncoderException, CorruptedFrameException, etc.)
  buffer/       -> ByteBuf, ByteBufAllocator (Unpooled/Pooled)
  handler/      -> ChannelHandler, codecs, protolib bridge, SSL,
                    timeouts/idle
  channel/      -> Channel, DatagramChannel, ChannelPipeline,
                    EventLoop/Group, ChannelFuture, ChannelOption,
                    flow control (backpressure + concurrency)
  bootstrap/    -> ServerBootstrap (server), Bootstrap (client)

Example: TCP echo server

import asyncio
from pynetty import ServerBootstrap, ChannelInboundHandler

class EchoHandler(ChannelInboundHandler):
    async def channel_active(self, ctx):
        print("Client connected:", ctx.channel.remote_address())

    async def channel_read(self, ctx, msg):
        await ctx.write(msg)  # echo

    async def exception_caught(self, ctx, exc):
        print("Error:", exc)
        await ctx.close()

async def main():
    def init_channel(channel):
        channel.pipeline.add_last("echo", EchoHandler())

    server = await ServerBootstrap().child_handler(init_channel).bind("0.0.0.0", 9000)
    await server.serve_forever()

asyncio.run(main())

Exceptions

pynetty raises the same exception hierarchy Netty does, under io.netty.*. Every exception's str() is prefixed with its fully-qualified Java-style path, so logs and exception_caught output look exactly like a real Netty stack:

from pynetty import DecoderException

async def exception_caught(self, ctx, exc):
    print(exc)
    # io.netty.handler.codec.DecoderException: Failed to decode packet

Available exceptions (all subclass NettyException):

Class	Netty path
ChannelException	io.netty.channel.ChannelException
DuplicateHandlerNameError	io.netty.channel.ChannelPipelineException
CodecException	io.netty.handler.codec.CodecException
DecoderException	io.netty.handler.codec.DecoderException
EncoderException	io.netty.handler.codec.EncoderException
CorruptedFrameException	io.netty.handler.codec.CorruptedFrameException
TooLongFrameException	io.netty.handler.codec.TooLongFrameException
ReadTimeoutError	io.netty.handler.timeout.ReadTimeoutException
WriteTimeoutError	io.netty.handler.timeout.WriteTimeoutException
IndexOutOfBoundsError	io.netty.buffer.IndexOutOfBoundsException

LengthFieldBasedFrameDecoder and ByteToMessageCodec automatically wrap any unexpected error raised inside decode() into a DecoderException, and LengthFieldPrepender does the same for EncoderException — matching Netty's own behavior of never letting a raw internal error escape a codec unwrapped.

Socket options

from pynetty import ServerBootstrap, ChannelOption

server = (
    ServerBootstrap()
    .child_handler(init_channel)
    .option(ChannelOption.SO_REUSEADDR, True)
    .option(ChannelOption.SO_BACKLOG, 128)
    .option(ChannelOption.TCP_NODELAY, True)  # low latency, typical for games
)
await server.bind("0.0.0.0", 9000)

TLS/SSL

from pynetty import ServerBootstrap, Bootstrap, SslContextBuilder

# Server
server_ctx = SslContextBuilder.for_server("cert.pem", "key.pem").build()
server = ServerBootstrap().child_handler(init_channel).ssl(server_ctx)
await server.bind("0.0.0.0", 9443)

# Client
client_ctx = SslContextBuilder.for_client().trust_manager("cert.pem").build()
channel = await Bootstrap().handler(init_client).ssl(client_ctx).connect("myserver.com", 9443)

UDP

from pynetty import DatagramBootstrap, ChannelInboundHandler

class UdpHandler(ChannelInboundHandler):
    async def channel_read(self, ctx, msg):
        data, addr = msg  # UDP delivers (bytes, (host, port))
        await ctx.write((b"pong", addr))

channel = await DatagramBootstrap().handler(
    lambda ch: ch.pipeline.add_last("udp", UdpHandler())
).bind("0.0.0.0", 9001)

Idle / Timeouts

from pynetty import IdleStateHandler, ReadTimeoutHandler

async def on_idle(ctx, event):
    print("Idle channel:", event.state)
    await ctx.write(b"PING")  # e.g. a game protocol heartbeat

def init_channel(channel):
    idle = IdleStateHandler(reader_idle_seconds=30)
    idle.on_idle = on_idle
    channel.pipeline.add_last("idle", idle)
    channel.pipeline.add_last("read_timeout", ReadTimeoutHandler(60))  # closes if no data in 60s
    channel.pipeline.add_last("my_handler", MyHandler())

EventLoopGroup (real multi-threaded concurrency)

from pynetty import EventLoopGroup

worker_group = EventLoopGroup(num_threads=4)
worker_group.start()

loop = worker_group.next_loop()  # round-robin
future = loop.submit(lambda: my_heavy_coroutine())
result = future.result(timeout=5)  # blocking, from any thread

ChannelFuture / ChannelPromise

from pynetty import ChannelFuture

future = channel.new_future()
future.add_listener(lambda f: print("Finished:", f.is_success()))

# Still directly awaitable, Python-style:
result = await future

Backpressure (WriteBufferWaterMark)

from pynetty import ServerBootstrap, WriteBufferWaterMark

server = (
    ServerBootstrap()
    .child_handler(init_channel)
    .water_mark(WriteBufferWaterMark(low=32*1024, high=64*1024))
)

# In the handler:
if not ctx.channel.is_writable():
    # pause sending more data until it becomes writable again
    ...

Concurrency (ChannelExecutor)

Guarantees a channel's messages are processed one at a time, in order, even across await points (avoids race conditions on shared handler state):

from pynetty import ChannelExecutor

executor = ChannelExecutor()
executor.start()

async def channel_read(self, ctx, msg):
    await executor.submit(self._process(ctx, msg))

Allocator (buffer pool)

from pynetty import PooledByteBufAllocator

allocator = PooledByteBufAllocator()
buf = allocator.buffer(256)
buf.write_int(42)
# ... use the buffer ...
allocator.release(buf)  # goes back to the pool, ready to be reused

Protocol framing (length-prefixed packets)

Very common in game protocols (Minecraft, Free Fire, etc.):

from pynetty import LengthFieldBasedFrameDecoder, LengthFieldPrepender

def init_channel(channel):
    channel.pipeline.add_last("frame_decoder", LengthFieldBasedFrameDecoder(4))
    channel.pipeline.add_last("frame_prepender", LengthFieldPrepender(4))
    channel.pipeline.add_last("my_handler", MyHandler())

Protocol decoding with protolib (declarative packets)

LengthFieldBasedFrameDecoder/LengthFieldPrepender above only solve framing (where one packet ends and the next begins) — you still have to hand-write the code that turns those raw bytes into a meaningful packet. protolib solves that second half: you describe every packet's fields in a .yml/.json file, and ProtolibCodec plugs that description directly into the pipeline. channel_read then hands your handler a ready-made {"name": ..., "params": {...}} dict instead of raw bytes, and write accepts the same shape (or a (name, params) tuple) going out.

ProtolibCodec has two framing modes, controlled by framed:

from pynetty import ServerBootstrap, ChannelInboundHandler, ProtolibCodec

# framed=True (default): the codec frames the stream itself via
# protolib's own PacketFramer (varint length-prefix, Minecraft-style).
# Don't also add LengthFieldBasedFrameDecoder in this mode.
def init_channel(channel):
    channel.pipeline.add_last(
        "protolib", ProtolibCodec("my_protocol.yml", state="play",
                                   direction_in="toServer", direction_out="toClient"),
    )
    channel.pipeline.add_last("game", GameHandler())

class GameHandler(ChannelInboundHandler):
    async def channel_read(self, ctx, msg):
        print(msg["name"], msg["params"])          # ready-made dict
        await ctx.write(("keep_alive", {"id": 1}))  # (name, params) out

# framed=False: for fixed-size / non-varint protocols (e.g. Minecraft
# Classic/ClassiCube), keep your own LengthFieldBasedFrameDecoder in
# front and let ProtolibCodec just parse/serialize the complete frame
# it's handed.
def init_channel(channel):
    channel.pipeline.add_last("frame_decoder", LengthFieldBasedFrameDecoder(1))
    channel.pipeline.add_last(
        "protolib", ProtolibCodec("classicube_protocol.yml", framed=False),
    )
    channel.pipeline.add_last("game", GameHandler())

protocol accepts a protolib.Protocol instance you already built, or anything Protocol(...) itself accepts (a .yml/.json path, an in-memory string, or a parsed dict) — ProtolibCodec builds the Protocol for you in that case. Any parsing/serialization error protolib raises is wrapped into DecoderException/EncoderException, same as the rest of the codecs in this library.

ByteBuf

from pynetty import ByteBuf

buf = ByteBuf()
buf.write_varint(1000)
buf.write_string("hello")
buf.write_int(-42)

data = buf.to_bytes()

read_buf = ByteBuf.wrapped(data)
n = read_buf.read_varint()
s = read_buf.read_string()
i = read_buf.read_int()

Supports: byte, unsigned_byte, short, unsigned_short, int, unsigned_int, long, float, double, boolean, varint, varlong (Minecraft/protobuf style), string, and raw bytes.

Tests

python3 tests/test_bytebuf.py
python3 tests/test_echo.py
python3 tests/test_core_extras.py      # EventLoop, ChannelFuture, Allocator
python3 tests/test_network_extras.py   # UDP, socket options, timeouts, idle
python3 tests/test_tls.py              # end-to-end TLS (generates temporary certs)
python3 tests/test_concurrency.py      # ChannelExecutor

Design notes

• The entire pipeline is 100% async/await.
• EventLoopGroup uses real OS threads (each with its own asyncio loop), unlike the rest of the library which normally runs on a single loop — it's the honest way to replicate Netty's model (thread pool) in Python.
• ByteBuf is a simple implementation backed by bytearray, with no manual refcounting (Python already has GC).
• PooledByteBufAllocator recycles buffers by size "bucket" (powers of 2), useful in high-traffic game servers to reduce GC pressure.
• Exceptions mirror Netty's own io.netty.* hierarchy (see Exceptions above), so error messages and logs read the same as a real Netty stack trace.
• Designed for use both in game projects (custom servers, protocol reversing) and as a general-purpose library.