Environment Module ¶

The Environment module provides an interface for the agent to interact with the outside world. It handles tool execution, approval requests, and resource management.

Overview ¶

The Environment class is a network implementation of the ToolEnvironmentConnection interface. It routes all calls through a specific Protocol instance that has a registered environment. This allows the agent to:

Execute tools with the given parameters
Request approval for tool usage
Clean up resources when done

Key Concepts ¶

Tool Execution ¶

The environment executes tools through the execute_tool() method, which:

Takes a ToolCallPart containing the tool name and parameters
Sends a request to the protocol to execute the tool
Returns a response containing the tool execution result

Approval Requests ¶

Before executing a tool, the environment may request approval through the approve_tool_use() method, which:

Takes a ToolCallPart containing the tool name and parameters
Sends a request to the protocol to approve the tool
Returns a response indicating whether the tool was approved

Response Types ¶

The environment uses several response types to communicate with the agent:

EnvironmentResponse: A generic response container for environment operations
WorkingDirectoryResponse: Response data for get_working_directory
ToolMetadataResponse: Response data for get_all_tool_metadata
ToolNamesResponse: Response data for get_available_tool_names
ApprovalResponse: Response data for approval operations
ValidatePathResponse: Response data for validate_path operation

Implementation Details ¶

The Environment class is initialized with a Protocol instance, which it uses to communicate with the client. It maintains a logger for logging events and a cache for tool metadata.

The execute_tool() and approve_tool_use() methods use the protocol's request_response() method to send requests and receive responses.

Usage Example ¶

Here's a simplified example of how to use the Environment:

# Create an environment
environment = Environment(protocol)

# Execute a tool
tool_call = ToolCallPart(
    tool_call_id="123",
    tool_name="write_file",
    args={"file_path": "example.txt", "content": "Hello, world!"}
)
response = await environment.execute_tool(tool_call)

# Check if the tool execution was successful
if response.success:
    print(f"Tool execution successful: {response.data}")
else:
    print(f"Tool execution failed: {response.message}")

Important Considerations ¶

When working with the Environment:

Protocol Initialization: Make sure the protocol is initialized before creating the environment.
Tool Approval: Some tools may require approval before execution. Always check if a tool requires approval and request it if needed.
Error Handling: Handle errors from tool execution appropriately. The environment returns a success flag and an error message if the tool execution fails.
Resource Cleanup: Call the cleanup() method when done with the environment to clean up any resources.

T = TypeVar('T')

success: bool

message `class-attribute` `instance-attribute` ¶

message: str | None = None

data `class-attribute` `instance-attribute` ¶

data: T | None = None

model_config `class-attribute` `instance-attribute` ¶

model_config = {'arbitrary_types_allowed': True}

directory: str

metadata: List[ToolMetadata]

tool_names: List[str]

approved: bool

feedback `class-attribute` `instance-attribute` ¶

feedback: Optional[str] = Field(default=None, description='Feedback message')

images `class-attribute` `instance-attribute` ¶

images: List[ImagePart] = Field(default_factory=list, description='List of image URLs')

valid: bool

Environment ¶

Environment(protocol: Protocol)

Network implementation of ToolEnvironmentConnection.

This class provides a network implementation of the ToolEnvironmentConnection interface by routing all calls through a specific Protocol instance that has a registered environment.

Initialize the connection.

Parameters:

protocol (Protocol) –

Protocol instance to use for communication

Source code in azad/environment.py

def __init__(self, protocol: Protocol):
    """Initialize the connection.

    Args:
        protocol: Protocol instance to use for communication
    """
    self._protocol = protocol
    self.logger = logging.getLogger(__name__)
    self._cached_tool_metadata = None

Attributes ¶

logger `instance-attribute` ¶

logger = getLogger(__name__)

Functions ¶

execute_tool `async` ¶

execute_tool(tool_call: ToolCallPart) -> EnvironmentResponse[ToolResultPart]

Execute a tool with the given parameters using request/response pattern.

Parameters:

tool_name –

Name of the tool to execute
parameters –

Tool parameters as primitive types

Returns:

EnvironmentResponse[ToolResultPart] –

EnvironmentResponse[ToolResultPart]: Response containing tool execution result

Source code in azad/environment.py

async def execute_tool(self, tool_call: ToolCallPart) -> EnvironmentResponse[ToolResultPart]:
    """Execute a tool with the given parameters using request/response pattern.

    Args:
        tool_name: Name of the tool to execute
        parameters: Tool parameters as primitive types

    Returns:
        EnvironmentResponse[ToolResultPart]: Response containing tool execution result
    """
    result = await self._protocol.request_response({
        "operation": "tool_env_execute_tool",
        "tool_call": tool_call.model_dump()
    })
    if not result.get("success", False):
        return EnvironmentResponse[ToolResultPart](
            success=False,
            message=result.get("error", "Failed to execute tool"),
            data=None
        )
    return EnvironmentResponse[ToolResultPart](
        success=True,
        message="Tool execution successful",
        data=ToolResultPart.model_validate(result.get("data", {}))
    )

approve_tool_use `async` ¶

approve_tool_use(tool_call: ToolCallPart) -> EnvironmentResponse[ApprovalResponse]

Request approval for tool usage.

Parameters:

tool_name –

Name of the tool to approve
parameters –

Tool parameters to approve

Returns:

EnvironmentResponse[ApprovalResponse] –

EnvironmentResponse[ApprovalResponse]: Response containing approval status

Source code in azad/environment.py

async def approve_tool_use(self, tool_call: ToolCallPart) -> EnvironmentResponse[ApprovalResponse]:
    """Request approval for tool usage.

    Args:
        tool_name: Name of the tool to approve
        parameters: Tool parameters to approve

    Returns:
        EnvironmentResponse[ApprovalResponse]: Response containing approval status
    """
    if not tool_call:
        self.logger.error("NetworkToolEnvironmentConnection.approve_tool_use called with no tool_name")
        return EnvironmentResponse[ApprovalResponse](
            success=False,
            message="No tool name provided",
            data=None
        )

    result = await self._protocol.request_response({
        "operation": "tool_env_approve_tool",
        "tool_call": tool_call.model_dump()
    })
    return EnvironmentResponse[ApprovalResponse].model_validate(result)

cleanup `async` ¶

cleanup() -> None

Clean up resources used by the connection.

Source code in azad/environment.py

async def cleanup(self) -> None:
    """Clean up resources used by the connection."""
    # Nothing to clean up for network connection
    pass