MCP SQL Agent

Instructions

Implementation Reference

• servers/mcp_server_sqlite.py:128-150 (handler)

  Handler implementation for query_data tool in SQLite server. Executes arbitrary SQL queries using sqlite3 and formats results.

  @mcp.tool()
  def query_data(sql: str) -> str:
      """Execute SQL queries safely. Use get_schema() first to understand the database structure."""
      logger.info(f"Executing SQL query: {sql}")
      conn = sqlite3.connect(db_path)
      try:
          result = conn.execute(sql).fetchall()
          conn.commit()
          
          if not result:
              return "Query executed successfully but returned no results."
          
          # Format results nicely
          output = f"Query returned {len(result)} row(s):\n\n"
          for i, row in enumerate(result, 1):
              output += f"Row {i}: {row}\n"
          
          return output
      except Exception as e:
          return f"Error: {str(e)}"
      finally:
          conn.close()

• servers/mcp_server_mysql.py:185-217 (handler)

  Handler implementation for query_data tool in MySQL server. Executes SQL queries using mysql.connector, handles SELECT vs mutations differently.

  @mcp.tool()
  def query_data(sql: str) -> str:
      """Execute SQL queries safely. Use get_schema() first to understand the database structure."""
      logger.info(f"Executing SQL query: {sql}")
      try:
          conn = mysql.connector.connect(**db_config)
          cursor = conn.cursor()
          
          cursor.execute(sql)
          
          # Handle different types of queries
          if sql.strip().upper().startswith(('SELECT', 'SHOW', 'DESCRIBE', 'EXPLAIN')):
              result = cursor.fetchall()
              if not result:
                  return "Query executed successfully but returned no results."
              
              # Format results nicely
              output = f"Query returned {len(result)} row(s):\n\n"
              for i, row in enumerate(result, 1):
                  output += f"Row {i}: {row}\n"
          else:
              # For INSERT, UPDATE, DELETE, etc.
              conn.commit()
              output = f"Query executed successfully. Affected rows: {cursor.rowcount}"
          
          return output
      except Exception as e:
          return f"Error: {str(e)}"
      finally:
          if 'cursor' in locals():
              cursor.close()
          if 'conn' in locals():
              conn.close()

• servers/mcp_server_oracle.py:201-232 (handler)

  Handler implementation for query_data tool in Oracle server. Executes SQL queries using cx_Oracle, distinguishes SELECT/WITH queries.

  @mcp.tool()
  def query_data(sql: str) -> str:
      """Execute SQL queries safely. Use get_schema() first to understand the database structure."""
      logger.info(f"Executing SQL query: {sql}")
      try:
          conn = cx_Oracle.connect(**db_config)
          cursor = conn.cursor()
          
          cursor.execute(sql)
          
          # Handle different types of queries
          if sql.strip().upper().startswith(('SELECT', 'WITH')):
              result = cursor.fetchall()
              if not result:
                  return "Query executed successfully but returned no results."
              
              # Format results nicely
              output = f"Query returned {len(result)} row(s):\n\n"
              for i, row in enumerate(result, 1):
                  output += f"Row {i}: {row}\n"
          else:
              # For INSERT, UPDATE, DELETE, etc.
              conn.commit()
              output = f"Query executed successfully. Affected rows: {cursor.rowcount}"
          
          return output
      except Exception as e:
          return f"Error: {str(e)}"
      finally:
          if 'cursor' in locals():
              cursor.close()
          if 'conn' in locals():