teslamate-mcp

Instructions

Implementation Reference

• main.py:22-28 (handler)

  Factory that dynamically creates the handler function for the "get_most_visited_locations" tool (and others), which executes the SQL query from the specified file synchronously.
  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
• src/tools.py:82-86 (registration)

  ToolDefinition that provides the name, description, and SQL file path for the "get_most_visited_locations" tool, used during dynamic registration.
  ToolDefinition( name="get_most_visited_locations", description="Get the most visited locations for each car. Shows frequently visited places with visit counts and durations.", sql_file="most_visited_locations.sql", ),
• main.py:32-39 (registration)

  Dynamic registration code that instantiates the handler for "get_most_visited_locations" using its ToolDefinition and registers it with the FastMCP server.
  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:179-186 (schema)

  Defines the input schema (empty object, no parameters) for the "get_most_visited_locations" tool (and other predefined tools) in the list_tools implementation for HTTP server.
  for tool_def in TOOL_DEFINITIONS: tools.append( types.Tool( name=tool_def.name, description=tool_def.description, inputSchema={"type": "object", "properties": {}}, ) )
• src/database.py:27-34 (helper)

  Helper method invoked by the tool handler to load the SQL from "most_visited_locations.sql" and execute it against 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()