"""Advanced example showing explicit executor usage and custom grep tool."""
import os
import shlex
from collections.abc import Sequence
from pydantic import Field, SecretStr
from openhands.sdk import (
LLM,
Action,
Agent,
Conversation,
Event,
ImageContent,
LLMConvertibleEvent,
Observation,
TextContent,
ToolDefinition,
get_logger,
)
from openhands.sdk.tool import (
Tool,
ToolExecutor,
register_tool,
)
from openhands.tools.file_editor import FileEditorTool
from openhands.tools.terminal import (
BashExecutor,
ExecuteBashAction,
TerminalTool,
)
logger = get_logger(__name__)
# --- Action / Observation ---
class GrepAction(Action):
pattern: str = Field(description="Regex to search for")
path: str = Field(
default=".", description="Directory to search (absolute or relative)"
)
include: str | None = Field(
default=None, description="Optional glob to filter files (e.g. '*.py')"
)
class GrepObservation(Observation):
matches: list[str] = Field(default_factory=list)
files: list[str] = Field(default_factory=list)
count: int = 0
@property
def to_llm_content(self) -> Sequence[TextContent | ImageContent]:
if not self.count:
return [TextContent(text="No matches found.")]
files_list = "\n".join(f"- {f}" for f in self.files[:20])
sample = "\n".join(self.matches[:10])
more = "\n..." if self.count > 10 else ""
ret = (
f"Found {self.count} matching lines.\n"
f"Files:\n{files_list}\n"
f"Sample:\n{sample}{more}"
)
return [TextContent(text=ret)]
# --- Executor ---
class GrepExecutor(ToolExecutor[GrepAction, GrepObservation]):
def __init__(self, bash: BashExecutor):
self.bash: BashExecutor = bash
def __call__(self, action: GrepAction, conversation=None) -> GrepObservation: # noqa: ARG002
root = os.path.abspath(action.path)
pat = shlex.quote(action.pattern)
root_q = shlex.quote(root)
# Use grep -r; add --include when provided
if action.include:
inc = shlex.quote(action.include)
cmd = f"grep -rHnE --include {inc} {pat} {root_q} 2>/dev/null | head -100"
else:
cmd = f"grep -rHnE {pat} {root_q} 2>/dev/null | head -100"
result = self.bash(ExecuteBashAction(command=cmd))
matches: list[str] = []
files: set[str] = set()
# grep returns exit code 1 when no matches; treat as empty
output_text = result.text
if output_text.strip():
for line in output_text.strip().splitlines():
matches.append(line)
# Expect "path:line:content" — take the file part before first ":"
file_path = line.split(":", 1)[0]
if file_path:
files.add(os.path.abspath(file_path))
return GrepObservation(matches=matches, files=sorted(files), count=len(matches))
# Tool description
_GREP_DESCRIPTION = """Fast content search tool.
* Searches file contents using regular expressions
* Supports full regex syntax (eg. "log.*Error", "function\\s+\\w+", etc.)
* Filter files by pattern with the include parameter (eg. "*.js", "*.{ts,tsx}")
* Returns matching file paths sorted by modification time.
* Only the first 100 results are returned. Consider narrowing your search with stricter regex patterns or provide path parameter if you need more results.
* Use this tool when you need to find files containing specific patterns
* When you are doing an open ended search that may require multiple rounds of globbing and grepping, use the Agent tool instead
""" # noqa: E501
# --- Tool Definition ---
class GrepTool(ToolDefinition[GrepAction, GrepObservation]):
"""A custom grep tool that searches file contents using regular expressions."""
@classmethod
def create(
cls, conv_state, bash_executor: BashExecutor | None = None
) -> Sequence[ToolDefinition]:
"""Create GrepTool instance with a GrepExecutor.
Args:
conv_state: Conversation state to get working directory from.
bash_executor: Optional bash executor to reuse. If not provided,
a new one will be created.
Returns:
A sequence containing a single GrepTool instance.
"""
if bash_executor is None:
bash_executor = BashExecutor(working_dir=conv_state.workspace.working_dir)
grep_executor = GrepExecutor(bash_executor)
return [
cls(
description=_GREP_DESCRIPTION,
action_type=GrepAction,
observation_type=GrepObservation,
executor=grep_executor,
)
]
# Configure LLM
api_key = os.getenv("LLM_API_KEY")
assert api_key is not None, "LLM_API_KEY environment variable is not set."
model = os.getenv("LLM_MODEL", "openhands/claude-sonnet-4-5-20250929")
base_url = os.getenv("LLM_BASE_URL")
llm = LLM(
usage_id="agent",
model=model,
base_url=base_url,
api_key=SecretStr(api_key),
)
# Tools - demonstrating both simplified and advanced patterns
cwd = os.getcwd()
def _make_bash_and_grep_tools(conv_state) -> list[ToolDefinition]:
"""Create terminal and custom grep tools sharing one executor."""
bash_executor = BashExecutor(working_dir=conv_state.workspace.working_dir)
# bash_tool = terminal_tool.set_executor(executor=bash_executor)
bash_tool = TerminalTool.create(conv_state, executor=bash_executor)[0]
# Use the GrepTool.create() method with shared bash_executor
grep_tool = GrepTool.create(conv_state, bash_executor=bash_executor)[0]
return [bash_tool, grep_tool]
register_tool("BashAndGrepToolSet", _make_bash_and_grep_tools)
tools = [
Tool(name=FileEditorTool.name),
Tool(name="BashAndGrepToolSet"),
]
# Agent
agent = Agent(llm=llm, tools=tools)
llm_messages = [] # collect raw LLM messages
def conversation_callback(event: Event):
if isinstance(event, LLMConvertibleEvent):
llm_messages.append(event.to_llm_message())
conversation = Conversation(
agent=agent, callbacks=[conversation_callback], workspace=cwd
)
conversation.send_message(
"Hello! Can you use the grep tool to find all files "
"containing the word 'class' in this project, then create a summary file listing them? " # noqa: E501
"Use the pattern 'class' to search and include only Python files with '*.py'." # noqa: E501
)
conversation.run()
conversation.send_message("Great! Now delete that file.")
conversation.run()
print("=" * 100)
print("Conversation finished. Got the following LLM messages:")
for i, message in enumerate(llm_messages):
print(f"Message {i}: {str(message)[:200]}")
# Report cost
cost = llm.metrics.accumulated_cost
print(f"EXAMPLE_COST: {cost}")
