diff --git a/.env.example b/.env.example
index 940d602..21165a8 100644
--- a/.env.example
+++ b/.env.example
@@ -39,6 +39,11 @@ OPENROUTER_API_KEY=replace-me
# Get a free API key at: https://brave.com/search/api/
# BRAVE_SEARCH_API_KEY=your_brave_search_api_key_here
+# Google Maps Routes API (optional distance, ETA, and traffic tools)
+# Use a separate server-side key restricted to the Routes API and apply
+# project quotas or budget alerts before enabling traffic-aware requests.
+# GOOGLE_MAPS_ROUTES_API_KEY=your_google_maps_routes_api_key_here
+
# ---------------------------------------------------------------------------
# Browser Use Cloud (optional browser automation for the agent)
# ---------------------------------------------------------------------------
diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml
index 0c35c5a..5686886 100644
--- a/.github/workflows/docker-publish.yml
+++ b/.github/workflows/docker-publish.yml
@@ -109,6 +109,7 @@ jobs:
TELEGRAM_TOOL_NOTIFICATIONS: ${{ secrets.TELEGRAM_TOOL_NOTIFICATIONS }}
EXA_API_KEY: ${{ secrets.EXA_API_KEY }}
BRAVE_SEARCH_API_KEY: ${{ secrets.BRAVE_SEARCH_API_KEY }}
+ GOOGLE_MAPS_ROUTES_API_KEY: ${{ secrets.GOOGLE_MAPS_ROUTES_API_KEY }}
BROWSER_USE_API_KEY: ${{ secrets.BROWSER_USE_API_KEY }}
SANDBOX_ENABLED: ${{ secrets.SANDBOX_ENABLED }}
SANDBOX_DOMAIN: ${{ secrets.SANDBOX_DOMAIN }}
@@ -171,6 +172,7 @@ jobs:
TELEGRAM_TOOL_NOTIFICATIONS \
EXA_API_KEY \
BRAVE_SEARCH_API_KEY \
+ GOOGLE_MAPS_ROUTES_API_KEY \
BROWSER_USE_API_KEY \
SANDBOX_ENABLED \
SANDBOX_DOMAIN \
diff --git a/docs/architecture.md b/docs/architecture.md
index c592663..c7c2c47 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -59,6 +59,8 @@ Optional tools follow the project's cloud-first principle:
- models use OpenRouter or Google;
- search can use Exa with Brave as a fallback;
+- distance, ETA, alternatives, and point-in-time traffic can use Google Maps
+ Routes;
- browser automation can use Browser Use Cloud;
- vector memory can use Qdrant Cloud; and
- code execution can use an OpenSandbox server.
diff --git a/docs/base-infra/environment-variables.md b/docs/base-infra/environment-variables.md
index e6864d0..a70dcfa 100644
--- a/docs/base-infra/environment-variables.md
+++ b/docs/base-infra/environment-variables.md
@@ -92,6 +92,7 @@ The token is required and format-validated when Telegram is enabled.
| --- | --- | --- |
| `EXA_API_KEY` | unset | Primary Exa search integration |
| `BRAVE_SEARCH_API_KEY` | unset | Brave search fallback |
+| `GOOGLE_MAPS_ROUTES_API_KEY` | unset | Distance, ETA, and traffic estimates |
| `BROWSER_USE_API_KEY` | unset | Browser Use Cloud automation |
These integrations are optional. Their absence should not replace the required
diff --git a/docs/google-maps-routes.md b/docs/google-maps-routes.md
new file mode 100644
index 0000000..df80691
--- /dev/null
+++ b/docs/google-maps-routes.md
@@ -0,0 +1,87 @@
+# Google Maps Routes
+
+Blacki can use the Google Maps Routes API for fresh distance, ETA, traffic, and
+route-comparison questions. The integration is optional and remains disabled
+unless `GOOGLE_MAPS_ROUTES_API_KEY` is configured.
+
+## Configure the API
+
+1. Enable billing and the Routes API in the Google Cloud project.
+2. Create a server-side API key dedicated to this integration.
+3. Restrict the key to the Routes API and to the deployed server where
+ practical.
+4. Configure quotas and billing alerts.
+5. Set the key in `.env`:
+
+```dotenv
+GOOGLE_MAPS_ROUTES_API_KEY=replace-me
+```
+
+Do not reuse the Gemini `GOOGLE_API_KEY`. Separating the keys allows independent
+restrictions, rotation, and quotas.
+
+For the repository's production deployment workflow, add the same value as the
+GitHub Actions environment secret `GOOGLE_MAPS_ROUTES_API_KEY`. Code-quality
+jobs do not need this secret because provider calls are mocked.
+
+## Agent capabilities
+
+`get_route_estimate` returns:
+
+- distance in meters and kilometers;
+- traffic-aware and static durations;
+- calculated traffic delay;
+- optional alternate routes;
+- provider fallback and route warnings;
+- Google Maps attribution.
+
+`compare_route_scenarios` compares up to five explicitly named scenarios for
+the same endpoints. A scenario can vary departure time, travel mode, traffic
+model, and avoid options. Requests run with bounded concurrency to limit burst
+traffic and cost.
+
+For a current driving estimate, the agent uses:
+
+- travel mode `DRIVE`;
+- departure time `now`; and
+- traffic model `BEST_GUESS`.
+
+`OPTIMISTIC` and `PESSIMISTIC` are also supported for driving. Non-driving
+modes use `NONE` because Google traffic models are limited to driving routes.
+
+## Location and time inputs
+
+Plain location strings are sent as addresses. A known Google place ID can be
+supplied using the `place_id:` prefix:
+
+```text
+place_id:ChIJ...
+```
+
+Future departure times must be RFC 3339 timestamps containing a timezone
+offset. This keeps an instruction such as "8:30" from being interpreted in the
+wrong timezone.
+
+## Operational boundaries
+
+- Route responses are point-in-time estimates. The Routes API does not provide
+ continuous tracking or a traffic push subscription.
+- Avoid-toll, highway, and ferry options are preferences, not guarantees.
+- Walking, bicycling, and two-wheeler results are beta and include a warning.
+- The integration requests a fixed minimal response field mask. It does not
+ request toll pricing, eco routes, traffic-colored polylines, or route
+ matrices.
+- Route responses and traffic snapshots are not persisted. Google Maps
+ Platform storage and attribution policies still apply to downstream uses.
+- Provider errors are normalized without logging the API key, request payload,
+ exact locations, or resolved place IDs. When Routes is enabled, OpenInference
+ input and output capture is disabled for the process.
+
+See the official [Compute Routes
+reference](https://developers.google.com/maps/documentation/routes/reference/rest/v2/TopLevel/computeRoutes),
+[traffic model
+guide](https://developers.google.com/maps/documentation/routes/traffic-model),
+[field-mask
+guidance](https://developers.google.com/maps/documentation/routes/choose_fields),
+and [Routes
+policies](https://developers.google.com/maps/documentation/routes/policies).
diff --git a/mkdocs.yml b/mkdocs.yml
index fc1849d..004061a 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -57,6 +57,7 @@ nav:
- Observability: base-infra/observability.md
- Understand:
- Architecture: architecture.md
+ - Google Maps Routes: google-maps-routes.md
- Docker Compose: base-infra/docker-compose-workflow.md
- Docker image: base-infra/dockerfile-strategy.md
- Reference:
diff --git a/src/blacki/callbacks.py b/src/blacki/callbacks.py
index a953a19..7614595 100644
--- a/src/blacki/callbacks.py
+++ b/src/blacki/callbacks.py
@@ -23,6 +23,12 @@
from .telegram.api import TelegramApiClient, TelegramApiError
from .telegram.formatting import escape_markdown, format_for_telegram
from .telegram.types import ParseMode
+from .utils.privacy import (
+ REDACTED_ROUTE_DETAILS,
+ ROUTE_TOOL_NAMES,
+ redact_route_tool_payload,
+ route_data_redaction_enabled,
+)
logger = logging.getLogger(__name__)
@@ -241,7 +247,7 @@ async def notify_telegram_before_tool(
return None
escaped_name = escape_markdown(tool.name)
- args_text = _format_tool_args(args)
+ args_text = _format_tool_args(redact_route_tool_payload(tool.name, args))
text = f"🔧 Using tool: *{escaped_name}*{args_text}"
try:
@@ -391,7 +397,9 @@ def before_agent(self, callback_context: CallbackContext) -> None:
)
self.logger.debug(f"State keys: {callback_context.state.to_dict().keys()}")
- if user_content := callback_context.user_content:
+ if route_data_redaction_enabled():
+ self.logger.debug(f"User Content: {REDACTED_ROUTE_DETAILS}")
+ elif user_content := callback_context.user_content:
content_data = user_content.model_dump(exclude_none=True, mode="json")
self.logger.debug(f"User Content: {content_data}")
@@ -410,7 +418,9 @@ def after_agent(self, callback_context: CallbackContext) -> None:
)
self.logger.debug(f"State keys: {callback_context.state.to_dict().keys()}")
- if user_content := callback_context.user_content:
+ if route_data_redaction_enabled():
+ self.logger.debug(f"User Content: {REDACTED_ROUTE_DETAILS}")
+ elif user_content := callback_context.user_content:
content_data = user_content.model_dump(exclude_none=True, mode="json")
self.logger.debug(f"User Content: {content_data}")
@@ -435,15 +445,21 @@ def before_model(
)
self.logger.debug(f"State keys: {callback_context.state.to_dict().keys()}")
- if user_content := callback_context.user_content:
+ redact_content = route_data_redaction_enabled()
+ if redact_content:
+ self.logger.debug(f"User Content: {REDACTED_ROUTE_DETAILS}")
+ elif user_content := callback_context.user_content:
content_data = user_content.model_dump(exclude_none=True, mode="json")
self.logger.debug(f"User Content: {content_data}")
self.logger.debug(f"LLM request contains {len(llm_request.contents)} messages:")
- for i, content in enumerate(llm_request.contents, start=1):
- self.logger.debug(
- f"Content {i}: {content.model_dump(exclude_none=True, mode='json')}"
- )
+ if redact_content:
+ self.logger.debug(f"LLM request content: {REDACTED_ROUTE_DETAILS}")
+ else:
+ for i, content in enumerate(llm_request.contents, start=1):
+ self.logger.debug(
+ f"Content {i}: {content.model_dump(exclude_none=True, mode='json')}"
+ )
return None
@@ -465,11 +481,16 @@ def after_model(
)
self.logger.debug(f"State keys: {callback_context.state.to_dict().keys()}")
- if user_content := callback_context.user_content:
+ redact_content = route_data_redaction_enabled()
+ if redact_content:
+ self.logger.debug(f"User Content: {REDACTED_ROUTE_DETAILS}")
+ elif user_content := callback_context.user_content:
content_data = user_content.model_dump(exclude_none=True, mode="json")
self.logger.debug(f"User Content: {content_data}")
- if llm_content := llm_response.content:
+ if redact_content and llm_response.content is not None:
+ self.logger.debug(f"LLM response: {REDACTED_ROUTE_DETAILS}")
+ elif llm_content := llm_response.content:
response_data = llm_content.model_dump(exclude_none=True, mode="json")
self.logger.debug(f"LLM response: {response_data}")
@@ -496,14 +517,17 @@ def before_tool(
)
self.logger.debug(f"State keys: {tool_context.state.to_dict().keys()}")
- if content := tool_context.user_content:
+ redact_content = route_data_redaction_enabled() or tool.name in ROUTE_TOOL_NAMES
+ if redact_content:
+ self.logger.debug(f"User Content: {REDACTED_ROUTE_DETAILS}")
+ elif content := tool_context.user_content:
self.logger.debug(
f"User Content: {content.model_dump(exclude_none=True, mode='json')}"
)
actions_data = tool_context.actions.model_dump(exclude_none=True, mode="json")
self.logger.debug(f"EventActions: {actions_data}")
- self.logger.debug(f"args: {args}")
+ self.logger.debug(f"args: {redact_route_tool_payload(tool.name, args)}")
return None
@@ -530,14 +554,19 @@ def after_tool(
)
self.logger.debug(f"State keys: {tool_context.state.to_dict().keys()}")
- if content := tool_context.user_content:
+ redact_content = route_data_redaction_enabled() or tool.name in ROUTE_TOOL_NAMES
+ if redact_content:
+ self.logger.debug(f"User Content: {REDACTED_ROUTE_DETAILS}")
+ elif content := tool_context.user_content:
self.logger.debug(
f"User Content: {content.model_dump(exclude_none=True, mode='json')}"
)
actions_data = tool_context.actions.model_dump(exclude_none=True, mode="json")
self.logger.debug(f"EventActions: {actions_data}")
- self.logger.debug(f"args: {args}")
- self.logger.debug(f"Tool response: {tool_response}")
+ self.logger.debug(f"args: {redact_route_tool_payload(tool.name, args)}")
+ self.logger.debug(
+ f"Tool response: {redact_route_tool_payload(tool.name, tool_response)}"
+ )
return None
diff --git a/src/blacki/prompt.py b/src/blacki/prompt.py
index 0d05646..1518566 100644
--- a/src/blacki/prompt.py
+++ b/src/blacki/prompt.py
@@ -124,6 +124,23 @@
"""
+ROUTES_POLICY = """\
+
+Use the dedicated route tools for distance, travel time, current traffic, route
+alternatives, and route-scenario comparisons. Do not use general web search,
+browser automation, or memory for those values. A request for current or live
+traffic requires a fresh route lookup; the result is a point-in-time estimate,
+not continuous tracking.
+
+Use get_route_estimate for one route and compare_route_scenarios only when the
+user asks to compare departure times, modes, traffic assumptions, or avoid
+options. For current driving traffic use DRIVE, now, and BEST_GUESS. Use NONE
+as the traffic model for non-driving modes. Treat avoid options as preferences,
+not guarantees, and preserve all provider warnings and Google Maps attribution.
+Ask one focused question when an endpoint or required departure time is missing.
+"""
+
+
DOMAIN_PATTERNS = {
"nutrition": re.compile(
r"\b(?:ate|eaten|eating|drank|drink|food|meal|breakfast|lunch|dinner|"
@@ -139,6 +156,16 @@
r"\b(?:remind|reminder|schedule|alarm|notify|notification)\b",
re.IGNORECASE,
),
+ "routes": re.compile(
+ r"\b(?:route|routes|directions?|distance\s+(?:from|to|between)|how\s+far|"
+ r"travel\s+time|traffic|"
+ r"commute|avoid\s+(?:tolls?|highways?|ferries)|get\s+there|on\s+foot|"
+ r"by\s+(?:car|bike|bicycle|transit)|"
+ r"eta\s+(?:to|from|between|for\s+(?:the\s+)?(?:route|trip|commute))|"
+ r"(?:drive|driving|walk|walking|bicycle|bicycling|bike|biking|"
+ r"two[-\s]wheeler)\s+(?:to|from|between))\b",
+ re.IGNORECASE,
+ ),
"search": re.compile(
r"\b(?:latest|current|news|recent|today|as of|verify|verified|search|"
r"look up|source|sources|citation|citations)\b",
@@ -172,6 +199,7 @@
}
),
"reminder": frozenset({"schedule_reminder", "list_reminders", "cancel_reminder"}),
+ "routes": frozenset({"get_route_estimate", "compare_route_scenarios"}),
"search": frozenset({"exa_search", "brave_search"}),
}
@@ -231,12 +259,18 @@ def select_domain_policy_names(
) -> tuple[str, ...]:
"""Select request-relevant domains that also have enabled tools."""
selected = []
- for domain in ("nutrition", "workout", "reminder", "search"):
+ for domain in ("nutrition", "workout", "reminder", "routes"):
if (
DOMAIN_PATTERNS[domain].search(user_text)
and DOMAIN_TOOL_NAMES[domain] & available_tool_names
):
selected.append(domain)
+ if (
+ "routes" not in selected
+ and DOMAIN_PATTERNS["search"].search(user_text)
+ and DOMAIN_TOOL_NAMES["search"] & available_tool_names
+ ):
+ selected.append("search")
return tuple(selected)
@@ -258,6 +292,8 @@ def build_domain_instruction(
blocks.append(workout_policy)
elif domain == "reminder":
blocks.append(REMINDER_POLICY)
+ elif domain == "routes":
+ blocks.append(ROUTES_POLICY)
elif domain == "search": # pragma: no branch - search is the final domain
blocks.append(_build_search_policy(available_tool_names))
return "\n\n".join(blocks)
@@ -308,15 +344,18 @@ async def before_model_callback(
if not user_text:
return
- instruction = build_domain_instruction(
- user_text, frozenset(llm_request.tools_dict)
- )
+ available_tools = frozenset(llm_request.tools_dict)
+ selected_domains = select_domain_policy_names(user_text, available_tools)
+ instruction = build_domain_instruction(user_text, available_tools)
if instruction:
llm_request.append_instructions([instruction])
- if "search" in select_domain_policy_names(
- user_text, frozenset(llm_request.tools_dict)
- ):
+ if "routes" in selected_domains:
+ _hide_tools(
+ llm_request,
+ set(DOMAIN_TOOL_NAMES["search"] & available_tools),
+ )
+ elif "search" in selected_domains:
_apply_search_tool_budget(callback_context, llm_request)
async def before_tool_callback(
diff --git a/src/blacki/registry.py b/src/blacki/registry.py
index 14e71fc..0976bd5 100644
--- a/src/blacki/registry.py
+++ b/src/blacki/registry.py
@@ -24,6 +24,7 @@ class ToolConfig:
Attributes:
exa_api_key: API key for Exa Search.
brave_search_api_key: API key for Brave Search.
+ google_maps_routes_api_key: API key for Google Maps Routes.
sqlite_path: Path to SQLite database for storage-backed tools.
sandbox_enabled: Whether to enable sandbox tools.
skills_dir: Directory containing skill definitions.
@@ -32,6 +33,7 @@ class ToolConfig:
exa_api_key: str | None = None
brave_search_api_key: str | None = None
+ google_maps_routes_api_key: str | None = None
sqlite_path: str | None = None
sandbox_enabled: bool = False
skills_dir: Path | None = None
@@ -58,6 +60,10 @@ def build_tools(config: ToolConfig) -> list[Any]:
tools.extend(_build_brave_search_tools())
logger.info("Brave Search tool enabled")
+ if config.google_maps_routes_api_key:
+ tools.extend(_build_google_routes_tools())
+ logger.info("Google Maps Routes tools enabled")
+
if config.sqlite_path:
tools.extend(_build_reminder_tools())
tools.extend(_build_calorie_tools())
@@ -103,6 +109,17 @@ def _build_brave_search_tools() -> list[Any]:
return []
+def _build_google_routes_tools() -> list[Any]:
+ """Build Google Maps Routes tools."""
+ try:
+ from blacki.routes import compare_route_scenarios, get_route_estimate
+
+ return [get_route_estimate, compare_route_scenarios]
+ except ImportError as e:
+ logger.warning("Failed to load Google Maps Routes tools: %s", e)
+ return []
+
+
def _build_reminder_tools() -> list[Any]:
"""Build reminder tools."""
try:
@@ -304,6 +321,8 @@ def build_tool_config_from_env() -> ToolConfig:
return ToolConfig(
exa_api_key=os.getenv("EXA_API_KEY", "").strip() or None,
brave_search_api_key=os.getenv("BRAVE_SEARCH_API_KEY", "").strip() or None,
+ google_maps_routes_api_key=os.getenv("GOOGLE_MAPS_ROUTES_API_KEY", "").strip()
+ or None,
sqlite_path=sqlite_path,
sandbox_enabled=os.getenv("SANDBOX_ENABLED", "false").strip().lower()
in ("true", "1", "yes"),
diff --git a/src/blacki/routes/__init__.py b/src/blacki/routes/__init__.py
new file mode 100644
index 0000000..0c8000a
--- /dev/null
+++ b/src/blacki/routes/__init__.py
@@ -0,0 +1,11 @@
+"""Google Maps Routes API tools."""
+
+from .client import close_shared_routes_client
+from .tools import RouteScenario, compare_route_scenarios, get_route_estimate
+
+__all__ = [
+ "RouteScenario",
+ "close_shared_routes_client",
+ "compare_route_scenarios",
+ "get_route_estimate",
+]
diff --git a/src/blacki/routes/client.py b/src/blacki/routes/client.py
new file mode 100644
index 0000000..0ba9f4e
--- /dev/null
+++ b/src/blacki/routes/client.py
@@ -0,0 +1,158 @@
+"""Minimal async client for the Google Maps Routes API."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+COMPUTE_ROUTES_URL = "https://routes.googleapis.com/directions/v2:computeRoutes"
+ROUTES_FIELD_MASK = ",".join(
+ (
+ "routes.distanceMeters",
+ "routes.duration",
+ "routes.staticDuration",
+ "routes.description",
+ "routes.routeLabels",
+ "routes.warnings",
+ "fallbackInfo.routingMode",
+ "fallbackInfo.reason",
+ "geocodingResults.origin.placeId",
+ "geocodingResults.origin.partialMatch",
+ "geocodingResults.destination.placeId",
+ "geocodingResults.destination.partialMatch",
+ )
+)
+MAX_ATTEMPTS = 3
+RETRY_BASE_SECONDS = 0.25
+
+_routes_client_lock = asyncio.Lock()
+_routes_client: httpx.AsyncClient | None = None
+
+
+class RoutesAPIError(RuntimeError):
+ """Stable, credential-safe error raised by the Routes client."""
+
+ def __init__(self, code: str, message: str, *, retryable: bool = False) -> None:
+ super().__init__(message)
+ self.code = code
+ self.retryable = retryable
+
+
+async def reset_routes_client_cache() -> None:
+ """Close and clear the shared Routes client between tests."""
+ await close_shared_routes_client()
+
+
+async def close_shared_routes_client() -> None:
+ """Close the process-wide Routes client during application shutdown."""
+ global _routes_client
+ async with _routes_client_lock:
+ if _routes_client is not None:
+ try:
+ await _routes_client.aclose()
+ except Exception:
+ logger.exception("Error while closing shared Google Routes client")
+ _routes_client = None
+
+
+async def _get_shared_routes_client() -> httpx.AsyncClient:
+ """Return a process-wide async client for Google Routes requests."""
+ global _routes_client
+ async with _routes_client_lock:
+ if _routes_client is not None:
+ return _routes_client
+ _routes_client = httpx.AsyncClient(timeout=15.0)
+ return _routes_client
+
+
+def _response_error(status_code: int) -> RoutesAPIError | None:
+ """Map an HTTP status to a stable tool-facing error."""
+ if 200 <= status_code < 300:
+ return None
+ if status_code in (401, 403):
+ return RoutesAPIError(
+ "authentication_failed",
+ "Google Maps Routes authentication failed. Check the configured API key.",
+ )
+ if status_code == 429:
+ return RoutesAPIError(
+ "rate_limited",
+ "Google Maps Routes rate limit exceeded. Try again later.",
+ retryable=True,
+ )
+ if status_code >= 500:
+ return RoutesAPIError(
+ "unavailable",
+ "Google Maps Routes is temporarily unavailable.",
+ retryable=True,
+ )
+ if 400 <= status_code < 500:
+ return RoutesAPIError(
+ "invalid_request",
+ "Google Maps Routes rejected the route request.",
+ )
+ return RoutesAPIError(
+ "unavailable",
+ "Google Maps Routes returned an unexpected HTTP response.",
+ )
+
+
+async def compute_routes(
+ payload: dict[str, Any],
+ api_key: str,
+) -> dict[str, Any]:
+ """Submit one Compute Routes request with bounded retries."""
+ headers = {
+ "Content-Type": "application/json",
+ "X-Goog-Api-Key": api_key,
+ "X-Goog-FieldMask": ROUTES_FIELD_MASK,
+ }
+ client = await _get_shared_routes_client()
+
+ for attempt in range(MAX_ATTEMPTS): # pragma: no branch - always returns or raises
+ try:
+ response = await client.post(
+ COMPUTE_ROUTES_URL,
+ headers=headers,
+ json=payload,
+ )
+ error = _response_error(response.status_code)
+ except httpx.RequestError:
+ logger.warning(
+ "Google Routes network request failed on attempt %d",
+ attempt + 1,
+ )
+ error = RoutesAPIError(
+ "unavailable",
+ "Google Maps Routes could not be reached.",
+ retryable=True,
+ )
+
+ if error is None:
+ try:
+ data = response.json()
+ except ValueError as exc:
+ raise RoutesAPIError(
+ "invalid_response",
+ "Google Maps Routes returned an invalid response.",
+ ) from exc
+ if not isinstance(data, dict):
+ raise RoutesAPIError(
+ "invalid_response",
+ "Google Maps Routes returned an invalid response.",
+ )
+ return data
+
+ if not error.retryable or attempt == MAX_ATTEMPTS - 1:
+ raise error
+
+ await asyncio.sleep(RETRY_BASE_SECONDS * (2**attempt))
+
+ raise AssertionError( # pragma: no cover - defensive unreachable guard
+ "Routes retry loop exhausted without returning"
+ )
diff --git a/src/blacki/routes/tools.py b/src/blacki/routes/tools.py
new file mode 100644
index 0000000..09a9afb
--- /dev/null
+++ b/src/blacki/routes/tools.py
@@ -0,0 +1,560 @@
+"""Read-only tools backed by the Google Maps Routes API."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from datetime import UTC, datetime
+from typing import Any
+
+from google.adk.tools import ToolContext
+from pydantic import BaseModel, ConfigDict, Field
+
+from blacki.utils.timezone import now_utc, utc_iso_seconds
+
+from .client import RoutesAPIError, compute_routes
+
+SUPPORTED_TRAVEL_MODES = frozenset(
+ {"DRIVE", "WALK", "BICYCLE", "TWO_WHEELER", "TRANSIT"}
+)
+SUPPORTED_TRAFFIC_MODELS = frozenset({"BEST_GUESS", "OPTIMISTIC", "PESSIMISTIC"})
+MODE_ALIASES = {
+ "DRIVING": "DRIVE",
+ "WALKING": "WALK",
+ "BIKING": "BICYCLE",
+ "CYCLING": "BICYCLE",
+ "MOTORCYCLE": "TWO_WHEELER",
+}
+MAX_SCENARIOS = 5
+SCENARIO_CONCURRENCY = 3
+GOOGLE_MAPS_ATTRIBUTION = "Google Maps"
+
+
+class RouteScenario(BaseModel):
+ """One explicitly named route-comparison scenario."""
+
+ model_config = ConfigDict(extra="forbid")
+
+ label: str = Field(description="Short unique label for this scenario.")
+ travel_mode: str = Field(
+ description=(
+ "DRIVE, WALK, BICYCLE, TWO_WHEELER, or TRANSIT. Common aliases such "
+ "as driving and walking are accepted."
+ )
+ )
+ departure_time: str = Field(
+ description="Use 'now' or an RFC 3339 timestamp with a timezone offset."
+ )
+ traffic_model: str = Field(
+ description=(
+ "BEST_GUESS, OPTIMISTIC, or PESSIMISTIC for DRIVE; use NONE for "
+ "other travel modes."
+ )
+ )
+ avoid_tolls: bool = Field(description="Prefer routes without tolls.")
+ avoid_highways: bool = Field(description="Prefer routes without highways.")
+ avoid_ferries: bool = Field(description="Prefer routes without ferries.")
+
+
+class RouteValidationError(ValueError):
+ """Invalid user-controlled route input."""
+
+
+def _error_result(
+ code: str,
+ message: str,
+ origin: str,
+ destination: str,
+) -> dict[str, Any]:
+ """Build the stable error contract returned to the agent."""
+ return {
+ "status": "error",
+ "error_code": code,
+ "error": message,
+ "origin": origin,
+ "destination": destination,
+ "routes": [],
+ "attribution": GOOGLE_MAPS_ATTRIBUTION,
+ }
+
+
+def _normalize_travel_mode(value: str) -> str:
+ normalized = value.strip().upper().replace("-", "_").replace(" ", "_")
+ normalized = MODE_ALIASES.get(normalized, normalized)
+ if normalized not in SUPPORTED_TRAVEL_MODES:
+ supported = ", ".join(sorted(SUPPORTED_TRAVEL_MODES))
+ raise RouteValidationError(f"Travel mode must be one of: {supported}.")
+ return normalized
+
+
+def _normalize_departure_time(value: str) -> str | None:
+ normalized = value.strip()
+ if normalized.lower() == "now":
+ return None
+ if not normalized:
+ raise RouteValidationError(
+ "Departure time must be 'now' or an RFC 3339 timestamp."
+ )
+
+ try:
+ parsed = datetime.fromisoformat(normalized.replace("Z", "+00:00"))
+ except ValueError as exc:
+ raise RouteValidationError(
+ "Departure time must be 'now' or an RFC 3339 timestamp."
+ ) from exc
+ if parsed.tzinfo is None:
+ raise RouteValidationError("Departure time must include a timezone offset.")
+ return parsed.astimezone(UTC).isoformat().replace("+00:00", "Z")
+
+
+def _normalize_traffic_model(value: str, travel_mode: str) -> str | None:
+ normalized = value.strip().upper().replace("-", "_").replace(" ", "_")
+ if normalized == "NONE":
+ return None
+ if travel_mode != "DRIVE":
+ raise RouteValidationError(
+ "Traffic models are available only when travel mode is DRIVE."
+ )
+ if normalized not in SUPPORTED_TRAFFIC_MODELS:
+ supported = ", ".join(sorted(SUPPORTED_TRAFFIC_MODELS))
+ raise RouteValidationError(
+ f"Traffic model must be NONE or one of: {supported}."
+ )
+ return normalized
+
+
+def _waypoint(value: str, label: str) -> dict[str, str]:
+ normalized = value.strip()
+ if not normalized:
+ raise RouteValidationError(f"{label} cannot be empty.")
+ if len(normalized) > 512:
+ raise RouteValidationError(f"{label} is too long.")
+ prefix, separator, place_id = normalized.partition(":")
+ if prefix.lower() == "place_id" and separator:
+ if not place_id.strip():
+ raise RouteValidationError(f"{label} place ID cannot be empty.")
+ return {"placeId": place_id.strip()}
+ return {"address": normalized}
+
+
+def _build_payload(
+ *,
+ origin: str,
+ destination: str,
+ travel_mode: str,
+ departure_time: str,
+ traffic_model: str,
+ avoid_tolls: bool,
+ avoid_highways: bool,
+ avoid_ferries: bool,
+ include_alternatives: bool,
+) -> tuple[dict[str, Any], str, str]:
+ mode = _normalize_travel_mode(travel_mode)
+ normalized_departure = _normalize_departure_time(departure_time)
+ normalized_traffic_model = _normalize_traffic_model(traffic_model, mode)
+
+ has_route_modifiers = avoid_tolls or avoid_highways or avoid_ferries
+ if has_route_modifiers and mode not in {"DRIVE", "TWO_WHEELER"}:
+ raise RouteValidationError(
+ "Toll, highway, and ferry avoidance require DRIVE or TWO_WHEELER."
+ )
+
+ payload: dict[str, Any] = {
+ "origin": _waypoint(origin, "Origin"),
+ "destination": _waypoint(destination, "Destination"),
+ "travelMode": mode,
+ "computeAlternativeRoutes": include_alternatives,
+ }
+ if normalized_departure is not None:
+ payload["departureTime"] = normalized_departure
+
+ if mode == "DRIVE":
+ if normalized_traffic_model is None:
+ payload["routingPreference"] = "TRAFFIC_AWARE"
+ else:
+ payload["routingPreference"] = "TRAFFIC_AWARE_OPTIMAL"
+ payload["trafficModel"] = normalized_traffic_model
+ elif mode == "TWO_WHEELER":
+ payload["routingPreference"] = "TRAFFIC_AWARE"
+
+ if has_route_modifiers:
+ payload["routeModifiers"] = {
+ "avoidTolls": avoid_tolls,
+ "avoidHighways": avoid_highways,
+ "avoidFerries": avoid_ferries,
+ }
+
+ requested_departure = normalized_departure or "now"
+ return payload, mode, requested_departure
+
+
+def _duration_seconds(value: object, field_name: str) -> float:
+ if not isinstance(value, str) or not value.endswith("s"):
+ raise RoutesAPIError(
+ "invalid_response",
+ f"Google Maps Routes omitted a valid {field_name}.",
+ )
+ try:
+ return float(value[:-1])
+ except ValueError as exc:
+ raise RoutesAPIError(
+ "invalid_response",
+ f"Google Maps Routes returned an invalid {field_name}.",
+ ) from exc
+
+
+def _resolved_waypoint(data: object) -> dict[str, Any] | None:
+ if not isinstance(data, dict):
+ return None
+ place_id = data.get("placeId")
+ if not isinstance(place_id, str):
+ return None
+ return {
+ "place_id": place_id,
+ "partial_match": data.get("partialMatch") is True,
+ }
+
+
+def _normalize_route(
+ route: object,
+ route_index: int,
+) -> dict[str, Any]:
+ if not isinstance(route, dict):
+ raise RoutesAPIError(
+ "invalid_response",
+ "Google Maps Routes returned an invalid route.",
+ )
+
+ distance = route.get("distanceMeters")
+ if not isinstance(distance, int) or isinstance(distance, bool):
+ raise RoutesAPIError(
+ "invalid_response",
+ "Google Maps Routes omitted a valid route distance.",
+ )
+ duration = _duration_seconds(route.get("duration"), "route duration")
+ raw_static_duration = route.get("staticDuration")
+ static_duration = (
+ _duration_seconds(raw_static_duration, "static route duration")
+ if raw_static_duration is not None
+ else None
+ )
+ traffic_delay = duration - static_duration if static_duration is not None else None
+ traffic_delay_percent = (
+ round((traffic_delay / static_duration) * 100, 1)
+ if (
+ traffic_delay is not None
+ and static_duration is not None
+ and static_duration > 0
+ )
+ else None
+ )
+
+ labels = route.get("routeLabels")
+ warnings = route.get("warnings")
+ description = route.get("description")
+ return {
+ "route_index": route_index,
+ "route_labels": (
+ [label for label in labels if isinstance(label, str)]
+ if isinstance(labels, list)
+ else []
+ ),
+ "description": description if isinstance(description, str) else "",
+ "distance_meters": distance,
+ "distance_kilometers": round(distance / 1000, 2),
+ "duration_seconds": duration,
+ "duration_minutes": round(duration / 60, 1),
+ "static_duration_seconds": static_duration,
+ "static_duration_minutes": (
+ round(static_duration / 60, 1) if static_duration is not None else None
+ ),
+ "traffic_delay_seconds": traffic_delay,
+ "traffic_delay_minutes": (
+ round(traffic_delay / 60, 1) if traffic_delay is not None else None
+ ),
+ "traffic_delay_percent": traffic_delay_percent,
+ "warnings": (
+ [warning for warning in warnings if isinstance(warning, str)]
+ if isinstance(warnings, list)
+ else []
+ ),
+ }
+
+
+def _normalize_response(
+ data: dict[str, Any],
+ *,
+ origin: str,
+ destination: str,
+ travel_mode: str,
+ departure_time: str,
+ used_route_modifiers: bool,
+) -> dict[str, Any]:
+ raw_routes = data.get("routes")
+ if not isinstance(raw_routes, list):
+ raise RoutesAPIError(
+ "invalid_response",
+ "Google Maps Routes returned an invalid routes collection.",
+ )
+ if not raw_routes:
+ raise RoutesAPIError(
+ "no_route",
+ "Google Maps Routes could not find a route for these locations.",
+ )
+
+ geocoding = data.get("geocodingResults")
+ geocoding = geocoding if isinstance(geocoding, dict) else {}
+ fallback = data.get("fallbackInfo")
+ fallback = fallback if isinstance(fallback, dict) else {}
+
+ mode_warning = (
+ "Walking, bicycling, and two-wheeler routes are beta; use caution."
+ if travel_mode in {"WALK", "BICYCLE", "TWO_WHEELER"}
+ else None
+ )
+ return {
+ "status": "success",
+ "origin": origin,
+ "destination": destination,
+ "travel_mode": travel_mode,
+ "departure_time": departure_time,
+ "computed_at": utc_iso_seconds(now_utc()),
+ "routes": [
+ _normalize_route(route, index) for index, route in enumerate(raw_routes)
+ ],
+ "resolved_waypoints": {
+ "origin": _resolved_waypoint(geocoding.get("origin")),
+ "destination": _resolved_waypoint(geocoding.get("destination")),
+ },
+ "fallback": (
+ {
+ "routing_mode": fallback.get("routingMode"),
+ "reason": fallback.get("reason"),
+ }
+ if fallback
+ else None
+ ),
+ "mode_warning": mode_warning,
+ "modifier_warning": (
+ "Avoid options are preferences, not guarantees."
+ if used_route_modifiers
+ else None
+ ),
+ "attribution": GOOGLE_MAPS_ATTRIBUTION,
+ }
+
+
+def _public_route_result(result: dict[str, Any]) -> dict[str, Any]:
+ """Keep exact endpoints and provider place IDs out of public tool results."""
+ for key in ("origin", "destination", "resolved_waypoints"):
+ result.pop(key, None)
+ return result
+
+
+async def _estimate_route(
+ *,
+ api_key: str,
+ origin: str,
+ destination: str,
+ travel_mode: str,
+ departure_time: str,
+ traffic_model: str,
+ avoid_tolls: bool,
+ avoid_highways: bool,
+ avoid_ferries: bool,
+ include_alternatives: bool,
+) -> dict[str, Any]:
+ try:
+ payload, mode, requested_departure = _build_payload(
+ origin=origin,
+ destination=destination,
+ travel_mode=travel_mode,
+ departure_time=departure_time,
+ traffic_model=traffic_model,
+ avoid_tolls=avoid_tolls,
+ avoid_highways=avoid_highways,
+ avoid_ferries=avoid_ferries,
+ include_alternatives=include_alternatives,
+ )
+ response = await compute_routes(payload, api_key)
+ return _normalize_response(
+ response,
+ origin=origin.strip(),
+ destination=destination.strip(),
+ travel_mode=mode,
+ departure_time=requested_departure,
+ used_route_modifiers=avoid_tolls or avoid_highways or avoid_ferries,
+ )
+ except RouteValidationError as exc:
+ return _error_result("invalid_input", str(exc), origin, destination)
+ except RoutesAPIError as exc:
+ return _error_result(exc.code, str(exc), origin, destination)
+
+
+async def get_route_estimate(
+ origin: str,
+ destination: str,
+ travel_mode: str,
+ departure_time: str,
+ traffic_model: str,
+ avoid_tolls: bool,
+ avoid_highways: bool,
+ avoid_ferries: bool,
+ include_alternatives: bool,
+ tool_context: ToolContext,
+) -> dict[str, Any]:
+ """Get a fresh route distance, ETA, traffic delay, and optional alternatives.
+
+ Use this for one route between an origin and destination. For current
+ driving traffic, set travel_mode to DRIVE, departure_time to now, and
+ traffic_model to BEST_GUESS. Use NONE as the traffic model for non-driving
+ modes. Prefix a Google place ID with ``place_id:``; otherwise locations are
+ treated as addresses. Avoid options express preferences, not guarantees.
+
+ Args:
+ origin: Starting address or a place ID prefixed with ``place_id:``.
+ destination: Ending address or a place ID prefixed with ``place_id:``.
+ travel_mode: DRIVE, WALK, BICYCLE, TWO_WHEELER, or TRANSIT.
+ departure_time: ``now`` or an RFC 3339 timestamp with timezone offset.
+ traffic_model: BEST_GUESS, OPTIMISTIC, PESSIMISTIC, or NONE.
+ avoid_tolls: Whether to prefer routes without tolls.
+ avoid_highways: Whether to prefer routes without highways.
+ avoid_ferries: Whether to prefer routes without ferries.
+ include_alternatives: Whether to request alternate routes.
+
+ Returns:
+ A dictionary with normalized routes, durations, traffic delay, warnings,
+ and Google Maps attribution.
+ """
+ _ = tool_context
+ api_key = os.environ.get("GOOGLE_MAPS_ROUTES_API_KEY", "").strip()
+ if not api_key:
+ return _public_route_result(
+ _error_result(
+ "not_configured",
+ "GOOGLE_MAPS_ROUTES_API_KEY is not configured.",
+ origin,
+ destination,
+ )
+ )
+ result = await _estimate_route(
+ api_key=api_key,
+ origin=origin,
+ destination=destination,
+ travel_mode=travel_mode,
+ departure_time=departure_time,
+ traffic_model=traffic_model,
+ avoid_tolls=avoid_tolls,
+ avoid_highways=avoid_highways,
+ avoid_ferries=avoid_ferries,
+ include_alternatives=include_alternatives,
+ )
+ return _public_route_result(result)
+
+
+async def compare_route_scenarios(
+ origin: str,
+ destination: str,
+ scenarios: list[RouteScenario],
+ tool_context: ToolContext,
+) -> dict[str, Any]:
+ """Compare up to five fresh route scenarios for the same endpoints.
+
+ Use this only when the user asks to compare departure times, travel modes,
+ traffic assumptions, or avoid options. Each scenario must have a short
+ unique label and all of its route settings. The tool returns the primary
+ route for each scenario and preserves individual failures.
+
+ Args:
+ origin: Starting address or a place ID prefixed with ``place_id:``.
+ destination: Ending address or a place ID prefixed with ``place_id:``.
+ scenarios: One to five explicitly named route scenarios.
+
+ Returns:
+ A dictionary with overall status and one normalized result per scenario.
+ """
+ _ = tool_context
+ if not 1 <= len(scenarios) <= MAX_SCENARIOS:
+ return _public_route_result(
+ _error_result(
+ "invalid_input",
+ f"Provide between 1 and {MAX_SCENARIOS} route scenarios.",
+ origin,
+ destination,
+ )
+ )
+
+ labels = [scenario.label.strip() for scenario in scenarios]
+ if any(not label for label in labels):
+ return _public_route_result(
+ _error_result(
+ "invalid_input",
+ "Every route scenario requires a non-empty label.",
+ origin,
+ destination,
+ )
+ )
+ normalized_labels = [label.casefold() for label in labels]
+ if len(set(normalized_labels)) != len(normalized_labels):
+ return _public_route_result(
+ _error_result(
+ "invalid_input",
+ "Route scenario labels must be unique.",
+ origin,
+ destination,
+ )
+ )
+
+ api_key = os.environ.get("GOOGLE_MAPS_ROUTES_API_KEY", "").strip()
+ if not api_key:
+ return _public_route_result(
+ _error_result(
+ "not_configured",
+ "GOOGLE_MAPS_ROUTES_API_KEY is not configured.",
+ origin,
+ destination,
+ )
+ )
+
+ semaphore = asyncio.Semaphore(SCENARIO_CONCURRENCY)
+
+ async def run_scenario(
+ scenario: RouteScenario,
+ label: str,
+ ) -> dict[str, Any]:
+ async with semaphore:
+ result = await _estimate_route(
+ api_key=api_key,
+ origin=origin,
+ destination=destination,
+ travel_mode=scenario.travel_mode,
+ departure_time=scenario.departure_time,
+ traffic_model=scenario.traffic_model,
+ avoid_tolls=scenario.avoid_tolls,
+ avoid_highways=scenario.avoid_highways,
+ avoid_ferries=scenario.avoid_ferries,
+ include_alternatives=False,
+ )
+ return {"label": label, **_public_route_result(result)}
+
+ results = await asyncio.gather(
+ *(
+ run_scenario(scenario, label)
+ for scenario, label in zip(scenarios, labels, strict=True)
+ )
+ )
+ success_count = sum(result.get("status") == "success" for result in results)
+ if success_count == len(results):
+ status = "success"
+ elif success_count:
+ status = "partial"
+ else:
+ status = "error"
+
+ return {
+ "status": status,
+ "scenario_count": len(results),
+ "successful_scenarios": success_count,
+ "scenarios": results,
+ "attribution": GOOGLE_MAPS_ATTRIBUTION,
+ }
diff --git a/src/blacki/server.py b/src/blacki/server.py
index 0a7baaf..c9aaf5c 100644
--- a/src/blacki/server.py
+++ b/src/blacki/server.py
@@ -15,6 +15,7 @@
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from google.adk.cli.fast_api import get_fast_api_app
+from openinference.instrumentation import TraceConfig
from openinference.instrumentation.google_adk import GoogleADKInstrumentor
from .adk_runtime import create_adk_runtime
@@ -28,6 +29,7 @@
setup_tracing,
validation,
)
+from .utils.privacy import route_data_redaction_enabled
logger = logging.getLogger(__name__)
@@ -37,7 +39,13 @@
agent_name=env.agent_name,
)
-GoogleADKInstrumentor().instrument()
+_route_data_redaction = route_data_redaction_enabled()
+GoogleADKInstrumentor().instrument(
+ config=TraceConfig(
+ hide_inputs=True if _route_data_redaction else None,
+ hide_outputs=True if _route_data_redaction else None,
+ )
+)
setup_logging(log_level=env.log_level)
setup_tracing()
@@ -187,6 +195,10 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
await close_shared_exa_search_client()
+ from .routes import close_shared_routes_client
+
+ await close_shared_routes_client()
+
from .callbacks import close_shared_notify_client
await close_shared_notify_client()
diff --git a/src/blacki/utils/privacy.py b/src/blacki/utils/privacy.py
new file mode 100644
index 0000000..e9ef8dc
--- /dev/null
+++ b/src/blacki/utils/privacy.py
@@ -0,0 +1,40 @@
+"""Privacy helpers for sensitive external-tool data."""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+ROUTE_TOOL_NAMES = frozenset(
+ {
+ "get_route_estimate",
+ "compare_route_scenarios",
+ }
+)
+REDACTED_ROUTE_DETAILS = ""
+
+
+def route_data_redaction_enabled() -> bool:
+ """Return whether this process has enabled the Google Routes integration."""
+ return bool(os.environ.get("GOOGLE_MAPS_ROUTES_API_KEY", "").strip())
+
+
+def redact_route_tool_payload(
+ tool_name: str,
+ payload: dict[str, Any],
+) -> dict[str, Any]:
+ """Remove locations, place IDs, and route content from observable payloads."""
+ if tool_name not in ROUTE_TOOL_NAMES:
+ return payload
+
+ redacted: dict[str, Any] = {"details": REDACTED_ROUTE_DETAILS}
+ for key in (
+ "status",
+ "error_code",
+ "scenario_count",
+ "successful_scenarios",
+ "attribution",
+ ):
+ if key in payload:
+ redacted[key] = payload[key]
+ return redacted
diff --git a/tests/conftest.py b/tests/conftest.py
index d53b4fa..2687215 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -387,6 +387,7 @@ def clean_environment(monkeypatch: pytest.MonkeyPatch) -> None:
"HOST",
"PORT",
"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT",
+ "GOOGLE_MAPS_ROUTES_API_KEY",
]
for var in env_vars_to_clean:
diff --git a/tests/eval/blacki_eval/agent.py b/tests/eval/blacki_eval/agent.py
index e33a787..7b3b2f9 100644
--- a/tests/eval/blacki_eval/agent.py
+++ b/tests/eval/blacki_eval/agent.py
@@ -25,6 +25,34 @@
_active_invocations = 0
+async def _route_eval_compute_routes(
+ payload: dict[str, Any],
+ api_key: str,
+) -> dict[str, Any]:
+ """Return deterministic provider data for route behavior evaluations."""
+ del payload, api_key
+ return {
+ "routes": [
+ {
+ "distanceMeters": 12500,
+ "duration": "1800s",
+ "staticDuration": "1200s",
+ }
+ ]
+ }
+
+
+def _configure_route_eval_boundary() -> None:
+ """Replace only the external Maps boundary when explicitly requested."""
+ if os.environ.get("BLACKI_EVAL_ROUTES", "").strip().lower() != "true":
+ return
+
+ os.environ.setdefault("GOOGLE_MAPS_ROUTES_API_KEY", "eval-only")
+ from blacki.routes import tools as route_tools
+
+ route_tools.compute_routes = _route_eval_compute_routes
+
+
async def _ensure_eval_container(*, callback_context: Any) -> None:
"""Initialize the real storage container once for stateful eval cases."""
global _active_invocations
@@ -128,4 +156,5 @@ async def after_tool_policy(
return eval_agent
+_configure_route_eval_boundary()
root_agent = create_eval_agent()
diff --git a/tests/eval/routes.evalset.json b/tests/eval/routes.evalset.json
new file mode 100644
index 0000000..7da4e72
--- /dev/null
+++ b/tests/eval/routes.evalset.json
@@ -0,0 +1,94 @@
+{
+ "eval_set_id": "routes",
+ "name": "Google Maps route behavior",
+ "description": "Validates dedicated route-tool selection and focused clarification.",
+ "eval_cases": [
+ {
+ "eval_id": "current_traffic_uses_route_tool",
+ "conversation": [
+ {
+ "invocation_id": "routes-current-1",
+ "user_content": {
+ "role": "user",
+ "parts": [
+ {
+ "text": "Call get_route_estimate exactly once for current driving traffic from 'Home address' to 'Office address'. Use travel_mode DRIVE, departure_time now, traffic_model BEST_GUESS, set every avoid option to false, and do not request alternatives."
+ }
+ ]
+ },
+ "intermediate_data": {
+ "tool_uses": [
+ {
+ "name": "get_route_estimate",
+ "args": {
+ "origin": "Home address",
+ "destination": "Office address",
+ "travel_mode": "DRIVE",
+ "departure_time": "now",
+ "traffic_model": "BEST_GUESS",
+ "avoid_tolls": false,
+ "avoid_highways": false,
+ "avoid_ferries": false,
+ "include_alternatives": false
+ }
+ }
+ ]
+ }
+ }
+ ],
+ "session_input": {
+ "app_name": "blacki",
+ "user_id": "routes-eval-user",
+ "state": {}
+ }
+ },
+ {
+ "eval_id": "missing_endpoint_does_not_guess",
+ "conversation": [
+ {
+ "invocation_id": "routes-missing-1",
+ "user_content": {
+ "role": "user",
+ "parts": [
+ {
+ "text": "How long will my commute take right now? I have not told you either endpoint. Ask one focused question and do not call any tool."
+ }
+ ]
+ },
+ "intermediate_data": {
+ "tool_uses": []
+ }
+ }
+ ],
+ "session_input": {
+ "app_name": "blacki",
+ "user_id": "routes-eval-user",
+ "state": {}
+ }
+ },
+ {
+ "eval_id": "non_route_distance_stays_read_only",
+ "conversation": [
+ {
+ "invocation_id": "routes-distance-1",
+ "user_content": {
+ "role": "user",
+ "parts": [
+ {
+ "text": "Explain Levenshtein distance without using any tool."
+ }
+ ]
+ },
+ "intermediate_data": {
+ "tool_uses": []
+ }
+ }
+ ],
+ "session_input": {
+ "app_name": "blacki",
+ "user_id": "routes-eval-user",
+ "state": {}
+ }
+ }
+ ]
+}
diff --git a/tests/eval/routes_eval_config.json b/tests/eval/routes_eval_config.json
new file mode 100644
index 0000000..4f9aa70
--- /dev/null
+++ b/tests/eval/routes_eval_config.json
@@ -0,0 +1,18 @@
+{
+ "criteria": {
+ "tool_trajectory_avg_score": {
+ "threshold": 1.0,
+ "match_type": "EXACT"
+ },
+ "concise_response_score": {
+ "threshold": 1.0
+ }
+ },
+ "custom_metrics": {
+ "concise_response_score": {
+ "code_config": {
+ "name": "tests.eval.prompt_metrics.concise_response_score"
+ }
+ }
+ }
+}
diff --git a/tests/eval/test_eval_agent.py b/tests/eval/test_eval_agent.py
index cd2215a..870e11d 100644
--- a/tests/eval/test_eval_agent.py
+++ b/tests/eval/test_eval_agent.py
@@ -1,5 +1,6 @@
"""Tests for the ADK CLI evaluation adapter."""
+import os
from unittest.mock import MagicMock
import pytest
@@ -7,7 +8,9 @@
from eval.blacki_eval.agent import (
_callback_list,
+ _configure_route_eval_boundary,
_ensure_eval_container,
+ _route_eval_compute_routes,
create_eval_agent,
)
@@ -50,3 +53,52 @@ async def test_eval_container_requires_explicit_sqlite_path(
assert str(error) == "SQLITE_PATH is required for prompt evaluations"
else:
raise AssertionError("missing SQLITE_PATH should fail")
+
+
+@pytest.mark.asyncio
+async def test_route_eval_boundary_is_deterministic() -> None:
+ result = await _route_eval_compute_routes(
+ {"origin": {"address": "private"}},
+ "eval-only",
+ )
+
+ assert result == {
+ "routes": [
+ {
+ "distanceMeters": 12500,
+ "duration": "1800s",
+ "staticDuration": "1200s",
+ }
+ ]
+ }
+
+
+def test_route_eval_boundary_requires_explicit_opt_in(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ from blacki.routes import tools as route_tools
+
+ original = route_tools.compute_routes
+ monkeypatch.delenv("BLACKI_EVAL_ROUTES", raising=False)
+ monkeypatch.delenv("GOOGLE_MAPS_ROUTES_API_KEY", raising=False)
+
+ _configure_route_eval_boundary()
+
+ assert route_tools.compute_routes is original
+ assert "GOOGLE_MAPS_ROUTES_API_KEY" not in os.environ
+
+
+def test_route_eval_boundary_replaces_only_maps_provider(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ from blacki.routes import tools as route_tools
+
+ original = route_tools.compute_routes
+ monkeypatch.setenv("BLACKI_EVAL_ROUTES", "true")
+ monkeypatch.delenv("GOOGLE_MAPS_ROUTES_API_KEY", raising=False)
+ monkeypatch.setattr(route_tools, "compute_routes", original)
+
+ _configure_route_eval_boundary()
+
+ assert route_tools.compute_routes is _route_eval_compute_routes
+ assert os.environ["GOOGLE_MAPS_ROUTES_API_KEY"] == "eval-only"
diff --git a/tests/routes/__init__.py b/tests/routes/__init__.py
new file mode 100644
index 0000000..0024052
--- /dev/null
+++ b/tests/routes/__init__.py
@@ -0,0 +1 @@
+"""Tests for Google Maps Routes tools."""
diff --git a/tests/routes/test_client.py b/tests/routes/test_client.py
new file mode 100644
index 0000000..7aa0c4e
--- /dev/null
+++ b/tests/routes/test_client.py
@@ -0,0 +1,267 @@
+"""Tests for the Google Maps Routes HTTP client."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from typing import Any
+from unittest.mock import AsyncMock, create_autospec, patch
+
+import httpx
+import pytest
+
+from blacki.routes import client as routes_client
+from blacki.routes.client import (
+ COMPUTE_ROUTES_URL,
+ MAX_ATTEMPTS,
+ ROUTES_FIELD_MASK,
+ RoutesAPIError,
+ _get_shared_routes_client,
+ close_shared_routes_client,
+ compute_routes,
+ reset_routes_client_cache,
+)
+
+
+@pytest.fixture(autouse=True)
+async def reset_client() -> AsyncGenerator[None, None]:
+ """Keep the process-wide client isolated between tests."""
+ await reset_routes_client_cache()
+ yield
+ await reset_routes_client_cache()
+
+
+def _strict_client() -> Any:
+ return create_autospec(httpx.AsyncClient, instance=True, spec_set=True)
+
+
+def _response(status_code: int, json: object | None = None) -> httpx.Response:
+ request = httpx.Request("POST", COMPUTE_ROUTES_URL)
+ if json is None:
+ return httpx.Response(status_code, request=request)
+ return httpx.Response(status_code, request=request, json=json)
+
+
+class TestSharedRoutesClient:
+ """Shared-client lifecycle behavior."""
+
+ @pytest.mark.asyncio
+ async def test_creates_and_reuses_client(self) -> None:
+ strict_client = _strict_client()
+ with patch(
+ "blacki.routes.client.httpx.AsyncClient",
+ autospec=True,
+ return_value=strict_client,
+ ) as client_class:
+ first = await _get_shared_routes_client()
+ second = await _get_shared_routes_client()
+
+ assert first is strict_client
+ assert second is strict_client
+ client_class.assert_called_once_with(timeout=15.0)
+
+ @pytest.mark.asyncio
+ async def test_close_handles_none_and_clears_client(self) -> None:
+ routes_client._routes_client = None
+
+ await close_shared_routes_client()
+
+ strict_client = _strict_client()
+ routes_client._routes_client = strict_client
+
+ await close_shared_routes_client()
+
+ strict_client.aclose.assert_awaited_once()
+ assert routes_client._routes_client is None
+
+ @pytest.mark.asyncio
+ async def test_close_logs_and_clears_after_error(
+ self, caplog: pytest.LogCaptureFixture
+ ) -> None:
+ strict_client = _strict_client()
+ strict_client.aclose.side_effect = RuntimeError("close failed")
+ routes_client._routes_client = strict_client
+
+ await close_shared_routes_client()
+
+ assert routes_client._routes_client is None
+ assert "Error while closing shared Google Routes client" in caplog.text
+
+
+class TestComputeRoutes:
+ """HTTP status mapping, field masks, and bounded retry behavior."""
+
+ @pytest.fixture
+ def strict_client(self) -> Any:
+ client = _strict_client()
+ routes_client._routes_client = client
+ return client
+
+ @pytest.mark.asyncio
+ async def test_success_uses_fixed_headers_and_payload(
+ self, strict_client: Any
+ ) -> None:
+ payload = {"origin": {"address": "A"}, "destination": {"address": "B"}}
+ strict_client.post.return_value = _response(200, {"routes": []})
+
+ result = await compute_routes(payload, "secret-key")
+
+ assert result == {"routes": []}
+ strict_client.post.assert_awaited_once_with(
+ COMPUTE_ROUTES_URL,
+ headers={
+ "Content-Type": "application/json",
+ "X-Goog-Api-Key": "secret-key",
+ "X-Goog-FieldMask": ROUTES_FIELD_MASK,
+ },
+ json=payload,
+ )
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("status_code", [401, 403])
+ async def test_authentication_errors_do_not_retry(
+ self, strict_client: Any, status_code: int
+ ) -> None:
+ strict_client.post.return_value = _response(status_code)
+
+ with pytest.raises(RoutesAPIError) as caught:
+ await compute_routes({}, "bad-key")
+
+ assert caught.value.code == "authentication_failed"
+ assert caught.value.retryable is False
+ assert strict_client.post.await_count == 1
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("status_code", [400, 404, 422])
+ async def test_client_errors_are_normalized(
+ self, strict_client: Any, status_code: int
+ ) -> None:
+ strict_client.post.return_value = _response(status_code)
+
+ with pytest.raises(RoutesAPIError) as caught:
+ await compute_routes({}, "key")
+
+ assert caught.value.code == "invalid_request"
+ assert strict_client.post.await_count == 1
+
+ @pytest.mark.asyncio
+ async def test_unexpected_redirect_is_normalized(self, strict_client: Any) -> None:
+ strict_client.post.return_value = _response(302)
+
+ with pytest.raises(RoutesAPIError) as caught:
+ await compute_routes({}, "key")
+
+ assert caught.value.code == "unavailable"
+ assert caught.value.retryable is False
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("first_status", "expected_delay"),
+ [(429, 0.25), (500, 0.25)],
+ )
+ async def test_retryable_status_recovers(
+ self,
+ strict_client: Any,
+ first_status: int,
+ expected_delay: float,
+ ) -> None:
+ strict_client.post.side_effect = [
+ _response(first_status),
+ _response(200, {"routes": [{"distanceMeters": 1}]}),
+ ]
+ sleep = AsyncMock()
+
+ with patch("blacki.routes.client.asyncio.sleep", new=sleep):
+ result = await compute_routes({}, "key")
+
+ assert result["routes"] == [{"distanceMeters": 1}]
+ sleep.assert_awaited_once_with(expected_delay)
+ assert strict_client.post.await_count == 2
+
+ @pytest.mark.asyncio
+ async def test_retries_use_exponential_delays(self, strict_client: Any) -> None:
+ strict_client.post.side_effect = [
+ _response(500),
+ _response(503),
+ _response(200, {"routes": []}),
+ ]
+ sleep = AsyncMock()
+
+ with patch("blacki.routes.client.asyncio.sleep", new=sleep):
+ await compute_routes({}, "key")
+
+ assert [call.args[0] for call in sleep.await_args_list] == [0.25, 0.5]
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ ("status_code", "expected_code"),
+ [(429, "rate_limited"), (503, "unavailable")],
+ )
+ async def test_retryable_status_exhaustion(
+ self,
+ strict_client: Any,
+ status_code: int,
+ expected_code: str,
+ ) -> None:
+ strict_client.post.return_value = _response(status_code)
+
+ with (
+ patch("blacki.routes.client.asyncio.sleep", new=AsyncMock()),
+ pytest.raises(RoutesAPIError) as caught,
+ ):
+ await compute_routes({}, "key")
+
+ assert caught.value.code == expected_code
+ assert strict_client.post.await_count == MAX_ATTEMPTS
+
+ @pytest.mark.asyncio
+ async def test_network_failure_retries_then_recovers(
+ self, strict_client: Any
+ ) -> None:
+ request = httpx.Request("POST", COMPUTE_ROUTES_URL)
+ strict_client.post.side_effect = [
+ httpx.ConnectError("offline", request=request),
+ _response(200, {"routes": []}),
+ ]
+ sleep = AsyncMock()
+
+ with patch("blacki.routes.client.asyncio.sleep", new=sleep):
+ result = await compute_routes({}, "key")
+
+ assert result == {"routes": []}
+ sleep.assert_awaited_once_with(0.25)
+
+ @pytest.mark.asyncio
+ async def test_network_failure_exhaustion(self, strict_client: Any) -> None:
+ request = httpx.Request("POST", COMPUTE_ROUTES_URL)
+ strict_client.post.side_effect = httpx.ConnectError("offline", request=request)
+
+ with (
+ patch("blacki.routes.client.asyncio.sleep", new=AsyncMock()),
+ pytest.raises(RoutesAPIError) as caught,
+ ):
+ await compute_routes({}, "key")
+
+ assert caught.value.code == "unavailable"
+ assert "could not be reached" in str(caught.value)
+ assert strict_client.post.await_count == MAX_ATTEMPTS
+
+ @pytest.mark.asyncio
+ async def test_invalid_json_is_rejected(self, strict_client: Any) -> None:
+ request = httpx.Request("POST", COMPUTE_ROUTES_URL)
+ strict_client.post.return_value = httpx.Response(
+ 200, request=request, content=b"not-json"
+ )
+
+ with pytest.raises(RoutesAPIError) as caught:
+ await compute_routes({}, "key")
+
+ assert caught.value.code == "invalid_response"
+
+ @pytest.mark.asyncio
+ async def test_non_object_json_is_rejected(self, strict_client: Any) -> None:
+ strict_client.post.return_value = _response(200, [])
+
+ with pytest.raises(RoutesAPIError) as caught:
+ await compute_routes({}, "key")
+
+ assert caught.value.code == "invalid_response"
diff --git a/tests/routes/test_tools.py b/tests/routes/test_tools.py
new file mode 100644
index 0000000..d7222f8
--- /dev/null
+++ b/tests/routes/test_tools.py
@@ -0,0 +1,593 @@
+"""Tests for the read-only Google Maps Routes tools."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import AsyncGenerator
+from datetime import UTC, datetime
+from typing import Any, cast
+from unittest.mock import create_autospec, patch
+
+import httpx
+import pytest
+from conftest import MockState, MockToolContext
+from google.adk.tools import FunctionTool, ToolContext
+
+from blacki.routes import client as routes_client
+from blacki.routes.client import (
+ COMPUTE_ROUTES_URL,
+ RoutesAPIError,
+ reset_routes_client_cache,
+)
+from blacki.routes.tools import (
+ MAX_SCENARIOS,
+ RouteScenario,
+ RouteValidationError,
+ _build_payload,
+ _duration_seconds,
+ _normalize_response,
+ _normalize_route,
+ _resolved_waypoint,
+ compare_route_scenarios,
+ get_route_estimate,
+)
+
+FIXED_NOW = datetime(2026, 7, 24, 8, 30, tzinfo=UTC)
+
+
+def _tool_context() -> ToolContext:
+ return cast(ToolContext, MockToolContext(state=MockState({})))
+
+
+def _strict_client() -> Any:
+ client = create_autospec(httpx.AsyncClient, instance=True, spec_set=True)
+ routes_client._routes_client = client
+ return client
+
+
+def _response(status_code: int, json: object | None = None) -> httpx.Response:
+ request = httpx.Request("POST", COMPUTE_ROUTES_URL)
+ if json is None:
+ return httpx.Response(status_code, request=request)
+ return httpx.Response(status_code, request=request, json=json)
+
+
+def _route_response(
+ *,
+ duration: str = "1800s",
+ static_duration: str | None = "1200s",
+) -> dict[str, object]:
+ route: dict[str, object] = {
+ "distanceMeters": 12500,
+ "duration": duration,
+ "description": "Main Road",
+ "routeLabels": ["DEFAULT_ROUTE", 7],
+ "warnings": ["Road closures may apply.", None],
+ }
+ if static_duration is not None:
+ route["staticDuration"] = static_duration
+ return {
+ "routes": [route],
+ "fallbackInfo": {
+ "routingMode": "FALLBACK_TRAFFIC_AWARE",
+ "reason": "LATENCY_EXCEEDED",
+ },
+ "geocodingResults": {
+ "origin": {"placeId": "origin-id", "partialMatch": True},
+ "destination": {"placeId": "destination-id"},
+ },
+ }
+
+
+def _scenario(
+ label: str,
+ *,
+ travel_mode: str = "DRIVE",
+ traffic_model: str = "BEST_GUESS",
+) -> RouteScenario:
+ return RouteScenario(
+ label=label,
+ travel_mode=travel_mode,
+ departure_time="now",
+ traffic_model=traffic_model,
+ avoid_tolls=False,
+ avoid_highways=False,
+ avoid_ferries=False,
+ )
+
+
+@pytest.fixture(autouse=True)
+async def reset_shared_client() -> AsyncGenerator[None, None]:
+ await reset_routes_client_cache()
+ yield
+ await reset_routes_client_cache()
+
+
+class TestGetRouteEstimate:
+ """Public estimate-tool behavior and normalized output."""
+
+ @pytest.mark.asyncio
+ async def test_missing_api_key_is_safe(
+ self, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ monkeypatch.delenv("GOOGLE_MAPS_ROUTES_API_KEY", raising=False)
+
+ result = await get_route_estimate(
+ "A",
+ "B",
+ "DRIVE",
+ "now",
+ "BEST_GUESS",
+ False,
+ False,
+ False,
+ False,
+ _tool_context(),
+ )
+
+ assert result["status"] == "error"
+ assert result["error_code"] == "not_configured"
+ assert result["routes"] == []
+ assert result["attribution"] == "Google Maps"
+
+ @pytest.mark.asyncio
+ async def test_success_normalizes_traffic_without_exposing_place_ids(
+ self, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ monkeypatch.setenv("GOOGLE_MAPS_ROUTES_API_KEY", "test-key")
+ client = _strict_client()
+ client.post.return_value = _response(200, _route_response())
+
+ with patch("blacki.routes.tools.now_utc", return_value=FIXED_NOW):
+ result = await get_route_estimate(
+ " place_id:origin-id ",
+ " Destination address ",
+ "driving",
+ "2026-07-24T14:00:00+05:30",
+ "best guess",
+ True,
+ False,
+ True,
+ True,
+ _tool_context(),
+ )
+
+ assert result["status"] == "success"
+ assert "origin" not in result
+ assert "destination" not in result
+ assert result["travel_mode"] == "DRIVE"
+ assert result["departure_time"] == "2026-07-24T08:30:00Z"
+ assert result["computed_at"] == "2026-07-24T08:30:00+00:00"
+ assert result["attribution"] == "Google Maps"
+ assert result["modifier_warning"] == (
+ "Avoid options are preferences, not guarantees."
+ )
+ assert result["mode_warning"] is None
+ assert result["fallback"] == {
+ "routing_mode": "FALLBACK_TRAFFIC_AWARE",
+ "reason": "LATENCY_EXCEEDED",
+ }
+ assert "resolved_waypoints" not in result
+ assert "origin-id" not in str(result)
+ assert "destination-id" not in str(result)
+
+ route = result["routes"][0]
+ assert route["route_labels"] == ["DEFAULT_ROUTE"]
+ assert route["warnings"] == ["Road closures may apply."]
+ assert route["distance_meters"] == 12500
+ assert route["distance_kilometers"] == 12.5
+ assert route["duration_minutes"] == 30.0
+ assert route["static_duration_minutes"] == 20.0
+ assert route["traffic_delay_minutes"] == 10.0
+ assert route["traffic_delay_percent"] == 50.0
+
+ request_payload = client.post.await_args.kwargs["json"]
+ assert request_payload == {
+ "origin": {"placeId": "origin-id"},
+ "destination": {"address": "Destination address"},
+ "travelMode": "DRIVE",
+ "computeAlternativeRoutes": True,
+ "departureTime": "2026-07-24T08:30:00Z",
+ "routingPreference": "TRAFFIC_AWARE_OPTIMAL",
+ "trafficModel": "BEST_GUESS",
+ "routeModifiers": {
+ "avoidTolls": True,
+ "avoidHighways": False,
+ "avoidFerries": True,
+ },
+ }
+
+ @pytest.mark.asyncio
+ async def test_api_error_uses_stable_contract(
+ self, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ monkeypatch.setenv("GOOGLE_MAPS_ROUTES_API_KEY", "bad-key")
+ client = _strict_client()
+ client.post.return_value = _response(403)
+
+ result = await get_route_estimate(
+ "A",
+ "B",
+ "DRIVE",
+ "now",
+ "BEST_GUESS",
+ False,
+ False,
+ False,
+ False,
+ _tool_context(),
+ )
+
+ assert result["status"] == "error"
+ assert result["error_code"] == "authentication_failed"
+ assert "bad-key" not in str(result)
+
+
+class TestPayloadValidation:
+ """Local validation prevents avoidable and invalid billable requests."""
+
+ @pytest.mark.parametrize(
+ ("travel_mode", "traffic_model", "expected_mode", "routing_preference"),
+ [
+ ("driving", "BEST_GUESS", "DRIVE", "TRAFFIC_AWARE_OPTIMAL"),
+ ("DRIVE", "NONE", "DRIVE", "TRAFFIC_AWARE"),
+ ("motorcycle", "NONE", "TWO_WHEELER", "TRAFFIC_AWARE"),
+ ("walking", "NONE", "WALK", None),
+ ("biking", "NONE", "BICYCLE", None),
+ ("TRANSIT", "NONE", "TRANSIT", None),
+ ],
+ )
+ def test_supported_modes_build_expected_routing(
+ self,
+ travel_mode: str,
+ traffic_model: str,
+ expected_mode: str,
+ routing_preference: str | None,
+ ) -> None:
+ payload, mode, departure = _build_payload(
+ origin="A",
+ destination="B",
+ travel_mode=travel_mode,
+ departure_time="now",
+ traffic_model=traffic_model,
+ avoid_tolls=False,
+ avoid_highways=False,
+ avoid_ferries=False,
+ include_alternatives=False,
+ )
+
+ assert mode == expected_mode
+ assert departure == "now"
+ assert payload.get("routingPreference") == routing_preference
+ assert "departureTime" not in payload
+
+ @pytest.mark.parametrize(
+ ("kwargs", "message"),
+ [
+ ({"travel_mode": "spaceship"}, "Travel mode must be"),
+ ({"departure_time": ""}, "Departure time must"),
+ ({"departure_time": "tomorrow morning"}, "Departure time must"),
+ (
+ {"departure_time": "2026-07-24T08:00:00"},
+ "must include a timezone",
+ ),
+ ({"traffic_model": "average"}, "Traffic model must"),
+ (
+ {"travel_mode": "WALK", "traffic_model": "BEST_GUESS"},
+ "only when travel mode is DRIVE",
+ ),
+ (
+ {
+ "travel_mode": "BICYCLE",
+ "traffic_model": "NONE",
+ "avoid_tolls": True,
+ },
+ "require DRIVE or TWO_WHEELER",
+ ),
+ ({"origin": ""}, "Origin cannot be empty"),
+ ({"destination": ""}, "Destination cannot be empty"),
+ ({"origin": "x" * 513}, "Origin is too long"),
+ ({"origin": "place_id: "}, "Origin place ID cannot be empty"),
+ ],
+ )
+ def test_invalid_inputs_are_rejected_locally(
+ self, kwargs: dict[str, object], message: str
+ ) -> None:
+ request: dict[str, object] = {
+ "origin": "A",
+ "destination": "B",
+ "travel_mode": "DRIVE",
+ "departure_time": "now",
+ "traffic_model": "BEST_GUESS",
+ "avoid_tolls": False,
+ "avoid_highways": False,
+ "avoid_ferries": False,
+ "include_alternatives": False,
+ }
+ request.update(kwargs)
+
+ with pytest.raises(RouteValidationError, match=message):
+ _build_payload(**request) # type: ignore[arg-type]
+
+ @pytest.mark.asyncio
+ async def test_public_tool_returns_validation_error_without_http(
+ self, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ monkeypatch.setenv("GOOGLE_MAPS_ROUTES_API_KEY", "test-key")
+ client = _strict_client()
+
+ result = await get_route_estimate(
+ "",
+ "B",
+ "DRIVE",
+ "now",
+ "BEST_GUESS",
+ False,
+ False,
+ False,
+ False,
+ _tool_context(),
+ )
+
+ assert result["error_code"] == "invalid_input"
+ client.post.assert_not_awaited()
+
+
+class TestResponseNormalization:
+ """Response validation, optional fields, and warning behavior."""
+
+ @pytest.mark.parametrize("value", [None, 10, "10"])
+ def test_duration_requires_protobuf_duration(self, value: object) -> None:
+ with pytest.raises(RoutesAPIError, match="valid duration"):
+ _duration_seconds(value, "duration")
+
+ def test_duration_rejects_non_numeric_seconds(self) -> None:
+ with pytest.raises(RoutesAPIError, match="invalid duration"):
+ _duration_seconds("tenseconds", "duration")
+
+ @pytest.mark.parametrize(
+ ("value", "expected"),
+ [
+ (None, None),
+ ({}, None),
+ ({"placeId": 3}, None),
+ (
+ {"placeId": "abc", "partialMatch": False},
+ {"place_id": "abc", "partial_match": False},
+ ),
+ ],
+ )
+ def test_resolved_waypoint_validation(
+ self, value: object, expected: dict[str, object] | None
+ ) -> None:
+ assert _resolved_waypoint(value) == expected
+
+ @pytest.mark.parametrize(
+ ("route", "message"),
+ [
+ (None, "invalid route"),
+ ({"distanceMeters": True, "duration": "1s"}, "route distance"),
+ ({"distanceMeters": 1, "duration": "bad"}, "route duration"),
+ ],
+ )
+ def test_invalid_routes_are_rejected(self, route: object, message: str) -> None:
+ with pytest.raises(RoutesAPIError, match=message):
+ _normalize_route(route, 0)
+
+ def test_optional_route_fields_and_zero_static_duration(self) -> None:
+ normalized = _normalize_route(
+ {
+ "distanceMeters": 10,
+ "duration": "3.5s",
+ "staticDuration": "0s",
+ "description": 4,
+ "routeLabels": "DEFAULT_ROUTE",
+ "warnings": "warning",
+ },
+ 2,
+ )
+
+ assert normalized["route_index"] == 2
+ assert normalized["description"] == ""
+ assert normalized["route_labels"] == []
+ assert normalized["warnings"] == []
+ assert normalized["duration_seconds"] == 3.5
+ assert normalized["static_duration_seconds"] == 0.0
+ assert normalized["traffic_delay_seconds"] == 3.5
+ assert normalized["traffic_delay_percent"] is None
+
+ def test_missing_static_duration_has_no_traffic_delta(self) -> None:
+ normalized = _normalize_route(
+ {
+ "distanceMeters": 100,
+ "duration": "60s",
+ "routeLabels": ["DEFAULT_ROUTE", 3],
+ "warnings": ["Use caution", 4],
+ },
+ 0,
+ )
+
+ assert normalized["static_duration_seconds"] is None
+ assert normalized["static_duration_minutes"] is None
+ assert normalized["traffic_delay_seconds"] is None
+ assert normalized["traffic_delay_minutes"] is None
+ assert normalized["route_labels"] == ["DEFAULT_ROUTE"]
+ assert normalized["warnings"] == ["Use caution"]
+
+ @pytest.mark.parametrize(
+ ("data", "message"),
+ [
+ ({}, "invalid routes collection"),
+ ({"routes": "invalid"}, "invalid routes collection"),
+ ({"routes": []}, "could not find a route"),
+ ],
+ )
+ def test_invalid_route_collections(
+ self, data: dict[str, object], message: str
+ ) -> None:
+ with pytest.raises(RoutesAPIError, match=message):
+ _normalize_response(
+ data,
+ origin="A",
+ destination="B",
+ travel_mode="DRIVE",
+ departure_time="now",
+ used_route_modifiers=False,
+ )
+
+ @pytest.mark.parametrize("mode", ["WALK", "BICYCLE", "TWO_WHEELER"])
+ def test_beta_modes_include_warning(self, mode: str) -> None:
+ result = _normalize_response(
+ {
+ "routes": [
+ {
+ "distanceMeters": 100,
+ "duration": "60s",
+ }
+ ],
+ "geocodingResults": "invalid",
+ "fallbackInfo": "invalid",
+ },
+ origin="A",
+ destination="B",
+ travel_mode=mode,
+ departure_time="now",
+ used_route_modifiers=False,
+ )
+
+ assert "beta" in result["mode_warning"]
+ assert result["resolved_waypoints"] == {
+ "origin": None,
+ "destination": None,
+ }
+ assert result["fallback"] is None
+ assert result["modifier_warning"] is None
+
+
+class TestCompareRouteScenarios:
+ """Scenario limits, partial failures, and concurrency."""
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize("count", [0, MAX_SCENARIOS + 1])
+ async def test_scenario_count_is_bounded(self, count: int) -> None:
+ result = await compare_route_scenarios(
+ "A",
+ "B",
+ [_scenario(str(index)) for index in range(count)],
+ _tool_context(),
+ )
+
+ assert result["status"] == "error"
+ assert result["error_code"] == "invalid_input"
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize(
+ "scenarios",
+ [
+ [_scenario(" ")],
+ [_scenario("Morning"), _scenario(" morning ")],
+ ],
+ )
+ async def test_labels_must_be_nonempty_and_unique(
+ self, scenarios: list[RouteScenario]
+ ) -> None:
+ result = await compare_route_scenarios("A", "B", scenarios, _tool_context())
+
+ assert result["status"] == "error"
+ assert result["error_code"] == "invalid_input"
+
+ @pytest.mark.asyncio
+ async def test_missing_api_key_is_safe(
+ self, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ monkeypatch.delenv("GOOGLE_MAPS_ROUTES_API_KEY", raising=False)
+
+ result = await compare_route_scenarios(
+ "A", "B", [_scenario("Now")], _tool_context()
+ )
+
+ assert result["error_code"] == "not_configured"
+
+ @pytest.mark.asyncio
+ async def test_partial_and_all_failed_statuses(
+ self, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ monkeypatch.setenv("GOOGLE_MAPS_ROUTES_API_KEY", "test-key")
+ client = _strict_client()
+ client.post.return_value = _response(200, _route_response())
+
+ partial = await compare_route_scenarios(
+ "A",
+ "B",
+ [
+ _scenario("Drive"),
+ _scenario("Invalid walk", travel_mode="WALK"),
+ ],
+ _tool_context(),
+ )
+ failed = await compare_route_scenarios(
+ "A",
+ "B",
+ [
+ _scenario("Walk", travel_mode="WALK"),
+ _scenario("Transit", travel_mode="TRANSIT"),
+ ],
+ _tool_context(),
+ )
+
+ assert partial["status"] == "partial"
+ assert partial["successful_scenarios"] == 1
+ assert partial["scenarios"][0]["label"] == "Drive"
+ assert partial["scenarios"][1]["error_code"] == "invalid_input"
+ assert failed["status"] == "error"
+ assert failed["successful_scenarios"] == 0
+ assert client.post.await_count == 1
+
+ @pytest.mark.asyncio
+ async def test_success_preserves_order_and_bounds_concurrency(
+ self, monkeypatch: pytest.MonkeyPatch
+ ) -> None:
+ monkeypatch.setenv("GOOGLE_MAPS_ROUTES_API_KEY", "test-key")
+ client = _strict_client()
+ active = 0
+ max_active = 0
+ three_started = asyncio.Event()
+
+ async def post(*_args: object, **_kwargs: object) -> httpx.Response:
+ nonlocal active, max_active
+ active += 1
+ max_active = max(max_active, active)
+ if active == 3:
+ three_started.set()
+ await three_started.wait()
+ await asyncio.sleep(0)
+ active -= 1
+ return _response(200, _route_response())
+
+ client.post.side_effect = post
+ labels = ["Now", "Later", "Optimistic", "Pessimistic", "No tolls"]
+
+ result = await compare_route_scenarios(
+ " A ",
+ " B ",
+ [_scenario(label) for label in labels],
+ _tool_context(),
+ )
+
+ assert result["status"] == "success"
+ assert "origin" not in result
+ assert "destination" not in result
+ assert result["scenario_count"] == 5
+ assert result["successful_scenarios"] == 5
+ assert [scenario["label"] for scenario in result["scenarios"]] == labels
+ assert max_active == 3
+ assert "resolved_waypoints" not in str(result)
+
+ def test_adk_declaration_keeps_structured_scenario_schema(self) -> None:
+ declaration = FunctionTool(compare_route_scenarios)._get_declaration()
+
+ assert declaration is not None
+ schema = declaration.parameters_json_schema
+ assert schema is not None
+ assert schema["properties"]["scenarios"]["type"] == "array"
+ assert "RouteScenario" in schema["$defs"]
diff --git a/tests/test_deployment_contract.py b/tests/test_deployment_contract.py
index e48ed74..62cca7a 100644
--- a/tests/test_deployment_contract.py
+++ b/tests/test_deployment_contract.py
@@ -452,6 +452,7 @@ def test_production_deployment_preflights_before_stopping_service() -> None:
assert "--env OPENROUTER_API_KEY" not in workflow
for setting in (
+ "GOOGLE_MAPS_ROUTES_API_KEY",
"OTEL_EXPORTER_OTLP_TRACES_PROTOCOL",
"OTEL_EXPORTER_OTLP_TRACES_HEADERS",
):
diff --git a/tests/test_logging_callbacks.py b/tests/test_logging_callbacks.py
index 52dab80..f26ec4b 100644
--- a/tests/test_logging_callbacks.py
+++ b/tests/test_logging_callbacks.py
@@ -324,6 +324,62 @@ def test_after_tool_without_user_content(
assert "User Content:" not in caplog.text
assert "Tool response: {'status': 'success'}" in caplog.text
+ def test_route_tool_logs_redact_locations_and_place_ids(
+ self,
+ caplog: pytest.LogCaptureFixture,
+ ) -> None:
+ """Route callback logs contain status metadata but no exact locations."""
+ caplog.set_level(logging.DEBUG)
+ callbacks = LoggingCallbacks()
+ tool = MockBaseTool(name="get_route_estimate")
+ context = MockToolContext(
+ user_content=MockContent({"text": "home-address-canary"})
+ )
+ args = {
+ "origin": "home-address-canary",
+ "destination": "office-address-canary",
+ }
+ response = {
+ "status": "success",
+ "resolved_waypoints": {
+ "origin": {"place_id": "place-id-canary"},
+ },
+ "attribution": "Google Maps",
+ }
+
+ callbacks.before_tool(tool, args, context) # type: ignore[arg-type]
+ callbacks.after_tool(tool, args, context, response) # type: ignore[arg-type]
+
+ assert "route details redacted" in caplog.text
+ assert "status" in caplog.text
+ assert "Google Maps" in caplog.text
+ assert "canary" not in caplog.text
+
+ def test_routes_enabled_redacts_agent_and_model_content(
+ self,
+ monkeypatch: pytest.MonkeyPatch,
+ caplog: pytest.LogCaptureFixture,
+ ) -> None:
+ """A Routes-enabled process never logs message or model content."""
+ monkeypatch.setenv("GOOGLE_MAPS_ROUTES_API_KEY", "configured")
+ caplog.set_level(logging.DEBUG)
+ callbacks = LoggingCallbacks()
+ context = MockLoggingCallbackContext(
+ user_content=MockContent({"text": "home-address-canary"})
+ )
+ request = MockLlmRequest(
+ contents=[MockContent({"text": "office-address-canary"})]
+ )
+ response = MockLlmResponse(content=MockContent({"text": "place-id-canary"}))
+
+ callbacks.before_agent(context) # type: ignore[arg-type]
+ callbacks.after_agent(context) # type: ignore[arg-type]
+ callbacks.before_model(context, request) # type: ignore[arg-type]
+ callbacks.after_model(context, response) # type: ignore[arg-type]
+
+ assert "route details redacted" in caplog.text
+ assert "canary" not in caplog.text
+
class TestEdgeCases:
"""Tests for edge cases and special scenarios."""
diff --git a/tests/test_privacy.py b/tests/test_privacy.py
new file mode 100644
index 0000000..b068780
--- /dev/null
+++ b/tests/test_privacy.py
@@ -0,0 +1,56 @@
+"""Tests for sensitive external-tool privacy helpers."""
+
+from typing import Any
+
+import pytest
+
+from blacki.utils.privacy import (
+ REDACTED_ROUTE_DETAILS,
+ redact_route_tool_payload,
+ route_data_redaction_enabled,
+)
+
+
+@pytest.mark.parametrize(
+ ("value", "expected"), [(None, False), (" ", False), ("key", True)]
+)
+def test_route_data_redaction_tracks_routes_configuration(
+ monkeypatch: pytest.MonkeyPatch,
+ value: str | None,
+ expected: bool,
+) -> None:
+ if value is None:
+ monkeypatch.delenv("GOOGLE_MAPS_ROUTES_API_KEY", raising=False)
+ else:
+ monkeypatch.setenv("GOOGLE_MAPS_ROUTES_API_KEY", value)
+
+ assert route_data_redaction_enabled() is expected
+
+
+def test_non_route_payload_is_unchanged() -> None:
+ payload = {"query": "public information"}
+
+ assert redact_route_tool_payload("exa_search", payload) is payload
+
+
+def test_route_payload_preserves_only_safe_operational_metadata() -> None:
+ payload: dict[str, Any] = {
+ "status": "success",
+ "origin": "home-address-canary",
+ "destination": "office-address-canary",
+ "resolved_waypoints": {"origin": {"place_id": "place-id-canary"}},
+ "scenario_count": 2,
+ "successful_scenarios": 1,
+ "attribution": "Google Maps",
+ }
+
+ redacted = redact_route_tool_payload("get_route_estimate", payload)
+
+ assert redacted == {
+ "details": REDACTED_ROUTE_DETAILS,
+ "status": "success",
+ "scenario_count": 2,
+ "successful_scenarios": 1,
+ "attribution": "Google Maps",
+ }
+ assert "canary" not in str(redacted)
diff --git a/tests/test_prompt.py b/tests/test_prompt.py
index 7bbb364..b733dcb 100644
--- a/tests/test_prompt.py
+++ b/tests/test_prompt.py
@@ -36,6 +36,8 @@
"schedule_reminder",
"list_reminders",
"cancel_reminder",
+ "get_route_estimate",
+ "compare_route_scenarios",
"exa_search",
"brave_search",
}
@@ -123,7 +125,14 @@ def test_global_instruction_uses_application_timezone(
("I ate a sandwich for lunch", ("nutrition",)),
("Log my resistance workout", ("workout",)),
("Suggest a reminder schedule", ("reminder",)),
+ ("What is the current traffic on my commute?", ("routes",)),
+ ("How far is Pune from Mumbai?", ("routes",)),
+ ("Drive from Pune to Mumbai", ("routes",)),
+ ("What is the ETA to the airport?", ("routes",)),
("What is the latest verified Python news?", ("search",)),
+ ("Walk me through the latest Python news", ("search",)),
+ ("Find current Google Drive news", ("search",)),
+ ("Explain Levenshtein distance", ()),
("Explain dependency injection", ()),
],
)
@@ -177,6 +186,22 @@ def test_reminder_discussion_is_read_only(self) -> None:
assert "discussing a possible schedule is read-only" in instruction
assert "Ask for a missing required\ntime" in instruction
+ def test_routes_policy_uses_fresh_dedicated_data(self) -> None:
+ instruction = build_domain_instruction(
+ "Compare current traffic for my commute",
+ {"get_route_estimate", "compare_route_scenarios", "exa_search"},
+ )
+
+ assert "" in instruction
+ assert "requires a fresh route lookup" in instruction
+ assert "not continuous tracking" in instruction
+ assert "Use get_route_estimate for one route" in instruction
+ assert "compare_route_scenarios only when" in instruction
+ assert (
+ "For current driving traffic use DRIVE, now, and BEST_GUESS" in instruction
+ )
+ assert "" not in instruction
+
@pytest.mark.parametrize(
("tools", "expected", "unexpected"),
[
@@ -295,6 +320,39 @@ async def test_search_initially_exposes_only_primary_and_hides_sandbox(
assert second_tool.function_declarations is None
assert request.config.tools[-1] is opaque_tool
+ @pytest.mark.asyncio
+ async def test_route_request_hides_generic_search_declarations(self) -> None:
+ plugin = DomainPolicyPlugin()
+ request = _request_with_tools(
+ "get_route_estimate",
+ "compare_route_scenarios",
+ "exa_search",
+ "brave_search",
+ )
+ context = SimpleNamespace(
+ user_content=_user_content(
+ "Compare the current traffic for my drive to work"
+ ),
+ state={},
+ )
+
+ await plugin.before_model_callback(
+ callback_context=context, # type: ignore[arg-type]
+ llm_request=request,
+ )
+
+ assert "" in str(request.config.system_instruction)
+ assert "temp:blacki_search_primary" not in context.state
+ assert request.config.tools is not None
+ first_tool = request.config.tools[0]
+ assert isinstance(first_tool, types.Tool)
+ declarations = first_tool.function_declarations
+ assert declarations is not None
+ assert [declaration.name for declaration in declarations] == [
+ "get_route_estimate",
+ "compare_route_scenarios",
+ ]
+
@pytest.mark.asyncio
async def test_successful_search_removes_search_tools_on_next_model_call(
self,
diff --git a/tests/test_registry.py b/tests/test_registry.py
index 54a168b..dadd00a 100644
--- a/tests/test_registry.py
+++ b/tests/test_registry.py
@@ -15,6 +15,7 @@ def test_default_values(self) -> None:
assert config.exa_api_key is None
assert config.brave_search_api_key is None
+ assert config.google_maps_routes_api_key is None
assert config.sqlite_path is None
assert config.sandbox_enabled is False
assert config.skills_dir is None
@@ -26,6 +27,7 @@ def test_custom_values(self) -> None:
config = ToolConfig(
exa_api_key="exa-key",
brave_search_api_key="test-key",
+ google_maps_routes_api_key="routes-key",
sqlite_path="/tmp/blacki.db",
sandbox_enabled=True,
skills_dir=skills_path,
@@ -34,6 +36,7 @@ def test_custom_values(self) -> None:
assert config.exa_api_key == "exa-key"
assert config.brave_search_api_key == "test-key"
+ assert config.google_maps_routes_api_key == "routes-key"
assert config.sqlite_path == "/tmp/blacki.db"
assert config.sandbox_enabled is True
assert config.skills_dir == skills_path
@@ -70,6 +73,16 @@ def test_exa_search_tools_added_first(self) -> None:
assert [tool.__name__ for tool in tools[:2]] == ["exa_search", "brave_search"]
assert len(tools) == 10
+ def test_google_routes_tools_added_when_key_provided(self) -> None:
+ """Should add both read-only Routes tools when configured."""
+ config = ToolConfig(google_maps_routes_api_key="routes-key")
+
+ tools = build_tools(config)
+
+ tool_names = {tool.__name__ for tool in tools}
+ assert {"get_route_estimate", "compare_route_scenarios"} <= tool_names
+ assert len(tools) == 10
+
def test_database_tools_added(self) -> None:
"""Should add database-backed tools when sqlite path provided."""
config = ToolConfig(sqlite_path="/tmp/blacki.db")
@@ -139,6 +152,7 @@ def test_empty_env(self) -> None:
assert config.exa_api_key is None
assert config.brave_search_api_key is None
+ assert config.google_maps_routes_api_key is None
assert config.sqlite_path is not None
assert config.sqlite_path.endswith(".adk/tools.db")
assert config.sandbox_enabled is False
@@ -183,6 +197,26 @@ def test_brave_search_api_key_empty_string_becomes_none(self) -> None:
assert config.brave_search_api_key is None
+ def test_google_routes_api_key_is_stripped(self) -> None:
+ """Should normalize the optional Google Maps Routes API key."""
+ with patch.dict(
+ "os.environ",
+ {"GOOGLE_MAPS_ROUTES_API_KEY": " routes-key "},
+ clear=False,
+ ):
+ config = build_tool_config_from_env()
+
+ assert config.google_maps_routes_api_key == "routes-key"
+
+ with patch.dict(
+ "os.environ",
+ {"GOOGLE_MAPS_ROUTES_API_KEY": " "},
+ clear=False,
+ ):
+ config = build_tool_config_from_env()
+
+ assert config.google_maps_routes_api_key is None
+
def test_sqlite_path_from_env(self) -> None:
"""Should read SQLITE_PATH from env."""
with patch.dict("os.environ", {"SQLITE_PATH": "/tmp/blacki.db"}, clear=False):
@@ -262,6 +296,30 @@ def test_returns_tool_when_available(self) -> None:
assert tools[0].__name__ == "exa_search"
+class TestBuildGoogleRoutesTools:
+ """Tests for _build_google_routes_tools."""
+
+ def test_returns_routes_tools_when_available(self) -> None:
+ """Should return both read-only Routes tools."""
+ from blacki.registry import _build_google_routes_tools
+
+ tools = _build_google_routes_tools()
+
+ assert [tool.__name__ for tool in tools] == [
+ "get_route_estimate",
+ "compare_route_scenarios",
+ ]
+
+ def test_returns_empty_on_import_error(self) -> None:
+ """Should omit Routes tools if the module cannot be imported."""
+ from blacki.registry import _build_google_routes_tools
+
+ with patch.dict("sys.modules", {"blacki.routes": None}):
+ tools = _build_google_routes_tools()
+
+ assert tools == []
+
+
class TestBuildReminderTools:
"""Tests for _build_reminder_tools."""
diff --git a/tests/test_server_config.py b/tests/test_server_config.py
index dd8b734..0902af7 100644
--- a/tests/test_server_config.py
+++ b/tests/test_server_config.py
@@ -18,7 +18,9 @@ def mock_dependencies() -> Generator[MagicMock]:
patch("google.adk.cli.fast_api.get_fast_api_app") as mock_get_app,
patch("blacki.utils.initialize_environment") as mock_init_env,
patch("blacki.utils.configure_otel_resource"),
- patch("openinference.instrumentation.google_adk.GoogleADKInstrumentor"),
+ patch(
+ "openinference.instrumentation.google_adk.GoogleADKInstrumentor"
+ ) as mock_instrumentor,
patch("blacki.utils.setup_logging"),
patch("blacki.utils.setup_tracing"),
):
@@ -35,6 +37,7 @@ def mock_dependencies() -> Generator[MagicMock]:
mock_init_env.return_value = mock_env
mock_get_app.return_value = FastAPI()
+ mock_get_app.instrumentor = mock_instrumentor
yield mock_get_app
@@ -52,6 +55,24 @@ def test_server_session_service_uri_is_none(mock_dependencies: MagicMock) -> Non
assert call_kwargs["session_service_uri"] is None
+def test_routes_enabled_redacts_openinference_content(
+ mock_dependencies: MagicMock,
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """Routes-enabled tracing hides tool/model inputs and outputs."""
+ monkeypatch.setenv("GOOGLE_MAPS_ROUTES_API_KEY", "configured")
+ if "blacki.server" in sys.modules:
+ del sys.modules["blacki.server"]
+
+ import blacki.server # noqa: F401
+
+ instrument = mock_dependencies.instrumentor.return_value.instrument
+ instrument.assert_called_once()
+ config = instrument.call_args.kwargs["config"]
+ assert config.hide_inputs is True
+ assert config.hide_outputs is True
+
+
@pytest.mark.asyncio
async def test_server_lifespan_closes_search_clients(
mock_dependencies: MagicMock,
@@ -68,6 +89,7 @@ async def test_server_lifespan_closes_search_clients(
close_container = AsyncMock()
close_brave = AsyncMock()
close_exa = AsyncMock()
+ close_routes = AsyncMock()
close_notify = AsyncMock()
log_warning = MagicMock()
@@ -85,6 +107,7 @@ async def test_server_lifespan_closes_search_clients(
patch.object(server.logger, "warning", new=log_warning),
patch("blacki.tools.close_shared_brave_search_client", new=close_brave),
patch("blacki.search.close_shared_exa_search_client", new=close_exa),
+ patch("blacki.routes.close_shared_routes_client", new=close_routes),
patch("blacki.callbacks.close_shared_notify_client", new=close_notify),
):
async with server.lifespan(server.app):
@@ -94,6 +117,7 @@ async def test_server_lifespan_closes_search_clients(
close_container.assert_awaited_once()
close_brave.assert_awaited_once()
close_exa.assert_awaited_once()
+ close_routes.assert_awaited_once()
close_notify.assert_awaited_once()
log_warning.assert_called_once_with("test warning")
@@ -115,6 +139,7 @@ async def test_lifespan_cleans_up_after_validation_failure(
stop_scheduler = AsyncMock()
close_brave = AsyncMock()
close_exa = AsyncMock()
+ close_routes = AsyncMock()
close_notify = AsyncMock()
with (
@@ -133,6 +158,7 @@ async def test_lifespan_cleans_up_after_validation_failure(
),
patch("blacki.tools.close_shared_brave_search_client", new=close_brave),
patch("blacki.search.close_shared_exa_search_client", new=close_exa),
+ patch("blacki.routes.close_shared_routes_client", new=close_routes),
patch("blacki.callbacks.close_shared_notify_client", new=close_notify),
pytest.raises(server.ConfigurationError, match="invalid"),
):
@@ -144,6 +170,7 @@ async def test_lifespan_cleans_up_after_validation_failure(
close_container.assert_awaited_once()
close_brave.assert_awaited_once()
close_exa.assert_awaited_once()
+ close_routes.assert_awaited_once()
close_notify.assert_awaited_once()
@@ -162,6 +189,7 @@ async def test_lifespan_tolerates_container_closed_during_runtime(
close_container = AsyncMock()
close_brave = AsyncMock()
close_exa = AsyncMock()
+ close_routes = AsyncMock()
close_notify = AsyncMock()
with (
@@ -177,6 +205,7 @@ async def test_lifespan_tolerates_container_closed_during_runtime(
patch.object(server.validation, "validate_configuration", return_value=[]),
patch("blacki.tools.close_shared_brave_search_client", new=close_brave),
patch("blacki.search.close_shared_exa_search_client", new=close_exa),
+ patch("blacki.routes.close_shared_routes_client", new=close_routes),
patch("blacki.callbacks.close_shared_notify_client", new=close_notify),
):
async with server.lifespan(server.app):
@@ -185,6 +214,7 @@ async def test_lifespan_tolerates_container_closed_during_runtime(
close_container.assert_not_awaited()
close_brave.assert_awaited_once()
close_exa.assert_awaited_once()
+ close_routes.assert_awaited_once()
close_notify.assert_awaited_once()
diff --git a/tests/test_task_worker.py b/tests/test_task_worker.py
index 500fa76..59a52f7 100644
--- a/tests/test_task_worker.py
+++ b/tests/test_task_worker.py
@@ -38,6 +38,7 @@ def _task_worker_test_config() -> ToolConfig:
"""Build a deterministic config with sandbox and skill toolsets enabled."""
skills_dir = Path(__file__).parents[1] / "src" / "blacki" / "skills"
return ToolConfig(
+ google_maps_routes_api_key="routes-key",
sandbox_enabled=True,
skills_dir=skills_dir,
weather_enabled=False,
@@ -110,6 +111,8 @@ def test_enabled_task_worker_has_equivalent_isolated_toolsets(
worker_capabilities.pop("finish_task")
assert root_capabilities == worker_capabilities
+ assert root_capabilities["get_route_estimate"] == 1
+ assert root_capabilities["compare_route_scenarios"] == 1
root_toolsets = [tool for tool in agent.tools if isinstance(tool, BaseToolset)]
worker_toolsets = [tool for tool in worker.tools if isinstance(tool, BaseToolset)]
diff --git a/tests/test_telegram_tool_notifications.py b/tests/test_telegram_tool_notifications.py
index 44b4240..a8eb366 100644
--- a/tests/test_telegram_tool_notifications.py
+++ b/tests/test_telegram_tool_notifications.py
@@ -386,6 +386,36 @@ async def test_notify_sends_to_telegram_with_chat_and_thread(
assert "Using tool" in kwargs["text"]
+@pytest.mark.asyncio
+async def test_route_notification_redacts_exact_locations(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ """Tool notices identify route use without repeating private endpoints."""
+ monkeypatch.setenv("TELEGRAM_ENABLED", "true")
+ monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "secret-token")
+ monkeypatch.setenv("TELEGRAM_TOOL_NOTIFICATIONS", "true")
+
+ mock_client = MagicMock()
+ mock_client.send_message = AsyncMock()
+ context = MockToolContext(
+ state=MockState({"telegram_chat_id": "4242"}),
+ )
+
+ with patch("blacki.callbacks.TelegramApiClient", return_value=mock_client):
+ await notify_telegram_before_tool(
+ cast(BaseTool, MockBaseTool("get_route_estimate")),
+ {
+ "origin": "home-address-canary",
+ "destination": "office-address-canary",
+ },
+ cast(ToolContext, context),
+ )
+
+ text = mock_client.send_message.await_args.kwargs["text"]
+ assert "route details redacted" in text
+ assert "canary" not in text
+
+
@pytest.mark.asyncio
async def test_notify_sends_for_each_tool_call(
monkeypatch: pytest.MonkeyPatch,