MCP Trino Server

Instructions

Implementation Reference

• src/server.py:344-375 (handler)

  The main handler function for the 'show_manifests' MCP tool. Decorated with @mcp.tool(), it accepts catalog, schema_name, table, and all_snapshots parameters with Field descriptions for validation. Delegates to client.show_manifests().

  @mcp.tool(description="Show Iceberg table manifests")
  def show_manifests(
      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 manifests for current or all snapshots.
  
      The manifests table contains:
      - path: Manifest file location
      - length: Manifest file length
      - partition_spec_id: ID of partition spec used
      - added_snapshot_id: ID of snapshot when manifest was added
      - added_data_files_count: Number of data files with status ADDED
      - added_rows_count: Total rows in ADDED files
      - existing_data_files_count: Number of EXISTING files
      - existing_rows_count: Total rows in EXISTING files
      - deleted_data_files_count: Number of DELETED files
      - deleted_rows_count: Total rows in DELETED files
      - partition_summaries: Partition range metadata
  
      Args:
          catalog: catalog name
          schema_name: schema name
          table: The name of the table
          all_snapshots: If True, show manifests from all snapshots
  
      Returns:
          str: JSON-formatted table manifests
      """
      return client.show_manifests(catalog, schema_name, table, all_snapshots)

• src/server.py:344-350 (schema)

  Input schema definition for the show_manifests tool using Pydantic Field descriptions. Defines catalog (str), schema_name (str), table (str), and all_snapshots (bool) parameters.

  @mcp.tool(description="Show Iceberg table manifests")
  def show_manifests(
      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:448-480 (helper)

  The actual implementation of show_manifests in the TrinoClient class. Builds and executes a SQL query to fetch Iceberg table manifests from either 'manifests' or 'all_manifests' metadata tables. Handles catalog/schema defaults and validation.

  def show_manifests(self, table: str, catalog: str, schema: str, all_snapshots: bool = False) -> str:
      """Show Iceberg table manifests for current or all snapshots.
  
      The manifests table contains:
      - path: VARCHAR - Manifest file location
      - length: BIGINT - Manifest file length
      - partition_spec_id: INTEGER - ID of partition spec used
      - added_snapshot_id: BIGINT - ID of snapshot when manifest was added
      - added_data_files_count: INTEGER - Number of data files with status ADDED
      - added_rows_count: BIGINT - Total rows in ADDED files
      - existing_data_files_count: INTEGER - Number of EXISTING files
      - existing_rows_count: BIGINT - Total rows in EXISTING files
      - deleted_data_files_count: INTEGER - Number of DELETED files
      - deleted_rows_count: BIGINT - Total rows in DELETED files
      - partition_summaries: ARRAY(ROW(...)) - Partition range metadata
  
      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 manifests from all snapshots
  
      Returns:
          str: JSON-formatted string containing table manifests
      """
      catalog = catalog or self.config.catalog
      schema = schema or self.config.schema
      if not catalog or not schema:
          raise CatalogSchemaError
      table_type = "all_manifests" if all_snapshots else "manifests"
      query = 'SELECT * FROM "{}${}"'
      table_identifier = f"{catalog}.{schema}.{table}"
      return self.execute_query(query.format(table_identifier, table_type))