Lenses MCP Server

Instructions

Implementation Reference

• src/lenses_mcp/tools/topics.py:310-382 (handler)

  The core handler function for the 'list_datasets' tool. It is decorated with @mcp.tool(), which registers it with the MCP server when register_topics(mcp) is called. The function builds a query string from parameters and makes a GET request to the API endpoint to retrieve paginated datasets.

  @mcp.tool()
  async def list_datasets(
      environment: str,
      page: int = 1,
      page_size: int = 25,
      search: Optional[str] = None,
      connections: Optional[List[str]] = None,
      tags: Optional[List[str]] = None,
      sort_field: Optional[str] = None,
      sort_order: str = "asc",
      include_system: bool = False,
      search_fields: bool = True,
      schema_format: Optional[str] = None,
      has_records: Optional[bool] = None,
      is_compacted: Optional[bool] = None
  ) -> Dict[str, Any]:
      """
      Retrieves a paginated list of datasets (topics and other data sources).
      
      Args:
          environment: The environment name.
          page: Page number (default: 1).
          page_size: Items per page (default: 25).
          search: Search keyword for dataset, fields and description.
          connections: List of connection names to filter by.
          tags: List of tag names to filter by.
          sort_field: Field to sort results by.
          sort_order: Sorting order - "asc" or "desc" (default: "asc").
          include_system: Include system entities (default: False).
          search_fields: Search field names/documentation (default: True).
          schema_format: Schema format filter for SchemaRegistrySubject.
          has_records: Filter based on whether dataset has records.
          is_compacted: Filter based on compacted status (Kafka only).
      
      Returns:
          Paginated list of datasets with source types.
      """
      params = {
          "page": page,
          "pageSize": page_size,
          "sortOrder": sort_order,
          "includeSystemEntities": include_system,
          "searchFields": search_fields
      }
      
      if search:
          params["search"] = search
      if connections:
          params["connections"] = connections
      if tags:
          params["tags"] = tags
      if sort_field:
          params["sortField"] = sort_field
      if schema_format:
          params["schemaFormat"] = schema_format
      if has_records is not None:
          params["hasRecords"] = has_records
      if is_compacted is not None:
          params["isCompacted"] = is_compacted
      
      # Build query string
      query_params = []
      for key, value in params.items():
          if isinstance(value, list):
              for item in value:
                  query_params.append(f"{key}={item}")
          else:
              query_params.append(f"{key}={value}")
      
      query_string = "&".join(query_params)
      endpoint = f"/api/v1/environments/{environment}/proxy/api/v1/datasets?{query_string}"
      
      return await api_client._make_request("GET", endpoint)