8.3 KiB
catalog_title, catalog_description, catalog_icon, catalog_tags
| catalog_title | catalog_description | catalog_icon | catalog_tags | ||
|---|---|---|---|---|---|
| Environment Simulator | Intercept tool calls to mock responses and inject faults for offline evaluation | /integrations/assets/environmentsimulator.png |
|
Environment Simulator plugin for ADK
The Environment Simulator plugin intercepts tool calls during agent execution. It allows you to mock external APIs and inject faults without altering your agent code, which enables hermetic, deterministic testing, cost savings, and offline evaluation for ADK agents.
Use cases
- Hermetic Offline Testing: Evaluate agent behavior without requiring live third-party services, databases, or API keys.
- Fault Injection and Resilience: Programmatically inject HTTP status codes, error messages, and latency to verify agent recovery paths.
- Stateful Tool Simulation: Use an LLM to maintain consistent state across sequential tool calls, such as consuming IDs generated by preceding tool calls.
- Cost and Rate-Limit Optimization: Replace expensive downstream API invocations with schema-driven mock responses during CI/CD test runs.
Prerequisites
- Python 3.10 or higher
google-adkpackage version 1.24.0 or higher- Access to Gemini models configured in your environment, for LLM-driven mock generation and schema analysis
Installation
Environment Simulation is included directly with ADK Python:
pip install google-adk
Use with agent
The EnvironmentSimulationFactory provides two integration mechanisms: as an application-level plugin, for global evaluation runs, or as an agent-level callback.
=== "App plugin"
```python
from google.adk.agents import Agent
from google.adk.apps import App
from google.adk.tools import FunctionTool
from google.adk.tools.environment_simulation import EnvironmentSimulationFactory
from google.adk.tools.environment_simulation.environment_simulation_config import (
EnvironmentSimulationConfig,
InjectedError,
InjectionConfig,
MockStrategy,
ToolSimulationConfig,
)
# 1. Define a tool to be simulated
def get_ticket(ticket_id: str) -> dict[str, str]:
"""Retrieves support ticket details by ID."""
return {"ticket_id": ticket_id, "status": "open"}
ticket_tool = FunctionTool(get_ticket)
# 2. Define tool simulation configuration with fault injection and LLM mock fallback
tool_config = ToolSimulationConfig(
tool_name="get_ticket",
injection_configs=[
InjectionConfig(
injection_probability=0.25,
match_args={"ticket_id": "FAIL_123"},
injected_error=InjectedError(
injected_http_error_code=500,
error_message="Internal Server Error",
),
)
],
mock_strategy_type=MockStrategy.MOCK_STRATEGY_TOOL_SPEC,
)
simulation_config = EnvironmentSimulationConfig(
tool_simulation_configs=[tool_config],
simulation_model="gemini-2.5-flash",
)
# 3. Create the plugin
simulation_plugin = EnvironmentSimulationFactory.create_plugin(simulation_config)
# 4. Attach the plugin to the application
root_agent = Agent(
name="support_agent",
model="gemini-2.5-flash",
instruction="You are a customer support agent.",
tools=[ticket_tool],
)
app = App(
name="support_app",
root_agent=root_agent,
plugins=[simulation_plugin],
)
```
=== "Agent callback"
```python
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.tools.environment_simulation import EnvironmentSimulationFactory
from google.adk.tools.environment_simulation.environment_simulation_config import (
EnvironmentSimulationConfig,
MockStrategy,
ToolSimulationConfig,
)
# 1. Define a tool to be simulated
def database_query(query: str) -> dict[str, str]:
"""Executes a database query."""
return {"result": "sample_data"}
db_tool = FunctionTool(database_query)
# 2. Define simulation configuration
tool_config = ToolSimulationConfig(
tool_name="database_query",
mock_strategy_type=MockStrategy.MOCK_STRATEGY_TOOL_SPEC,
)
simulation_config = EnvironmentSimulationConfig(
tool_simulation_configs=[tool_config],
simulation_model="gemini-2.5-flash",
)
# 3. Create the before_tool_callback
callback = EnvironmentSimulationFactory.create_callback(simulation_config)
# 4. Attach directly to the agent
root_agent = Agent(
name="analyst_agent",
model="gemini-2.5-flash",
instruction="Analyze operational database metrics.",
tools=[db_tool],
before_tool_callback=callback,
)
```
Available components
The factory and plugin expose the following integration points:
| Component | Type | Description |
|---|---|---|
EnvironmentSimulationFactory.create_plugin |
Method | Creates an EnvironmentSimulationPlugin instance to attach to an App. |
EnvironmentSimulationFactory.create_callback |
Method | Generates an async before_tool_callback to attach directly to an Agent. |
EnvironmentSimulationPlugin |
Plugin | ADK plugin implementing before_tool_callback to intercept calls before tool execution. |
Configuration
The simulator is configured using structured Pydantic models from google.adk.tools.environment_simulation.environment_simulation_config:
EnvironmentSimulationConfig
The root configuration object holds simulation parameters and internal model settings:
| Field | Type | Default | Description |
|---|---|---|---|
tool_simulation_configs |
List[ToolSimulationConfig] |
[] |
List of simulation rules per tool. Must contain at least one config. |
simulation_model |
str |
gemini-flash-latest |
The model used for tool connection analysis and mock generation. |
simulation_model_configuration |
genai_types.GenerateContentConfig |
ThinkingConfig(10240) |
Generation parameters passed to the simulation model. |
tracing |
Optional[str] |
None |
Historical trace data in JSON string format providing context for mocks. |
environment_data |
Optional[str] |
None |
Seed environment data, such as mock database tables, passed to mock strategies. |
ToolSimulationConfig
Defines the simulation policy for an individual tool:
| Field | Type | Default | Description |
|---|---|---|---|
tool_name |
str |
(Required) | The exact name of the tool to simulate. |
injection_configs |
List[InjectionConfig] |
[] |
List of fault or response injections evaluated in order. |
mock_strategy_type |
MockStrategy |
UNSPECIFIED |
Fallback strategy used when no injection condition is met. |
InjectionConfig and InjectedError
Controls conditional fault injection and deterministic response overrides:
injection_probability: A float from0.0to1.0specifying the likelihood of triggering the injection.match_args: Optional dictionary of arguments that must match the tool input for the injection to fire.injected_latency_seconds: Float specifying sleep duration before returning the response.random_seed: Optional integer seed for deterministic probabilistic evaluation runs.injected_error: AnInjectedErrorinstance containinginjected_http_error_code(int) anderror_message(str).injected_response: A dictionary returning a fixed mock payload. Exactly one ofinjected_errororinjected_responsemust be set.
Stateful tool connection analysis
When MockStrategy.MOCK_STRATEGY_TOOL_SPEC is enabled, the simulation engine uses ToolConnectionAnalyzer with the configured simulation model to inspect tool schemas and build a ToolConnectionMap.
The analyzer identifies shared parameter dependencies across tools, such as an ID created by create_ticket and consumed by get_ticket, maintaining consistent state values throughout multi-turn agent interactions.