Adb MySQL MCP Server

Instructions

Implementation Reference

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

  The @app.call_tool() function serves as the handler for the get_query_plan tool (among others). For get_query_plan, it retrieves the input query, prefixes it with 'EXPLAIN ', executes it using pymysql on the configured MySQL connection, fetches the results, formats them as CSV text, and returns as TextContent.

  @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:137-150 (registration)

  The get_query_plan tool is registered in the list_tools() function, specifying its name, description, and input schema which requires a 'query' string parameter.

  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"]
      },
  ),