get_current_car_status

Retrieve real-time vehicle status including location, battery level, and state for each car from your TeslaMate database.

Instructions

Get the current car status for each car. Returns real-time vehicle status including location, battery level, and state.

Input Schema

TableJSON Schema

Name	Required	Description	Default
No arguments

Implementation Reference

src/tools.py:47-51 (registration)

Registration of the 'get_current_car_status' tool in the TOOL_DEFINITIONS list, specifying its name, description, and the SQL file containing the query logic.

ToolDefinition(
    name="get_current_car_status",
    description="Get the current car status for each car. Returns real-time vehicle status including location, battery level, and state.",
    sql_file="current_car_status.sql",
),

main.py:22-29 (handler)

Factory that creates the handler function for 'get_current_car_status', which executes the associated SQL query synchronously using 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)

Dynamically registers the handler for 'get_current_car_status' (and all tools) with the MCP server using the 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)

main_remote.py:118-127 (handler)

Handler for executing predefined tools including 'get_current_car_status' in the remote HTTP server by retrieving the tool definition and running 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
    )

src/database.py:27-34 (helper)

Core helper function that implements the tool logic by reading the SQL file and executing 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()

teslamate-mcp