get_charging_by_location
Retrieve Tesla vehicle charging data by location from the TeslaMate database. Analyze charging patterns and usage for efficient energy management.
Instructions
Get the charging by location for each car.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/tools.py:42-46 (schema)Schema definition for the 'get_charging_by_location' tool, specifying name, description, and the SQL file containing the query logic.ToolDefinition( name="get_charging_by_location", description="Get the charging by location for each car. Shows charging patterns and statistics grouped by location.", sql_file="charging_by_location.sql", ),
- main.py:22-29 (handler)Handler factory that creates the execution function for the tool by reading and executing the associated SQL file via the database manager.def create_tool_handler(sql_file: str): """Factory function to create tool handlers""" def handler() -> List[Dict[str, Any]]: return db_manager.execute_query_sync(sql_file) return handler
- main.py:31-39 (registration)Registration loop that dynamically creates and registers the MCP tool handler for each defined tool, including 'get_charging_by_location', using FastMCP's tool decorator.# Register all tools from definitions for tool_def in TOOL_DEFINITIONS: tool_func = create_tool_handler(tool_def.sql_file) tool_func.__doc__ = tool_def.description tool_func.__name__ = tool_def.name # Register the tool with the MCP server mcp.tool()(tool_func)
- src/database.py:27-34 (helper)Core helper function called by the tool handler to read the SQL file and execute the query on the database.def execute_query_sync(self, sql_file_path: str) -> List[Dict[str, Any]]: """Execute SQL query synchronously""" sql_query = self.read_sql_file(sql_file_path) with psycopg.connect(self.connection_string, row_factory=dict_row) as conn: with conn.cursor() as cur: cur.execute(sql_query) return cur.fetchall()
- main_remote.py:179-186 (registration)In the remote HTTP server, the list_tools method registers the tool schema (name and description) for 'get_charging_by_location' among others.for tool_def in TOOL_DEFINITIONS: tools.append( types.Tool( name=tool_def.name, description=tool_def.description, inputSchema={"type": "object", "properties": {}}, ) )
- main_remote.py:118-127 (handler)Handler function in the remote server that looks up the tool definition and executes its SQL query asynchronously.async def execute_predefined_tool(tool_name: str) -> List[Dict[str, Any]]: """Execute a predefined tool by name""" if not app_context: raise RuntimeError("Application context not initialized") tool = get_tool_by_name(tool_name) return await app_context.db_manager.execute_query_async( tool.sql_file, app_context.db_pool )