File size: 5,585 Bytes
8b96826 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | """Tool registry: neutral JSON-schema specs + a dispatcher.
The same specs are converted to Gemini FunctionDeclarations or OpenAI-style
tools by the provider adapters, so tools are defined exactly once.
"""
import json
from app.tools import advisories, events, festivals, geocode, traffic, weather
from app.history import get_route_history, get_route_patterns
TOOL_SPECS = [
{
"name": "geocode_place",
"description": "Resolve a place name in Lucknow to coordinates and a full address.",
"parameters": {
"type": "object",
"properties": {
"place": {"type": "string", "description": "Place name, e.g. 'Hazratganj' or 'Charbagh station'"},
},
"required": ["place"],
},
},
{
"name": "get_route",
"description": "Get driving routes (with live traffic delay) between two coordinates in Lucknow.",
"parameters": {
"type": "object",
"properties": {
"o_lat": {"type": "number"}, "o_lon": {"type": "number"},
"d_lat": {"type": "number"}, "d_lon": {"type": "number"},
"depart_at": {"type": "string", "description": "Optional ISO datetime for future departure"},
},
"required": ["o_lat", "o_lon", "d_lat", "d_lon"],
},
},
{
"name": "compare_departures",
"description": (
"Simulate the same route at multiple departure times before a deadline "
"and return the ETA curve plus the recommended departure time."
),
"parameters": {
"type": "object",
"properties": {
"o_lat": {"type": "number"}, "o_lon": {"type": "number"},
"d_lat": {"type": "number"}, "d_lon": {"type": "number"},
"arrive_by_iso": {"type": "string", "description": "ISO datetime deadline, e.g. 2026-06-12T09:30:00"},
},
"required": ["o_lat", "o_lon", "d_lat", "d_lon", "arrive_by_iso"],
},
},
{
"name": "get_weather",
"description": "Weather forecast summary (rain, wind, temperature) for a time window at a location.",
"parameters": {
"type": "object",
"properties": {
"lat": {"type": "number"}, "lon": {"type": "number"},
"start_iso": {"type": "string"}, "end_iso": {"type": "string"},
},
"required": ["lat", "lon", "start_iso", "end_iso"],
},
},
{
"name": "get_festivals",
"description": "Festivals and public holidays affecting Lucknow traffic on a date (includes Bada Mangal, Muharram).",
"parameters": {
"type": "object",
"properties": {"date_str": {"type": "string", "description": "YYYY-MM-DD"}},
"required": ["date_str"],
},
},
{
"name": "get_events",
"description": "Public events (concerts, matches) in Lucknow on a date that may cause traffic.",
"parameters": {
"type": "object",
"properties": {"date_str": {"type": "string", "description": "YYYY-MM-DD"}},
"required": ["date_str"],
},
},
{
"name": "get_police_advisories",
"description": "Active Lucknow Traffic Police route diversions and road closure advisories.",
"parameters": {"type": "object", "properties": {}},
},
{
"name": "get_route_history",
"description": (
"Retrieve past commute results stored in MongoDB for a specific route. "
"Use when the user asks about their usual trips, past delays, or historical patterns."
),
"parameters": {
"type": "object",
"properties": {
"from_name": {"type": "string", "description": "Origin place name, e.g. 'Charbagh'"},
"to_name": {"type": "string", "description": "Destination place name, e.g. 'KGMU'"},
"limit": {"type": "integer", "description": "Max results to return (default 10, max 50)"},
},
"required": ["from_name", "to_name"],
},
},
{
"name": "get_route_patterns",
"description": (
"Aggregate historical delay and risk by day-of-week for a route pair from MongoDB. "
"Use when the user asks which day is worst, or what their typical delays look like."
),
"parameters": {
"type": "object",
"properties": {
"from_name": {"type": "string"},
"to_name": {"type": "string"},
},
"required": ["from_name", "to_name"],
},
},
]
_DISPATCH = {
"geocode_place": geocode.geocode_place,
"get_route": traffic.get_route,
"compare_departures": traffic.compare_departures,
"get_weather": weather.get_weather,
"get_festivals": festivals.get_festivals,
"get_events": events.get_events,
"get_police_advisories": advisories.get_police_advisories,
"get_route_history": get_route_history,
"get_route_patterns": get_route_patterns,
}
def dispatch(name, args):
"""Execute a tool by name; always returns a JSON string for the LLM."""
func = _DISPATCH.get(name)
if func is None:
return json.dumps({"error": f"unknown tool: {name}"})
try:
return json.dumps(func(**(args or {})), default=str)
except Exception as error: # tool failures go back to the model, not up the stack
return json.dumps({"error": str(error)})
|