chore: Let tools to handle root directory resolvement and model only knows the project name and always use relative path

PiperOrigin-RevId: 816213558
This commit is contained in:
Xiang (Sean) Zhou
2025-10-07 08:11:06 -07:00
committed by Copybara-Service
parent 3f4bd67b49
commit 3be9cd844c
11 changed files with 237 additions and 268 deletions
@@ -34,7 +34,6 @@ from .tools.delete_files import delete_files
from .tools.explore_project import explore_project
from .tools.read_config_files import read_config_files
from .tools.read_files import read_files
from .tools.resolve_root_directory import resolve_root_directory
from .tools.search_adk_source import search_adk_source
from .tools.write_config_files import write_config_files
from .tools.write_files import write_files
@@ -60,9 +59,7 @@ class AgentBuilderAssistant:
Configured LlmAgent with embedded ADK AgentConfig schema
"""
# Load full ADK AgentConfig schema directly into instruction context
instruction = AgentBuilderAssistant._load_instruction_with_schema(
model, working_directory
)
instruction = AgentBuilderAssistant._load_instruction_with_schema(model)
# TOOL ARCHITECTURE: Hybrid approach using both AgentTools and FunctionTools
#
@@ -95,8 +92,6 @@ class AgentBuilderAssistant:
write_config_files
), # Write/validate multiple YAML configs
FunctionTool(explore_project), # Analyze project structure
# Working directory context tools
FunctionTool(resolve_root_directory),
# File management tools (multi-file support)
FunctionTool(read_files), # Read multiple files
FunctionTool(write_files), # Write multiple files
@@ -135,7 +130,6 @@ class AgentBuilderAssistant:
# ADK AgentConfig schema loading with caching and error handling.
schema_content = load_agent_config_schema(
raw_format=True, # Get as JSON string
escape_braces=True, # Escape braces for template embedding
)
# Format as indented code block for instruction embedding
@@ -157,7 +151,6 @@ class AgentBuilderAssistant:
@staticmethod
def _load_instruction_with_schema(
model: Union[str, BaseLlm],
working_directory: Optional[str] = None,
) -> Callable[[ReadonlyContext], str]:
"""Load instruction template and embed ADK AgentConfig schema content."""
instruction_template = (
@@ -172,19 +165,43 @@ class AgentBuilderAssistant:
else getattr(model, "model_name", str(model))
)
# Fill the instruction template with ADK AgentConfig schema content and default model
instruction_text = instruction_template.format(
schema_content=schema_content, default_model=model_str
)
# Return a function that accepts ReadonlyContext and returns the instruction
def instruction_provider(context: ReadonlyContext) -> str:
return AgentBuilderAssistant._compile_instruction_with_context(
instruction_text, context, working_directory
# Extract project folder name from session state
project_folder_name = AgentBuilderAssistant._extract_project_folder_name(
context
)
# Fill the instruction template with all variables
instruction_text = instruction_template.format(
schema_content=schema_content,
default_model=model_str,
project_folder_name=project_folder_name,
)
return instruction_text
return instruction_provider
@staticmethod
def _extract_project_folder_name(context: ReadonlyContext) -> str:
"""Extract project folder name from session state using resolve_file_path."""
from .utils.resolve_root_directory import resolve_file_path
session_state = context._invocation_context.session.state
# Use resolve_file_path to get the full resolved path for "."
# This handles all the root_directory resolution logic consistently
resolved_path = resolve_file_path(".", session_state)
# Extract the project folder name from the resolved path
project_folder_name = resolved_path.name
# Fallback to "project" if we somehow get an empty name
if not project_folder_name:
project_folder_name = "project"
return project_folder_name
@staticmethod
def _load_embedded_schema_instruction_template() -> str:
"""Load instruction template for embedded ADK AgentConfig schema mode."""
@@ -197,70 +214,3 @@ class AgentBuilderAssistant:
with open(template_path, "r", encoding="utf-8") as f:
return f.read()
@staticmethod
def _compile_instruction_with_context(
instruction_text: str,
context: ReadonlyContext,
working_directory: Optional[str] = None,
) -> str:
"""Compile instruction with session context and working directory information.
This method enhances instructions with:
1. Working directory information for path resolution
2. Session-based root directory binding if available
Args:
instruction_text: Base instruction text
context: ReadonlyContext from the agent session
working_directory: Optional working directory for path resolution
Returns:
Enhanced instruction text with context information
"""
import os
# Get working directory (use provided or current working directory)
actual_working_dir = working_directory or os.getcwd()
# Check for existing root directory in session state
session_root_directory = context._invocation_context.session.state.get(
"root_directory"
)
# Compile additional context information
context_info = f"""
## SESSION CONTEXT
**Working Directory**: `{actual_working_dir}`
- Use this as the base directory for path resolution when calling resolve_root_directory
- Pass this as the working_directory parameter to resolve_root_directory tool
"""
if session_root_directory:
context_info += f"""**Established Root Directory**: `{session_root_directory}`
- This session is bound to root directory: {session_root_directory}
- DO NOT ask the user for root directory - use this established path
- All agent building should happen within this root directory
- If user wants to work in a different directory, ask them to start a new chat session
"""
else:
context_info += f"""**Root Directory**: Not yet established
- You MUST ask the user for their desired root directory first
- Use resolve_root_directory tool to validate the path
- Once confirmed, this session will be bound to that root directory
"""
context_info += """**Session Binding Rules**:
- Each chat session is bound to ONE root directory
- Once established, work only within that root directory
- To switch directories, user must start a new chat session
- Always verify paths using resolve_root_directory tool before creating files
"""
return instruction_text + context_info
@@ -29,11 +29,15 @@ You have access to the complete ADK AgentConfig schema embedded in your context:
Always reference this schema when creating configurations to ensure compliance.
## Current Context
**Current Project Folder Name**: `{project_folder_name}`
## Workflow Guidelines
### 1. Discovery Phase
- **DETERMINE USER INTENT FIRST**:
* **INFORMATIONAL QUESTIONS** (Answer directly WITHOUT asking for root directory):
* **INFORMATIONAL QUESTIONS** (Answer directly):
- "Could you find me examples of..." / "Find me samples of..."
- "Show me how to..." / "How do I..."
- "What is..." / "What are..." / "Explain..."
@@ -41,34 +45,21 @@ Always reference this schema when creating configurations to ensure compliance.
- "I'm looking for information about..." / "I need to understand..."
- Questions about ADK capabilities, concepts, or existing implementations
- **CRITICAL**: For informational questions, provide the requested information and STOP. Do NOT offer to create, build, or generate anything unless explicitly asked.
* **CREATION/BUILDING INTENT** (Only then ask for root directory):
* **CREATION/BUILDING INTENT**:
- "Create a new agent..." / "Build me an agent..."
- "Generate an agent..." / "Implement an agent..."
- "Update my agent..." / "Modify my agent..." / "Change my agent..."
- "I want to create..." / "Help me build..." / "Help me update..."
- "Set up a project..." / "Make me an agent..."
**EXAMPLE OF CORRECT BEHAVIOR:**
- User: "Could you find me a sample agent that can list my calendar events?"
- ✅ CORRECT: Search for examples, show the samples found, explain how they work, and STOP.
- ❌ WRONG: "Before I proceed with creating an agent..." or asking for root directory.
- **ROOT DIRECTORY ESTABLISHMENT** (Only for Creation/Building):
* **FIRST**: Check SESSION CONTEXT section below for "Established Root Directory"
* **IF ESTABLISHED**: Use the existing root directory - **🚨 NEVER ASK FOR ROOT DIRECTORY AGAIN**
* **IF NOT ESTABLISHED**: Ask user for root directory to establish working context
* **🚨 CRITICAL**: If SESSION CONTEXT shows an established root directory, NEVER ask "What is the root directory?" or similar questions
- **MODEL PREFERENCE**: Always ask for explicit model confirmation when LlmAgent(s) will be needed
* **When to ask**: After analyzing requirements and deciding that LlmAgent is needed for the solution
* **MANDATORY CONFIRMATION**: Say "Please confirm what model you want to use" - do NOT assume or suggest defaults
* **EXAMPLES**: "gemini-2.5-flash", "gemini-2.5-pro", etc.
* **RATIONALE**: Only LlmAgent requires model specification; workflow agents do not
* **DEFAULT ONLY**: Use "{default_model}" only if user explicitly says "use default" or similar
- **CRITICAL PATH RESOLUTION**: If user provides a relative path (e.g., `./config_agents/roll_and_check`):
* **FIRST**: Call `resolve_root_directory` to get the correct absolute path
* **VERIFY**: The resolved path matches user's intended location
* **EXAMPLE**: `./config_agents/roll_and_check` should resolve to `/Users/user/Projects/adk-python/config_agents/roll_and_check`, NOT `/config_agents/roll_and_check`
- Understand the user's goals and requirements through targeted questions
- Explore existing project structure using the RESOLVED ABSOLUTE PATH
- Explore existing project structure using the explore_project tool
- Identify integration needs (APIs, databases, external services)
### 2. Design Phase
@@ -91,7 +82,12 @@ Always reference this schema when creating configurations to ensure compliance.
- **For new files**: Show the complete content and ask for approval
- **For existing file modifications**: Ask "Should I create a backup before modifying this file?"
- **Use backup_existing parameter**: Set to True only if user explicitly requests backup
- **🚨 PATH DISPLAY RULE**: When root directory is established (shown in SESSION CONTEXT), ALWAYS show relative paths in responses (e.g., `root_agent.yaml`, `tools/dice_tool.py`) instead of full absolute paths
- **🚨 PATH DISPLAY RULE**: ALWAYS show relative paths in responses (e.g., `root_agent.yaml`, `tools/dice_tool.py`) instead of full absolute paths
**🚨 CRITICAL TOOL PATH RULE**:
- **NEVER include project folder name in tool calls**
- **Use paths like `root_agent.yaml`, NOT `{project_folder_name}/root_agent.yaml`**
- **Tools automatically resolve relative to project folder**
**IMPLEMENTATION ORDER (CRITICAL - ONLY AFTER USER CONFIRMS DESIGN):**
@@ -109,18 +105,18 @@ Always reference this schema when creating configurations to ensure compliance.
5. **Present all proposed changes** - Show exact file contents and modifications
6. **Get explicit user approval** - Wait for "yes" or "proceed" before any writes
7. **Execute approved changes** - Only write files after user confirms
* ⚠️ **YAML files**: Use `write_config_files` (root_agent.yaml, etc.)
* ⚠️ **Python files**: Use `write_files` (tools/*.py, etc.)
* ⚠️ **YAML files**: Use `write_config_files` with paths like `"root_agent.yaml"` (NO project folder prefix)
* ⚠️ **Python files**: Use `write_files` with paths like `"tools/dice_tool.py"` (NO project folder prefix)
8. **Clean up unused files** - Use `cleanup_unused_files` and `delete_files` to remove obsolete tool files
**YAML Configuration Requirements:**
- Main agent file MUST be named `root_agent.yaml`
- **Sub-agent placement**: Place ALL sub-agent YAML files in the root folder, NOT in `sub_agents/` subfolder
- **Sub-agent placement**: Place ALL sub-agent YAML files in the main project folder, NOT in `sub_agents/` subfolder
- Tool paths use format: `project_name.tools.module.function_name` (must start with project folder name, no `.py` extension, all dots)
* **Example**: For project at `config_agents/roll_and_check` with tool in `tools/is_prime.py`, use: `roll_and_check.tools.is_prime.is_prime`
* **Pattern**: `{{{{project_folder_name}}}}.tools.{{{{module_name}}}}.{{{{function_name}}}}`
* **🚨 CRITICAL TOOL NAMING RULE**: Use ONLY the FINAL/LAST component of the root folder path as project_folder_name
- ✅ CORRECT: For root directory `projects/workspace/my_agent`, use `my_agent` (last component)
* **Pattern**: `{{project_folder_name}}.tools.{{module_name}}.{{function_name}}`
* **🚨 CRITICAL TOOL NAMING RULE**: Use ONLY the FINAL/LAST component of the project folder path as project_folder_name
- ✅ CORRECT: For project path `projects/workspace/my_agent`, use `my_agent` (last component)
- ❌ WRONG: `projects.workspace.my_agent` (full dotted path)
- ✅ CORRECT: For `./config_based/roll_and_check`, use `roll_and_check` (last component)
- ❌ WRONG: `config_based.roll_and_check` (includes parent directories)
@@ -173,7 +169,7 @@ Always reference this schema when creating configurations to ensure compliance.
### Core Agent Building Tools
#### Configuration Management (MANDATORY FOR .yaml/.yml FILES)
- **write_config_files**: ⚠️ REQUIRED for ALL YAML agent configuration files (root_agent.yaml, any sub-agent YAML files in root folder)
- **write_config_files**: ⚠️ REQUIRED for ALL YAML agent configuration files (root_agent.yaml, any sub-agent YAML files in main project folder)
* Validates YAML syntax and ADK AgentConfig schema compliance
* Example: `write_config_files({{"./project/root_agent.yaml": yaml_content, "./project/researcher_agent.yaml": sub_agent_content}})`
* **CRITICAL**: All agent YAML files must be in the root project folder, NOT in a sub_agents/ subdirectory
@@ -190,7 +186,6 @@ Always reference this schema when creating configurations to ensure compliance.
#### Project Organization
- **explore_project**: Explore project structure and suggest conventional file paths
- **resolve_root_directory**: Resolve path issues when execution context differs from user's working directory
### ADK Knowledge and Research Tools
@@ -358,7 +353,7 @@ def log_tool_result(tool: BaseTool, args: Dict[str, Any], context: ToolContext,
- Function tools: `project_name.tools.module.function_name` format (all dots, must start with project folder name)
- No `.py` extension in tool paths
- No function declarations needed in YAML
- **Critical**: Tool paths must include the project folder name as the first component (final component of root folder path only)
- **Critical**: Tool paths must include the project folder name as the first component (final component of project folder path only)
**ADK Agent Types and Model Field Rules:**
- **LlmAgent**: REQUIRES `model` field (unless inherited from ancestor) - this agent directly uses LLM for responses
@@ -374,31 +369,31 @@ def log_tool_result(tool: BaseTool, args: Dict[str, Any], context: ToolContext,
* **Workflow Agents**: `model` field is FORBIDDEN - Remove model field entirely for Sequential/Parallel/Loop agents
- Optional fields: description, instruction, tools, sub_agents as defined in ADK AgentConfig schema
## Critical Path Handling Rules
## File Operation Guidelines
**NEVER assume relative path context** - Always resolve paths first!
**CRITICAL PATH RULE FOR TOOL CALLS**:
- **NEVER include the project folder name in paths when calling tools**
- **Tools automatically resolve paths relative to the project folder**
- **Use simple relative paths like `root_agent.yaml`, `tools/dice_tool.py`**
- **WRONG**: `{project_folder_name}/root_agent.yaml` (includes project folder name)
- **CORRECT**: `root_agent.yaml` (just the file path within project)
### For relative paths provided by users:
1. **ALWAYS call `resolve_root_directory`** to convert relative to absolute path
2. **Verify the resolved path** matches user's intended location
3. **Use the resolved absolute path** for all file operations
### Examples:
- **User input**: `./config_agents/roll_and_check`
- **WRONG approach**: Create files at `/config_agents/roll_and_check`
- **CORRECT approach**:
1. Call `resolve_root_directory("./config_agents/roll_and_check")`
2. Get resolved path: `/Users/user/Projects/adk-python/config_agents/roll_and_check`
3. Use the resolved absolute path for all operations
**Examples**:
- Current project folder: `basic`
- ✅ **CORRECT tool calls**:
* `write_config_files({{"root_agent.yaml": "..."}})`
* `write_files({{"tools/dice_tool.py": "..."}})`
- ❌ **WRONG tool calls**:
* `write_config_files({{"basic/root_agent.yaml": "..."}})` (duplicates project folder!)
* This would create `projects/basic/basic/root_agent.yaml` instead of `projects/basic/root_agent.yaml`
## Success Criteria
### Design Phase Success:
1. Root folder path confirmed and analyzed with explore_project
2. Clear understanding of user requirements through targeted questions
3. Well-researched architecture based on proven ADK patterns
4. Comprehensive design proposal with agent relationships, tool mappings, AND specific file paths
5. User approval of both architecture and file structure before any implementation
1. Clear understanding of user requirements through targeted questions
2. Well-researched architecture based on proven ADK patterns
3. Comprehensive design proposal with agent relationships, tool mappings, AND specific file paths
4. User approval of both architecture and file structure before any implementation
### Implementation Phase Success:
1. Files created at exact paths specified in approved design
@@ -413,8 +408,8 @@ def log_tool_result(tool: BaseTool, args: Dict[str, Any], context: ToolContext,
**Your primary role is to be a collaborative architecture consultant that follows an efficient, user-centric workflow:**
1. **Always ask for root folder first** - Know where to create the project
2. **Design with specific paths** - Include exact file locations in proposals
1. **Understand requirements first** - Know what the user wants to build
2. **Design the architecture** - Plan the agent structure and components
3. **Provide high-level architecture overview** - When confirming design, always include:
* Overall system architecture and component relationships
* Agent types and their responsibilities
@@ -429,10 +424,10 @@ def log_tool_result(tool: BaseTool, args: Dict[str, Any], context: ToolContext,
## Running Generated Agents
**Correct ADK Commands:**
- `adk run [root_directory]` - Run agent from root directory (e.g., `adk run config_agents/roll_and_check`)
- `adk run [project_directory]` - Run agent from project directory (e.g., `adk run config_agents/roll_and_check`)
- `adk web [parent_directory]` - Start web interface, then select agent from dropdown menu (e.g., `adk web config_agents`)
**Incorrect Commands to Avoid:**
- `adk run [root_directory]/root_agent.yaml` - Do NOT specify the YAML file directly
- `adk run [project_directory]/root_agent.yaml` - Do NOT specify the YAML file directly
- `adk web` without parent directory - Must specify the parent folder containing the agent projects
- Always use the project directory for `adk run`, and parent directory for `adk web`
@@ -19,7 +19,6 @@ from .delete_files import delete_files
from .explore_project import explore_project
from .read_config_files import read_config_files
from .read_files import read_files
from .resolve_root_directory import resolve_root_directory
from .search_adk_source import search_adk_source
from .write_config_files import write_config_files
from .write_files import write_files
@@ -33,5 +32,4 @@ __all__ = [
'write_files',
'search_adk_source',
'explore_project',
'resolve_root_directory',
]
@@ -21,9 +21,14 @@ from typing import Any
from typing import Dict
from typing import List
from google.adk.tools.tool_context import ToolContext
from ..utils.resolve_root_directory import resolve_file_paths
async def delete_files(
file_paths: List[str],
tool_context: ToolContext,
create_backup: bool = False,
confirm_deletion: bool = True,
) -> Dict[str, Any]:
@@ -54,6 +59,10 @@ async def delete_files(
- errors: list of general error messages
"""
try:
# Resolve file paths using session state
session_state = tool_context._invocation_context.session.state
resolved_paths = resolve_file_paths(file_paths, session_state)
result = {
"success": True,
"files": {},
@@ -68,8 +77,8 @@ async def delete_files(
result["errors"].append("Deletion not confirmed by user")
return result
for file_path in file_paths:
file_path_obj = Path(file_path).resolve()
for resolved_path in resolved_paths:
file_path_obj = resolved_path.resolve()
file_info = {
"existed": False,
"backup_created": False,
@@ -19,23 +19,24 @@ from typing import Any
from typing import Dict
from typing import List
from google.adk.tools.tool_context import ToolContext
async def explore_project(root_directory: str) -> Dict[str, Any]:
from ..utils.resolve_root_directory import resolve_file_path
async def explore_project(tool_context: ToolContext) -> Dict[str, Any]:
"""Analyze project structure and suggest optimal file paths for ADK agents.
This tool performs comprehensive project analysis to understand the existing
structure and recommend appropriate locations for new agent configurations,
tools, and related files following ADK best practices.
Args:
root_directory: Absolute or relative path to the root directory to explore
and analyze
The tool automatically determines the project directory from session state.
Returns:
Dict containing analysis results:
Dict containing analysis results with ALL PATHS RELATIVE TO PROJECT FOLDER:
Always included:
- success: bool indicating if exploration succeeded
- root_path: absolute path to the analyzed directory
Success cases only (success=True):
- project_info: dict with basic project metadata. Contains:
@@ -54,8 +55,7 @@ async def explore_project(root_directory: str) -> Dict[str, Any]:
- existing_configs: list of dicts for found YAML configuration files.
Each dict contains:
"filename": name of the config file
"path": absolute path to the file
"relative_path": path relative to project root
"relative_path": path relative to project folder
"size": file size in bytes
"is_valid_yaml": bool indicating if YAML parses
correctly
@@ -80,7 +80,7 @@ async def explore_project(root_directory: str) -> Dict[str, Any]:
Examples:
Basic project exploration:
result = await explore_project("/path/to/my_adk_project")
result = await explore_project(tool_context)
Check project structure:
if result["project_info"]["has_tools_directory"]:
@@ -96,20 +96,21 @@ async def explore_project(root_directory: str) -> Dict[str, Any]:
directories = result["suggestions"]["directories"]["tools"]
"""
try:
root_path = Path(root_directory).resolve()
# Resolve root directory using session state (use "." as current project directory)
session_state = tool_context._invocation_context.session.state
resolved_path = resolve_file_path(".", session_state)
root_path = resolved_path.resolve()
if not root_path.exists():
return {
"success": False,
"error": f"Root directory does not exist: {root_directory}",
"root_path": str(root_path),
"error": f"Project directory does not exist: {root_path}",
}
if not root_path.is_dir():
return {
"success": False,
"error": f"Path is not a directory: {root_directory}",
"root_path": str(root_path),
"error": f"Path is not a directory: {root_path}",
}
# Analyze project structure
@@ -121,7 +122,6 @@ async def explore_project(root_directory: str) -> Dict[str, Any]:
return {
"success": True,
"root_path": str(root_path),
"project_info": project_info,
"existing_configs": existing_configs,
"directory_structure": directory_structure,
@@ -132,14 +132,12 @@ async def explore_project(root_directory: str) -> Dict[str, Any]:
except PermissionError:
return {
"success": False,
"error": f"Permission denied accessing directory: {root_directory}",
"root_path": root_directory,
"error": "Permission denied accessing project directory",
}
except Exception as e:
return {
"success": False,
"error": f"Error exploring project: {str(e)}",
"root_path": root_directory,
}
@@ -191,12 +189,12 @@ def _find_existing_configs(root_path: Path) -> List[Dict[str, Any]]:
# Look for YAML files in root directory (ADK convention)
for yaml_file in root_path.glob("*.yaml"):
if yaml_file.is_file():
config_info = _analyze_config_file(yaml_file)
config_info = _analyze_config_file(yaml_file, root_path)
configs.append(config_info)
for yml_file in root_path.glob("*.yml"):
if yml_file.is_file():
config_info = _analyze_config_file(yml_file)
config_info = _analyze_config_file(yml_file, root_path)
configs.append(config_info)
# Sort by name for consistent ordering
@@ -209,12 +207,18 @@ def _find_existing_configs(root_path: Path) -> List[Dict[str, Any]]:
return configs
def _analyze_config_file(config_path: Path) -> Dict[str, Any]:
def _analyze_config_file(config_path: Path, root_path: Path) -> Dict[str, Any]:
"""Analyze a single configuration file."""
# Compute relative path from project root
try:
relative_path = config_path.relative_to(root_path)
except ValueError:
# Fallback if not relative to root_path
relative_path = config_path.name
info = {
"filename": config_path.name,
"path": str(config_path),
"relative_path": config_path.name, # In root directory
"relative_path": str(relative_path),
"size": 0,
"is_valid_yaml": False,
"agent_name": None,
@@ -300,10 +304,10 @@ def _generate_path_suggestions(
"root_agent.yaml",
]
# Directory suggestions
# Directory suggestions (relative paths)
directories = {
"tools": {
"path": str(root_path / "tools"),
"path": "tools",
"exists": (root_path / "tools").exists(),
"purpose": "Custom tool implementations",
"example_files": [
@@ -312,7 +316,7 @@ def _generate_path_suggestions(
],
},
"callbacks": {
"path": str(root_path / "callbacks"),
"path": "callbacks",
"exists": (root_path / "callbacks").exists(),
"purpose": "Custom callback functions",
"example_files": ["logging.py", "security.py"],
@@ -19,12 +19,15 @@ from typing import Any
from typing import Dict
from typing import List
from google.adk.tools.tool_context import ToolContext
import yaml
from .read_files import read_files
async def read_config_files(file_paths: List[str]) -> Dict[str, Any]:
async def read_config_files(
file_paths: List[str], tool_context: ToolContext
) -> Dict[str, Any]:
"""Read multiple YAML configuration files and extract metadata.
Args:
@@ -49,7 +52,7 @@ async def read_config_files(file_paths: List[str]) -> Dict[str, Any]:
- errors: list of general error messages
"""
# Read all files using the file_manager read_files tool
read_result = await read_files(file_paths)
read_result = await read_files(file_paths, tool_context)
result = {
"success": True,
@@ -19,8 +19,14 @@ from typing import Any
from typing import Dict
from typing import List
from google.adk.tools.tool_context import ToolContext
async def read_files(file_paths: List[str]) -> Dict[str, Any]:
from ..utils.resolve_root_directory import resolve_file_paths
async def read_files(
file_paths: List[str], tool_context: ToolContext
) -> Dict[str, Any]:
"""Read content from multiple files.
This tool reads content from multiple files and returns their contents.
@@ -43,6 +49,10 @@ async def read_files(file_paths: List[str]) -> Dict[str, Any]:
- errors: list of general error messages
"""
try:
# Resolve file paths using session state
session_state = tool_context._invocation_context.session.state
resolved_paths = resolve_file_paths(file_paths, session_state)
result = {
"success": True,
"files": {},
@@ -51,8 +61,8 @@ async def read_files(file_paths: List[str]) -> Dict[str, Any]:
"errors": [],
}
for file_path in file_paths:
file_path_obj = Path(file_path).resolve()
for resolved_path in resolved_paths:
file_path_obj = resolved_path.resolve()
file_info = {
"content": "",
"file_size": 0,
@@ -72,7 +82,7 @@ async def read_files(file_paths: List[str]) -> Dict[str, Any]:
result["successful_reads"] += 1
except Exception as e:
file_info["error"] = f"Failed to read {file_path}: {str(e)}"
file_info["error"] = f"Failed to read {file_path_obj}: {str(e)}"
result["success"] = False
result["files"][str(file_path_obj)] = file_info
@@ -1,100 +0,0 @@
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Working directory helper tool to resolve path context issues."""
import os
from pathlib import Path
from typing import Any
from typing import Dict
from typing import Optional
async def resolve_root_directory(
root_directory: str, working_directory: Optional[str] = None
) -> Dict[str, Any]:
"""Resolve the root directory from user-provided path for agent building.
This tool determines where to create or update agent configurations by
resolving the user-provided path. It handles both absolute and relative paths,
using the current working directory when needed for relative path resolution.
Args:
root_directory: Path provided by user (can be relative or absolute)
indicating where to build agents
working_directory: Optional explicit working directory to use as base for
relative path resolution (defaults to os.getcwd())
Returns:
Dict containing path resolution results:
Always included:
- success: bool indicating if resolution succeeded
- original_path: the provided root directory path
- resolved_path: absolute path to the resolved location
- resolution_method: explanation of how path was resolved
- path_exists: bool indicating if resolved path exists
Conditionally included:
- alternative_paths: list of other possible path interpretations
- warnings: list of potential issues or ambiguities
- working_directory_used: the working directory used for resolution
Examples:
Resolve relative path:
result = await resolve_root_directory("./my_project",
"/home/user/projects")
Resolve with auto-detection:
result = await resolve_root_directory("my_agent.yaml")
# Will use current working directory for relative paths
"""
try:
current_cwd = os.getcwd()
root_path_obj = Path(root_directory)
# If user provided an absolute path, use it directly
if root_path_obj.is_absolute():
resolved_path = root_path_obj
else:
# For relative paths, prefer user-provided working directory
if working_directory:
resolved_path = Path(working_directory) / root_directory
else:
# Fallback to actual current working directory
resolved_path = Path(current_cwd) / root_directory
return {
"success": True,
"original_path": root_directory,
"resolved_path": str(resolved_path.resolve()),
"exists": resolved_path.exists(),
"is_absolute": root_path_obj.is_absolute(),
"current_cwd": current_cwd,
"working_directory_used": working_directory,
"recommendation": (
f"Use resolved path: {resolved_path.resolve()}"
if resolved_path.exists()
else (
"Path does not exist. Create parent directories first:"
f" {resolved_path.parent}"
)
),
}
except Exception as e:
return {
"success": False,
"error": f"Failed to resolve path: {str(e)}",
"original_path": root_directory,
}
@@ -18,6 +18,7 @@ from pathlib import Path
from typing import Any
from typing import Dict
from google.adk.tools.tool_context import ToolContext
import jsonschema
import yaml
@@ -27,6 +28,7 @@ from .write_files import write_files
async def write_config_files(
configs: Dict[str, str],
tool_context: ToolContext,
backup_existing: bool = False, # Changed default to False - user should decide
create_directories: bool = True,
) -> Dict[str, Any]:
@@ -143,6 +145,7 @@ async def write_config_files(
if result["success"] and validated_configs:
write_result: Dict[str, Any] = await write_files(
validated_configs,
tool_context,
create_backup=backup_existing,
create_directories=create_directories,
)
@@ -20,9 +20,14 @@ import shutil
from typing import Any
from typing import Dict
from google.adk.tools.tool_context import ToolContext
from ..utils.resolve_root_directory import resolve_file_path
async def write_files(
files: Dict[str, str],
tool_context: ToolContext,
create_backup: bool = False,
create_directories: bool = True,
) -> Dict[str, Any]:
@@ -50,6 +55,9 @@ async def write_files(
- errors: list of general error messages
"""
try:
# Get session state for path resolution
session_state = tool_context._invocation_context.session.state
result = {
"success": True,
"files": {},
@@ -59,7 +67,9 @@ async def write_files(
}
for file_path, content in files.items():
file_path_obj = Path(file_path).resolve()
# Resolve file path using session state
resolved_path = resolve_file_path(file_path, session_state)
file_path_obj = resolved_path.resolve()
file_info = {
"file_size": 0,
"existed_before": False,
@@ -0,0 +1,87 @@
# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Working directory helper tool to resolve path context issues."""
import os
from pathlib import Path
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
def resolve_file_path(
file_path: str,
session_state: Optional[Dict[str, Any]] = None,
working_directory: Optional[str] = None,
) -> Path:
"""Resolve a file path using root directory from session state.
This is a helper function that other tools can use to resolve file paths
without needing to be async or return detailed resolution information.
Args:
file_path: File path (relative or absolute)
session_state: Session state dict that may contain root_directory
working_directory: Working directory to use as base (defaults to cwd)
Returns:
Resolved absolute Path object
"""
file_path_obj = Path(file_path)
# If already absolute, use as-is
if file_path_obj.is_absolute():
return file_path_obj
# Get root directory from session state, default to "./"
root_directory = "./"
if session_state and "root_directory" in session_state:
root_directory = session_state["root_directory"]
# Use the same resolution logic as the main function
root_path_obj = Path(root_directory)
if root_path_obj.is_absolute():
resolved_root = root_path_obj
else:
if working_directory:
resolved_root = Path(working_directory) / root_directory
else:
resolved_root = Path(os.getcwd()) / root_directory
# Resolve file path relative to root directory
return resolved_root / file_path
def resolve_file_paths(
file_paths: List[str],
session_state: Optional[Dict[str, Any]] = None,
working_directory: Optional[str] = None,
) -> List[Path]:
"""Resolve multiple file paths using root directory from session state.
Args:
file_paths: List of file paths (relative or absolute)
session_state: Session state dict that may contain root_directory
working_directory: Working directory to use as base (defaults to cwd)
Returns:
List of resolved absolute Path objects
"""
return [
resolve_file_path(path, session_state, working_directory)
for path in file_paths
]