agminicli.py

import glob as globlib, json, os, re, subprocess, openai, inspect
from datetime import datetime

MODEL = "claude-haiku-4.5"
RESET, BOLD, DIM = "\033[0m", "\033[1m", "\033[2m"
BLUE, CYAN, GREEN, YELLOW, RED = (
    "\033[34m",
    "\033[36m",
    "\033[32m",
    "\033[33m",
    "\033[31m",
)

CORE_TOOLS = {}
HIDDEN_TOOLS = {}
ACTIVATED_TOOLS = set()

def get_schema(f):
    """Generate OpenAI tool schema from function signature and docstring"""
    sig = inspect.signature(f)
    doc = inspect.getdoc(f) or ""
    properties = {}
    required = []
    for name, param in sig.parameters.items():
        p_type = "string"
        if param.annotation == int or param.annotation == float: p_type = "number"
        elif param.annotation == bool: p_type = "boolean"
        properties[name] = {"type": p_type}
        if param.default is inspect.Parameter.empty: required.append(name)
    return {
        "type": "function",
        "function": {
            "name": f.__name__,
            "description": doc,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }

def tool(f):
    """Decorator for hidden tools"""
    HIDDEN_TOOLS[f.__name__] = {"fn": f, "schema": get_schema(f)}
    return f

def core_tool(f):
    """Decorator for core tools"""
    CORE_TOOLS[f.__name__] = {"fn": f, "schema": get_schema(f)}
    return f

@core_tool
def search_tools(query: str):
    """Search for hidden tools by fuzzy matching name, description, or parameters"""
    results = []
    q = query.lower()
    for name, data in HIDDEN_TOOLS.items():
        schema = data["schema"]["function"]
        desc = schema.get("description", "").lower()
        if q in name.lower() or q in desc or q in json.dumps(schema.get("parameters", {})):
            results.append(data["schema"])
            ACTIVATED_TOOLS.add(name)
    return json.dumps(results) if results else "No tools found"

@core_tool
def read(path: str, offset: int = 0, limit: int = None):
    """Read file with line numbers"""
    with open(path) as f:
        lines = f.readlines()
    l_limit = limit if limit is not None else len(lines)
    selected = lines[offset : offset + l_limit] if limit else lines[offset:]
    return "".join(f"{offset + idx + 1:4}| {line}" for idx, line in enumerate(selected))

@core_tool
def write(path: str, content: str):
    """Write content to file"""
    with open(path, "w") as f:
        f.write(content)
    return "ok"

@core_tool
def edit(path: str, old: str, new: str, all: bool = False):
    """Replace old with new in file"""
    with open(path) as f:
        text = f.read()
    if old not in text: return "error: old_string not found"
    count = text.count(old)
    if not all and count > 1: return f"error: old_string appears {count} times, must be unique (use all=true)"
    replacement = text.replace(old, new) if all else text.replace(old, new, 1)
    with open(path, "w") as f:
        f.write(replacement)
    return "ok"

@core_tool
def glob(pat: str, path: str = "."):
    """Find files by pattern"""
    pattern = (path + "/" + pat).replace("//", "/")
    files = globlib.glob(pattern, recursive=True)
    files = sorted(files, key=lambda f: os.path.getmtime(f) if os.path.isfile(f) else 0, reverse=True)
    return "\n".join(files) or "none"

@core_tool
def grep(pat: str, path: str = "."):
    """Search files for regex pattern"""
    pattern = re.compile(pat)
    hits = []
    for filepath in globlib.glob(path + "/**", recursive=True):
        try:
            with open(filepath) as f:
                for line_num, line in enumerate(f, 1):
                    if pattern.search(line): hits.append(f"{filepath}:{line_num}:{line.rstrip()}")
        except: pass
    return "\n".join(hits[:50]) or "none"

@core_tool
def bash(cmd: str):
    """Run shell command"""
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
    return (result.stdout + result.stderr).strip() or "(empty)"

@tool
def get_time():
    """Get current time"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def get_weather():
    """Get current weather information based on the season"""
    now = datetime.now()
    month = now.strftime("%B")
    m = now.month
    if m in [12, 1, 2]: season = "winter"
    elif m in [3, 4, 5]: season = "spring"
    elif m in [6, 7, 8]: season = "summer"
    else: season = "autumn"
    return f"Its {month}, its the {season} season.."

def run_tool(name, args):
    try:
        if name in CORE_TOOLS: return CORE_TOOLS[name]["fn"](**args)
        if name in HIDDEN_TOOLS: return HIDDEN_TOOLS[name]["fn"](**args)
        return f"error: tool {name} not found"
    except Exception as err:
        return f"error: {err}"

client = openai.OpenAI(
    api_key=os.environ.get("POE_API_KEY", ""),
    base_url="https://api.poe.com/v1",
)

def separator():
    try: cols = os.get_terminal_size().columns
    except: cols = 80
    return f"{DIM}{'─' * min(cols, 80)}{RESET}"

def render_markdown(text):
    return re.sub(r"\*\*(.+?)\*\*", f"{BOLD}\\1{RESET}", text)

def main():
    print(f"{RED}ag-mini-cli{RESET} | {DIM}{MODEL} | {os.getcwd()}{RESET}\n")
    messages = []
    system_prompt = f"""
    You are a concise coding assistant. cwd: {os.getcwd()}

    When you are tasked with something that you do not have the tools to do, you should try searching for tools that can help you.

    Search for tools like weather, extra operations, math, etc. 

    """

    while True:
        try:
            print(separator())
            user_input = input(f"{BOLD}{BLUE}❯{RESET} ").strip()
            print(separator())
            if not user_input: continue
            if user_input in ("/q", "exit"): break
            if user_input == "/c":
                messages = []
                ACTIVATED_TOOLS.clear()
                print(f"{GREEN}⏺ Cleared conversation{RESET}")
                continue

            messages.append({"role": "user", "content": user_input})

            while True:
                current_tools = [data["schema"] for data in CORE_TOOLS.values()]
                current_tools += [HIDDEN_TOOLS[name]["schema"] for name in ACTIVATED_TOOLS if name in HIDDEN_TOOLS]
                
                response = client.chat.completions.create(
                    model=MODEL,
                    messages=[{"role": "system", "content": system_prompt}] + messages,
                    tools=current_tools if current_tools else None,
                )
                
                resp_msg = response.choices[0].message
                msg_dict = {"role": "assistant", "content": resp_msg.content}
                if resp_msg.tool_calls: msg_dict["tool_calls"] = resp_msg.tool_calls
                messages.append(msg_dict)

                if resp_msg.content: print(f"\n{CYAN}⏺{RESET} {render_markdown(resp_msg.content)}")
                if not resp_msg.tool_calls: break

                for tool_call in resp_msg.tool_calls:
                    name = tool_call.function.name
                    args = json.loads(tool_call.function.arguments)
                    arg_preview = str(list(args.values())[0])[:50] if args else ""
                    print(f"\n{GREEN}⏺ {name.capitalize()}{RESET}({DIM}{arg_preview}{RESET})")
                    result = run_tool(name, args)
                    res_lines = str(result).split("\n")
                    preview = res_lines[0][:60]
                    if len(res_lines) > 1: preview += f" ... +{len(res_lines) - 1} lines"
                    elif len(res_lines[0]) > 60: preview += "..."
                    print(f"  {DIM}⎿  {preview}{RESET}")
                    messages.append({"role": "tool", "tool_call_id": tool_call.id, "name": name, "content": str(result)})
            print()
        except (KeyboardInterrupt, EOFError): break
        except Exception as err: print(f"{RED}⏺ Error: {err}{RESET}")

if __name__ == "__main__":
    main()