← Back to Gists

Minimal Harness Example (micro claude code).

📝 Python
example
example · Level 1 ·

This 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.