I built a small spec-to-code agent that reads a project brief, drafts an architecture, writes the code, and reviews itself before handing anything back. It saves me from staring at a blank file when I need to bootstrap a complex module. I run it on Oxlo.ai because request-based pricing keeps the cost flat even when I paste a long spec and a multi-file plan. See https://oxlo.ai/pricing for details.
What you'll need
- Python 3.10 or newer
- The OpenAI SDK:
pip install openai - An Oxlo.ai API key from https://portal.oxlo.ai
Step 1: Initialize the Oxlo.ai client
First, we create a client pointing at Oxlo.ai. Because Oxlo.ai is fully OpenAI-compatible, the only difference is the base URL and the model names. Oxlo.ai also serves popular models with no cold starts, so three sequential calls feel instant.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ["OXLO_API_KEY"],
)Step 2: Define the system prompt
The system prompt locks the agent into a strict workflow: plan, implement, review. I keep it in its own variable so I can tune it without touching business logic.
SYSTEM_PROMPT = """You are a senior staff engineer. Turn a user specification into production-ready Python.
Workflow:
1. PLAN: List files, classes, functions, and key data structures. Note concurrency or I/O concerns.
2. IMPLEMENTATION: Write complete, typed Python source for every file. Prefer the standard library.
3. REVIEW: Flag bugs, type errors, or race conditions. Suggest concrete fixes.
Format your response with exactly these headers: PLAN, IMPLEMENTATION, REVIEW.
"""Step 3: Generate the plan
The planner sends the spec to Oxlo.ai and returns only the PLAN section. This gives us a roadmap we can inspect before burning tokens on a full implementation.
def generate_plan(client, spec: str) -> str:
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Return only the PLAN section for this spec:\n\n{spec}"},
],
temperature=0.2,
max_tokens=2000,
)
return response.choices[0].message.contentStep 4: Generate the implementation
Next, we feed the plan back into the model and ask for the IMPLEMENTATION. Replaying the plan in the message history keeps the model consistent with its own architecture decisions.
def generate_implementation(client, plan: str, spec: str) -> str:
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Return only the PLAN section for this spec:\n\n{spec}"},
{"role": "assistant", "content": plan},
{"role": "user", "content": "Now return only the IMPLEMENTATION section based on the plan above."},
],
temperature=0.2,
max_tokens=4000,
)
return response.choices[0].message.contentStep 5: Generate the review
The reviewer stage catches logical errors I might miss. I pass both the plan and the implementation back so the critique stays grounded in the original requirements.
def generate_review(client, plan: str, implementation: str, spec: str) -> str:
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Return only the PLAN section for this spec:\n\n{spec}"},
{"role": "assistant", "content": plan},
{"role": "user", "content": "Now return only the IMPLEMENTATION section based on the plan above."},
{"role": "assistant", "content": implementation},
{"role": "user", "content": "Now return only the REVIEW section for the implementation above."},
],
temperature=0.2,
max_tokens=2000,
)
return response.choices[0].message.contentStep 6: Orchestrate the pipeline
A single runner function feeds the spec through each stage and prints labeled markdown. I also added a small guard so the plan is not empty before we ask for code.
def run_pipeline(client, spec: str) -> dict:
print("=== PLAN ===")
plan = generate_plan(client, spec)
print(plan)
print("\n=== IMPLEMENTATION ===")
impl = generate_implementation(client, plan, spec)
print(impl)
print("\n=== REVIEW ===")
review = generate_review(client, plan, impl, spec)
print(review)
return {"plan": plan, "implementation": impl, "review": review}Run it
Here is the full script assembled, followed by the output for a sample spec. I chose a task that requires concurrency awareness and file I/O so the agent has to think beyond a single function.
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.environ["OXLO_API_KEY"],
)
SYSTEM_PROMPT = """You are a senior staff engineer. Turn a user specification into production-ready Python.
Workflow:
1. PLAN: List files, classes, functions, and key data structures. Note concurrency or I/O concerns.
2. IMPLEMENTATION: Write complete, typed Python source for every file. Prefer the standard library.
3. REVIEW: Flag bugs, type errors, or race conditions. Suggest concrete fixes.
Format your response with exactly these headers: PLAN, IMPLEMENTATION, REVIEW.
"""
def generate_plan(client, spec: str) -> str:
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Return only the PLAN section for this spec:\n\n{spec}"},
],
temperature=0.2,
max_tokens=2000,
)
return response.choices[0].message.content
def generate_implementation(client, plan: str, spec: str) -> str:
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Return only the PLAN section for this spec:\n\n{spec}"},
{"role": "assistant", "content": plan},
{"role": "user", "content": "Now return only the IMPLEMENTATION section based on the plan above."},
],
temperature=0.2,
max_tokens=4000,
)
return response.choices[0].message.content
def generate_review(client, plan: str, implementation: str, spec: str) -> str:
response = client.chat.completions.create(
model="deepseek-v3.2",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Return only the PLAN section for this spec:\n\n{spec}"},
{"role": "assistant", "content": plan},
{"role": "user", "content": "Now return only the IMPLEMENTATION section based on the plan above."},
{"role": "assistant", "content": implementation},
{"role": "user", "content": "Now return only the REVIEW section for the implementation above."},
],
temperature=0.2,
max_tokens=2000,
)
return response.choices[0].message.content
def run_pipeline(client, spec: str) -> dict:
print("=== PLAN ===")
plan = generate_plan(client, spec)
print(plan)
print("\n=== IMPLEMENTATION ===")
impl = generate_implementation(client, plan, spec)
print(impl)
print("\n=== REVIEW ===")
review = generate_review(client, plan, impl, spec)
print(review)
return {"plan": plan, "implementation": impl, "review": review}
spec = """
Build a thread-safe LRU cache with TTL expiration and JSON persistence to disk.
Requirements:
- Max size 100 items.
- Items expire after 300 seconds.
- Save to cache.json on every write.
- Reload from disk on startup if the file exists.
- Include type hints and docstrings.
"""
run_pipeline(client, spec)Example output:
=== PLAN ===
Files
- cache.py: LRUTTLCache class using collections.OrderedDict, threading.RLock, and time.time.
Key functions
- __init__(self, capacity=100, ttl=300, filepath="cache.json")
- get(self, key: str) -> Any | None
- put(self, key: str, value: Any) -> None
- _persist(self) -> None
- _load(self) -> None
Concerns
- Need threading.RLock around all reads and writes to OrderedDict.
- JSON serialization may fail on non-primitive values; document that limitation.
=== IMPLEMENTATION ===
import json
import os
import threading
import time
from collections import OrderedDict
from typing import Any, Optional
class LRUTTLCache:
def __init__(self, capacity: int = 100, ttl: int = 300, filepath: str = "cache.json"):
self.capacity = capacity
self.ttl = ttl
self.filepath = filepath
self._cache: OrderedDict[str, Any] = OrderedDict()
self._timestamps: dict[str, float] = {}
self._lock = threading.RLock()
self._load()
def get(self, key: str) -> Optional[Any]:
with self._lock:
if key not in self._cache:
return None
if time.time() - self._timestamps[key] > self.ttl:
self._cache.pop(key, None)
self._timestamps.pop(key, None)
self._persist()
return None
self._cache.move_to_end(key)
return self._cache[key]
def put(self, key: str, value: Any) -> None:
with self._lock:
if key in self._cache:
self._cache.move_to_end(key)
self._cache[key] = value
self._timestamps[key] = time.time()
if len(self._cache) > self.capacity:
oldest = next(iter(self._cache))
self._cache.pop(oldest)
self._timestamps.pop(oldest, None)
self._persist()
def _persist(self) -> None:
with open(self.filepath, "w") as f:
json.dump({"data": list(self._cache.items()), "timestamps": self._timestamps}, f)
def _load(self) -> None:
if not os.path.exists(self.filepath):
return
with open(self.filepath, "r") as f:
payload = json.load(f)
for k, v in payload.get("data", []):
self._cache[k] = v
self._timestamps.update(payload.get("timestamps", {}))
=== REVIEW ===
1. Race condition in _load: _load mutates self._cache and self._timestamps without acquiring self._lock. Wrap the body in with self._lock:.
2. JSON serialization: The spec does not restrict values to JSON-serializable types. Consider adding a custom encoder or documenting the constraint.
3. File I/O under lock: _persist holds the lock while writing to disk. For large caches this blocks all getters. Consider offloading the write or using a snapshot.
4. Type hint on _cache: OrderedDict[str, Any] is correct, but _timestamps should be dict[str, float].Wrap-up
This pipeline turns a rough spec into a reviewed draft in three API calls. A concrete next step is to write the returned code directly to disk with open("cache.py", "w") and invoke subprocess.run(["python", "-m", "pyright", "cache.py"]) to type-check the output automatically. You could also swap deepseek-v3.2 for qwen-3-32b or llama-3.3-70b on Oxlo.ai to compare planning depth against implementation speed.
Top comments (0)