run_query

Instructions

Implementation Reference

• main.py:39-52 (registration)

  Tool 'run_query' is registered in the list_tools() function with a schema requiring a 'query' string parameter.

  Tool(
      name="run_query",
      description="Run a read-only SQL query against the database. ONLY SELECT queries are allowed for safety.",
      inputSchema={
          "type": "object",
          "properties": {
              "query": {
                  "type": "string",
                  "description": "The read-only SQL query to execute"
              }
          },
          "required": ["query"]
      }
  )

• main.py:40-52 (schema)

  The inputSchema for 'run_query' defines a 'query' string property as required input.

      name="run_query",
      description="Run a read-only SQL query against the database. ONLY SELECT queries are allowed for safety.",
      inputSchema={
          "type": "object",
          "properties": {
              "query": {
                  "type": "string",
                  "description": "The read-only SQL query to execute"
              }
          },
          "required": ["query"]
      }
  )

• main.py:95-124 (handler)

  The handler for 'run_query' in call_tool() extracts the query, validates it's SELECT/WITH only, executes it in a read-only transaction via asyncpg, and formats the results as text.

  elif name == "run_query":
      query = arguments.get("query")
      if not query:
          return [TextContent(type="text", text="Error: query is required")]
          
      if not query.strip().upper().startswith("SELECT") and not query.strip().upper().startswith("WITH"):
           return [TextContent(type="text", text="Error: Only SELECT/WITH queries are permitted via this tool.")]
           
      try:
          async with pool.acquire() as conn:
              async with conn.transaction(readonly=True):
                  # Use direct fetch to avoid prepared statement argument issues for general queries
                  records = await conn.fetch(query)
                  
                  if not records:
                      return [TextContent(type="text", text="Query returned 0 rows.")]
                  
                  keys = list(records[0].keys())
                  header = " | ".join(keys)
                  separator = "-" * len(header)
                  
                  rows = []
                  for record in records:
                      rows.append(" | ".join(str(record[k]) for k in keys))
                      
                  result_text = f"{header}\n{separator}\n" + "\n".join(rows) + "\n\n(Limited to records fetched)"
                  return [TextContent(type="text", text=result_text)]
              
      except Exception as e:
           return [TextContent(type="text", text=f"Error executing query: {str(e)}")]