reqd_query
Retrieve data from Oracle databases using SELECT queries. Execute SQL commands to access and extract information directly from the database via the MCP Oracle Server.
Instructions
Execute SELECT queries to read data from the oracle database
Args:
query (string): The SELECT query to execute
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes |
Implementation Reference
- src/mcp_server_oracle/__init__.py:39-47 (handler)MCP tool handler and registration for 'reqd_query'. Thin wrapper that calls oracle_tools.read_query with the provided SQL SELECT query.@mcp.tool() async def reqd_query(query: str) -> str: """Execute SELECT queries to read data from the oracle database Args: query (string): The SELECT query to execute """ return await oracle_tools.read_query(query)
- MCP tool handler and registration for 'reqd_query' in the JYSD CX Oracle server. Thin wrapper that calls oracle_tools.read_query with the provided SQL SELECT query.@mcp.tool() async def reqd_query(query: str) -> str: """Execute SELECT queries to read data from the oracle database Args: query (string): The SELECT query to execute """ return await oracle_tools.read_query(query)
- Core helper function implementing the database query execution using cx_Oracle. Validates SELECT only, fetches results as CSV with headers, handles NULLs, runs in thread for async compatibility.async def read_query(query: str) -> str: try: # Check if the query is a SELECT statement if not query.strip().upper().startswith('SELECT'): return "Error: Only SELECT statements are supported." # Run database operations in a separate thread def db_operation(query): with cx_Oracle.connect(connection_string) as conn: cursor = conn.cursor() cursor.execute(query) # Execute query first # Get column names after executing the query columns = [col[0] for col in cursor.description] result = [','.join(columns)] # Add column headers # Process each row for row in cursor: # Convert each value in the tuple to string string_values = [ str(val) if val is not None else "NULL" for val in row] result.append(','.join(string_values)) return '\n'.join(result) return await asyncio.to_thread(db_operation, query) except cx_Oracle.DatabaseError as e: print('Error occurred:', e) return str(e)
- Core helper function implementing the database query execution using cx_Oracle (identical to the other server). Validates SELECT only, fetches results as CSV with headers, handles NULLs, runs in thread for async compatibility.async def read_query(query: str) -> str: try: # Check if the query is a SELECT statement if not query.strip().upper().startswith('SELECT'): return "Error: Only SELECT statements are supported." # Run database operations in a separate thread def db_operation(query): with cx_Oracle.connect(connection_string) as conn: cursor = conn.cursor() cursor.execute(query) # Execute query first # Get column names after executing the query columns = [col[0] for col in cursor.description] result = [','.join(columns)] # Add column headers # Process each row for row in cursor: # Convert each value in the tuple to string string_values = [ str(val) if val is not None else "NULL" for val in row] result.append(','.join(string_values)) return '\n'.join(result) return await asyncio.to_thread(db_operation, query) except cx_Oracle.DatabaseError as e: print('Error occurred:', e) return str(e)