Trino MCP Server

Instructions

    Get detailed metadata about a table.
    
    Args:
        catalog: Catalog name.
        schema: Schema name.
        table: Table name.
        
    Returns:
        Dict[str, Any]: Table metadata including columns, statistics, etc.
    

Implementation Reference

• src/trino_mcp/tools/__init__.py:113-184 (handler)

  The inspect_table tool handler: decorated with @mcp.tool(), fetches table metadata via TrinoClient.get_table_details, computes row count with a COUNT(*) query, enhances columns with data types, nullability, and defaults from information_schema.columns.

  @mcp.tool()
  def inspect_table(
      catalog: str, 
      schema: str, 
      table: str
  ) -> Dict[str, Any]:
      """
      Get detailed metadata about a table.
      
      Args:
          catalog: Catalog name.
          schema: Schema name.
          table: Table name.
          
      Returns:
          Dict[str, Any]: Table metadata including columns, statistics, etc.
      """
      logger.info(f"Inspecting table: {catalog}.{schema}.{table}")
      
      try:
          table_details = client.get_table_details(catalog, schema, table)
          
          # Try to get a row count (this might not work on all connectors)
          try:
              count_result = client.execute_query(
                  f"SELECT count(*) AS row_count FROM {catalog}.{schema}.{table}"
              )
              if count_result.rows and count_result.rows[0]:
                  table_details["row_count"] = count_result.rows[0][0]
          except Exception as e:
              logger.warning(f"Failed to get row count: {e}")
              
          # Get additional info from the information_schema if available
          try:
              info_schema_query = f"""
              SELECT column_name, data_type, is_nullable, column_default
              FROM {catalog}.information_schema.columns
              WHERE table_catalog = '{catalog}'
              AND table_schema = '{schema}'
              AND table_name = '{table}'
              """
              info_schema_result = client.execute_query(info_schema_query)
              
              enhanced_columns = []
              for col in table_details["columns"]:
                  enhanced_col = col.copy()
                  
                  # Find matching info_schema row
                  for row in info_schema_result.rows:
                      if row[0] == col["name"]:
                          enhanced_col["data_type"] = row[1]
                          enhanced_col["is_nullable"] = row[2]
                          enhanced_col["default"] = row[3]
                          break
                          
                  enhanced_columns.append(enhanced_col)
                  
              table_details["columns"] = enhanced_columns
          except Exception as e:
              logger.warning(f"Failed to get column details from information_schema: {e}")
              
          return table_details
      
      except Exception as e:
          error_msg = str(e)
          logger.error(f"Table inspection failed: {error_msg}")
          return {
              "error": error_msg,
              "catalog": catalog,
              "schema": schema,
              "table": table
          }

• src/trino_mcp/tools/__init__.py:13-22 (registration)

  The register_trino_tools function where inspect_table is registered via @mcp.tool() decorator among other Trino tools.

  def register_trino_tools(mcp: FastMCP, client: TrinoClient) -> None:
      """
      Register Trino tools with the MCP server.
      
      Args:
          mcp: The MCP server instance.
          client: The Trino client instance.
      """
      
      @mcp.tool()