mysqldb-mcp-server

Instructions

Implementation Reference

• src/mysqldb_mcp_server/server.py:218-230 (handler)

  The primary handler function for the 'execute_query' tool. Decorated with @mcp.tool() for registration in FastMCP. It orchestrates query execution via QueryExecutor and formats results as JSON.

  @mcp.tool()
  def execute_query(query: str, ctx: Context) -> str:
      """Execute MySQL queries"""
      try:
          executor = _get_executor(ctx)
          results = executor.execute_multiple_queries(query)
  
          if len(results) == 1:
              return json.dumps(results[0], indent=2)
          return json.dumps(results, indent=2)
      except (ConnectionError, QueryError) as e:
          return str(e)

• src/mysqldb_mcp_server/server.py:85-155 (helper)

  Core helper class QueryExecutor that handles the actual MySQL query execution, connection management, result processing, and error handling used by the execute_query tool.

  class QueryExecutor:
      """Handles MySQL query execution and result processing"""
  
      def __init__(self, context: MySQLContext):
          self.context = context
  
      def _format_datetime(self, value: Any) -> Any:
          """Format datetime values to string"""
          return value.strftime('%Y-%m-%d %H:%M:%S') if hasattr(value, 'strftime') else value
  
      def _process_row(self, row: Dict[str, Any]) -> Dict[str, Any]:
          """Process a single row of results"""
          return {key: self._format_datetime(value) for key, value in row.items()}
  
      def _process_results(self, cursor: MySQLCursor) -> Union[List[Dict[str, Any]], Dict[str, int]]:
          """Process query results"""
          if cursor.with_rows:
              results = cursor.fetchall()
              return [self._process_row(row) for row in results]
          return {"affected_rows": cursor.rowcount}
  
      def execute_single_query(self, query: str) -> Dict[str, Any]:
          """Execute a single query and return results"""
          self.context.ensure_connected()
          cursor = None
  
          try:
              cursor = self.context.connection.cursor(dictionary=True)
              query_type = QueryType(query.strip().upper().split()[0])
  
              # Handle readonly mode
              if self.context.readonly and QueryType.is_write_operation(query_type.value):
                  raise QueryError(
                      "Server is in read-only mode. Write operations are not allowed.")
  
              # Handle USE statements
              if query_type == QueryType.USE:
                  db_name = query.strip().split()[-1].strip('`').strip()
                  self.context.database = db_name
                  cursor.execute(query)
                  return {"message": f"Switched to database: {db_name}"}
  
              # Execute query
              cursor.execute(query)
              results = self._process_results(cursor)
  
              if not self.context.readonly:
                  self.context.connection.commit()
  
              return results
  
          except MySQLError as e:
              raise QueryError(f"Error executing query: {str(e)}")
          finally:
              if cursor:
                  cursor.close()
  
      def execute_multiple_queries(self, query: str) -> List[Dict[str, Any]]:
          """Execute multiple queries and return results"""
          queries = [q.strip() for q in query.split(';') if q.strip()]
          results = []
  
          for single_query in queries:
              try:
                  result = self.execute_single_query(single_query)
                  results.append(result)
              except QueryError as e:
                  results.append({"error": str(e)})
  
          return results

• src/mysqldb_mcp_server/server.py:201-204 (helper)

  Helper function to retrieve the QueryExecutor instance from the MCP context, used within the execute_query handler.

  def _get_executor(ctx: Context) -> QueryExecutor:
      """Helper function to get QueryExecutor from context"""
      mysql_ctx = ctx.request_context.lifespan_context
      return QueryExecutor(mysql_ctx)