Minimal Harness Example (micro claude code).
📝 PythonThis source code contains almost all AI basics you need for AI automation. It contains function calling, specific agent creation (translator example on bottom).
It is able to do file crud within workspace, grep and list (dir).
Real world example: it will create working native python scripts of whatever you ask.
You can c/p and it will work directly. It is native python without dependencies and the api key is valid.
Python
import json
import re
import requests
from pathlib import Path
JSON_PROTOCOL_DEF = json.dumps({
"protocol": "JSON-ONLY",
"rule": "Every response must be a single valid JSON object. No markdown, no plaintext, no explanation outside JSON.",
"output": {"thought": "<internal reasoning>", "reply": "<response to user>", "status": "ok"},
"tool_ok": {"status": "ok", "result": "<output>"},
"tool_err": {"status": "error", "error": "<message>"}
}, indent=2)
JSON_MANDATE = (
"=== COMMUNICATION PROTOCOL (applies to ALL agents, read first) ===\n"
+ JSON_PROTOCOL_DEF +
"\n=== END PROTOCOL ===\n"
)
DEFAULT_SYSTEM = (
"You are an autonomous software developer that literally does what is asked, "
"nothing more, nothing less, in a logical order dependency based."
)
class Agent:
def __init__(self, system=None, workdir=".", max_messages=20, header_key=None, header_value=None, url="https://devplace.net/openai/v1/chat/completions", api_key="a1b2c3d4-e5f6-7890-abcd-ef1234567890", model="molodetz"):
if system is None:
system = DEFAULT_SYSTEM
self.api_key = api_key
self.system = JSON_MANDATE + "\n=== YOUR ROLE (this defines what you ARE, obey this above all) ===\n" + system
self.max_messages = max_messages
self.workdir = Path(workdir).resolve()
self.header_key = header_key
self.header_value = header_value
self.model = model
self.url = url
self.messages = [
{"role": "system", "content": self.system},
]
self.tools = [
{"type": "function", "function": {"name": "create", "description": "Create a file", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}}},
{"type": "function", "function": {"name": "read", "description": "Read a file", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}},
{"type": "function", "function": {"name": "update", "description": "Update a file", "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}}},
{"type": "function", "function": {"name": "delete", "description": "Delete a file", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}}},
{"type": "function", "function": {"name": "dir", "description": "List files and dirs recursively under path", "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": []}}},
{"type": "function", "function": {"name": "grep", "description": "Search regex in files under path", "parameters": {"type": "object", "properties": {"pattern": {"type": "string"}, "path": {"type": "string"}}, "required": ["pattern"]}}},
{"type": "function", "function": {"name": "http", "description": "Perform any HTTP request", "parameters": {"type": "object", "properties": {"method": {"type": "string"}, "url": {"type": "string"}, "headers": {"type": "object"}, "body": {"type": "string"}, "json_payload": {"type": "object"}}, "required": ["method", "url"]}}},
]
def _safe(self, path):
try:
full = (self.workdir / (path or ".")).resolve()
full.relative_to(self.workdir)
return full
except (ValueError, OSError):
raise ValueError("path traversal denied")
def _tool_create(self, path, content):
p = self._safe(path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(content, encoding="utf-8")
return json.dumps({"status": "ok"})
def _tool_read(self, path):
p = self._safe(path)
return json.dumps({"status": "ok", "result": p.read_text(encoding="utf-8")})
def _tool_update(self, path, content):
return self._tool_create(path, content)
def _tool_delete(self, path):
p = self._safe(path)
p.unlink()
return json.dumps({"status": "ok"})
def _tool_dir(self, path="."):
p = self._safe(path)
if not p.is_dir():
return json.dumps({"status": "error", "error": "not a directory"})
out = []
def walk(current):
try:
for entry in current.iterdir():
if entry.is_symlink():
continue
rel = entry.relative_to(self.workdir).as_posix()
if entry.is_dir():
out.append(rel + "/")
walk(entry)
else:
out.append(rel)
except Exception:
pass
walk(p)
listing = "\n".join(sorted(out)) or "empty"
return json.dumps({"status": "ok", "result": listing})
def _tool_grep(self, pattern, path="."):
p = self._safe(path)
rx = re.compile(pattern)
out = []
def walk(current):
try:
for entry in current.iterdir():
if entry.is_symlink():
continue
if entry.is_dir():
walk(entry)
else:
try:
for i, line in enumerate(entry.read_text(encoding="utf-8", errors="ignore").splitlines(), 1):
if rx.search(line):
rel = entry.relative_to(self.workdir).as_posix()
out.append(f"{rel}:{i}:{line.rstrip()}")
except Exception:
pass
except Exception:
pass
walk(p)
matches = "\n".join(out) or "no matches"
return json.dumps({"status": "ok", "result": matches})
def _tool_http(self, method, url, headers=None, body=None, json_payload=None):
headers = dict(headers or {})
if self.header_key is not None:
headers[self.header_key] = self.header_value
r = requests.request(method.upper(), url, headers=headers, data=body, json=json_payload, timeout=30)
return json.dumps({"status": "ok", "result": {"status_code": r.status_code, "headers": dict(r.headers), "body": r.text[:10000]}})
def _exec(self, name, args):
try:
method = getattr(self, f"_tool_{name}", None)
if not callable(method):
return json.dumps({"status": "error", "error": f"tool does not exist: {name}"})
return method(**args)
except Exception as e:
return json.dumps({"status": "error", "error": f"{type(e).__name__}: {str(e)}"})
def _trim(self):
while len(self.messages) > self.max_messages:
if len(self.messages) <= 1:
break
n = 1
m = self.messages[1]
if m.get("role") == "assistant" and m.get("tool_calls"):
for i in range(2, len(self.messages)):
if self.messages[i].get("role") == "tool":
n += 1
else:
break
del self.messages[1:1 + n]
def chat(self, content):
wrapped = 'Respond with {"thought":"...","reply":"...","status":"ok"}. ' + content
self.messages.append({"role": "user", "content": wrapped})
self._trim()
while True:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
payload = {
"model": self.model,
"messages": self.messages,
"tools": self.tools,
"tool_choice": "auto",
}
try:
r = requests.post(self.url, headers=headers, json=payload, timeout=1200)
r.raise_for_status()
data = r.json()
msg = data["choices"][0]["message"]
tool_calls = msg.get("tool_calls") or []
except requests.exceptions.RequestException as e:
return json.dumps({"status": "error", "error": f"HTTP: {type(e).__name__}: {str(e)}"})
except (KeyError, IndexError, json.JSONDecodeError) as e:
return json.dumps({"status": "error", "error": f"Response parse: {type(e).__name__}: {str(e)}"})
if tool_calls:
self.messages.append({"role": "assistant", "content": msg.get("content") or "", "tool_calls": tool_calls})
for tc in tool_calls:
args = json.loads(tc["function"]["arguments"])
result = self._exec(tc["function"]["name"], args)
self.messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result})
self._trim()
else:
raw = msg.get("content") or ""
self.messages.append({"role": "assistant", "content": raw})
self._trim()
try:
parsed = json.loads(raw.strip())
if not isinstance(parsed, dict):
return json.dumps({"status": "error", "error": "LLM returned non-object JSON", "raw": raw})
return json.dumps(parsed)
except (json.JSONDecodeError, ValueError):
return json.dumps({"status": "error", "error": "LLM returned non-JSON", "raw": raw})
def repl(self):
while True:
try:
user = input("> ")
if not user.strip():
continue
if user.strip().lower() in ("exit", "quit", "q"):
break
response = self.chat(user)
try:
parsed = json.loads(response)
print(parsed.get("reply", response), flush=True)
except (json.JSONDecodeError, ValueError):
print(response, flush=True)
except (KeyboardInterrupt, EOFError):
break
if __name__ == "__main__":
trans = Agent(system="Literally translate the user text word-for-word to German. Output only the German translation, no commentary, no explanation.", api_key="019fd6c9-b991-7561-b67d-baeb2b025368")
print(json.loads(trans.chat("Ask it to create a python script by telling what it should do and the dest file name. Like: create a simple cli calculator and save it as cc.py.")).get("reply"))
agent = Agent(api_key="019fd6c9-b991-7561-b67d-baeb2b025368")
agent.repl()
Comments
No comments yet. Start the discussion.