MySQL MCP Server

Instructions

Implementation Reference

• src/mysql_mcp_server/server.py:127-174 (handler)

  The call_tool handler function that implements the execute_sql tool. It validates the tool name, extracts the SQL query, connects to MySQL, executes the query, handles results for SELECT/SHOW queries or reports rowcount for modifications, with special case for SHOW TABLES.

  @app.call_tool()
  async def call_tool(name: str, arguments: dict) -> list[TextContent]:
      """Execute SQL commands."""
      config = get_db_config()
      logger.info(f"Calling tool: {name} with arguments: {arguments}")
  
      if name != "execute_sql":
          raise ValueError(f"Unknown tool: {name}")
  
      query = arguments.get("query")
      if not query:
          raise ValueError("Query is required")
  
      try:
          logger.info(f"Connecting to MySQL with charset: {config.get('charset')}, collation: {config.get('collation')}")
          with connect(**config) as conn:
              logger.info(f"Successfully connected to MySQL server version: {conn.get_server_info()}")
              with conn.cursor() as cursor:
                  cursor.execute(query)
  
                  # Special handling for SHOW TABLES
                  if query.strip().upper().startswith("SHOW TABLES"):
                      tables = cursor.fetchall()
                      result = ["Tables_in_" + config["database"]]  # Header
                      result.extend([table[0] for table in tables])
                      return [TextContent(type="text", text="\n".join(result))]
  
                  # Handle all other queries that return result sets (SELECT, SHOW, DESCRIBE etc.)
                  elif cursor.description is not None:
                      columns = [desc[0] for desc in cursor.description]
                      try:
                          rows = cursor.fetchall()
                          result = [",".join(map(str, row)) for row in rows]
                          return [TextContent(type="text", text="\n".join([",".join(columns)] + result))]
                      except Error as e:
                          logger.warning(f"Error fetching results: {str(e)}")
                          return [TextContent(type="text", text=f"Query executed but error fetching results: {str(e)}")]
  
                  # Non-SELECT queries
                  else:
                      conn.commit()
                      return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}")]
  
      except Error as e:
          logger.error(f"Error executing SQL '{query}': {e}")
          logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")
          return [TextContent(type="text", text=f"Error executing query: {str(e)}")]

• src/mysql_mcp_server/server.py:114-123 (schema)

  Input schema definition for the execute_sql tool, specifying an object with a required 'query' property of type string.

  inputSchema={
      "type": "object",
      "properties": {
          "query": {
              "type": "string",
              "description": "The SQL query to execute"
          }
      },
      "required": ["query"]
  }

• src/mysql_mcp_server/server.py:106-126 (registration)

  Registration of the execute_sql tool via the list_tools decorator, defining its name, description, and input schema.

  @app.list_tools()
  async def list_tools() -> list[Tool]:
      """List available MySQL tools."""
      logger.info("Listing tools...")
      return [
          Tool(
              name="execute_sql",
              description="Execute an SQL query on the MySQL server",
              inputSchema={
                  "type": "object",
                  "properties": {
                      "query": {
                          "type": "string",
                          "description": "The SQL query to execute"
                      }
                  },
                  "required": ["query"]
              }
          )
      ]

• src/mysql_mcp_server/server.py:17-43 (helper)

  Helper function get_db_config() used by the handler to obtain MySQL connection parameters from environment variables.

  def get_db_config():
      """Get database configuration from environment variables."""
      config = {
          "host": os.getenv("MYSQL_HOST", "localhost"),
          "port": int(os.getenv("MYSQL_PORT", "3306")),
          "user": os.getenv("MYSQL_USER"),
          "password": os.getenv("MYSQL_PASSWORD"),
          "database": os.getenv("MYSQL_DATABASE"),
          # Add charset and collation to avoid utf8mb4_0900_ai_ci issues with older MySQL versions
          # These can be overridden via environment variables for specific MySQL versions
          "charset": os.getenv("MYSQL_CHARSET", "utf8mb4"),
          "collation": os.getenv("MYSQL_COLLATION", "utf8mb4_unicode_ci"),
          # Disable autocommit for better transaction control
          "autocommit": True,
          # Set SQL mode for better compatibility - can be overridden
          "sql_mode": os.getenv("MYSQL_SQL_MODE", "TRADITIONAL")
      }
  
      # Remove None values to let MySQL connector use defaults if not specified
      config = {k: v for k, v in config.items() if v is not None}
  
      if not all([config.get("user"), config.get("password"), config.get("database")]):
          logger.error("Missing required database configuration. Please check environment variables:")
          logger.error("MYSQL_USER, MYSQL_PASSWORD, and MYSQL_DATABASE are required")
          raise ValueError("Missing required database configuration")
  
      return config