MCP Trino Server

Instructions

Implementation Reference

• src/server.py:439-465 (handler)

  MCP tool handler function 'show_entries' decorated with @mcp.tool(). Defines the tool interface with parameters (catalog, schema_name, table, all_snapshots) using Pydantic Field for schema validation, includes detailed docstring explaining the entries table structure, and delegates to client.show_entries().

  @mcp.tool(description="Show Iceberg table manifest entries")
  def show_entries(
      catalog: str = Field(description="catalog name "),
      schema_name: str = Field(description="schema name "),
      table: str = Field(description="The name of the table"),
      all_snapshots: bool = False,
  ) -> str:
      """Show Iceberg table manifest entries for current or all snapshots.
  
      The entries table contains:
      - status: Status of entry (0=EXISTING, 1=ADDED, 2=DELETED)
      - snapshot_id: ID of the snapshot
      - sequence_number: Data sequence number
      - file_sequence_number: File sequence number
      - data_file: File metadata including path, format, size etc
      - readable_metrics: Human-readable file metrics
  
      Args:
          catalog: catalog name
          schema_name: schema name
          table: The name of the table
          all_snapshots: If True, show entries from all snapshots
  
      Returns:
          str: JSON-formatted manifest entries
      """
      return client.show_entries(catalog, schema_name, table, all_snapshots)

• src/server.py:439-445 (schema)

  Input schema defined via Pydantic Field() parameters: catalog (str), schema_name (str), table (str), all_snapshots (bool). These define the tool's input validation and JSON schema for MCP.

  @mcp.tool(description="Show Iceberg table manifest entries")
  def show_entries(
      catalog: str = Field(description="catalog name "),
      schema_name: str = Field(description="schema name "),
      table: str = Field(description="The name of the table"),
      all_snapshots: bool = False,
  ) -> str:

• src/trino_client.py:543-569 (helper)

  TrinoClient.show_entries() method - the core implementation that builds and executes the SQL query to retrieve Iceberg table manifest entries. Constructs table name with optional '$all_entries' suffix for all snapshots, validates catalog/schema, and executes query via execute_query().

  def show_entries(self, table: str, catalog: str, schema: str, all_snapshots: bool = False) -> str:
      """Show Iceberg table manifest entries for current or all snapshots.
  
      The entries table contains:
      - status: INTEGER - Status of entry (0=EXISTING, 1=ADDED, 2=DELETED)
      - snapshot_id: BIGINT - ID of the snapshot
      - sequence_number: BIGINT - Data sequence number
      - file_sequence_number: BIGINT - File sequence number
      - data_file: ROW(...) - File metadata including path, format, size etc
      - readable_metrics: JSON - Human-readable file metrics
  
      Args:
          table: The name of the table
          catalog: Optional catalog name (defaults to configured catalog)
          schema: Optional schema name (defaults to configured schema)
          all_snapshots: If True, show entries from all snapshots
  
      Returns:
          str: JSON-formatted string containing manifest entries
      """
      catalog = catalog or self.config.catalog
      schema = schema or self.config.schema
      if not catalog or not schema:
          raise CatalogSchemaError
      table_name = f"{catalog}.{schema}.{table}${'all_' if all_snapshots else ''}entries"
      query = 'SELECT * FROM "{}"'
      return self.execute_query(query.format(table_name))