get_schema

Instructions

Implementation Reference

• main.py:25-38 (registration)

  Tool registration via Tool object with name='get_schema', including description and inputSchema requiring table_name

  Tool(
      name="get_schema",
      description="Get the schema (columns, types) of a specific table",
      inputSchema={
          "type": "object",
          "properties": {
              "table_name": {
                  "type": "string",
                  "description": "Name of the table"
              }
          },
          "required": ["table_name"]
      }
  ),

• main.py:73-93 (handler)

  Handler implementation for get_schema: queries information_schema.columns for the given table_name and returns column names, data types, and nullability

  elif name == "get_schema":
      table_name = arguments.get("table_name")
      if not table_name:
          return [TextContent(type="text", text="Error: table_name is required")]
          
      async with pool.acquire() as conn:
          records = await conn.fetch("""
              SELECT column_name, data_type, is_nullable
              FROM information_schema.columns
              WHERE table_schema = 'public' AND table_name = $1
              ORDER BY ordinal_position
          """, table_name)
          
          if not records:
              return [TextContent(type="text", text=f"Table '{table_name}' not found or has no columns.")]
              
          schema_info = [f"Schema for {table_name}:"]
          for r in records:
              schema_info.append(f"- {r['column_name']} ({r['data_type']}, nullable: {r['is_nullable']})")
              
          return [TextContent(type="text", text="\n".join(schema_info))]

• main.py:14-53 (registration)

  Full list_tools() function that registers all three tools including get_schema

  @app.list_tools()
  async def list_tools() -> list[Tool]:
      return [
          Tool(
              name="list_tables",
              description="List all tables in the current database schema",
              inputSchema={
                  "type": "object",
                  "properties": {},
              }
          ),
          Tool(
              name="get_schema",
              description="Get the schema (columns, types) of a specific table",
              inputSchema={
                  "type": "object",
                  "properties": {
                      "table_name": {
                          "type": "string",
                          "description": "Name of the table"
                      }
                  },
                  "required": ["table_name"]
              }
          ),
          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:55-127 (handler)

  Full call_tool() handler dispatching function that routes to get_schema logic on line 73-93

  @app.call_tool()
  async def call_tool(name: str, arguments: dict) -> list[TextContent]:
      if not pool:
          return [TextContent(type="text", text="Error: Database connection pool not initialized. DATABASE_URL may be missing or invalid.")]
  
      if name == "list_tables":
          async with pool.acquire() as conn:
              records = await conn.fetch("""
                  SELECT table_name 
                  FROM information_schema.tables 
                  WHERE table_schema = 'public' 
                    AND table_type = 'BASE TABLE'
              """)
              tables = [record["table_name"] for record in records]
              if not tables:
                  return [TextContent(type="text", text="No tables found in public schema.")]
              return [TextContent(type="text", text=f"Tables in public schema:\n" + "\n".join(f"- {t}" for t in tables))]
  
      elif name == "get_schema":
          table_name = arguments.get("table_name")
          if not table_name:
              return [TextContent(type="text", text="Error: table_name is required")]
              
          async with pool.acquire() as conn:
              records = await conn.fetch("""
                  SELECT column_name, data_type, is_nullable
                  FROM information_schema.columns
                  WHERE table_schema = 'public' AND table_name = $1
                  ORDER BY ordinal_position
              """, table_name)
              
              if not records:
                  return [TextContent(type="text", text=f"Table '{table_name}' not found or has no columns.")]
                  
              schema_info = [f"Schema for {table_name}:"]
              for r in records:
                  schema_info.append(f"- {r['column_name']} ({r['data_type']}, nullable: {r['is_nullable']})")
                  
              return [TextContent(type="text", text="\n".join(schema_info))]
  
      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)}")]
  
      else:
          raise ValueError(f"Unknown tool: {name}")