execute_sql

Instructions

Implementation Reference

• src/tibero_mcp_server/server.py:212-261 (handler)

  Handler function for executing SQL queries. Supports SELECT (returns CSV results), DML (INSERT/UPDATE/DELETE with commit and rowcount), and other DDL queries.

  if name == "execute_sql":
      query = arguments.get("query")
      if not query:
          raise ValueError("Query is required")
      
      # Execute the query
      cursor.execute(query)
      
      # Handle different query types
      query_upper = query.strip().upper()
      
      # Data retrieval queries
      if (query_upper.startswith("SELECT") or 
          query_upper.startswith("SHOW") or 
          query_upper.startswith("DESC")):
          
          # Get column names
          columns = [desc[0] for desc in cursor.description] if cursor.description else []
          rows = cursor.fetchall()
          
          if not columns:
              return [TextContent(type="text", text="Query executed successfully, but returned no columns.")]
          
          # Format as CSV
          result = [",".join(map(str, columns))]
          result.extend([",".join(map(lambda x: str(x) if x is not None else "NULL", row)) for row in rows])
          
          return [TextContent(
              type="text", 
              text=f"Results ({len(rows)} rows):\n" + "\n".join(result)
          )]
      
      # Non-SELECT queries (DML/DDL)
      else:
          # For DML, commit the transaction
          if (query_upper.startswith("INSERT") or 
              query_upper.startswith("UPDATE") or 
              query_upper.startswith("DELETE")):
              conn.commit()
              return [TextContent(
                  type="text", 
                  text=f"Query executed successfully. Rows affected: {cursor.rowcount}"
              )]
          # For DDL, no commit needed (auto-commit)
          else:
              return [TextContent(
                  type="text", 
                  text="Query executed successfully."
              )]

• src/tibero_mcp_server/server.py:176-185 (schema)

  JSON schema defining the input for the execute_sql tool: an object with a required 'query' string.

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

• src/tibero_mcp_server/server.py:173-186 (registration)

  Registration of the execute_sql tool in the list_tools() function, including name, description, and input schema.

  Tool(
      name="execute_sql",
      description="Execute an SQL query on the Tibero server",
      inputSchema={
          "type": "object",
          "properties": {
              "query": {
                  "type": "string",
                  "description": "The SQL query to execute"
              }
          },
          "required": ["query"]
      }
  ),

• src/tibero_mcp_server/server.py:33-52 (helper)

  Helper function to establish JDBC connection to Tibero database, used by the execute_sql handler.

  def get_connection():
      """Create a JDBC connection to Tibero database."""
      config = get_db_config()
      connection_string = f"jdbc:tibero:thin:@{config['host']}:{config['port']}:{config['sid']}"
      
      try:
          # Make sure the JDBC driver is in the classpath or provide path
          driver_path = os.getenv("CLASSPATH", "drivers/tibero6-jdbc.jar")
          
          conn = jaydebeapi.connect(
              "com.tmax.tibero.jdbc.TbDriver",
              connection_string,
              [config["user"], config["password"]],
              driver_path
          )
          conn.jconn.setAutoCommit(False)  # Set to manual commit mode for safety
          return conn
      except Exception as e:
          logger.error(f"Failed to connect to Tibero: {str(e)}")
          raise