Arkitekt Gateway

A Python library for managing a Tailscale-based sidecar proxy. Arkitekt Gateway bundles a Go binary that handles secure networking via Tailscale, with full Python control over the process lifecycle, signal handling, and log streaming.

Features

• 🚀 Multi-platform support - Pre-built binaries for Linux, macOS, and Windows (x86_64 and ARM64)
• 🔄 Async-first API - Full asyncio support with context managers
• 📡 Signal handling - Send SIGINT, SIGTERM, or any signal to the sidecar
• 📝 Log streaming - Async iterator and callback-based log access
• 🖥️ CLI included - Typer-based command-line interface
• 🔒 Lifecycle management - Python controls when the sidecar starts and stops

Installation

pip install arkitekt-gateway

Or with uv:

uv add arkitekt-gateway

Quick Start

Command Line

# Run the sidecar proxy
arkitekt-gateway run \
  --authkey tskey-auth-xxxxx \
  --coordserver https://your-coordination-server.com \
  --hostname my-proxy \
  --port 8080

# Check binary info
arkitekt-gateway info

Environment variables are also supported:

export TS_AUTHKEY="tskey-auth-xxxxx"
export TS_COORDSERVER="https://your-coordination-server.com"
arkitekt-gateway run

Python API

Basic Usage

import asyncio
from arkitekt_gateway import Sidecar, SidecarConfig

async def main():
    config = SidecarConfig(
        authkey="tskey-auth-xxxxx",
        coordserver="https://your-coordination-server.com",
        hostname="my-proxy",
        port="8080",
    )
    
    async with Sidecar(config) as sidecar:
        # Wait for the proxy to be ready
        proxy_url = await sidecar.wait_ready()
        print(f"Proxy ready at: {proxy_url}")
        
        # Get status and peer info
        status = await sidecar.get_status()
        print(f"Connected peers: {len(status.peers)}")
        
        # Keep running...
        await asyncio.sleep(60)
    
    # Sidecar is automatically stopped when exiting the context

asyncio.run(main())

Streaming Logs

async def main():
    config = SidecarConfig(
        authkey="tskey-auth-xxxxx",
        coordserver="https://your-coordination-server.com",
    )
    
    async with Sidecar(config) as sidecar:
        async for log in sidecar.logs():
            print(f"[{log.stream}] {log.line}")

asyncio.run(main())

Using Log Callbacks

from arkitekt_gateway import Sidecar, SidecarConfig, LogLine

async def log_handler(log: LogLine):
    if "error" in log.line.lower():
        print(f"ERROR: {log.line}")

async def main():
    config = SidecarConfig(
        authkey="tskey-auth-xxxxx",
        coordserver="https://your-coordination-server.com",
    )
    
    sidecar = Sidecar(config)
    sidecar.on_log(log_handler)
    
    await sidecar.start()
    try:
        await sidecar.wait()
    finally:
        if sidecar.is_running:
            await sidecar.stop()

asyncio.run(main())

Sending Signals

import signal

async def main():
    config = SidecarConfig(
        authkey="tskey-auth-xxxxx",
        coordserver="https://your-coordination-server.com",
    )
    
    async with Sidecar(config) as sidecar:
        # Do some work...
        await asyncio.sleep(5)
        
        # Send interrupt signal
        await sidecar.interrupt()
        
        # Or send any signal
        await sidecar.send_signal(signal.SIGTERM)

asyncio.run(main())

API Reference

SidecarConfig

Configuration dataclass for the sidecar process.

Sidecar

Async wrapper for the Go sidecar binary.

Properties

Methods

LogLine

Represents a log line from the sidecar.

SidecarSignal

IPC signals emitted by the sidecar.

SidecarStatus

Status information from the status API.

Development

Local Setup

# Clone the repository
git clone https://github.com/jhnnsrs/arkitekt-gateway.git
cd arkitekt-gateway

# Install dependencies
uv sync --all-extras --dev

# Download the sidecar binary for local development
python dev.py

# Run tests
uv run pytest tests -v

Building

The package uses hatchling with a custom build hook that downloads the appropriate platform-specific binary during wheel building.

uv build

License

License

MIT