DataPilot MCP Server

Instructions

Implementation Reference

• src/main.py:139-173 (handler)

  MCP tool handler implementation for 'list_tables'. Decorated with @mcp.tool(), calls Snowflake client, formats TableInfo objects to dicts, provides user feedback via ctx.

  @mcp.tool()
  async def list_tables(database: Optional[str] = None, schema: Optional[str] = None, ctx: Context = None) -> List[Dict[str, Any]]:
      """List all tables in a database/schema"""
      context_msg = f"Retrieving tables"
      if database:
          context_msg += f" from database: {database}"
      if schema:
          context_msg += f", schema: {schema}"
      
      await ctx.info(context_msg)
      
      try:
          client = await get_snowflake_client()
          tables = await client.list_tables(database, schema)
          
          # Convert to dict for JSON serialization
          result = []
          for table in tables:
              result.append({
                  "table_name": table.table_name,
                  "schema_name": table.schema_name,
                  "database_name": table.database_name,
                  "table_type": table.table_type,
                  "row_count": table.row_count,
                  "bytes": table.bytes,
                  "comment": table.comment
              })
          
          await ctx.info(f"Found {len(result)} tables")
          return result
          
      except Exception as e:
          logger.error(f"Error listing tables: {str(e)}")
          await ctx.error(f"Failed to list tables: {str(e)}")
          return []

• src/snowflake_client.py:135-158 (helper)

  Core SnowflakeClient method that executes SHOW TABLES query, parses results into TableInfo objects. Used by the MCP tool handler.

  async def list_tables(self, database: Optional[str] = None, schema: Optional[str] = None) -> List[TableInfo]:
      """List all tables in a database/schema"""
      query = "SHOW TABLES"
      if database and schema:
          query += f" IN SCHEMA {database}.{schema}"
      elif database:
          query += f" IN DATABASE {database}"
      
      result = await self.execute_query(query)
      
      tables = []
      for row in result.data:
          if result.success:
              tables.append(TableInfo(
                  table_name=row.get('name', ''),
                  schema_name=row.get('schema_name', ''),
                  database_name=row.get('database_name', ''),
                  table_type=row.get('kind', ''),
                  row_count=row.get('rows'),
                  bytes=row.get('bytes'),
                  comment=row.get('comment')
              ))
      
      return tables

• src/models.py:24-33 (schema)

  Pydantic model defining the structure of table information returned by list_tables operations.

  class TableInfo(BaseModel):
      """Information about a Snowflake table"""
      table_name: str
      schema_name: str
      database_name: str
      table_type: str
      row_count: Optional[int] = None
      bytes: Optional[int] = None
      comment: Optional[str] = None

• src/main.py:139-139 (registration)

  @mcp.tool() decorator registers the list_tables function as an MCP tool.

  @mcp.tool()