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)