Ctx Geo¶
ctx_geo — G6 Geographic & Spatial Intelligence component.
Cluster: Uncategorised | Type: component | MCP Tools: 25
Overview¶
Geographic and spatial intelligence component. Provides geocoding, reverse geocoding, directions, place search, distance matrices, elevation, timezone lookup, weather, and air quality via a unified Tier 1 block and a 25-op MCP sub-package with SQLite persistence.
External Google Maps service
ctx_geo is an external-service adapter, not a standalone offline geo engine. Google-backed operations require GOOGLE_MAPS_API_KEY and may incur Google Maps Platform charges. Calls are blocked unless the request sets allow_paid_api=True or the environment sets ALLOW_PAID_API=1. Without a configured key, Google-backed operations fail with an actionable setup error.
Local persistence operations such as geo_store, geo_retrieve, geo_list, geo_delete, and geo_search remain available without a Google Maps key.
Degradation is surfaced through the canonical envelope fields completion_state, warning_card, evidence, available_with, request_id, and run_id. Google-backed live status remains qualified-draft until opt-in real-credential smoke tests pass for geocode, reverse_geocode, and geo_info.
When to use:
- Resolving addresses to coordinates (or vice versa) in agent workflows
- Finding nearby places or calculating travel routes between locations
- Enriching job agent outputs with geographic context
Example:
from mvp.ctx_geo import CtxGeoBlock, GeoInput
block = CtxGeoBlock(name="geo")
result = block.infer(GeoInput(
op="geocode",
address="1 Martin Place, Sydney",
allow_paid_api=True,
))
# result.ok -> True when GOOGLE_MAPS_API_KEY is configured.
Works well with: job_tourism, job_hospitality, adapt_diagrams
Weakest link: Google Maps credentials, quota, optional transport, and live endpoint verification determine whether Google-backed ops are usable. Use geo_info for the per-op capability map before relying on a paid external call.
Public API¶
CtxGeoDecisionError(ValueError)¶
The LLM did not produce a usable, validated geo-search-rerank decision.
GeoRerankDecision¶
Validated advisory rerank verdict over a returned geo-search-record set.
| Field | Type | Default |
|---|---|---|
ordered_indices | tuple[int, ...] | required |
dropped_indices | tuple[int, ...] | () |
rationale | str | '' |
eligible_fingerprint | str | '' |
confidence | float | 0.0 |
degraded | bool | False |
raw_response | str | '' |
LLMGeoRerankRuntime¶
Provider-neutral geo-search-result-rerank runtime backed by G6's LLM caller.
Constructor:
| Parameter | Type | Default |
|---|---|---|
llm | LLMCaller \| None | None |
Methods:
rerank(query: str, records: list[Any]) -> GeoRerankDecision¶
CtxGeoRerankPatternRuntime¶
Stateless, load-bearing returned-set-ceiling enforcement.
Methods:
enforce_eligibility(decision: GeoRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]¶
CtxGeoPlanner¶
Runtime-first advisory result-rerank facade with returned-order fallback.
Constructor:
| Parameter | Type | Default |
|---|---|---|
runtime | GeoRerankRuntime \| None | None |
pattern_runtime | CtxGeoRerankPatternRuntime \| None | None |
Methods:
rerank(query: str, records: list[Any]) -> list[Any]¶
CtxGeoBlock(AIBlock[GeoInput, GeoOutput, None])¶
Tier 1 geo block — validates input then delegates to GoogleMapsClient.
| Field | Type | Default |
|---|---|---|
name | str | 'ctx_geo' |
resource_bounds | ResourceBounds \| None | None |
usage | ResourceUsage | field(default_factory=ResourceUsage) |
Methods:
infer(data: GeoInput) -> Result[GeoOutput]¶
GeoLocation(BaseModel)¶
| Field | Type | Default |
|---|---|---|
lat | float | required |
lng | float | required |
address | str | '' |
GeoInput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | Literal['geocode', 'reverse_geocode', 'directions', 'search_places', 'search_nearby', 'distance_matrix', 'elevation', 'timezone', 'ops', 'help'] | required |
address | str | '' |
lat | float | 0.0 |
lng | float | 0.0 |
origin | str | '' |
destination | str | '' |
query | str | '' |
place_type | str | '' |
keyword | str | '' |
radius | int | 5000 |
origins | list[str] | Field(default_factory=list) |
destinations | list[str] | Field(default_factory=list) |
travel_mode | Literal['driving', 'walking', 'bicycling', 'transit'] | 'driving' |
allow_paid_api | bool | False |
run_mode | Literal['beta', 'production'] | 'beta' |
reviewer_signature | str | '' |
GeoOutput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | str | required |
location | dict | Field(default_factory=dict) |
results | list[dict] | Field(default_factory=list) |
summary | dict | Field(default_factory=dict) |
message | str | '' |
degraded | bool | False |
degradation_reason | str | '' |
available_with | str | '' |
completion_state | str | 'qualified-draft' |
warning_card | dict[str, Any] | Field(default_factory=dict) |
evidence | dict[str, Any] | Field(default_factory=dict) |
request_id | str | '' |
task_id | str | '' |
run_id | str | '' |
Functions¶
eligible_fingerprint(records: list[Any]) -> str¶
sha256 over the returned record texts (order-sensitive).
validate_geo_rerank_decision(decision: GeoRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶
Returned-set / anti-injection guard for a geo-search-rerank decision.
planner_is_llm_trusted(planner: Any) -> bool¶
Whether the BLOCK may report
llm_used=Trueforplanner.
MCP Tools¶
| Operation | Source |
|---|---|
search_nearby | geo_mcp |
search_places | geo_mcp |
place_details | geo_mcp |
explore_area | geo_mcp |
compare_places | geo_mcp |
local_rank_tracker | geo_mcp |
directions | geo_mcp |
distance_matrix | geo_mcp |
plan_route | geo_mcp |
search_along_route | geo_mcp |
geocode | geo_mcp |
reverse_geocode | geo_mcp |
batch_geocode | geo_mcp |
elevation | geo_mcp |
timezone | geo_mcp |
weather | geo_mcp |
air_quality | geo_mcp |
static_map | geo_mcp |
geo_store | geo_mcp |
geo_retrieve | geo_mcp |
geo_list | geo_mcp |
geo_delete | geo_mcp |
geo_search | geo_mcp |
geo_batch | geo_mcp |
geo_info | geo_mcp |