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()