I needed a terminal assistant that stays out of the way. In this walkthrough I will build a streaming shell helper using Oxlo.ai that translates natural language into commands with minimal latency. A tight feedback loop keeps you in flow without opening a browser or context switching.
What you'll need
- Python 3.10 or newer
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai
Step 1: Connect to Oxlo.ai with streaming
First I verify that streaming works against the Oxlo.ai endpoint. Streaming lets us print tokens as they arrive instead of blocking until the full response is generated.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
stream = client.chat.completions.create(
model="llama-3.3-70b",
messages=[
{"role": "system", "content": "You are a terse shell assistant."},
{"role": "user", "content": "list all python files modified today"},
],
stream=True,
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()
Step 2: Define a constrained system prompt
Latency grows when a model has to reason about open ended instructions. I lock the behavior down with a rigid system prompt so the model emits a short, parseable format and stops quickly.
SYSTEM_PROMPT = """You are a low-latency shell assistant. The user describes a task in natural language.
Rules:
- Reply with exactly one
```bash code block containing the command.
- Follow it with one sentence of explanation.
- End with [SAFE] or [DESTRUCTIVE] on its own line.
- No filler, no markdown headers, no apologies."""
Step 3: Build the interactive loop
Now I wrap the stream in a simple REPL. The OpenAI client reuses the underlying HTTP connection automatically, which removes TCP handshake overhead on every turn against Oxlo.ai.
from openai import OpenAI
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
SYSTEM_PROMPT = """You are a low-latency shell assistant. The user describes a task in natural language.
Rules:
- Reply with exactly one ```
bash code block containing the command.
- Follow it with one sentence of explanation.
- End with [SAFE] or [DESTRUCTIVE] on its own line.
- No filler, no markdown headers, no apologies."""
def stream_command(user_message):
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
]
stream = client.chat.completions.create(
model="llama-3.3-70b",
messages=messages,
stream=True,
)
collected = ""
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
collected += delta
print(delta, end="", flush=True)
print()
return collected
if __name__ == "__main__":
while True:
try:
user_input = input("shell-helper> ")
except (EOFError, KeyboardInterrupt):
break
if not user_input.strip():
continue
stream_command(user_input)
Step 4: Add a safety gate
Executing generated commands blindly is dangerous. I parse the assistant's output for the mandatory safety tag and require explicit confirmation before any destructive operation proceeds.
def run_assistant(user_message):
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
]
stream = client.chat.completions.create(
model="llama-3.3-70b",
messages=messages,
stream=True,
)
collected = ""
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
collected += delta
print(delta, end="", flush=True)
print()
if "[DESTRUCTIVE]" in collected:
confirm = input("This looks destructive. Run anyway? [y/N] ")
if confirm.lower() != "y":
print("Aborted.")
return None
return collected
if __name__ == "__main__":
while True:
try:
user_input = input("shell-helper> ")
except (EOFError, KeyboardInterrupt):
break
if not user_input.strip():
continue
run_assistant(user_input)
Step 5: Trim context to protect latency
Even though Oxlo.ai charges per request rather than per token, long context windows still increase time to first token. I keep a rolling buffer of the last three exchanges so the prompt stays short and fast. If you want predictable costs for continuous workloads, check Oxlo.ai's flat request pricing at https://oxlo.ai/pricing.
MAX_HISTORY_PAIRS = 3
class ShellAssistant:
def __init__(self):
self.client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
self.history = []
def ask(self, user_message):
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
messages.extend(self.history)
messages.append({"role": "user", "content": user_message})
stream = self.client.chat.completions.create(
model="llama-3.3-70b",
messages=messages,
stream=True,
)
collected = ""
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
collected += delta
print(delta, end="", flush=True)
print()
self.history.append({"role": "user", "content": user_message})
self.history.append({"role": "assistant", "content": collected})
while len(self.history) > MAX_HISTORY_PAIRS * 2:
self.history.pop(0)
self.history.pop(0)
if "[DESTRUCTIVE]" in collected:
confirm = input("Destructive command detected. Execute? [y/N] ")
if confirm.lower() != "y":
return None
return collected
if __name__ == "__main__":
assistant = ShellAssistant()
while True:
try:
user_input = input("shell-helper> ")
except (EOFError, KeyboardInterrupt):
break
if not user_input.strip():
continue
assistant.ask(user_input)
Run it
Save the final script as shell_helper.py, export your key, and run it.
export OXLO_API_KEY="sk-oxlo.ai-..."
python shell_helper.py
Here is a sample session. The assistant streams its response instantly and blocks only when it needs human confirmation.
$ python shell_helper.py
shell-helper> find all json files in /var/log over 1MB
```bash
find /var/log -name "*.json" -size +1M
```
Searches /var/log for JSON files larger than one megabyte.
[SAFE]
shell-helper> delete old log files in the current directory
```bash
find . -name "*.log" -mtime +30 -delete
```
Removes log files older than 30 days in the current directory.
[DESTRUCTIVE]
Destructive command detected. Execute? [y/N] n
Aborted.
Wrap-up and next steps
The assistant is usable now, but two upgrades make it sharper. First, wire in function calling so [SAFE] commands execute automatically and feed stdout back into the Oxlo.ai context loop. Second, add a voice layer by sending audio to Oxlo.ai's Whisper endpoint and piping the transcript into the same helper. Both stay inside Oxlo.ai's flat pricing, so longer prompts do not inflate your bill.
Top comments (0)