Adb MySQL MCP Server

Instructions

Implementation Reference

• src/adb_mysql_mcp_server/server.py:168-208 (handler)

  The @app.call_tool() handler implements the logic for 'execute_sql' by extracting the query from arguments, connecting to the MySQL database using get_db_config(), executing the query, formatting results as CSV text content, and handling errors.

  @app.call_tool()
  async def call_tool(name: str, arguments: dict) -> list[TextContent]:
      """Execute SQL commands."""
      config = get_db_config()
  
      if name == "execute_sql":
          query = arguments.get("query")
          if not query:
              raise ValueError("Query is required")
      elif name == "get_query_plan":
          query = arguments.get("query")
          if not query:
              raise ValueError("Query is required")
          query = f"EXPLAIN {query}"
      elif name == "get_execution_plan":
          query = arguments.get("query")
          if not query:
              raise ValueError("Query is required")
          query = f"EXPLAIN ANALYZE {query}"
      else:
          raise ValueError(f"Unknown tool: {name}")
  
      conn = pymysql.connect(**config)
      conn.autocommit(True)
      cursor = conn.cursor()
  
      try:
          # Execute the query
          cursor.execute(query)
  
          columns = [desc[0] for desc in cursor.description]
          rows = cursor.fetchall()
          result = [",".join(map(str, row)) for row in rows]
          return [TextContent(type="text", text="\n".join([",".join(columns)] + result))]
      except Exception as e:
          return [TextContent(type="text", text=f"Error executing query: {str(e)}")]
      finally:
          if cursor:
              cursor.close()
          if conn.open:
              conn.close()

• src/adb_mysql_mcp_server/server.py:123-136 (schema)

  The input schema definition for the 'execute_sql' tool, specifying a required 'query' string property.

  Tool(
      name="execute_sql",
      description="Execute a SQL query in the Adb MySQL Cluster",
      inputSchema={
          "type": "object",
          "properties": {
              "query": {
                  "type": "string",
                  "description": "The SQL query to execute"
              }
          },
          "required": ["query"]
      },
  ),

• src/adb_mysql_mcp_server/server.py:120-165 (registration)

  Registration of the 'execute_sql' tool (among others) via the @app.list_tools() decorator, which returns the list of available tools with their schemas.

  @app.list_tools()
  async def list_tools() -> list[Tool]:
      return [
          Tool(
              name="execute_sql",
              description="Execute a SQL query in the Adb MySQL Cluster",
              inputSchema={
                  "type": "object",
                  "properties": {
                      "query": {
                          "type": "string",
                          "description": "The SQL query to execute"
                      }
                  },
                  "required": ["query"]
              },
          ),
          Tool(
              name="get_query_plan",
              description="Get the query plan for a SQL query",
              inputSchema={
                  "type": "object",
                  "properties": {
                      "query": {
                          "type": "string",
                          "description": "The SQL query to analyze"
                      }
                  },
                  "required": ["query"]
              },
          ),
          Tool(
              name="get_execution_plan",
              description="Get the actual execution plan with runtime statistics for a SQL query",
              inputSchema={
                  "type": "object",
                  "properties": {
                      "query": {
                          "type": "string",
                          "description": "The SQL query to analyze"
                      }
                  },
                  "required": ["query"]
              }
          )
      ]

• src/adb_mysql_mcp_server/server.py:10-22 (helper)

  Helper function to retrieve MySQL database configuration from environment variables, used by the tool handler.

  def get_db_config():
      config = {
          "host": os.getenv("ADB_MYSQL_HOST", "localhost"),
          "port": int(os.getenv("ADB_MYSQL_PORT", 3306)),
          "user": os.getenv("ADB_MYSQL_USER"),
          "password": os.getenv("ADB_MYSQL_PASSWORD"),
          "database": os.getenv("ADB_MYSQL_DATABASE"),
      }
  
      if not all([config["user"], config["password"], config["database"]]):
          raise ValueError("Missing required database configuration")
  
      return config