teslamate-mcp

Instructions

Implementation Reference

• src/tools.py:52-56 (registration)

  Tool definition registration including name, description, and associated SQL query file 'daily_battery_usage.sql'.
  ToolDefinition( name="get_daily_battery_usage_patterns", description="Get the daily battery usage patterns for each car. Analyzes battery consumption patterns throughout the day.", sql_file="daily_battery_usage.sql", ),
• main.py:22-29 (handler)

  Handler factory that creates the execution function for the tool by running 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
• main.py:32-39 (registration)

  Dynamically generates and registers the tool handler with the MCP server using the @mcp.tool decorator for STDIO transport.
  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 that reads the SQL file and executes the query on the database, returning results as list of dicts.
  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:118-127 (handler)

  Async handler for predefined tools in HTTP transport server, resolves tool definition and executes corresponding SQL query.
  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 )
• main_remote.py:179-186 (registration)

  Registers the tool schema (name, description, empty input schema) in the list_tools handler for HTTP transport MCP server.
  for tool_def in TOOL_DEFINITIONS: tools.append( types.Tool( name=tool_def.name, description=tool_def.description, inputSchema={"type": "object", "properties": {}}, ) )