> ## Documentation Index
> Fetch the complete documentation index at: https://docs.boxlite.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Read and write raw bytes over a tunnel

> Drive a tunnel as a bidirectional byte stream when you are speaking a protocol of your own rather than HTTP.

A tunnel moves bytes and interprets nothing along the way. When your service speaks a protocol of its own, take the connection directly and read and write it yourself.

## Prerequisites

* An API key from the console, exported as `BOXLITE_API_KEY`. See [API keys](/cloud/api-keys).
* `pip install boxlite`, and the REST URL exported as `BOXLITE_REST_URL`. See [Quickstart](/cloud/quickstart).
* A box you can start, and a service inside it that listens on a TCP port.

Every example reads both values from the environment, so nothing hard-codes a credential.

`connect()` hands you the byte stream directly. This is the level you work at for a protocol that is not HTTP — a line protocol, a binary framing, a database wire protocol — or when you want full control over what goes on the wire.

```python theme={"theme":{"light":"github-light","dark":"github-dark"}}
# cloud_tunnel_bytes.py — talk to a non-HTTP TCP service inside a box
# Run: python cloud_tunnel_bytes.py
import asyncio
import os
import shlex
import time

from boxlite import ApiKeyCredential, Boxlite, BoxliteRestOptions, BoxOptions

GUEST_PORT = 18090
IMAGE = "ghcr.io/boxlite-ai/boxlite-agent-base:v0.1.0"

# A line protocol: read one newline-terminated line, reply with it upper-cased.
SERVER_CODE = f"""
import socketserver

class Handler(socketserver.StreamRequestHandler):
    def handle(self):
        line = self.rfile.readline()
        self.wfile.write(line.upper())

class Server(socketserver.ThreadingTCPServer):
    allow_reuse_address = True

Server(("0.0.0.0", {GUEST_PORT}), Handler).serve_forever()
"""


async def main() -> None:
    # export BOXLITE_API_KEY=<YOUR_API_KEY> before running
    rt = Boxlite.rest(BoxliteRestOptions(
        url=os.environ.get("BOXLITE_REST_URL", "https://api.boxlite.ai"),
        credential=ApiKeyCredential(os.environ["BOXLITE_API_KEY"]),
    ))

    box_id = None
    connection = None
    try:
        box = await rt.create(
            BoxOptions(image=IMAGE),
            name=f"tunnel-bytes-{int(time.time())}",
        )
        box_id = box.id
        await box.start()

        launch = await box.exec("sh", args=[
            "-c",
            f"nohup python3 -u -c {shlex.quote(SERVER_CODE)} "
            "> /tmp/line-server.log 2>&1 & echo launched",
        ])
        await launch.wait()
        await asyncio.sleep(2)  # give the server time to bind

        tunnel = await box.network.tunnel(GUEST_PORT)
        connection = await tunnel.connect()

        await connection.write(b"ping over a raw tunnel\n")

        # Signals that you have finished sending. This server replies to the
        # newline, so it does not depend on seeing the half-close -- see Limits.
        await connection.shutdown_write()

        reply = bytearray()
        while b"\n" not in reply:
            chunk = await connection.read(4096)  # max_bytes must be non-zero
            if not chunk:  # empty read: the far side closed the stream
                break
            reply.extend(chunk)

        print(f"reply: {bytes(reply)!r}")
    except Exception as exc:
        print(f"byte stream demo failed: {type(exc).__name__}: {exc}")
    finally:
        if connection is not None:
            try:
                await connection.close()
            except Exception as exc:
                print(f"connection close failed: {exc}")
        if box_id:
            try:
                await rt.remove(box_id, force=True)
            except Exception as exc:
                print(f"remove failed: {exc}")


if __name__ == "__main__":
    asyncio.run(main())
```

`read(max_bytes)` returns up to `max_bytes` bytes and an empty `bytes` object once the far side has closed. `max_bytes` of `0` raises `ValueError: max_bytes must be non-zero`, and reading after `close()` raises a BoxLite error carrying `connection is closed`.

A synchronous surface exists as well: `SyncBox.network` and `SyncBox.tunnel(port)` return the `SyncNetworkHandle` and `SyncTunnelForwarder` equivalents of the classes documented here.

## Next steps

<CardGroup cols={2}>
  <Card title="Network" icon="network-wired" href="/cloud/network">
    How tunnels work, the full parameter reference, and the outbound boundary.
  </Card>

  <Card title="Network policy" icon="shield-halved" href="/cloud/network-policy">
    Control what the box itself is allowed to reach.
  </Card>
</CardGroup>