agent-tty

exec() killed frame detection.
AI gets pure data through a pipe. Humans get a real terminal with colors and readline. Same namespace. Python-native runtime.

parse ANSI escape sequences
guess when a command finished
strip terminal control codes
exec(src) → return value → done
pip install agent-tty

Python 3.10+ · stdlib core, pywinpty on Windows · POSIX PTY / WinPTY / socket fallback · CLI: k

agent workflow
# start the runtime $ k daemon k daemon pid=12345 /tmp/k.sock mode=pty # one persistent Python namespace $ k new work OK work pid=12346 $ k run work "x = 41" $ k run work "x + 1" 42 # long tasks return immediately $ k fire work "import time; time.sleep(10); result = x * 2" {"cell_id": "a1b2c3d4e5f6", "status": "fired"} # assignment — no stdout; result lives in the variable $ k poll work {"cell_id": "a1b2c3d4e5f6", "status": "done", "output": ""} $ k run work "result" 84

Write files, then exec

Complex code with quotes, f-strings, SQL, or shell variables? Write a file with your shell tool, then load it. No manual escaping. The file is transport; the namespace is the workspace.

recommended pattern
# your agent writes complex code to a file (no escaping needed) $ cat > /tmp/task.py << 'EOF' config = json.load(open("settings.json")) config["rate_limit"] = 100 Path("settings.json").write_text(json.dumps(config, indent=2)) print(f"rate_limit set to {config['rate_limit']}") EOF # then loads it into the live session $ k run work "exec(open('/tmp/task.py').read())" rate_limit set to 100
The session remembers config. The next cell can read or change it. The file was just transport — the namespace is the workspace.

Two channels, one namespace

Your agent's commands enter through a structured pipe: source code in, captured output out. Humans attach through a real PTY: readline, tab completion, arrow keys, colors. Both paths share the same live Python process.

AI channel — pipe
# structured JSON protocol # no ANSI, no colors, no control codes # pure data in, pure data out $ k run work "len(users)" 1847 $ k run work "server.healthy()" True $ k status work {"state":"idle","running":[],"vars":12,"cells":3}
human channel — real PTY
$ k attach work shared with AI. Ctrl-] detaches. exit() kills session. >>> users[TAB] users.keys() users.values() users.items() >>> [ai] >>> blocked.add("1.2.3.4") [ai] >>> print(f"{len(blocked)} IPs blocked") 42 IPs blocked >>> len(blocked) 42

Stateful first

bash_tool is curl — every call forks a process, runs, and dies. k is a socket — one process stays alive, and every call is a function invocation inside it.

subprocess.run × 100 fork → exec → connect → run → die fork → exec → connect → run → die fork → exec → connect → run → die …97 more times ~15 seconds of overhead each call pays the full startup cost nothing survives between calls
k run × 100 exec(src) → return exec(src) → return exec(src) → return …97 more times ~0.15 seconds of overhead one process, one namespace imports, connections, state persist
Humans avoid state. State means cleanup, crash recovery, leaked resources. So human tools default to stateless — fork, run, die. But your agent isn't afraid of state. State is memory. Memory means it doesn't repeat work, doesn't re-parse configs, doesn't re-open connections. subprocess.run is amnesia. k run is accumulation.

What stays alive

Some things only exist in process memory. A database connection, a TCP socket, an SSH tunnel, a trained model, a Flask app serving requests in a daemon thread, a CDP browser session — none of these can be serialized to disk. subprocess.run kills them every call. k run keeps them alive. The process is the workspace.

Variables and imports

A pandas DataFrame, a trained model, a parsed config, a compiled regex — anything in the namespace survives across cells.

Connections and servers

Database handles, HTTP sessions, WebSocket connections, Flask apps running in daemon threads — open once, use from every cell.

Live control plane

Feature flags, rate limits, firewall sets, routing weights become Python variables. Patch one cell; the next request sees it. No restart.

Broadcast

When your agent runs a cell, humans see it prefixed with [ai] >>> on the PTY. Two-way visibility — nobody works blind.

Async cells

k fire queues background work. k poll checks the result. The session keeps running. Multiple sessions for parallelism.

Single Python runtime

Core runtime is Python stdlib. The process owns the session directly. pywinpty gives raw WinPTY on Windows — without it, a socket console works fine.

The REPL is Turing complete

You don't need a built-in file watcher, a notification framework, or a monitor callback. The session is Python. Python can do anything. Give your agent the primitives; it builds the rest.

your agent builds its own monitor
# start a long task $ k fire work "train_model(epochs=50)" {"cell_id": "c7d8e9f0a1b2", "status": "fired"} # poll from outside until done $ k poll work c7d8e9f0a1b2 {"cell_id": "c7d8e9f0a1b2", "status": "running", "output": ""} # or: set up a watcher inside the session $ k run work " import threading def on_done(): while not training_complete: time.sleep(5) open('/tmp/done', 'w').write('finished') threading.Thread(target=on_done, daemon=True).start() "
Frameworks provide features. A REPL provides physics. Every feature is a subset of what a Turing-complete session can express. k gives your agent fire + poll as primitives. Complex workflows are code, not configuration.

REPL patterns

Kill the prefix tax

from os import * — now listdir(".") instead of os.listdir("."). Every token your agent saves is money saved.

Print tax is zero

Expressions display automatically. k run w "len(data)" prints the result. No print() wrapper needed for the last expression.

Hot reload

exec(open("module.py").read()) or importlib.reload(m) — update code without restarting the session or losing state.

Incremental execution

Break a long script into cells. If step 3 crashes, fix and re-run just step 3 — steps 1 and 2's state is still in the namespace.

Catch, fix, retry

Exception in a cell? Read the traceback, fix the function, run again — all in the same session. State and data survive the error. No restart.

Shell via Python

Need the host OS? subprocess.run(["git","status"], capture_output=True, text=True).stdout — host commands return clean strings inside the session.

Commands

k daemon [--show-token] start daemon in foreground k stop stop daemon gracefully k new <name> create a Python session k int <name> interrupt running async cells k kill <name> terminate session k run <name> "code" sync eval/exec, raw output k fire <name> "code" async eval/exec, JSON cell_id k poll <name> [cell_id] JSON cell result k status <name> JSON session state k vars <name> JSON namespace names k complete <name> "text" JSON tab completions k ls list sessions k attach <name> human REPL (Ctrl-] to detach) k --version|-V|version print version TCP mode writes daemon.json for local token discovery. Shutdown removes that metadata file. Tokens are printed only with k daemon --show-token. Only one auto-discoverable TCP daemon owns daemon.json. K_TOKEN and K_PORT remain overrides. See README.
GitHub PyPI Docs