BigQuery Validator

Instructions

Implementation Reference

• src/mcp_bigquery/schema_explorer/datasets.py:17-72 (handler)

  The primary handler function that executes the tool logic for bq_list_datasets. It validates inputs, fetches datasets using BigQuery client, enriches with metadata, formats timestamps and returns structured JSON response with error handling.

  async def list_datasets(
      project_id: str | None = None,
      max_results: int | None = None,
  ) -> dict[str, Any]:
      """List datasets along with core metadata."""
      try:
          request = validate_request(
              ListDatasetsRequest,
              {"project_id": project_id, "max_results": max_results},
          )
      except MCPBigQueryError as exc:
          return {"error": format_error_response(exc)}
  
      try:
          return await _list_datasets_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 datasets")
          wrapped = MCPBigQueryError(str(exc), code="LIST_DATASETS_ERROR")
          return {"error": format_error_response(wrapped)}
  
  
  async def _list_datasets_impl(request: ListDatasetsRequest) -> dict[str, Any]:
      client = get_bigquery_client(project_id=request.project_id)
      project = request.project_id or client.project
  
      datasets = []
      list_kwargs: dict[str, Any] = {"project": project}
      if request.max_results is not None:
          list_kwargs["max_results"] = request.max_results
  
      iterator = client.list_datasets(**list_kwargs)
  
      for dataset in iterator:
          ref = client.get_dataset(dataset.reference)
          datasets.append(
              {
                  "dataset_id": dataset.dataset_id,
                  "project": dataset.project,
                  "location": ref.location,
                  "created": serialize_timestamp(ref.created),
                  "modified": serialize_timestamp(ref.modified),
                  "description": ref.description,
                  "labels": ref.labels or {},
                  "default_table_expiration_ms": ref.default_table_expiration_ms,
                  "default_partition_expiration_ms": ref.default_partition_expiration_ms,
              }
          )
  
      return {
          "project": project,
          "dataset_count": len(datasets),
          "datasets": datasets,
      }

• src/mcp_bigquery/server.py:173-189 (registration)

  Registers the bq_list_datasets tool in the MCP server's list_tools() callback, defining its name, description, and input schema.

  types.Tool(
      name="bq_list_datasets",
      description=("List all datasets in the BigQuery project"),
      inputSchema={
          "type": "object",
          "properties": {
              "project_id": {
                  "type": "string",
                  "description": "GCP project ID (uses default if not provided)",
              },
              "max_results": {
                  "type": "integer",
                  "description": "Maximum number of datasets to return",
              },
          },
      },
  ),

• src/mcp_bigquery/server.py:354-358 (handler)

  Dispatcher in the MCP server's call_tool() handler that routes bq_list_datasets calls to the list_datasets implementation function, serializes the result as JSON.

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

• src/mcp_bigquery/validators.py:101-106 (schema)

  Pydantic schema/model for input validation specific to list_datasets tool parameters (project_id with GCP pattern validation, max_results bounded). Used by the handler.

  class ListDatasetsRequest(BaseModel):
      """Request model for listing datasets."""
  
      project_id: str | None = Field(None, pattern=PROJECT_ID_PATTERN)
      max_results: int | None = Field(None, ge=1, le=10000)