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)