feat: add prerelease tools

Author: Palashio

.DS_Store

@@ -0,0 +1,51 @@
+import os
+import tempfile
+from typing import List, Dict, Any, Optional, Union, Literal
+from pathlib import Path
+
+from deepagents import create_deep_agent, SubAgent
+from post_model_hook import create_coding_agent_post_model_hook
+from subagents import code_reviewer_agent, test_generator_agent
+from langgraph.types import Command
+from state import CodingAgentState
+from coding_instructions import get_coding_instructions
+from tools import execute_bash, http_request, web_search
+
+from langsmith import Client
+from langsmith.wrappers import wrap_openai
+from langchain_core.tracers.langchain import LangChainTracer
+import dotenv
+
+dotenv.load_dotenv()
+
+langsmith_client = None
+langchain_tracer = None
+
+if os.environ.get("LANGCHAIN_API_KEY"):
+    try:
+        langsmith_client = Client()
+        langchain_tracer = LangChainTracer(
+            project_name=os.environ.get("LANGCHAIN_PROJECT", "coding-agent"),
+            client=langsmith_client
+        )
+    except Exception as e:
+        print(f"Warning: Failed to initialize LangSmith tracing: {e}")
+
+
+coding_instructions = get_coding_instructions()
+
+post_model_hook = create_coding_agent_post_model_hook()
+
+config = {"recursion_limit": 1000}
+
+if langchain_tracer:
+    config["callbacks"] = [langchain_tracer]
+
+agent = create_deep_agent(
+    [execute_bash, http_request, web_search],
+    coding_instructions,
+    subagents=[code_reviewer_agent, test_generator_agent],
+    local_filesystem=True,
+    state_schema=CodingAgentState,
+    post_model_hook=post_model_hook,
+).with_config(config)

examples/.DS_Store

@@ -0,0 +1,17 @@
+"""
+Constants for Open SWE V2 coding agent.
+"""
+
+# File operation commands that require approval in the approval system
+FILE_EDIT_COMMANDS = {"write_file", "str_replace_based_edit_tool", "edit_file"}
+
+# All commands that require approval (includes file operations plus other system operations)  
+WRITE_COMMANDS = {
+    "write_file", 
+    "execute_bash", 
+    "str_replace_based_edit_tool", 
+    "ls", 
+    "edit_file", 
+    "glob", 
+    "grep"
+}

examples/code/coding_agent.py

@@ -0,0 +1,7 @@
+{
+  "dependencies": ["."],
+  "graphs": {
+    "coding": "./coding_agent.py:agent"
+  },
+  "env": ".env"
+}

examples/code/coding_instructions.py

@@ -0,0 +1,135 @@
+"""
+Post model hook for the coding agent with intelligent caching for command approvals.
+Similar to Claude Code's approval system - only prompts for new commands or new directories.
+"""
+
+from typing import Dict, Any
+from langchain_core.messages import AIMessage
+from langgraph.types import interrupt
+import os
+
+
+def create_coding_agent_post_model_hook():
+    """Create a post model hook with intelligent approval caching."""
+    
+    def get_approval_key(command: str, args: Dict[str, Any]) -> str:
+        """Generate a cache key for command approval based on command type and target directory."""
+        # Extract directory from common file operations
+        target_dir = None
+        
+        if command in ["write_file", "str_replace_based_edit_tool", "edit_file"]:
+            file_path = args.get("file_path") or args.get("path")
+            if file_path:
+                target_dir = os.path.dirname(os.path.abspath(file_path))
+        elif command == "execute_bash":
+            # For bash commands, use current working directory or cwd from args
+            target_dir = args.get("cwd") or os.getcwd()
+        elif command in ["ls", "glob", "grep"]:
+            # For read operations, use the path if provided
+            target_dir = args.get("path") or args.get("directory") or os.getcwd()
+        
+        if not target_dir:
+            target_dir = os.getcwd()
+            
+        # Create a cache key: command_type:normalized_directory
+        normalized_dir = os.path.normpath(target_dir)
+        return f"{command}:{normalized_dir}"
+    
+    def is_operation_approved(approved_operations: Dict[str, Any], command: str, args: Dict[str, Any]) -> bool:
+        """Check if a command/directory combination has been previously approved."""
+        if not approved_operations:
+            return False
+            
+        approval_key = get_approval_key(command, args)
+        return approval_key in approved_operations.get("cached_approvals", set())
+    
+    def add_approved_operation(approved_operations: Dict[str, Any], command: str, args: Dict[str, Any]):
+        """Add a command/directory combination to the approved operations cache."""
+        if "cached_approvals" not in approved_operations:
+            approved_operations["cached_approvals"] = set()
+            
+        approval_key = get_approval_key(command, args)
+        approved_operations["cached_approvals"].add(approval_key)
+    
+    def post_model_hook(state: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Post model hook that checks for write tool calls and uses caching to avoid
+        redundant approval prompts for the same command/directory combinations.
+        """
+        # Get the last message from the state
+        messages = state.get("messages", [])
+        if not messages:
+            return state
+        
+        last_message = messages[-1]
+        
+        if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
+            return state
+        
+        # Get or initialize the approval cache from state
+        approved_operations = state.get("approved_operations", {"cached_approvals": set()})
+        
+        approved_tool_calls = []
+        
+        # Define write tools that need approval
+        write_tools = {
+            "write_file", "execute_bash", "str_replace_based_edit_tool", "ls", "edit_file", "glob", "grep" 
+        }
+        
+        for tool_call in last_message.tool_calls:
+            tool_name = tool_call.get("name", "")
+            tool_args = tool_call.get("args", {})
+            
+            if tool_name in write_tools:
+                # Check if this command/directory combination has been approved before
+                if is_operation_approved(approved_operations, tool_name, tool_args):
+                    # Already approved - execute without prompting
+                    approved_tool_calls.append(tool_call)
+                else:
+                    # New command or directory - ask for approval
+                    approval_key = get_approval_key(tool_name, tool_args)
+                    question = (
+                        f"New operation detected:\n\n"
+                        f"Command: {tool_name}\n"
+                        f"Directory: {approval_key.split(':', 1)[1]}\n"
+                        f"Args: {tool_args}\n\n"
+                        f"Approve this command for this directory? "
+                        f"(Future identical operations in this directory will be auto-approved)\n\n"
+                        f"Respond with True to approve or False to reject."
+                    )
+                    
+                    is_approved = interrupt({
+                        "question": question,
+                        "command": tool_name,
+                        "args": tool_args,
+                        "approval_key": approval_key
+                    })
+                    
+                    if is_approved:
+                        # Add to cache and approve the tool call
+                        add_approved_operation(approved_operations, tool_name, tool_args)
+                        approved_tool_calls.append(tool_call)
+                        
+                        # Update the main state with the new approval cache
+                        state["approved_operations"] = approved_operations
+                    else:
+                        # Rejected - skip this tool call
+                        continue
+            else:
+                # For all other tools (read-only operations), allow them to execute without approval
+                approved_tool_calls.append(tool_call)
+        
+        # Update the message if any tool calls were filtered out
+        if len(approved_tool_calls) != len(last_message.tool_calls):
+            new_message = AIMessage(
+                content=last_message.content,
+                tool_calls=approved_tool_calls,
+                additional_kwargs=last_message.additional_kwargs
+            )
+            
+            new_messages = messages[:-1] + [new_message]
+            state["messages"] = new_messages
+        
+        return state
+    
+    return post_model_hook

examples/code/constants.py

@@ -0,0 +1,6 @@
+# Core requirements
+deepagents
+langgraph-cli[inmem]
+tavily-python
+langchain-anthropic
+pydantic

examples/code/langgraph.json

@@ -0,0 +1,49 @@
+from deepagents.state import DeepAgentState
+from typing import Dict, Set, NotRequired
+from typing_extensions import TypedDict
+import os
+import hashlib
+from constants import FILE_EDIT_COMMANDS
+
+class CodingAgentState(DeepAgentState):
+    """Extended state for coding agent with approval operation caching."""
+    approved_operations: NotRequired[Dict[str, Set[str]]]
+    
+    def get_approval_key(self, command: str, args: Dict) -> str:
+        """Generate a cache key for command approval based on command type and target directory."""
+        target_dir = None
+        
+        if command in FILE_EDIT_COMMANDS:
+            file_path = args.get("file_path") or args.get("path")
+            if file_path:
+                target_dir = os.path.dirname(os.path.abspath(file_path))
+        elif command == "execute_bash":
+            target_dir = args.get("cwd") or os.getcwd()
+        elif command in ["ls", "glob", "grep"]:
+            target_dir = args.get("path") or args.get("directory") or os.getcwd()
+        
+        if not target_dir:
+            target_dir = os.getcwd()
+            
+        # Create a cache key: command_type:normalized_directory
+        normalized_dir = os.path.normpath(target_dir)
+        return f"{command}:{normalized_dir}"
+    
+    def is_operation_approved(self, command: str, args: Dict) -> bool:
+        """Check if a command/directory combination has been previously approved."""
+        if not hasattr(self, 'approved_operations') or not self.approved_operations:
+            return False
+            
+        approval_key = self.get_approval_key(command, args)
+        return approval_key in self.approved_operations.get("cached_approvals", set())
+    
+    def add_approved_operation(self, command: str, args: Dict):
+        """Add a command/directory combination to the approved operations cache."""
+        if not hasattr(self, 'approved_operations') or self.approved_operations is None:
+            self.approved_operations = {"cached_approvals": set()}
+        
+        if "cached_approvals" not in self.approved_operations:
+            self.approved_operations["cached_approvals"] = set()
+            
+        approval_key = self.get_approval_key(command, args)
+        self.approved_operations["cached_approvals"].add(approval_key)

examples/code/post_model_hook.py

@@ -0,0 +1,52 @@
+# Sub-agent for code review and analysis
+code_reviewer_prompt = """You are an expert code reviewer for all programming languages. Your job is to analyze code for:
+
+1. **Code Quality**: Check for clean, readable, and maintainable code
+2. **Best Practices**: Ensure adherence to language-specific best practices and conventions
+3. **Security**: Identify potential security vulnerabilities
+4. **Performance**: Suggest optimizations where applicable
+5. **Testing**: Evaluate test coverage and quality
+6. **Documentation**: Check for proper comments and documentation
+
+When reviewing code, provide:
+- Specific line-by-line feedback
+- Language-specific suggestions for improvements
+- Security concerns (if any)
+- Performance optimization opportunities
+- Overall assessment and rating (1-10)
+
+You can use bash commands to run linters, formatters, and other code analysis tools for any language.
+Be constructive and educational in your feedback. Focus on helping improve the code quality."""
+
+code_reviewer_agent = {
+    "name": "code-reviewer",
+    "description": "Expert code reviewer that analyzes code in any programming language for quality, security, performance, and best practices. Use this when you need detailed code analysis and improvement suggestions.",
+    "prompt": code_reviewer_prompt,
+    "tools": ["execute_bash"],
+}
+
+# Sub-agent for test generation
+test_generator_prompt = """You are an expert test engineer for all programming languages. Your job is to create comprehensive test suites for any codebase.
+
+When generating tests:
+1. **Test Coverage**: Create tests that cover all functions, methods, and edge cases
+2. **Test Types**: Include unit tests, integration tests, and edge case tests
+3. **Frameworks**: Use appropriate testing frameworks for each language (Jest, pytest, JUnit, Go test, etc.)
+4. **Assertions**: Write meaningful assertions that validate expected behavior
+5. **Documentation**: Include clear test descriptions and comments
+
+Test categories to consider:
+- **Happy Path**: Normal expected inputs and outputs
+- **Edge Cases**: Boundary conditions, empty inputs, large inputs
+- **Error Cases**: Invalid inputs, exception handling
+- **Integration**: How components work together
+
+Use bash commands to run language-specific test frameworks and verify that tests execute successfully.
+Always verify that your tests can run successfully and provide meaningful feedback."""
+
+test_generator_agent = {
+    "name": "test-generator",
+    "description": "Expert test engineer that creates comprehensive test suites for any programming language. Use when you need to generate thorough test suites for your code.",
+    "prompt": test_generator_prompt,
+    "tools": ["execute_bash"],
+}