diff --git a/.gitignore b/.gitignore index 5d381cc..ed95aad 100644 --- a/.gitignore +++ b/.gitignore @@ -160,3 +160,8 @@ cython_debug/ # option (not recommended) you can uncomment the following to ignore the entire idea folder. #.idea/ +# Project specific +config.json +*.jsonl +example_session.jsonl + diff --git a/README.md b/README.md index df9a72f..139d7a9 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,164 @@ -# picogent-py +# PicoGent - Minimal AI Coding Agent -Minimal AI coding agent in Python with minimal dependencies \ No newline at end of file +A lightweight AI coding assistant built in Python with minimal dependencies. PicoGent implements a ReAct (Reasoning and Acting) loop that allows LLMs to use tools to accomplish complex tasks. + +## Features + +- **Minimal Dependencies**: Only requires `httpx` - no heavy SDKs +- **ReAct Loop**: Classic reasoning and acting pattern for tool use +- **Anthropic Claude Support**: Direct API integration without SDK +- **Built-in Tools**: File operations (read/write/edit) and shell execution +- **Session Management**: JSONL-based conversation history +- **Skills System**: Extensible skills directory for custom prompts +- **Configurable**: JSON-based configuration with environment variable support + +## Installation + +```bash +# Clone the repository +git clone https://git.uix.su/markov/picogent-py.git +cd picogent-py + +# Install dependencies +pip install -r requirements.txt + +# Or install as a package +pip install -e . +``` + +## Quick Start + +1. **Set up your API key**: + ```bash + export ANTHROPIC_API_KEY="your-api-key-here" + ``` + +2. **Create a configuration file**: + ```bash + cp config.example.json config.json + ``` + +3. **Use the agent**: + ```python + import asyncio + from picogent import Agent, Config + + # Load configuration + config = Config.from_file("config.json") + + # Create agent + agent = Agent(config) + + # Run a task + async def main(): + response = await agent.chat("Create a hello world Python script") + print(response) + + asyncio.run(main()) + ``` + +## Architecture + +### Core Components + +- **Agent** (`agent.py`): Main ReAct loop implementation +- **Provider** (`providers/anthropic.py`): Anthropic API integration using httpx +- **Tools** (`tools/`): Built-in tools for file and system operations +- **Session** (`session.py`): JSONL-based conversation history +- **Config** (`config.py`): Configuration management + +### Built-in Tools + +- **read**: Read file contents with optional offset/limit +- **write**: Write files with automatic directory creation +- **edit**: Find and replace text in files +- **bash**: Execute shell commands with timeout + +### ReAct Loop + +1. Build context (system prompt + history + user message) +2. Send to LLM with tool definitions +3. If no tool calls → return text response (done) +4. If tool calls → execute each tool, add results to history +5. Repeat from step 2 (max 20 iterations) + +## Configuration + +The `config.json` file supports: + +```json +{ + "provider": "anthropic", + "model": "claude-sonnet-4-20250514", + "api_key": "env:ANTHROPIC_API_KEY", + "max_tokens": 8192, + "max_iterations": 20, + "workspace": ".", + "system_prompt": "You are a helpful coding assistant." +} +``` + +- `api_key` can reference environment variables with `env:VARIABLE_NAME` +- `workspace` sets the working directory for file operations +- `max_iterations` prevents infinite loops in the ReAct cycle + +## Skills System + +Add custom prompts and context to the `skills/` directory: + +``` +skills/ +├── python-expert.md # Python-specific knowledge +├── web-dev.md # Web development skills +└── debugging.md # Debugging techniques +``` + +Skills are automatically loaded and added to the system prompt. + +## Session Management + +Sessions are stored in JSONL format for easy inspection and debugging: + +```python +# Save conversation to file +response = await agent.run("Create a Python script", session_file="conversation.jsonl") + +# Load existing session +agent.session.load("conversation.jsonl") +``` + +## API Integration + +The Anthropic provider uses direct HTTP requests with `httpx`: +- No official SDK dependency +- Full control over API calls +- Support for tool use and tool result messages +- Proper error handling and timeouts + +## Development + +The codebase follows these principles: +- **Minimal dependencies**: Only essential packages +- **Type hints**: Full type annotations +- **Async/await**: Modern Python async patterns +- **Error handling**: Comprehensive exception management +- **Extensibility**: Easy to add new tools and providers + +## License + +MIT License - see LICENSE file for details. + +## Contributing + +1. Fork the repository +2. Create a feature branch +3. Add tests for new functionality +4. Submit a pull request + +## Roadmap + +- [ ] OpenAI provider support +- [ ] Streaming responses +- [ ] Plugin system for custom tools +- [ ] Web interface +- [ ] Multi-agent coordination \ No newline at end of file diff --git a/config.example.json b/config.example.json new file mode 100644 index 0000000..8a07e2f --- /dev/null +++ b/config.example.json @@ -0,0 +1,9 @@ +{ + "provider": "anthropic", + "model": "claude-sonnet-4-20250514", + "api_key": "env:ANTHROPIC_API_KEY", + "max_tokens": 8192, + "max_iterations": 20, + "workspace": ".", + "system_prompt": "You are a helpful coding assistant." +} \ No newline at end of file diff --git a/example.py b/example.py new file mode 100644 index 0000000..1cfef1b --- /dev/null +++ b/example.py @@ -0,0 +1,45 @@ +#!/usr/bin/env python3 +""" +Example usage of PicoGent +""" + +import asyncio +from picogent import Agent, Config + + +async def main(): + """Example usage""" + + # Load config + config = Config.from_file("config.example.json") + + # Create agent + agent = Agent(config) + + # Simple chat example + print("=== Simple Chat ===") + response = await agent.chat("List all files in the current directory") + print(response) + print() + + # Session-based conversation + print("=== Session-based Conversation ===") + agent.clear_session() # Start fresh + + response1 = await agent.run("Create a simple hello.py file", session_file="example_session.jsonl") + print("Step 1:", response1) + print() + + response2 = await agent.run("Now read the file you just created", session_file="example_session.jsonl") + print("Step 2:", response2) + print() + + # Show session history + print("=== Session History ===") + messages = agent.get_session_messages() + for i, msg in enumerate(messages): + print(f"{i+1}. {msg['role']}: {str(msg['content'])[:100]}...") + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/picogent/__init__.py b/picogent/__init__.py new file mode 100644 index 0000000..0bbbb8f --- /dev/null +++ b/picogent/__init__.py @@ -0,0 +1,11 @@ +""" +PicoGent - Minimal AI Coding Agent +""" + +from .agent import Agent, run_agent +from .config import Config +from .session import Session +from .context import Context + +__version__ = "0.1.0" +__all__ = ["Agent", "run_agent", "Config", "Session", "Context"] \ No newline at end of file diff --git a/picogent/agent.py b/picogent/agent.py new file mode 100644 index 0000000..f262212 --- /dev/null +++ b/picogent/agent.py @@ -0,0 +1,152 @@ +""" +Main agent loop with ReAct pattern +""" + +import uuid +import asyncio +from typing import Dict, Any, Optional, List +from .config import Config +from .session import Session +from .context import Context +from .providers.anthropic import AnthropicProvider +from .tools.registry import registry +from .tools.read import ReadTool +from .tools.write import WriteTool +from .tools.edit import EditTool +from .tools.bash import BashTool + + +class Agent: + """Main PicoGent agent with ReAct loop""" + + def __init__(self, config: Config, context: Optional[Context] = None): + self.config = config + self.context = context or Context(config.workspace) + self.provider = None + + # Initialize tools + self._initialize_tools() + + def _initialize_tools(self): + """Initialize and register all tools""" + # Register tools with workspace context + registry.register(ReadTool()) + registry.register(WriteTool()) + registry.register(EditTool()) + registry.register(BashTool(self.context.get_workspace())) + + async def __aenter__(self): + """Async context manager entry""" + self.provider = AnthropicProvider(self.config) + await self.provider.__aenter__() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + """Async context manager exit""" + if self.provider: + await self.provider.__aexit__(exc_type, exc_val, exc_tb) + + async def run_agent(self, prompt: str, session: Optional[Session] = None) -> str: + """ + Main ReAct loop + + Args: + prompt: User prompt/question + session: Optional session for conversation history + + Returns: + Final response from the agent + """ + # Create session if not provided + if session is None: + session = Session() + + # Add user message to session + session.add_message("user", prompt) + + # ReAct loop + for iteration in range(self.config.max_iterations): + try: + # Get messages and system prompt + messages = session.get_anthropic_messages() + system_prompt = self.context.get_system_prompt() + tools = registry.get_tool_definitions() + + # Call LLM + response = await self.provider.chat_completion( + messages=messages, + tools=tools, + system_prompt=system_prompt + ) + + # Handle errors + if "error" in response: + error_msg = f"Provider error: {response['error']}" + session.add_message("assistant", error_msg) + return error_msg + + # Get content and tool calls + content = response.get("content", "").strip() + tool_calls = self.provider.parse_tool_calls(response) + + # No tool calls - return text response + if not tool_calls: + if content: + session.add_message("assistant", content) + return content + else: + error_msg = "Empty response from provider" + session.add_message("assistant", error_msg) + return error_msg + + # Add assistant message with tool calls + if content: + # Add text content if present + session.add_message("assistant", content) + + # Execute each tool call + all_results = [] + for tool_call in tool_calls: + tool_id = tool_call.get("id") or str(uuid.uuid4()) + tool_name = tool_call.get("name", "") + tool_args = tool_call.get("arguments", {}) + + # Add tool use to session + session.add_tool_use(tool_name, tool_args, tool_id) + + # Execute tool + result = await registry.execute_tool(tool_name, tool_args) + all_results.append(result) + + # Add tool result to session + session.add_tool_result(tool_id, result) + + # Continue loop to process tool results + continue + + except Exception as e: + error_msg = f"Agent error in iteration {iteration + 1}: {str(e)}" + session.add_message("assistant", error_msg) + return error_msg + + # Max iterations reached + final_msg = f"Reached maximum iterations ({self.config.max_iterations}). Task may be incomplete." + session.add_message("assistant", final_msg) + return final_msg + + +# Convenience function +async def run_agent(prompt: str, config: Config, session: Optional[Session] = None) -> str: + """ + Convenience function to run agent + + Args: + prompt: User prompt + config: Agent configuration + session: Optional session for conversation history + + Returns: + Agent response + """ + async with Agent(config) as agent: + return await agent.run_agent(prompt, session) \ No newline at end of file diff --git a/picogent/cli.py b/picogent/cli.py new file mode 100644 index 0000000..99acf09 --- /dev/null +++ b/picogent/cli.py @@ -0,0 +1,64 @@ +""" +Command line interface for PicoGent +""" + +import asyncio +import argparse +import sys +import os +from .agent import Agent +from .config import Config + + +async def main(): + """Main CLI entry point""" + parser = argparse.ArgumentParser(description="PicoGent - Minimal AI Coding Agent") + parser.add_argument("message", nargs="*", help="Message to send to the agent") + parser.add_argument("--config", "-c", default="config.json", help="Config file path") + parser.add_argument("--session", "-s", help="Session file path (JSONL)") + parser.add_argument("--workspace", "-w", help="Workspace directory") + + args = parser.parse_args() + + # Join message parts + message = " ".join(args.message) if args.message else None + + # Read from stdin if no message provided + if not message: + if sys.stdin.isatty(): + print("Enter your message (or pipe input):") + message = input() + else: + message = sys.stdin.read().strip() + + if not message: + print("Error: No message provided") + sys.exit(1) + + # Load config + try: + if os.path.exists(args.config): + config = Config.from_file(args.config) + else: + print(f"Warning: Config file '{args.config}' not found, using defaults") + config = Config() + except Exception as e: + print(f"Error loading config: {e}") + sys.exit(1) + + # Override workspace if specified + if args.workspace: + config.workspace = args.workspace + + # Create and run agent + try: + agent = Agent(config) + response = await agent.run(message, session_file=args.session) + print(response) + except Exception as e: + print(f"Error: {e}") + sys.exit(1) + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/picogent/config.py b/picogent/config.py new file mode 100644 index 0000000..82455b8 --- /dev/null +++ b/picogent/config.py @@ -0,0 +1,59 @@ +""" +Configuration management for PicoGent +""" + +import json +import os +from dataclasses import dataclass +from typing import Optional, Dict, Any + + +@dataclass +class Config: + """Configuration for PicoGent agent""" + provider: str = "anthropic" + model: str = "claude-sonnet-4-20250514" + api_key: str = "env:ANTHROPIC_API_KEY" + max_tokens: int = 8192 + max_iterations: int = 20 + workspace: str = "." + system_prompt: str = "You are a helpful coding assistant." + + @classmethod + def from_file(cls, config_path: str) -> "Config": + """Load configuration from JSON file""" + if not os.path.exists(config_path): + raise FileNotFoundError(f"Config file not found: {config_path}") + + with open(config_path, 'r') as f: + data = json.load(f) + + # Resolve environment variables in API key + api_key = data.get('api_key', '') + if api_key.startswith('env:'): + env_var = api_key[4:] # Remove 'env:' prefix + api_key = os.getenv(env_var, '') + if not api_key: + raise ValueError(f"Environment variable {env_var} is not set") + + return cls( + provider=data.get('provider', 'anthropic'), + model=data.get('model', 'claude-sonnet-4-20250514'), + api_key=api_key, + max_tokens=data.get('max_tokens', 8192), + max_iterations=data.get('max_iterations', 20), + workspace=data.get('workspace', '.'), + system_prompt=data.get('system_prompt', 'You are a helpful coding assistant.') + ) + + def to_dict(self) -> Dict[str, Any]: + """Convert config to dictionary""" + return { + 'provider': self.provider, + 'model': self.model, + 'api_key': self.api_key, + 'max_tokens': self.max_tokens, + 'max_iterations': self.max_iterations, + 'workspace': self.workspace, + 'system_prompt': self.system_prompt + } \ No newline at end of file diff --git a/picogent/context.py b/picogent/context.py new file mode 100644 index 0000000..831554d --- /dev/null +++ b/picogent/context.py @@ -0,0 +1,69 @@ +""" +Context management for PicoGent +""" + +import os +from typing import Dict, Any, List, Optional + + +class Context: + """Context manager for agent workspace and environment""" + + def __init__(self, workspace: str = "."): + self.workspace = os.path.abspath(workspace) + self.context_data: Dict[str, Any] = {} + + def set_workspace(self, workspace: str): + """Set the workspace directory""" + self.workspace = os.path.abspath(workspace) + if not os.path.exists(self.workspace): + os.makedirs(self.workspace, exist_ok=True) + + def get_workspace(self) -> str: + """Get the current workspace directory""" + return self.workspace + + def resolve_path(self, path: str) -> str: + """Resolve a path relative to workspace""" + if os.path.isabs(path): + return path + return os.path.join(self.workspace, path) + + def set_context(self, key: str, value: Any): + """Set context data""" + self.context_data[key] = value + + def get_context(self, key: str, default: Any = None) -> Any: + """Get context data""" + return self.context_data.get(key, default) + + def clear_context(self): + """Clear all context data""" + self.context_data.clear() + + def get_system_prompt(self) -> str: + """Generate system prompt with context""" + prompt_parts = [ + "You are PicoGent, a helpful coding assistant.", + f"Your workspace is: {self.workspace}", + "", + "Available tools:", + "- read: Read file contents with optional offset/limit", + "- write: Write content to files (creates directories as needed)", + "- edit: Replace text in files (old_string -> new_string)", + "- bash: Execute shell commands (30s timeout)", + "", + "Guidelines:", + "1. Always use relative paths when possible", + "2. Be precise with file operations", + "3. Ask for confirmation before destructive operations", + "4. Provide helpful error messages", + "5. Keep responses concise and actionable" + ] + + # Add custom context if available + custom_context = self.get_context("system_additions") + if custom_context: + prompt_parts.extend(["", "Additional context:", custom_context]) + + return "\n".join(prompt_parts) \ No newline at end of file diff --git a/picogent/providers/__init__.py b/picogent/providers/__init__.py new file mode 100644 index 0000000..6804f81 --- /dev/null +++ b/picogent/providers/__init__.py @@ -0,0 +1,8 @@ +""" +PicoGent Providers Package +""" + +from .base import BaseProvider +from .anthropic import AnthropicProvider + +__all__ = ["BaseProvider", "AnthropicProvider"] \ No newline at end of file diff --git a/picogent/providers/anthropic.py b/picogent/providers/anthropic.py new file mode 100644 index 0000000..684e243 --- /dev/null +++ b/picogent/providers/anthropic.py @@ -0,0 +1,118 @@ +""" +Anthropic provider using httpx +""" + +import httpx +import json +import uuid +from typing import List, Dict, Any, Optional +from .base import BaseProvider + + +class AnthropicProvider(BaseProvider): + """Anthropic provider using httpx directly""" + + API_BASE = "https://api.anthropic.com/v1" + API_VERSION = "2023-06-01" + + def __init__(self, config): + super().__init__(config) + self.client = httpx.AsyncClient() + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self.client.aclose() + + async def chat_completion( + self, + messages: List[Dict[str, Any]], + tools: Optional[List[Dict[str, Any]]] = None, + system_prompt: Optional[str] = None + ) -> Dict[str, Any]: + """Get chat completion from Anthropic API""" + + headers = { + "x-api-key": self.config.api_key, + "anthropic-version": self.API_VERSION, + "content-type": "application/json" + } + + payload = { + "model": self.config.model, + "max_tokens": self.config.max_tokens, + "messages": messages + } + + if system_prompt: + payload["system"] = system_prompt + + if tools: + payload["tools"] = tools + + try: + response = await self.client.post( + f"{self.API_BASE}/messages", + headers=headers, + json=payload, + timeout=60.0 + ) + + if response.status_code != 200: + error_text = response.text + return { + "error": f"API Error {response.status_code}: {error_text}", + "content": f"Error: Anthropic API returned {response.status_code}" + } + + result = response.json() + + # Extract content and tool calls + content_blocks = result.get("content", []) + text_content = "" + tool_calls = [] + + for block in content_blocks: + if block.get("type") == "text": + text_content += block.get("text", "") + elif block.get("type") == "tool_use": + tool_calls.append({ + "id": block.get("id"), + "name": block.get("name"), + "arguments": block.get("input", {}) + }) + + return { + "content": text_content, + "tool_calls": tool_calls, + "raw_response": result + } + + except httpx.TimeoutException: + return { + "error": "Request timed out", + "content": "Error: Request to Anthropic API timed out" + } + except Exception as e: + return { + "error": str(e), + "content": f"Error: {str(e)}" + } + + def parse_tool_calls(self, response: Dict[str, Any]) -> List[Dict[str, Any]]: + """Parse tool calls from Anthropic response""" + return response.get("tool_calls", []) + + def format_tool_result(self, tool_call_id: str, result: str) -> Dict[str, Any]: + """Format tool result for Anthropic API""" + return { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": tool_call_id, + "content": result + } + ] + } \ No newline at end of file diff --git a/picogent/providers/base.py b/picogent/providers/base.py new file mode 100644 index 0000000..0454c78 --- /dev/null +++ b/picogent/providers/base.py @@ -0,0 +1,61 @@ +""" +Base provider class for AI providers +""" + +from abc import ABC, abstractmethod +from typing import List, Dict, Any, Optional +from ..config import Config + + +class BaseProvider(ABC): + """Base class for AI providers""" + + def __init__(self, config: Config): + self.config = config + + @abstractmethod + async def chat_completion( + self, + messages: List[Dict[str, Any]], + tools: Optional[List[Dict[str, Any]]] = None, + system_prompt: Optional[str] = None + ) -> Dict[str, Any]: + """ + Get chat completion from the provider + + Args: + messages: List of messages in the conversation + tools: Optional list of available tools + system_prompt: Optional system prompt + + Returns: + Dictionary containing response with 'content' and optional 'tool_calls' + """ + pass + + @abstractmethod + def parse_tool_calls(self, response: Dict[str, Any]) -> List[Dict[str, Any]]: + """ + Parse tool calls from provider response + + Args: + response: Provider response + + Returns: + List of tool calls with 'id', 'name', and 'arguments' + """ + pass + + @abstractmethod + def format_tool_result(self, tool_call_id: str, result: str) -> Dict[str, Any]: + """ + Format tool result for next request + + Args: + tool_call_id: ID of the tool call + result: Result of tool execution + + Returns: + Formatted message for the provider + """ + pass \ No newline at end of file diff --git a/picogent/session.py b/picogent/session.py new file mode 100644 index 0000000..46c4a4a --- /dev/null +++ b/picogent/session.py @@ -0,0 +1,98 @@ +""" +Session management with JSONL format +""" + +import json +import os +from datetime import datetime +from typing import List, Dict, Any, Optional + + +class Session: + """Session manager for conversation history""" + + def __init__(self, session_file: Optional[str] = None): + self.session_file = session_file + self.messages: List[Dict[str, Any]] = [] + + def add_message(self, role: str, content: str, timestamp: Optional[str] = None): + """Add a message to the session""" + if timestamp is None: + timestamp = datetime.utcnow().isoformat() + + message = { + 'role': role, + 'content': content, + 'timestamp': timestamp + } + + self.messages.append(message) + + def add_tool_use(self, tool_name: str, tool_input: Dict[str, Any], tool_id: str): + """Add a tool use message""" + content = [{ + 'type': 'tool_use', + 'id': tool_id, + 'name': tool_name, + 'input': tool_input + }] + self.add_message('assistant', content) + + def add_tool_result(self, tool_id: str, result: str): + """Add a tool result message""" + content = [{ + 'type': 'tool_result', + 'tool_use_id': tool_id, + 'content': result + }] + self.add_message('user', content) + + def get_messages(self) -> List[Dict[str, Any]]: + """Get all messages in the session""" + return self.messages.copy() + + def save(self, filename: Optional[str] = None): + """Save session to JSONL file""" + if filename is None and self.session_file is None: + raise ValueError("No filename specified for saving session") + + save_file = filename or self.session_file + + with open(save_file, 'w') as f: + for message in self.messages: + f.write(json.dumps(message) + '\n') + + def load(self, filename: Optional[str] = None): + """Load session from JSONL file""" + load_file = filename or self.session_file + + if not os.path.exists(load_file): + return # Empty session if file doesn't exist + + self.messages = [] + with open(load_file, 'r') as f: + for line in f: + line = line.strip() + if line: + message = json.loads(line) + self.messages.append(message) + + def clear(self): + """Clear all messages from the session""" + self.messages = [] + + def get_anthropic_messages(self) -> List[Dict[str, Any]]: + """Format messages for Anthropic API""" + anthropic_messages = [] + + for message in self.messages: + # Skip system messages (handled separately) + if message['role'] == 'system': + continue + + anthropic_messages.append({ + 'role': message['role'], + 'content': message['content'] + }) + + return anthropic_messages \ No newline at end of file diff --git a/picogent/tools/__init__.py b/picogent/tools/__init__.py new file mode 100644 index 0000000..c8c52ca --- /dev/null +++ b/picogent/tools/__init__.py @@ -0,0 +1,11 @@ +""" +PicoGent Tools Package +""" + +from .registry import Tool, ToolRegistry, registry +from .read import ReadTool +from .write import WriteTool +from .edit import EditTool +from .bash import BashTool + +__all__ = ["Tool", "ToolRegistry", "registry", "ReadTool", "WriteTool", "EditTool", "BashTool"] \ No newline at end of file diff --git a/picogent/tools/bash.py b/picogent/tools/bash.py new file mode 100644 index 0000000..8da5a03 --- /dev/null +++ b/picogent/tools/bash.py @@ -0,0 +1,106 @@ +""" +Bash execution tool +""" + +import subprocess +import asyncio +import os +from typing import Dict, Any +from .registry import Tool + + +class BashTool(Tool): + """Tool for executing shell commands""" + + def __init__(self): + super().__init__( + name="bash", + description="Execute shell commands with timeout. Use for running system commands, scripts, and terminal operations.", + parameters={ + "type": "object", + "properties": { + "command": { + "type": "string", + "description": "Shell command to execute" + }, + "timeout": { + "type": "number", + "description": "Timeout in seconds (default: 30)", + "default": 30 + }, + "cwd": { + "type": "string", + "description": "Working directory to execute command in (optional)" + } + }, + "required": ["command"] + } + ) + + async def execute(self, args: Dict[str, Any]) -> str: + """Execute the bash tool""" + command = args.get("command") + timeout = args.get("timeout", 30) + cwd = args.get("cwd") + + if not command: + return "Error: command is required" + + # Validate timeout + if timeout <= 0: + timeout = 30 + + # Validate and resolve cwd + if cwd: + if not os.path.isabs(cwd): + cwd = os.path.abspath(cwd) + if not os.path.exists(cwd): + return f"Error: Working directory '{cwd}' does not exist" + if not os.path.isdir(cwd): + return f"Error: Working directory '{cwd}' is not a directory" + + try: + # Execute command asynchronously + process = await asyncio.create_subprocess_shell( + command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + env=os.environ.copy() + ) + + try: + stdout, stderr = await asyncio.wait_for( + process.communicate(), + timeout=timeout + ) + except asyncio.TimeoutError: + # Kill the process if it times out + process.kill() + await process.wait() + return f"Error: Command timed out after {timeout} seconds" + + # Decode output + stdout_text = stdout.decode('utf-8', errors='replace').strip() + stderr_text = stderr.decode('utf-8', errors='replace').strip() + + # Format result + result_parts = [f"Command: {command}"] + if cwd: + result_parts.append(f"Working directory: {cwd}") + + result_parts.append(f"Exit code: {process.returncode}") + + if stdout_text: + result_parts.append(f"STDOUT:\n{stdout_text}") + + if stderr_text: + result_parts.append(f"STDERR:\n{stderr_text}") + + if not stdout_text and not stderr_text: + result_parts.append("(No output)") + + return "\n\n".join(result_parts) + + except Exception as e: + return f"Error: Could not execute command '{command}': {e}" \ No newline at end of file diff --git a/picogent/tools/edit.py b/picogent/tools/edit.py new file mode 100644 index 0000000..f89d84b --- /dev/null +++ b/picogent/tools/edit.py @@ -0,0 +1,91 @@ +""" +Edit file tool (find and replace) +""" + +import os +from typing import Dict, Any +from .registry import Tool + + +class EditTool(Tool): + """Tool for editing files using find and replace""" + + def __init__(self): + super().__init__( + name="edit", + description="Edit a file by finding and replacing exact text. The old_string must match exactly (including whitespace).", + parameters={ + "type": "object", + "properties": { + "file_path": { + "type": "string", + "description": "Path to the file to edit (relative or absolute)" + }, + "old_string": { + "type": "string", + "description": "Exact text to find and replace (must match exactly)" + }, + "new_string": { + "type": "string", + "description": "New text to replace the old text with" + } + }, + "required": ["file_path", "old_string", "new_string"] + } + ) + + async def execute(self, args: Dict[str, Any]) -> str: + """Execute the edit tool""" + file_path = args.get("file_path") + old_string = args.get("old_string") + new_string = args.get("new_string") + + if not file_path or old_string is None or new_string is None: + return "Error: file_path, old_string, and new_string are required" + + try: + # Convert to absolute path if relative + if not os.path.isabs(file_path): + file_path = os.path.abspath(file_path) + + if not os.path.exists(file_path): + return f"Error: File '{file_path}' does not exist" + + if not os.path.isfile(file_path): + return f"Error: '{file_path}' is not a file" + + # Read the current content + with open(file_path, 'r', encoding='utf-8') as f: + content = f.read() + + # Check if old_string exists in the content + if old_string not in content: + return f"Error: The specified text was not found in '{file_path}'" + + # Count occurrences + occurrence_count = content.count(old_string) + + # Perform the replacement + new_content = content.replace(old_string, new_string) + + # Write the modified content back + with open(file_path, 'w', encoding='utf-8') as f: + f.write(new_content) + + # Calculate changes + old_lines = len(content.splitlines()) + new_lines = len(new_content.splitlines()) + line_diff = new_lines - old_lines + + result_parts = [ + f"Successfully edited '{file_path}':", + f"- Replaced {occurrence_count} occurrence(s) of the specified text", + f"- File now has {new_lines} lines ({line_diff:+d} lines)" + ] + + return "\n".join(result_parts) + + except UnicodeDecodeError: + return f"Error: Could not read file '{file_path}' - file appears to be binary" + except Exception as e: + return f"Error: Could not edit file '{file_path}': {e}" \ No newline at end of file diff --git a/picogent/tools/read.py b/picogent/tools/read.py new file mode 100644 index 0000000..a9575e8 --- /dev/null +++ b/picogent/tools/read.py @@ -0,0 +1,94 @@ +""" +Read file tool +""" + +import os +from typing import Dict, Any +from .registry import Tool + + +class ReadTool(Tool): + """Tool for reading files with optional offset/limit""" + + def __init__(self): + super().__init__( + name="read", + description="Read the contents of a file. Supports optional offset/limit for reading specific lines.", + parameters={ + "type": "object", + "properties": { + "file_path": { + "type": "string", + "description": "Path to the file to read (relative or absolute)" + }, + "offset": { + "type": "integer", + "description": "Line number to start reading from (1-indexed, optional)", + "minimum": 1 + }, + "limit": { + "type": "integer", + "description": "Maximum number of lines to read (optional)", + "minimum": 1 + } + }, + "required": ["file_path"] + } + ) + + async def execute(self, args: Dict[str, Any]) -> str: + """Execute the read tool""" + file_path = args.get("file_path") + offset = args.get("offset", 1) # 1-indexed + limit = args.get("limit") + + if not file_path: + return "Error: file_path is required" + + try: + # Convert to absolute path if relative + if not os.path.isabs(file_path): + file_path = os.path.abspath(file_path) + + if not os.path.exists(file_path): + return f"Error: File '{file_path}' does not exist" + + if not os.path.isfile(file_path): + return f"Error: '{file_path}' is not a file" + + with open(file_path, 'r', encoding='utf-8') as f: + lines = f.readlines() + + # Apply offset (convert from 1-indexed to 0-indexed) + start_idx = max(0, offset - 1) + + # Apply limit + if limit is not None: + end_idx = start_idx + limit + selected_lines = lines[start_idx:end_idx] + else: + selected_lines = lines[start_idx:] + + content = ''.join(selected_lines) + + # Add metadata + total_lines = len(lines) + shown_lines = len(selected_lines) + + if offset > 1 or limit is not None: + metadata = f"# File: {file_path} (showing lines {offset}-{offset + shown_lines - 1} of {total_lines})\n\n" + else: + metadata = f"# File: {file_path} ({total_lines} lines)\n\n" + + return metadata + content + + except UnicodeDecodeError: + try: + # Try with different encoding + with open(file_path, 'r', encoding='latin1') as f: + content = f.read() + return f"# File: {file_path} (read with latin1 encoding)\n\n{content}" + except Exception as e: + return f"Error: Could not read file '{file_path}': {e}" + except Exception as e: + return f"Error: Could not read file '{file_path}': {e}" \ No newline at end of file diff --git a/picogent/tools/registry.py b/picogent/tools/registry.py new file mode 100644 index 0000000..99b8ec8 --- /dev/null +++ b/picogent/tools/registry.py @@ -0,0 +1,62 @@ +""" +Tool registry for managing available tools +""" + +from dataclasses import dataclass +from typing import Dict, Any, List, Callable +from abc import ABC, abstractmethod + + +@dataclass +class Tool(ABC): + """Base tool class""" + name: str + description: str + parameters: Dict[str, Any] # JSON Schema + + @abstractmethod + async def execute(self, args: Dict[str, Any]) -> str: + """Execute the tool with given arguments""" + pass + + def to_anthropic_format(self) -> Dict[str, Any]: + """Convert tool to Anthropic API format""" + return { + "name": self.name, + "description": self.description, + "input_schema": { + "type": "object", + "properties": self.parameters.get("properties", {}), + "required": self.parameters.get("required", []) + } + } + + +class ToolRegistry: + """Registry for managing available tools""" + + def __init__(self): + self.tools: Dict[str, Tool] = {} + + def register(self, tool: Tool): + """Register a tool""" + self.tools[tool.name] = tool + + def get_tool(self, name: str) -> Tool: + """Get a tool by name""" + if name not in self.tools: + raise ValueError(f"Tool '{name}' not found") + return self.tools[name] + + def get_all_tools(self) -> List[Tool]: + """Get all registered tools""" + return list(self.tools.values()) + + def get_anthropic_tools(self) -> List[Dict[str, Any]]: + """Get all tools in Anthropic API format""" + return [tool.to_anthropic_format() for tool in self.tools.values()] + + async def execute_tool(self, name: str, args: Dict[str, Any]) -> str: + """Execute a tool by name with arguments""" + tool = self.get_tool(name) + return await tool.execute(args) \ No newline at end of file diff --git a/picogent/tools/write.py b/picogent/tools/write.py new file mode 100644 index 0000000..cfee967 --- /dev/null +++ b/picogent/tools/write.py @@ -0,0 +1,62 @@ +""" +Write file tool +""" + +import os +from typing import Dict, Any +from .registry import Tool + + +class WriteTool(Tool): + """Tool for writing files with automatic directory creation""" + + def __init__(self): + super().__init__( + name="write", + description="Write content to a file. Creates parent directories if they don't exist. Overwrites existing files.", + parameters={ + "type": "object", + "properties": { + "file_path": { + "type": "string", + "description": "Path to the file to write (relative or absolute)" + }, + "content": { + "type": "string", + "description": "Content to write to the file" + } + }, + "required": ["file_path", "content"] + } + ) + + async def execute(self, args: Dict[str, Any]) -> str: + """Execute the write tool""" + file_path = args.get("file_path") + content = args.get("content", "") + + if not file_path: + return "Error: file_path is required" + + try: + # Convert to absolute path if relative + if not os.path.isabs(file_path): + file_path = os.path.abspath(file_path) + + # Create parent directories if they don't exist + parent_dir = os.path.dirname(file_path) + if parent_dir and not os.path.exists(parent_dir): + os.makedirs(parent_dir, exist_ok=True) + + # Write the file + with open(file_path, 'w', encoding='utf-8') as f: + f.write(content) + + # Get file info + file_size = os.path.getsize(file_path) + line_count = content.count('\n') + (1 if content and not content.endswith('\n') else 0) + + return f"Successfully wrote {file_size} bytes ({line_count} lines) to '{file_path}'" + + except Exception as e: + return f"Error: Could not write file '{file_path}': {e}" \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..486db2a --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +httpx>=0.24.0 \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..7b4579e --- /dev/null +++ b/setup.py @@ -0,0 +1,39 @@ +""" +Setup script for PicoGent +""" + +from setuptools import setup, find_packages + +with open("README.md", "r", encoding="utf-8") as fh: + long_description = fh.read() + +with open("requirements.txt", "r", encoding="utf-8") as fh: + requirements = [line.strip() for line in fh if line.strip() and not line.startswith("#")] + +setup( + name="picogent-py", + version="0.1.0", + author="PicoGent", + description="Minimal AI coding agent in Python with minimal dependencies", + long_description=long_description, + long_description_content_type="text/markdown", + packages=find_packages(), + classifiers=[ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + ], + python_requires=">=3.8", + install_requires=requirements, + entry_points={ + "console_scripts": [ + "picogent=picogent.cli:main", + ], + }, +) \ No newline at end of file diff --git a/skills/.gitkeep b/skills/.gitkeep new file mode 100644 index 0000000..334b83c --- /dev/null +++ b/skills/.gitkeep @@ -0,0 +1,2 @@ +# Empty skills directory +# Add your custom skill files here (.md or .txt) \ No newline at end of file