BigQuery Validator

Instructions

Implementation Reference

• src/mcp_bigquery/schema_explorer/tables.py:28-114 (handler)

  Core implementation of the `list_tables` handler: validates input, fetches tables from BigQuery dataset using client.list_tables, filters by type, enriches with metadata (partitioning, clustering, stats), and returns formatted JSON response.

  async def list_tables(
      dataset_id: str,
      project_id: str | None = None,
      max_results: int | None = None,
      table_type_filter: list[str] | None = None,
  ) -> dict[str, Any]:
      """List tables in a dataset."""
      try:
          request = validate_request(
              ListTablesRequest,
              {
                  "dataset_id": dataset_id,
                  "project_id": project_id,
                  "max_results": max_results,
                  "table_type_filter": table_type_filter,
              },
          )
      except MCPBigQueryError as exc:
          return {"error": format_error_response(exc)}
  
      try:
          return await _list_tables_impl(request)
      except MCPBigQueryError as exc:
          return {"error": format_error_response(exc)}
      except Exception as exc:  # pragma: no cover - defensive guard
          logger.exception("Unexpected error while listing tables")
          wrapped = MCPBigQueryError(str(exc), code="LIST_TABLES_ERROR")
          return {"error": format_error_response(wrapped)}
  
  
  async def _list_tables_impl(request: ListTablesRequest) -> dict[str, Any]:
      client = get_bigquery_client(project_id=request.project_id)
      project = request.project_id or client.project
  
      try:
          list_kwargs: dict[str, Any] = {"dataset": f"{project}.{request.dataset_id}"}
          if request.max_results is not None:
              list_kwargs["max_results"] = request.max_results
  
          iterator = client.list_tables(**list_kwargs)
      except NotFound as exc:
          raise DatasetNotFoundError(request.dataset_id, project) from exc
  
      allowed_types = set(request.table_type_filter) if request.table_type_filter else None
      tables: list[dict[str, Any]] = []
  
      for table in iterator:
          try:
              table_ref = client.get_table(table.reference)
          except NotFound as exc:
              raise TableNotFoundError(table.table_id, request.dataset_id, project) from exc
  
          table_type = table_ref.table_type
          if allowed_types and table_type not in allowed_types:
              continue
  
          partitioning = partitioning_overview(table_ref)
          clustering = clustering_fields(table_ref)
  
          table_info: dict[str, Any] = {
              "table_id": table.table_id,
              "dataset_id": table.dataset_id,
              "project": table.project,
              "table_type": table_type,
              "created": serialize_timestamp(table_ref.created),
              "modified": serialize_timestamp(table_ref.modified),
              "description": table_ref.description,
              "labels": table_ref.labels or {},
              "num_bytes": getattr(table_ref, "num_bytes", None),
              "num_rows": getattr(table_ref, "num_rows", None),
              "location": table_ref.location,
          }
  
          if partitioning:
              table_info["partitioning"] = partitioning
          if clustering:
              table_info["clustering_fields"] = clustering
  
          tables.append(table_info)
  
      return {
          "dataset_id": request.dataset_id,
          "project": project,
          "table_count": len(tables),
          "tables": tables,
      }

• src/mcp_bigquery/server.py:190-216 (schema)

  MCP Tool schema definition for 'bq_list_tables', including input schema with properties for dataset_id (required), project_id, max_results, and table_type_filter.

  types.Tool(
      name="bq_list_tables",
      description=("List all tables in a BigQuery dataset with metadata"),
      inputSchema={
          "type": "object",
          "properties": {
              "dataset_id": {
                  "type": "string",
                  "description": "The dataset ID",
              },
              "project_id": {
                  "type": "string",
                  "description": "GCP project ID (uses default if not provided)",
              },
              "max_results": {
                  "type": "integer",
                  "description": "Maximum number of tables to return",
              },
              "table_type_filter": {
                  "type": "array",
                  "items": {"type": "string"},
                  "description": "Filter by table types (TABLE, VIEW, EXTERNAL, MATERIALIZED_VIEW)",
              },
          },
          "required": ["dataset_id"],
      },
  ),

• src/mcp_bigquery/server.py:360-367 (registration)

  Registration and dispatch logic in the @server.call_tool() handler that maps 'bq_list_tables' calls to the list_tables implementation.

  elif name == "bq_list_tables":
      result = await list_tables(
          dataset_id=arguments["dataset_id"],
          project_id=arguments.get("project_id"),
          max_results=arguments.get("max_results"),
          table_type_filter=arguments.get("table_type_filter"),
      )
      return [types.TextContent(type="text", text=json.dumps(result, indent=2))]

• src/mcp_bigquery/validators.py:108-129 (schema)

  Pydantic validation model ListTablesRequest used by the handler for input validation, with field constraints and table_type_filter validator.

  class ListTablesRequest(BaseModel):
      """Request model for listing tables."""
  
      dataset_id: str = Field(..., min_length=1, max_length=1024)
      project_id: str | None = Field(None, pattern=PROJECT_ID_PATTERN)
      max_results: int | None = Field(None, ge=1, le=10000)
      table_type_filter: list[str] | None = Field(None)
  
      @field_validator("table_type_filter")
      @classmethod
      def validate_table_types(cls, v: list[str] | None) -> list[str] | None:
          """Validate table types are valid."""
          if v is None:
              return v
  
          valid_types = {t.value for t in TableType}
          for table_type in v:
              if table_type not in valid_types:
                  raise ValueError(f"Invalid table type: {table_type}. Must be one of {valid_types}")
  
          return v