From c498ad7bda55a81a7113bb7b00037f9c04fd6922 Mon Sep 17 00:00:00 2001 From: Paul Bailey Date: Sun, 22 Feb 2026 15:22:11 +0000 Subject: [PATCH] Main Python file --- ollama-cli.py | 1808 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1808 insertions(+) create mode 100755 ollama-cli.py diff --git a/ollama-cli.py b/ollama-cli.py new file mode 100755 index 0000000..ea8203c --- /dev/null +++ b/ollama-cli.py @@ -0,0 +1,1808 @@ +#!/usr/bin/env python3 +import requests +import json +import subprocess +import os +import re +import sys +import threading +import queue +from pathlib import Path + +OLLAMA_URL = "http://localhost:11434/api/chat" +MODEL = "llama3.2" # Default model +AVAILABLE_MODELS = [] # Cache available models +CONFIG_FILE = os.path.expanduser("~/.ollama-cli-config.json") +SESSION_DIR = os.path.expanduser("~/.ollama-cli-sessions") +CURRENT_SESSION = None +SESSION_CONTEXT = { + "created_files": [], + "last_output": "", + "context_files": [] +} +MCP_SERVERS = {} # Connected MCP servers +MCP_TOOLS = {} # Tools from MCP servers +LSP_SERVERS = {} # Language servers +KNOWLEDGE_BASE = {} # Simple knowledge store +SUBAGENTS = [] # Running subagents +NTFY_TOPIC = None # ntfy.sh topic for notifications + +def load_config(): + """Load config file""" + if os.path.exists(CONFIG_FILE): + try: + with open(CONFIG_FILE, 'r') as f: + return json.load(f) + except: + pass + return { + "excluded_models": [], + "git": { + "remote_url": "", + "username": "", + "password": "" + }, + "gitea": { + "url": "", + "token": "" + }, + "auto_save_sessions": True, + "mcp_servers": {}, + "ollama_url": "http://localhost:11434/api/chat", + "ntfy": { + "enabled": False, + "topic": "", + "server": "https://ntfy.sh" + } + } + +def send_notification(title, message, priority="default"): + """Send notification via ntfy.sh""" + config = load_config() + ntfy_config = config.get("ntfy", {}) + + if not ntfy_config.get("enabled") or not ntfy_config.get("topic"): + return + + try: + server = ntfy_config.get("server", "https://ntfy.sh") + topic = ntfy_config["topic"] + url = f"{server}/{topic}" + + requests.post(url, + data=message.encode('utf-8'), + headers={ + "Title": title, + "Priority": priority + }, + timeout=5 + ) + except: + pass # Silently fail + +def ntfy_listener(messages_list): + """Listen for ntfy notifications and process them""" + config = load_config() + ntfy_config = config.get("ntfy", {}) + + if not ntfy_config.get("enabled") or not ntfy_config.get("topic"): + return + + server = ntfy_config.get("server", "https://ntfy.sh") + topic = ntfy_config["topic"] + url = f"{server}/{topic}/json" + + try: + response = requests.get(url, stream=True, timeout=None) + for line in response.iter_lines(): + if line: + data = json.loads(line) + if data.get("event") == "message": + title = data.get("title", "") + message = data.get("message", "") + + # Skip messages sent by this script to avoid feedback loop + if title in ["Response", "Ollama CLI", "File Created", "Command Complete", "Test"]: + continue + + # Process notification as a user request + print(f"\n[📱 Notification: {title}]") + print(f"> {message}\n") + + messages_list.append({"role": "user", "content": message}) + + # Get response + selected_model = select_best_model(message, "initial") + response_text = call_ollama(messages_list, selected_model) + messages_list.append({"role": "assistant", "content": response_text}) + + # Extract clean output - remove tool calls and metadata + clean_output = re.sub(r'.*?', '', response_text, flags=re.DOTALL) + clean_output = '\n'.join([ + line for line in clean_output.split('\n') + if line.strip() and not any(line.strip().startswith(p) for p in + ['Result:', 'Output:', 'Exit code:', '[Executing', '[Result:', '[Using']) + ]) + + # Send response back as notification + send_notification("Response", clean_output.strip()[:1000] if clean_output.strip() else response_text[:1000]) + except: + pass + +def save_config(config): + """Save config file""" + with open(CONFIG_FILE, 'w') as f: + json.dump(config, f, indent=2) + +def save_session(messages, session_name=None): + """Save conversation session""" + os.makedirs(SESSION_DIR, exist_ok=True) + if not session_name: + from datetime import datetime + session_name = datetime.now().strftime("%Y%m%d_%H%M%S") + + session_file = os.path.join(SESSION_DIR, f"{session_name}.json") + with open(session_file, 'w') as f: + json.dump({"messages": messages, "timestamp": session_name}, f, indent=2) + return session_file + +def load_session(session_name): + """Load conversation session""" + session_file = os.path.join(SESSION_DIR, f"{session_name}.json") + if os.path.exists(session_file): + with open(session_file, 'r') as f: + data = json.load(f) + return data.get("messages", []) + return None + +def list_sessions(): + """List available sessions""" + if not os.path.exists(SESSION_DIR): + return [] + sessions = [f.replace('.json', '') for f in os.listdir(SESSION_DIR) if f.endswith('.json')] + return sorted(sessions, reverse=True) + +def connect_mcp_server(name, command, args=None): + """Connect to an MCP server via stdio""" + try: + import subprocess + cmd = [command] + (args or []) + process = subprocess.Popen( + cmd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1 + ) + + # Send initialize request + init_request = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "ollama-cli", "version": "2.0"} + } + } + + process.stdin.write(json.dumps(init_request) + "\n") + process.stdin.flush() + + # Read response + response = process.stdout.readline() + if response: + data = json.loads(response) + if "result" in data: + MCP_SERVERS[name] = process + + # List tools + tools_request = { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/list" + } + process.stdin.write(json.dumps(tools_request) + "\n") + process.stdin.flush() + + tools_response = process.stdout.readline() + if tools_response: + tools_data = json.loads(tools_response) + if "result" in tools_data and "tools" in tools_data["result"]: + for tool in tools_data["result"]["tools"]: + tool_name = f"mcp_{name}_{tool['name']}" + MCP_TOOLS[tool_name] = { + "server": name, + "name": tool["name"], + "description": tool.get("description", ""), + "schema": tool.get("inputSchema", {}) + } + + return f"Connected to MCP server '{name}' with {len([t for t in MCP_TOOLS if t.startswith(f'mcp_{name}_')])} tools" + + return f"Failed to initialize MCP server '{name}'" + except Exception as e: + return f"Error connecting to MCP server: {str(e)}" + +def call_mcp_tool(server_name, tool_name, arguments): + """Call a tool on an MCP server""" + try: + if server_name not in MCP_SERVERS: + return f"Error: MCP server '{server_name}' not connected" + + process = MCP_SERVERS[server_name] + + request = { + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": { + "name": tool_name, + "arguments": arguments + } + } + + process.stdin.write(json.dumps(request) + "\n") + process.stdin.flush() + + response = process.stdout.readline() + if response: + data = json.loads(response) + if "result" in data: + return json.dumps(data["result"], indent=2) + elif "error" in data: + return f"MCP Error: {data['error']}" + + return "No response from MCP server" + except Exception as e: + return f"Error calling MCP tool: {str(e)}" + +# === LSP SUPPORT === +def start_lsp_server(language, root_path="."): + """Start a language server""" + lsp_commands = { + "python": ["pyright-langserver", "--stdio"], + "javascript": ["typescript-language-server", "--stdio"], + "typescript": ["typescript-language-server", "--stdio"], + "rust": ["rust-analyzer"], + "go": ["gopls"], + } + + if language not in lsp_commands: + return f"No LSP server configured for {language}" + + try: + process = subprocess.Popen( + lsp_commands[language], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + cwd=os.path.expanduser(root_path) + ) + + # Initialize + init_request = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "processId": os.getpid(), + "rootUri": f"file://{os.path.abspath(os.path.expanduser(root_path))}", + "capabilities": {} + } + } + + msg = json.dumps(init_request) + process.stdin.write(f"Content-Length: {len(msg)}\r\n\r\n{msg}") + process.stdin.flush() + + LSP_SERVERS[language] = process + return f"Started LSP server for {language}" + except Exception as e: + return f"Error starting LSP: {str(e)}" + +def lsp_goto_definition(language, file_path, line, character): + """Get definition location""" + if language not in LSP_SERVERS: + return "LSP server not started" + + try: + process = LSP_SERVERS[language] + request = { + "jsonrpc": "2.0", + "id": 2, + "method": "textDocument/definition", + "params": { + "textDocument": {"uri": f"file://{os.path.abspath(file_path)}"}, + "position": {"line": line, "character": character} + } + } + + msg = json.dumps(request) + process.stdin.write(f"Content-Length: {len(msg)}\r\n\r\n{msg}") + process.stdin.flush() + + # Read response (simplified) + return "Definition lookup sent (response parsing not implemented)" + except Exception as e: + return f"LSP error: {str(e)}" + +def lsp_find_references(language, file_path, line, character): + """Find all references to symbol""" + if language not in LSP_SERVERS: + return "LSP server not started" + + try: + process = LSP_SERVERS[language] + request = { + "jsonrpc": "2.0", + "id": 3, + "method": "textDocument/references", + "params": { + "textDocument": {"uri": f"file://{os.path.abspath(file_path)}"}, + "position": {"line": line, "character": character}, + "context": {"includeDeclaration": True} + } + } + + msg = json.dumps(request) + process.stdin.write(f"Content-Length: {len(msg)}\r\n\r\n{msg}") + process.stdin.flush() + + return "References lookup sent" + except Exception as e: + return f"LSP error: {str(e)}" + +# === WEB SEARCH (Enhanced) === +def web_search_enhanced(query): + """Enhanced web search with DuckDuckGo""" + try: + import urllib.parse + encoded = urllib.parse.quote(query) + + # Use DuckDuckGo Instant Answer API + api_url = f"https://api.duckduckgo.com/?q={encoded}&format=json" + response = requests.get(api_url, timeout=10) + data = response.json() + + results = [] + if data.get("AbstractText"): + results.append(f"Summary: {data['AbstractText']}") + + for topic in data.get("RelatedTopics", [])[:5]: + if isinstance(topic, dict) and "Text" in topic: + results.append(f"- {topic['Text']}") + + return "\n".join(results) if results else "No results found" + except Exception as e: + return f"Search error: {str(e)}" + +# === KNOWLEDGE BASE === +def kb_add(key, content): + """Add to knowledge base""" + KNOWLEDGE_BASE[key] = { + "content": content, + "timestamp": subprocess.run("date", capture_output=True, text=True).stdout.strip() + } + return f"Added '{key}' to knowledge base" + +def kb_search(query): + """Search knowledge base (simple keyword match)""" + results = [] + query_lower = query.lower() + for key, data in KNOWLEDGE_BASE.items(): + if query_lower in key.lower() or query_lower in data["content"].lower(): + results.append(f"{key}: {data['content'][:200]}...") + return "\n".join(results) if results else "No matches found" + +def kb_get(key): + """Get from knowledge base""" + if key in KNOWLEDGE_BASE: + return KNOWLEDGE_BASE[key]["content"] + return f"Key '{key}' not found" + +# === SUBAGENTS === +def spawn_subagent(task, model=None): + """Spawn a subagent to handle a task""" + try: + agent_id = len(SUBAGENTS) + 1 + + # Create a simple prompt for the subagent + messages = [ + {"role": "system", "content": "You are a helpful assistant. Complete the task concisely."}, + {"role": "user", "content": task} + ] + + # Run in thread + result_queue = queue.Queue() + + def run_agent(): + try: + response = call_ollama(messages, model) + result_queue.put({"success": True, "result": response}) + except Exception as e: + result_queue.put({"success": False, "error": str(e)}) + + thread = threading.Thread(target=run_agent, daemon=True) + thread.start() + + agent = { + "id": agent_id, + "task": task, + "thread": thread, + "queue": result_queue, + "status": "running" + } + + SUBAGENTS.append(agent) + return f"Spawned subagent #{agent_id} for task: {task[:50]}..." + except Exception as e: + return f"Error spawning subagent: {str(e)}" + +def check_subagents(): + """Check status of subagents""" + results = [] + for agent in SUBAGENTS: + if not agent["queue"].empty(): + result = agent["queue"].get() + agent["status"] = "completed" + agent["result"] = result + results.append(f"Agent #{agent['id']}: {result.get('result', result.get('error'))[:200]}") + elif agent["thread"].is_alive(): + results.append(f"Agent #{agent['id']}: Still running...") + else: + results.append(f"Agent #{agent['id']}: Completed") + + return "\n".join(results) if results else "No active subagents" + +# === ADVANCED CODE TOOLS === +def code_search_symbol(symbol_name, path="."): + """Search for symbol definitions""" + try: + # Simple grep-based search + result = subprocess.run( + f'grep -rn "def {symbol_name}\\|class {symbol_name}\\|function {symbol_name}" {path}', + shell=True, capture_output=True, text=True + ) + return result.stdout if result.stdout else f"Symbol '{symbol_name}' not found" + except Exception as e: + return f"Error: {str(e)}" + +def code_analyze_file(file_path): + """Analyze code structure and suggest improvements""" + try: + with open(os.path.expanduser(file_path), 'r') as f: + content = f.read() + + lines = content.split('\n') + functions = [l for l in lines if 'def ' in l] + classes = [l for l in lines if 'class ' in l] + + # Analyze issues + issues = [] + if len(lines) > 500: + issues.append(f"Large file ({len(lines)} lines) - consider splitting into modules") + if len(functions) > 30: + issues.append(f"Many functions ({len(functions)}) - consider organizing into classes") + + bare_excepts = sum(1 for l in lines if l.strip() == 'except:') + if bare_excepts > 0: + issues.append(f"{bare_excepts} bare except clauses - should catch specific exceptions") + + todos = sum(1 for l in lines if 'TODO' in l or 'FIXME' in l) + if todos > 0: + issues.append(f"{todos} TODO/FIXME comments") + + long_funcs = [] + in_func = False + func_lines = 0 + func_name = "" + for line in lines: + if line.strip().startswith('def '): + if in_func and func_lines > 50: + long_funcs.append(f"{func_name} ({func_lines} lines)") + in_func = True + func_lines = 0 + func_name = line.strip().split('(')[0].replace('def ', '') + elif in_func: + func_lines += 1 + + if long_funcs: + issues.append(f"Long functions: {', '.join(long_funcs[:3])}") + + return f"""File: {file_path} +Lines: {len(lines)} | Functions: {len(functions)} | Classes: {len(classes)} + +Issues found: +{chr(10).join(f'- {i}' for i in issues) if issues else '- No major issues detected'} + +Suggestions: +- Add type hints for better code clarity +- Add docstrings to undocumented functions +- Consider using logging instead of print statements +- Run a linter (pylint/flake8) for detailed analysis""" + except Exception as e: + return f"Error: {str(e)}" + +# === DOCUMENTATION GENERATION === +def generate_readme(path="."): + """Generate README.md from project structure""" + try: + project_name = os.path.basename(os.path.abspath(path)) + + # Analyze project + files = [] + for root, dirs, filenames in os.walk(path): + # Skip common ignore dirs + dirs[:] = [d for d in dirs if d not in ['.git', 'node_modules', '__pycache__', 'venv']] + for f in filenames: + if f.endswith(('.py', '.js', '.ts', '.go', '.rs')): + files.append(os.path.join(root, f)) + + # Detect language + extensions = [os.path.splitext(f)[1] for f in files] + main_lang = max(set(extensions), key=extensions.count) if extensions else '.txt' + lang_map = {'.py': 'Python', '.js': 'JavaScript', '.ts': 'TypeScript', '.go': 'Go', '.rs': 'Rust'} + language = lang_map.get(main_lang, 'Unknown') + + readme = f"""# {project_name} + +## Description +A {language} project with {len(files)} source files. + +## Project Structure +``` +{chr(10).join([f'- {os.path.relpath(f, path)}' for f in files[:10]])} +{'...' if len(files) > 10 else ''} +``` + +## Installation +```bash +# Add installation instructions here +``` + +## Usage +```bash +# Add usage examples here +``` + +## License +MIT +""" + + readme_path = os.path.join(path, "README.md") + with open(readme_path, 'w') as f: + f.write(readme) + + return f"Generated README.md at {readme_path}" + except Exception as e: + return f"Error generating README: {str(e)}" + +def generate_docstrings(file_path): + """Generate docstrings for functions in a file""" + try: + with open(os.path.expanduser(file_path), 'r') as f: + lines = f.readlines() + + new_lines = [] + i = 0 + while i < len(lines): + line = lines[i] + # Detect function definition + if line.strip().startswith('def ') and ':' in line: + # Check if next line is already a docstring + if i + 1 < len(lines) and '"""' not in lines[i + 1]: + # Extract function name + func_name = line.strip().split('(')[0].replace('def ', '') + indent = len(line) - len(line.lstrip()) + docstring = f'{" " * (indent + 4)}"""TODO: Document {func_name}"""\n' + new_lines.append(line) + new_lines.append(docstring) + i += 1 + continue + new_lines.append(line) + i += 1 + + # Write back + with open(os.path.expanduser(file_path), 'w') as f: + f.writelines(new_lines) + + return f"Added docstring placeholders to {file_path}" + except Exception as e: + return f"Error: {str(e)}" + +# === CODE QUALITY TOOLS === +def run_linter(file_path, linter="auto"): + """Run linter on file""" + try: + file_path = os.path.expanduser(file_path) + ext = os.path.splitext(file_path)[1] + + # Auto-detect linter + if linter == "auto": + linter_map = { + '.py': 'pylint', + '.js': 'eslint', + '.ts': 'eslint', + '.go': 'golint', + '.rs': 'cargo clippy' + } + linter = linter_map.get(ext, 'pylint') + + # Run linter + result = subprocess.run( + f"{linter} {file_path}", + shell=True, capture_output=True, text=True, timeout=30 + ) + + return result.stdout + result.stderr + except Exception as e: + return f"Error running linter: {str(e)}" + +def format_code(file_path, formatter="auto"): + """Format code file""" + try: + file_path = os.path.expanduser(file_path) + ext = os.path.splitext(file_path)[1] + + # Auto-detect formatter + if formatter == "auto": + formatter_map = { + '.py': 'black', + '.js': 'prettier --write', + '.ts': 'prettier --write', + '.go': 'gofmt -w', + '.rs': 'rustfmt' + } + formatter = formatter_map.get(ext, 'cat') + + # Run formatter + result = subprocess.run( + f"{formatter} {file_path}", + shell=True, capture_output=True, text=True, timeout=30 + ) + + return f"Formatted {file_path}\n{result.stdout + result.stderr}" + except Exception as e: + return f"Error formatting: {str(e)}" + +def check_security(path="."): + """Run security checks""" + try: + # Try bandit for Python + result = subprocess.run( + f"bandit -r {path} 2>&1 || echo 'Install: pip install bandit'", + shell=True, capture_output=True, text=True, timeout=60 + ) + return result.stdout + except Exception as e: + return f"Error: {str(e)}" + +# === PROJECT TEMPLATES === +def create_project(name, template="python"): + """Create new project from template""" + try: + project_path = os.path.expanduser(f"./{name}") + os.makedirs(project_path, exist_ok=True) + + templates = { + "python": { + "main.py": "#!/usr/bin/env python3\n\ndef main():\n print('Hello, World!')\n\nif __name__ == '__main__':\n main()\n", + "requirements.txt": "# Add dependencies here\n", + "README.md": f"# {name}\n\nA Python project.\n", + ".gitignore": "__pycache__/\n*.pyc\nvenv/\n.env\n" + }, + "javascript": { + "index.js": "console.log('Hello, World!');\n", + "package.json": f'{{"name": "{name}", "version": "1.0.0", "main": "index.js"}}', + "README.md": f"# {name}\n\nA JavaScript project.\n", + ".gitignore": "node_modules/\n.env\n" + }, + "flask": { + "app.py": "from flask import Flask\n\napp = Flask(__name__)\n\n@app.route('/')\ndef hello():\n return 'Hello, World!'\n\nif __name__ == '__main__':\n app.run(debug=True)\n", + "requirements.txt": "flask\n", + "README.md": f"# {name}\n\nA Flask web application.\n", + ".gitignore": "__pycache__/\nvenv/\n*.pyc\n" + }, + "express": { + "server.js": "const express = require('express');\nconst app = express();\n\napp.get('/', (req, res) => {\n res.send('Hello, World!');\n});\n\napp.listen(3000, () => console.log('Server running on port 3000'));\n", + "package.json": f'{{"name": "{name}", "version": "1.0.0", "main": "server.js", "dependencies": {{"express": "^4.18.0"}}}}', + "README.md": f"# {name}\n\nAn Express.js application.\n", + ".gitignore": "node_modules/\n.env\n" + } + } + + if template not in templates: + return f"Unknown template: {template}. Available: {', '.join(templates.keys())}" + + # Create files + for filename, content in templates[template].items(): + filepath = os.path.join(project_path, filename) + with open(filepath, 'w') as f: + f.write(content) + + # Initialize git + subprocess.run("git init", shell=True, cwd=project_path, capture_output=True) + + return f"Created {template} project at {project_path}\nFiles: {', '.join(templates[template].keys())}" + except Exception as e: + return f"Error creating project: {str(e)}" + +# Model selection based on task type +MODEL_PREFERENCES = { + "code": ["qwen3-coder:30b", "mistral:latest", "llama3:latest", "deepseek-coder-v2:16b", "granite-code:latest"], + "general": ["mistral:latest", "llama3:latest", "llama3.2:latest"], + "fast": ["llama3.2:3b", "llama3.2:latest"], +} + +def get_available_models(): + """Get list of models available in Ollama""" + config = load_config() + excluded = config.get("excluded_models", []) + + try: + result = subprocess.run("ollama list", shell=True, capture_output=True, text=True, timeout=5) + models = [] + for line in result.stdout.split('\n')[1:]: # Skip header + if line.strip(): + model_name = line.split()[0] + # Skip base models, tiny models, and excluded models + if ('base' not in model_name.lower() and + ':1b' not in model_name.lower() and + model_name not in excluded): + models.append(model_name) + return models + except: + return [] + +def select_best_model(user_input, task_type="initial"): + """Select best model based on user input and task type""" + if not AVAILABLE_MODELS: + return MODEL + + # For tool execution, prefer fast general models + if task_type == "tool_response": + for model in MODEL_PREFERENCES["general"]: + if any(model in m for m in AVAILABLE_MODELS): + return next(m for m in AVAILABLE_MODELS if model in m) + + # Detect task type from user input + code_keywords = ['code', 'script', 'program', 'function', 'debug', 'python', 'javascript', 'write a', 'create a'] + + if any(keyword in user_input.lower() for keyword in code_keywords): + # Prefer code models + for model in MODEL_PREFERENCES["code"]: + if any(model in m for m in AVAILABLE_MODELS): + return next(m for m in AVAILABLE_MODELS if model in m) + + # Default to general models + for model in MODEL_PREFERENCES["general"]: + if any(model in m for m in AVAILABLE_MODELS): + return next(m for m in AVAILABLE_MODELS if model in m) + + # Fallback to first available + return AVAILABLE_MODELS[0] if AVAILABLE_MODELS else MODEL + +def execute_bash(command): + """Execute bash command and return output""" + # Check for dangerous commands + dangerous = ['rm -rf', 'dd if=', 'mkfs', '> /dev/', 'format', 'sudo rm', ':(){:|:&};:'] + if any(d in command.lower() for d in dangerous): + return f"⚠️ DANGEROUS COMMAND BLOCKED: {command}\nThis command could cause data loss. If you really need to run it, do so manually." + + try: + result = subprocess.run(command, shell=True, capture_output=True, text=True, + cwd=os.getcwd(), timeout=30) + output = result.stdout + result.stderr + return f"Exit code: {result.returncode}\n{output}" if output else f"Exit code: {result.returncode} (no output)" + except subprocess.TimeoutExpired: + return "Error: Command timed out after 30 seconds" + except Exception as e: + return f"Error: {str(e)}" + +def read_file(path): + """Read file contents""" + try: + with open(os.path.expanduser(path), 'r') as f: + return f.read() + except Exception as e: + return f"Error reading file: {str(e)}" + +def write_file(path, content): + """Write content to file""" + try: + expanded_path = os.path.expanduser(path) + + # For Python files, try to validate and fix basic issues + if path.endswith('.py'): + # Try to compile to check for syntax errors + try: + compile(content, expanded_path, 'exec') + except SyntaxError as e: + return f"Error: Generated Python code has syntax error at line {e.lineno}: {e.msg}" + + os.makedirs(os.path.dirname(expanded_path), exist_ok=True) + with open(expanded_path, 'w') as f: + f.write(content) + + # Track created file + SESSION_CONTEXT["created_files"].append(expanded_path) + + return f"Successfully written to {path}" + except Exception as e: + return f"Error writing file: {str(e)}" + +def edit_file(path, old_str, new_str): + """Edit file by replacing old_str with new_str""" + try: + expanded_path = os.path.expanduser(path) + with open(expanded_path, 'r') as f: + content = f.read() + + if old_str not in content: + return f"Error: Could not find the specified text in {path}" + + new_content = content.replace(old_str, new_str, 1) + + # Validate Python syntax if applicable + if path.endswith('.py'): + try: + compile(new_content, expanded_path, 'exec') + except SyntaxError as e: + return f"Error: Edited code has syntax error at line {e.lineno}: {e.msg}" + + with open(expanded_path, 'w') as f: + f.write(new_content) + return f"Successfully edited {path}" + except Exception as e: + return f"Error editing file: {str(e)}" + +def append_file(path, content): + """Append content to file""" + try: + expanded_path = os.path.expanduser(path) + with open(expanded_path, 'a') as f: + f.write(content) + return f"Successfully appended to {path}" + except Exception as e: + return f"Error appending to file: {str(e)}" + +def search_files(pattern, path="."): + """Search for pattern in files""" + try: + result = subprocess.run( + f'grep -r "{pattern}" {path}', + shell=True, capture_output=True, text=True, timeout=10 + ) + return result.stdout if result.stdout else "No matches found" + except Exception as e: + return f"Error searching: {str(e)}" + +def git_operation(operation, message=None, **kwargs): + """Perform git operations""" + config = load_config() + git_config = config.get("git", {}) + gitea_config = config.get("gitea", {}) + + try: + if operation == "status": + result = subprocess.run("git status", shell=True, capture_output=True, text=True) + elif operation == "diff": + result = subprocess.run("git diff", shell=True, capture_output=True, text=True) + elif operation == "commit": + msg = message or kwargs.get("message", "Auto commit") + result = subprocess.run(f'git add -A && git commit -m "{msg}"', + shell=True, capture_output=True, text=True) + elif operation == "push": + remote = git_config.get("remote_url", "origin") + username = git_config.get("username", "") + password = git_config.get("password", "") + + if username and password: + # Inject credentials into URL + result = subprocess.run(f"git push {remote}", + shell=True, capture_output=True, text=True) + else: + result = subprocess.run("git push", shell=True, capture_output=True, text=True) + elif operation == "pull": + result = subprocess.run("git pull", shell=True, capture_output=True, text=True) + elif operation == "list": + # Check if Gitea is configured + if gitea_config.get("url") and gitea_config.get("token"): + # Query Gitea API + gitea_url = gitea_config["url"].rstrip("/") + token = gitea_config["token"] + + try: + response = requests.get( + f"{gitea_url}/api/v1/user/repos", + headers={"Authorization": f"token {token}"}, + timeout=10, + verify=False # Skip SSL verification for localhost + ) + + if response.status_code == 200: + repos = response.json() + result_text = "Repositories on Gitea server:\n" + for repo in repos: + result_text += f"- {repo['full_name']} ({repo['clone_url']})\n" + return result_text + else: + return f"Gitea API error: {response.status_code}" + except Exception as e: + return f"Error querying Gitea: {str(e)}" + else: + # Fallback to local search + result = subprocess.run("find . -maxdepth 2 -name '.git' -type d | sed 's|/.git||'", + shell=True, capture_output=True, text=True) + else: + return f"Unknown git operation: {operation}" + + return result.stdout + result.stderr + except Exception as e: + return f"Git error: {str(e)}" + +def install_package(package, manager="pip"): + """Install package using pip or npm""" + try: + if manager == "pip": + result = subprocess.run(f"pip install {package}", + shell=True, capture_output=True, text=True, timeout=60) + elif manager == "npm": + result = subprocess.run(f"npm install {package}", + shell=True, capture_output=True, text=True, timeout=60) + else: + return f"Unknown package manager: {manager}" + + return result.stdout + result.stderr + except Exception as e: + return f"Error installing package: {str(e)}" + +def web_search(query): + """Simple web search using curl""" + try: + import urllib.parse + encoded_query = urllib.parse.quote(query) + result = subprocess.run( + f'curl -s "https://html.duckduckgo.com/html/?q={encoded_query}"', + shell=True, capture_output=True, text=True, timeout=10 + ) + # Extract snippets (very basic) + content = result.stdout[:3000] + return f"Search results for '{query}':\n{content}" + except Exception as e: + return f"Error searching: {str(e)}" + +def list_directory(path="."): + """List directory contents""" + try: + expanded = os.path.expanduser(path) + files = os.listdir(expanded) + return "\n".join(files) if files else "Directory is empty" + except Exception as e: + return f"Error listing directory: {str(e)}" + +def get_file_info(path): + """Get file metadata""" + try: + p = os.path.expanduser(path) + stat = os.stat(p) + import time + return f"Path: {p}\nSize: {stat.st_size} bytes\nModified: {time.ctime(stat.st_mtime)}\nPermissions: {oct(stat.st_mode)}" + except Exception as e: + return f"Error getting file info: {str(e)}" + +TOOLS = { + "execute_bash": execute_bash, + "read_file": read_file, + "write_file": write_file, + "edit_file": edit_file, + "append_file": append_file, + "web_search": web_search, + "web_search_enhanced": web_search_enhanced, + "list_directory": list_directory, + "get_file_info": get_file_info, + "search_files": search_files, + "git_operation": git_operation, + "install_package": install_package, + "lsp_goto_definition": lsp_goto_definition, + "lsp_find_references": lsp_find_references, + "kb_add": kb_add, + "kb_search": kb_search, + "kb_get": kb_get, + "spawn_subagent": spawn_subagent, + "check_subagents": check_subagents, + "code_search_symbol": code_search_symbol, + "code_analyze_file": code_analyze_file, + "generate_readme": generate_readme, + "generate_docstrings": generate_docstrings, + "run_linter": run_linter, + "format_code": format_code, + "check_security": check_security, + "create_project": create_project, +} + +SYSTEM_PROMPT = """You are a helpful AI assistant with access to tools. + +WHEN TO USE TOOLS: +- User says "create a FILE/SCRIPT" → use write_file +- User says "edit/modify FILE" → use edit_file +- User says "add to FILE" → use append_file +- User asks to READ a file → use read_file +- User asks to RUN/EXECUTE a command → use execute_bash +- User asks to LIST files/directories → use list_directory +- User asks to LIST GIT PROJECTS/REPOS → use git_operation with operation="list" +- User asks to SEARCH in files → use search_files +- User asks about GIT (status, commit, push, pull) → use git_operation +- User asks to INSTALL package → use install_package +- User asks to SEARCH the web → use web_search_enhanced +- User asks about code definitions/references → use lsp_goto_definition, lsp_find_references +- User wants to save/retrieve knowledge → use kb_add, kb_search, kb_get +- User has complex multi-step task → use spawn_subagent +- User wants code analysis → use code_search_symbol, code_analyze_file +- User wants documentation → use generate_readme, generate_docstrings +- User wants code quality checks → use run_linter, format_code, check_security +- User wants to create new project → use create_project + +WHEN NOT TO USE TOOLS (just answer normally): +- "how do I make FOOD" → answer with recipe +- "how to do X" → answer with instructions +- "what is X" → answer with explanation +- General questions about cooking, advice, information + +Available tools: +- write_file, edit_file, append_file, read_file +- execute_bash, list_directory, get_file_info, search_files +- git_operation, install_package +- web_search_enhanced: Better web search with API +- lsp_goto_definition: Find symbol definition. Params: {"language": "python", "file_path": "file.py", "line": 0, "character": 0} +- lsp_find_references: Find all references. Params: {"language": "python", "file_path": "file.py", "line": 0, "character": 0} +- kb_add: Add to knowledge base. Params: {"key": "string", "content": "string"} +- kb_search: Search knowledge base. Params: {"query": "string"} +- kb_get: Get from knowledge base. Params: {"key": "string"} +- spawn_subagent: Spawn parallel task. Params: {"task": "string", "model": "optional"} +- check_subagents: Check subagent status +- code_search_symbol: Find symbol in code. Params: {"symbol_name": "string", "path": "."} +- code_analyze_file: Analyze code structure. Params: {"file_path": "string"} +- generate_readme: Generate README.md. Params: {"path": "."} +- generate_docstrings: Add docstrings to file. Params: {"file_path": "string"} +- run_linter: Run linter. Params: {"file_path": "string", "linter": "auto|pylint|eslint"} +- format_code: Format code. Params: {"file_path": "string", "formatter": "auto|black|prettier"} +- check_security: Security scan. Params: {"path": "."} +- create_project: Create project from template. Params: {"name": "string", "template": "python|javascript|flask|express"} + +MCP Tools: {mcp_tools_list} + +Tool call format: + +write_file +{"path": "~/file.py", "content": "code here"} + + +Output ONLY the tool call when using a tool. Use \\n for newlines, proper indentation (4 spaces for Python).""" + +def extract_code_blocks(text): + """Extract code blocks from markdown""" + pattern = r'```(\w+)?\n(.*?)```' + matches = re.findall(pattern, text, re.DOTALL) + return [(lang, code.strip()) for lang, code in matches] + +def auto_save_code(response, user_input): + """Automatically save code blocks if user asked to create a file""" + create_keywords = ['create', 'write', 'make', 'generate', 'build'] + file_keywords = ['script', 'file', 'program', '.py', '.js', '.html', '.sh'] + + should_save = any(k in user_input.lower() for k in create_keywords) and \ + any(k in user_input.lower() for k in file_keywords) + + if not should_save: + return None + + code_blocks = extract_code_blocks(response) + if not code_blocks: + return None + + # Get the largest code block (likely the main code) + lang, code = max(code_blocks, key=lambda x: len(x[1])) + + if not code: + return None + + # Determine filename from language or user input + extensions = { + 'python': '.py', + 'javascript': '.js', + 'html': '.html', + 'bash': '.sh', + 'sh': '.sh', + } + + # Try to extract filename from user input + words = user_input.lower().split() + filename = None + for i, word in enumerate(words): + if any(ext in word for ext in ['.py', '.js', '.html', '.sh']): + filename = word + break + if word in ['called', 'named'] and i + 1 < len(words): + filename = words[i + 1] + break + + if not filename: + # Generate filename based on language + ext = extensions.get(lang.lower(), '.txt') + if 'calculator' in user_input.lower(): + filename = f'calculator{ext}' + elif 'script' in user_input.lower(): + filename = f'script{ext}' + else: + filename = f'output{ext}' + + if not filename: + return None + + return filename, code + +def parse_tool_calls(text): + """Extract tool calls from model response""" + calls = [] + + # Try standard format first + pattern = r'\s*(.*?)\s*(.*?)\s*' + matches = re.findall(pattern, text, re.DOTALL) + + # Try DeepSeek format: <｜tool▁call▁begin｜>function<｜tool▁sep｜>TOOL_NAME\n```json\n{...}\n``` + if not matches: + ds_pattern = r'<｜tool▁call▁begin｜>function<｜tool▁sep｜>(\w+)\s*```json\s*(\{.*?\})\s*```' + ds_matches = re.findall(ds_pattern, text, re.DOTALL) + if ds_matches: + matches = ds_matches + + for name, params in matches: + params_str = params.strip() + name = name.strip() + + # Try JSON first + try: + calls.append((name, json.loads(params_str))) + continue + except json.JSONDecodeError: + pass + + # Try ast.literal_eval (handles Python literals) + try: + import ast + calls.append((name, ast.literal_eval(params_str))) + continue + except: + pass + + # Try fixing common issues: literal newlines in JSON strings + try: + # Replace literal newlines with \n escape sequence + fixed = params_str.replace('\n', '\\n').replace('\r', '\\r').replace('\t', '\\t') + calls.append((name, json.loads(fixed))) + continue + except: + pass + + print(f"Warning: Failed to parse parameters for {name}") + print(f"Raw params: {params_str[:100]}...") + + return calls + +def call_ollama(messages, model=None): + """Call Ollama API with streaming""" + config = load_config() + ollama_url = config.get("ollama_url", OLLAMA_URL) + selected_model = model or MODEL + try: + response = requests.post(ollama_url, json={ + "model": selected_model, + "messages": messages, + "stream": True + }, stream=True, timeout=60) + response.raise_for_status() + + full_response = "" + for line in response.iter_lines(): + if line: + chunk = json.loads(line) + if "message" in chunk and "content" in chunk["message"]: + content = chunk["message"]["content"] + print(content, end="", flush=True) + full_response += content + print() # Newline after streaming + return full_response + except requests.exceptions.ConnectionError: + print("\nError: Cannot connect to Ollama. Make sure it's running with 'ollama serve'") + sys.exit(1) + except Exception as e: + return f"Error calling Ollama: {str(e)}" + +def trim_messages(messages, max_messages=20): + """Keep only recent messages to avoid context overflow""" + if len(messages) > max_messages: + # Keep system prompt + last (max_messages - 1) messages + return [messages[0]] + messages[-(max_messages-1):] + return messages + +def main(): + global AVAILABLE_MODELS, CURRENT_SESSION, MODEL + AVAILABLE_MODELS = get_available_models() + + print(f"Ollama CLI v2.0 - Default model: {MODEL}") + + config = load_config() + ollama_url = config.get("ollama_url", OLLAMA_URL) + print(f"Server: {ollama_url}") + + if AVAILABLE_MODELS: + print(f"Available models: {', '.join(AVAILABLE_MODELS)}") + else: + print("Warning: Could not detect available models, using default") + + config = load_config() + if config.get("excluded_models"): + print(f"Excluded models: {', '.join(config['excluded_models'])}") + + print(f"Current directory: {os.getcwd()}") + print(f"Home directory: {os.path.expanduser('~')}") + print("\nCommands:") + print(" /quit - Exit") + print(" /clear - Clear history") + print(" /save [name] - Save session") + print(" /load - Load session") + print(" /sessions - List sessions") + print(" /model - Switch model") + print(" /models - List models") + print(" /exclude - Exclude model") + print(" /retry - Retry last request") + print(" /run - Execute file") + print(" /config git - Configure git") + print(" /config gitea - Configure Gitea server") + print(" /export - Export to markdown") + print(" /mcp connect [args] - Connect MCP server") + print(" /mcp list - List MCP servers and tools") + print(" /mcp disconnect - Disconnect MCP server") + print(" /context - Add file to context") + print(" /context list - List context files") + print(" /context clear - Clear context files") + print(" /lsp start [path] - Start LSP server") + print(" /kb add - Add to knowledge base") + print(" /kb search - Search knowledge base") + print(" /agents - Check subagent status") + print(" /server - Set Ollama server URL") + print(" /notify setup - Enable ntfy.sh notifications") + print(" /notify off - Disable notifications") + print(" /notify test - Send test notification") + print("\nWhen notifications are enabled, send messages to your topic to interact remotely!\n") + + # Add system context to initial prompt + system_context = f"""Operating System: {sys.platform} +Current Directory: {os.getcwd()} +Home Directory: {os.path.expanduser('~')} +User: {os.environ.get('USER', 'unknown')} + +{SYSTEM_PROMPT}""" + + # Add MCP tools to system prompt + if MCP_TOOLS: + mcp_tools_desc = "\n".join([f"- {name}: {tool['description']}" for name, tool in MCP_TOOLS.items()]) + system_context = system_context.replace("{mcp_tools_list}", mcp_tools_desc) + else: + system_context = system_context.replace("{mcp_tools_list}", "None connected") + + messages = [{"role": "system", "content": system_context}] + + # Auto-connect MCP servers from config + config = load_config() + for name, server_config in config.get("mcp_servers", {}).items(): + result = connect_mcp_server(name, server_config["command"], server_config.get("args")) + print(f"[{result}]") + + # Start ntfy listener if enabled + if config.get("ntfy", {}).get("enabled"): + topic = config["ntfy"]["topic"] + print(f"[Starting ntfy listener for topic: {topic}]") + listener_thread = threading.Thread(target=ntfy_listener, args=(messages,), daemon=True) + listener_thread.start() + + while True: + try: + user_input = input("\n> ").strip() + except (EOFError, KeyboardInterrupt): + print("\nGoodbye!") + break + + if not user_input: + continue + + if user_input.lower() in ['/quit', 'exit']: + print("Goodbye!") + break + + if user_input.lower() == '/clear': + system_context = f"""Operating System: {sys.platform} +Current Directory: {os.getcwd()} +Home Directory: {os.path.expanduser('~')} +User: {os.environ.get('USER', 'unknown')} + +{SYSTEM_PROMPT}""" + messages = [{"role": "system", "content": system_context}] + print("History cleared.") + continue + + if user_input.lower() == '/save': + try: + with open('chat_history.json', 'w') as f: + json.dump(messages, f, indent=2) + print("History saved to chat_history.json") + except Exception as e: + print(f"Error saving history: {e}") + continue + + if user_input.lower().startswith('/exclude '): + model_to_exclude = user_input[9:].strip() + config = load_config() + if model_to_exclude not in config["excluded_models"]: + config["excluded_models"].append(model_to_exclude) + save_config(config) + print(f"Excluded {model_to_exclude}. Restart to apply changes.") + else: + print(f"{model_to_exclude} is already excluded.") + continue + + if user_input.lower() == '/list-excluded': + config = load_config() + excluded = config.get("excluded_models", []) + if excluded: + print(f"Excluded models: {', '.join(excluded)}") + else: + print("No models excluded.") + continue + + if user_input.lower().startswith('/save'): + parts = user_input.split(maxsplit=1) + session_name = parts[1] if len(parts) > 1 else None + session_file = save_session(messages, session_name) + print(f"Session saved to {session_file}") + continue + + if user_input.lower().startswith('/load '): + session_name = user_input[6:].strip() + loaded = load_session(session_name) + if loaded: + messages = loaded + print(f"Session '{session_name}' loaded.") + else: + print(f"Session '{session_name}' not found.") + continue + + if user_input.lower() == '/sessions': + sessions = list_sessions() + if sessions: + print("Available sessions:") + for s in sessions[:10]: # Show last 10 + print(f" {s}") + else: + print("No saved sessions.") + continue + + if user_input.lower().startswith('/model '): + new_model = user_input[7:].strip() + if new_model in AVAILABLE_MODELS: + MODEL = new_model + print(f"Switched to model: {MODEL}") + else: + print(f"Model '{new_model}' not available.") + continue + + if user_input.lower() == '/models': + print("Available models:") + for m in AVAILABLE_MODELS: + print(f" {m}") + continue + + if user_input.lower() == '/retry': + if len(messages) > 1: + # Remove last assistant response and retry + if messages[-1]["role"] == "assistant": + messages.pop() + user_input = messages[-1]["content"] if messages[-1]["role"] == "user" else "" + if not user_input: + print("Nothing to retry.") + continue + messages.pop() # Remove user message, will be re-added + else: + print("Nothing to retry.") + continue + + if user_input.lower().startswith('/run '): + filepath = user_input[5:].strip() + result = execute_bash(f"python3 {filepath}" if filepath.endswith('.py') else filepath) + print(result) + continue + + if user_input.lower().startswith('/config git '): + parts = user_input[12:].split() + if len(parts) >= 3: + config = load_config() + config["git"] = { + "remote_url": parts[0], + "username": parts[1], + "password": parts[2] + } + save_config(config) + print("Git configuration saved.") + else: + print("Usage: /config git ") + continue + + if user_input.lower().startswith('/config gitea '): + parts = user_input[14:].split() + if len(parts) >= 2: + config = load_config() + config["gitea"] = { + "url": parts[0], + "token": parts[1] + } + save_config(config) + print("Gitea configuration saved.") + else: + print("Usage: /config gitea ") + continue + + if user_input.lower().startswith('/export '): + filename = user_input[8:].strip() + try: + with open(filename, 'w') as f: + f.write("# Ollama CLI Conversation\n\n") + for msg in messages[1:]: # Skip system prompt + role = msg["role"].upper() + content = msg["content"] + f.write(f"## {role}\n\n{content}\n\n") + print(f"Conversation exported to {filename}") + except Exception as e: + print(f"Error exporting: {e}") + continue + + if user_input.lower().startswith('/mcp connect '): + parts = user_input[13:].split() + if len(parts) >= 2: + name = parts[0] + command = parts[1] + args = parts[2:] if len(parts) > 2 else None + result = connect_mcp_server(name, command, args) + print(result) + # Save to config + config = load_config() + config["mcp_servers"][name] = {"command": command, "args": args} + save_config(config) + else: + print("Usage: /mcp connect [args...]") + continue + + if user_input.lower() == '/mcp list': + if MCP_SERVERS: + print("Connected MCP servers:") + for server in MCP_SERVERS: + tools = [t for t in MCP_TOOLS if t.startswith(f"mcp_{server}_")] + print(f" {server}: {len(tools)} tools") + for tool_key in tools[:5]: # Show first 5 + tool = MCP_TOOLS[tool_key] + print(f" - {tool['name']}: {tool['description'][:60]}") + else: + print("No MCP servers connected.") + continue + + if user_input.lower().startswith('/mcp disconnect '): + name = user_input[16:].strip() + if name in MCP_SERVERS: + MCP_SERVERS[name].terminate() + del MCP_SERVERS[name] + # Remove tools + MCP_TOOLS.clear() + for key in list(MCP_TOOLS.keys()): + if key.startswith(f"mcp_{name}_"): + del MCP_TOOLS[key] + print(f"Disconnected from MCP server '{name}'") + else: + print(f"MCP server '{name}' not connected.") + continue + + if user_input.lower().startswith('/context '): + arg = user_input[9:].strip() + + if arg == 'list': + if SESSION_CONTEXT["context_files"]: + print("Context files:") + for cf in SESSION_CONTEXT["context_files"]: + print(f" {cf['path']} ({len(cf['content'])} chars)") + else: + print("No context files loaded.") + continue + + if arg == 'clear': + SESSION_CONTEXT["context_files"].clear() + print("Context files cleared.") + continue + + # Load file into context + filepath = os.path.expanduser(arg) + try: + with open(filepath, 'r') as f: + content = f.read() + SESSION_CONTEXT["context_files"].append({ + "path": filepath, + "content": content + }) + print(f"Added {filepath} to context ({len(content)} chars)") + except Exception as e: + print(f"Error loading context file: {e}") + continue + + if user_input.lower().startswith('/lsp start '): + parts = user_input[11:].split() + if len(parts) >= 1: + language = parts[0] + path = parts[1] if len(parts) > 1 else "." + result = start_lsp_server(language, path) + print(result) + else: + print("Usage: /lsp start [path]") + continue + + if user_input.lower().startswith('/kb add '): + parts = user_input[8:].split(maxsplit=1) + if len(parts) == 2: + key, content = parts + result = kb_add(key, content) + print(result) + else: + print("Usage: /kb add ") + continue + + if user_input.lower().startswith('/kb search '): + query = user_input[11:].strip() + result = kb_search(query) + print(result) + continue + + if user_input.lower() == '/agents': + result = check_subagents() + print(result) + continue + + if user_input.lower().startswith('/server '): + url = user_input[8:].strip() + if not url.startswith('http'): + url = f"http://{url}" + if '/api/chat' not in url: + url = f"{url}/api/chat" + + config = load_config() + config["ollama_url"] = url + save_config(config) + print(f"Ollama server set to: {url}") + print("Restart to apply changes.") + continue + + if user_input.lower().startswith('/notify '): + parts = user_input[8:].split(maxsplit=1) + if len(parts) >= 1: + cmd = parts[0] + if cmd == 'setup' and len(parts) == 2: + topic = parts[1].strip() + config = load_config() + config["ntfy"] = { + "enabled": True, + "topic": topic, + "server": "https://ntfy.sh" + } + save_config(config) + print(f"Notifications enabled for topic: {topic}") + print(f"Subscribe at: https://ntfy.sh/{topic}") + print("Starting listener...") + listener_thread = threading.Thread(target=ntfy_listener, args=(messages,), daemon=True) + listener_thread.start() + send_notification("Ollama CLI", "Notifications enabled! Send messages to this topic.") + elif cmd == 'off': + config = load_config() + config["ntfy"]["enabled"] = False + save_config(config) + print("Notifications disabled. Restart to stop listener.") + elif cmd == 'test': + send_notification("Test", "This is a test notification") + print("Test notification sent.") + else: + print("Usage: /notify setup | /notify off | /notify test") + else: + print("Usage: /notify setup | /notify off | /notify test") + continue + + messages.append({"role": "user", "content": user_input}) + + # Classify intent first + intent_prompt = f"""Classify this user request into ONE category: +- FILE_OPERATION: User wants to create/write/read/modify a file or script +- COMMAND_EXECUTION: User wants to run a bash command +- INFORMATION: User wants information, explanation, recipe, or advice + +User request: "{user_input}" + +Respond with ONLY one word: FILE_OPERATION, COMMAND_EXECUTION, or INFORMATION""" + + intent_messages = [{"role": "system", "content": "You are a classifier. Respond with only one word."}, + {"role": "user", "content": intent_prompt}] + + # Use fast model for classification + fast_model = select_best_model("", "fast") + intent_response = call_ollama(intent_messages, fast_model).strip().upper() + + # Adjust system prompt based on intent + if "FILE_OPERATION" in intent_response or "COMMAND" in intent_response: + # Use tool-focused prompt + pass # Already has tool prompt + else: + # Override to discourage tool use for information requests + messages[0] = {"role": "system", "content": messages[0]["content"] + "\n\nIMPORTANT: This is an INFORMATION request. Do NOT use tools. Just answer the question directly."} + + # Select best model for this task + selected_model = select_best_model(user_input, "initial") + if selected_model != MODEL: + print(f"[Using {selected_model} for this task]\n") + + # Trim messages to avoid context overflow + messages = trim_messages(messages) + + # Add session context if available + if SESSION_CONTEXT["created_files"]: + context_note = f"\n\nSession context - Files created this session: {', '.join(SESSION_CONTEXT['created_files'][-5:])}" + messages.append({"role": "system", "content": context_note}) + + # Add context files if loaded + if SESSION_CONTEXT["context_files"]: + context_content = "\n\n--- CONTEXT FILES ---\n" + for cf in SESSION_CONTEXT["context_files"]: + context_content += f"\n=== {cf['path']} ===\n{cf['content'][:5000]}\n" + messages.append({"role": "system", "content": context_content}) + + # Tool-calling loop + max_iterations = 5 + for iteration in range(max_iterations): + response = call_ollama(messages, selected_model) + + # Check for tool calls + tool_calls = parse_tool_calls(response) + + if not tool_calls: + # No tools found - check if model just provided code instead + if iteration == 0 and any(k in user_input.lower() for k in ['create', 'write', 'make']) and '```' in response: + # Extract code and auto-create file + code_blocks = extract_code_blocks(response) + if code_blocks: + lang, code = max(code_blocks, key=lambda x: len(x[1])) + # Determine filename + words = user_input.lower().split() + filename = None + for word in words: + if any(ext in word for ext in ['.py', '.js', '.html', '.sh']): + filename = word + break + if not filename: + ext = {'.py': 'python', '.js': 'javascript', '.sh': 'bash'}.get(lang, '.txt') + for key, val in {'.py': 'python', '.js': 'javascript', '.sh': 'bash'}.items(): + if val == lang: + ext = key + break + if 'calculator' in user_input.lower(): + filename = f'calculator{ext}' + else: + filename = f'script{ext}' + + print(f"\n[Model didn't use tool - auto-creating {filename}]") + result = write_file(filename, code) + print(f"[{result}]") + messages.append({"role": "assistant", "content": response}) + messages.append({"role": "user", "content": f"Tool: write_file\nResult:\n{result}"}) + continue + + # No tools needed, show response + if '' in response or any(tool in response for tool in TOOLS.keys()): + print(f"\n[Warning: Model attempted tool use but format was incorrect]") + + messages.append({"role": "assistant", "content": response}) + + # Remove the old auto-save logic since we handle it above now + break + + # Execute tools + print() # Newline before tool execution + tool_results = [] + for tool_name, params in tool_calls: + # Check if it's an MCP tool + if tool_name.startswith("mcp_") and tool_name in MCP_TOOLS: + mcp_tool = MCP_TOOLS[tool_name] + server_name = mcp_tool["server"] + actual_tool_name = mcp_tool["name"] + print(f"[Executing MCP tool: {actual_tool_name} on {server_name}]") + result = call_mcp_tool(server_name, actual_tool_name, params) + print(f"[Result: {result[:200]}{'...' if len(result) > 200 else ''}]") + tool_results.append(f"Tool: {tool_name}\nResult:\n{result}") + elif tool_name in TOOLS: + print(f"[Executing: {tool_name}({json.dumps(params)[:100]}...)]") + result = TOOLS[tool_name](**params) + # Show more for git list operations + display_limit = 1000 if tool_name == "git_operation" else 200 + print(f"[Result: {result[:display_limit]}{'...' if len(result) > display_limit else ''}]") + tool_results.append(f"Tool: {tool_name}\nResult:\n{result}") + + # Send notifications for key events + if tool_name == "write_file" and "Successfully" in result: + send_notification("File Created", f"Created: {params.get('path', 'file')}") + elif tool_name == "execute_bash" and "Exit code: 0" in result: + send_notification("Command Complete", f"Executed: {params.get('command', 'command')[:50]}") + else: + tool_results.append(f"Error: Unknown tool '{tool_name}'") + + # Add assistant message and tool results + messages.append({"role": "assistant", "content": response}) + + # Check if there was a syntax error + has_syntax_error = any('syntax error' in r.lower() for r in tool_results) + + # For successful file creation, confirm and stop + if any('write_file' in r and 'Successfully' in r for r in tool_results): + print("\nFile created successfully.") + + # Check if it's an executable script and offer to run it + created_files = [params.get('path') for name, params in tool_calls if name == 'write_file'] + for filepath in created_files: + if filepath and (filepath.endswith('.py') or filepath.endswith('.sh')): + try: + response = input(f"\nRun {filepath}? (y/n): ").strip().lower() + if response == 'y': + print(f"\n[Running {filepath}...]") + result = execute_bash(f"python3 {filepath}" if filepath.endswith('.py') else f"bash {filepath}") + print(result) + except: + pass + + messages.append({"role": "user", "content": "\n\n".join(tool_results)}) + break + + # For other successful tool executions, show result and stop (don't loop) + if not has_syntax_error and iteration == 0: + print() # Newline after tool results + messages.append({"role": "user", "content": "\n\n".join(tool_results)}) + break + + # If syntax error, let model see it and try to fix + if has_syntax_error: + print("\n[Syntax error detected - asking model to fix]") + messages.append({"role": "user", "content": "\n\n".join(tool_results) + "\n\nPlease fix the syntax error and try again."}) + else: + messages.append({"role": "user", "content": "\n\n".join(tool_results)}) + + # Switch to a general model for interpreting tool results + response_model = select_best_model("", "tool_response") + if response_model != selected_model: + print(f"[Switching to {response_model} for response]\n") + selected_model = response_model + + # If last iteration, get final response + if iteration == max_iterations - 1: + final_response = call_ollama(messages, selected_model) + messages.append({"role": "assistant", "content": final_response}) + + # Auto-save session after each interaction + config = load_config() + if config.get("auto_save_sessions", True): + save_session(messages, "autosave") + +if __name__ == "__main__": + main()