DataPilot MCP Server

Instructions

Implementation Reference

• src/main.py:122-137 (handler)

  The MCP tool handler function for 'list_schemas'. It retrieves schemas for a given database using the SnowflakeClient and provides user feedback via context.

  @mcp.tool()
  async def list_schemas(database: str, ctx: Context) -> List[str]:
      """List all schemas in a database"""
      await ctx.info(f"Retrieving schemas for database: {database}")
      
      try:
          client = await get_snowflake_client()
          schemas = await client.list_schemas(database)
          await ctx.info(f"Found {len(schemas)} schemas in {database}")
          return schemas
          
      except Exception as e:
          logger.error(f"Error listing schemas: {str(e)}")
          await ctx.error(f"Failed to list schemas: {str(e)}")
          return []

• src/snowflake_client.py:130-134 (helper)

  Helper method in SnowflakeClient that executes the SHOW SCHEMAS SQL query and extracts schema names.

  async def list_schemas(self, database: str) -> List[str]:
      """List all schemas in a database"""
      result = await self.execute_query(f"SHOW SCHEMAS IN DATABASE {database}")
      return [row.get('name', '') for row in result.data if result.success]

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

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

  @mcp.tool()

• src/main.py:417-426 (helper)

  MCP resource that also uses list_schemas functionality for snowflake://schemas/{database}.

  @mcp.resource("snowflake://schemas/{database}")
  async def get_schemas_resource(database: str) -> List[str]:
      """Resource to get list of schemas in a database"""
      try:
          client = await get_snowflake_client()
          return await client.list_schemas(database)
      except Exception as e:
          logger.error(f"Error getting schemas resource: {str(e)}")
          return []