get_efficiency_by_month_and_temperature
Analyze Tesla vehicle efficiency trends by month and temperature using detailed data from the TeslaMate database. Identify patterns to optimize performance and energy usage.
Instructions
Get the efficiency by month and temperature for each car.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/tools.py:67-71 (schema)Tool definition specifying name, description, and the SQL query file to execute for this parameterless tool.ToolDefinition( name="get_efficiency_by_month_and_temperature", description="Get the efficiency by month and temperature for each car. Analyzes how seasonal temperature changes affect vehicle efficiency.", sql_file="efficiency_by_month_and_temperature.sql", ),
- main.py:22-28 (handler)Dynamic handler factory that creates a parameterless function executing the SQL query synchronously from the specified file.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-38 (registration)Registration loop that dynamically creates and registers the handler for this tool (and others) with the FastMCP server using mcp.tool().# 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)
- main_remote.py:118-127 (handler)Async handler for predefined tools in the remote HTTP MCP server, retrieves tool definition and executes the 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 )
- src/database.py:27-34 (helper)Core helper method that reads the SQL file and executes it synchronously against the PostgreSQL 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()