Describe the bug:

Hello! I've been implementing async mode of google-ads and found that is_async=True doesn't work in GoogleAdsClient.get_service() due to multiple architectural issues.

Root Causes

1. Wrong Transport Class Retrieved

CustomerServiceAsyncClient.get_transport_class() returns sync transport instead of async:

• Line 201 in async_client.py:

  get_transport_class = CustomerServiceClient.get_transport_class()

• It calls sync client's method without label='grpc_asyncio' parameter
• Should be: get_transport_class = staticmethod(lambda name='grpc_asyncio': ...)

2. GoogleAdsClient Creates Sync Channel for Async Services

GoogleAdsClient.get_service() always creates synchronous gRPC channel even when is_async=True:

# Line 420-427 in client.py
service_transport_class = service_client_class.get_transport_class()  # ← Returns SYNC transport!
channel = service_transport_class.create_channel(...)  # ← Creates grpc.Channel (sync)

Should detect is_async and create grpc.aio.Channel instead.

3. Sync Interceptors Used with Async Channel

Lines 433-444 in client.py:

interceptors = interceptors + [
    MetadataInterceptor(...),      # ← Sync interceptor
    LoggingInterceptor(...),       # ← Sync interceptor
    ExceptionInterceptor(...),     # ← Sync interceptor
]
channel = grpc.intercept_channel(channel, *interceptors)  # ← Sync function, fails for aio

Problems:

• grpc.intercept_channel doesn't exist in grpc.aio
• For grpc.aio, interceptors must be passed when creating the channel:
  grpc.aio.secure_channel(..., interceptors=[...])
• All three interceptors (MetadataInterceptor, LoggingInterceptor, ExceptionInterceptor) are synchronous - they don't inherit from grpc.aio.UnaryUnaryClientInterceptor

4. No Public Async Interceptors

The library has internal _LoggingClientAIOInterceptor in each grpc_asyncio.py transport file, but:

• It's private (name starts with _)
• Not exported or reusable
• No async versions of MetadataInterceptor or ExceptionInterceptor exist

Impact

When using is_async=True:

1. ✅ CustomerServiceAsyncClient is created
2. ❌ But it wraps sync CustomerServiceClient with sync transport
3. ❌ Methods like await service.list_accessible_customers() fail with:

   TypeError: object ListAccessibleCustomersResponse can't be used in 'await' expression
   

4. ❌ Or fail during channel creation with interceptor type mismatch:

   ValueError: Interceptor must be UnaryUnaryClientInterceptor or ...
   

Proposed Fix

Option 1: Fix in GoogleAdsClient (Recommended)

Detect is_async and create appropriate channel + interceptors:

def get_service(self, name, version, interceptors=None, is_async=False):
    # ...existing code...
    
    if is_async:
        # Create async channel and attach async interceptors at creation
        import grpc.aio

        async_interceptors = [
            AsyncMetadataInterceptor(...),
            AsyncLoggingInterceptor(...),
            AsyncExceptionInterceptor(...),
        ]

        channel = grpc.aio.secure_channel(
            endpoint,
            grpc.ssl_channel_credentials(),
            options=_GRPC_CHANNEL_OPTIONS,
            interceptors=async_interceptors,
        )

        service_transport = service_transport_class(channel=channel, ...)
    else:
        # Existing sync logic
        channel = service_transport_class.create_channel(...)
        channel = grpc.intercept_channel(channel, *sync_interceptors)
        service_transport = service_transport_class(channel=channel, ...)
    
    return service_client_class(transport=service_transport)

Option 2: Provide Async Interceptors

Export async versions of interceptors from google.ads.googleads.interceptors:

• AsyncMetadataInterceptor(grpc.aio.UnaryUnaryClientInterceptor)
• AsyncLoggingInterceptor(grpc.aio.UnaryUnaryClientInterceptor) (make _LoggingClientAIOInterceptor public)
• AsyncExceptionInterceptor(grpc.aio.UnaryUnaryClientInterceptor)

Option 3: Fix get_transport_class

In each *AsyncClient:

# Instead of:
get_transport_class = CustomerServiceClient.get_transport_class()

# Do:
@classmethod
def get_transport_class(cls, label='grpc_asyncio'):
    return CustomerServiceTransport._transport_registry.get(label)

Steps to Reproduce:

from google.ads.googleads.client import GoogleAdsClient

config = {
    'developer_token': 'YOUR_DEV_TOKEN',
    'client_id': 'YOUR_CLIENT_ID',
    'client_secret': 'YOUR_CLIENT_SECRET',
    'refresh_token': 'YOUR_REFRESH_TOKEN',
    'use_proto_plus': True,
}

client = GoogleAdsClient.load_from_dict(config, version='v22')

# Try to use async service
customer_service = client.get_service('CustomerService', is_async=True)

# This fails:
import asyncio
async def test():
    response = await customer_service.list_accessible_customers()
    print(response.resource_names)

asyncio.run(test())

Error 1 (if sync interceptors passed to async channel):

ValueError: Interceptor <MetadataInterceptor object at 0x...> must be 
UnaryUnaryClientInterceptor or UnaryStreamClientInterceptor or ...

Error 2 (if sync transport used):

TypeError: object ListAccessibleCustomersResponse can't be used in 'await' expression

Expected behavior:

1. client.get_service('CustomerService', is_async=True) should create fully async service
2. Channel should be grpc.aio.Channel, not grpc.Channel
3. Transport should be CustomerServiceGrpcAsyncIOTransport
4. Interceptors should be async-compatible
5. await service.list_accessible_customers() should work without errors

Client library version and API version:

• google-ads: 28.4.0
• Python: 3.12
• grpcio: 1.68.1
• Google Ads API version: v22

Environment:

• OS: macOS
• Async framework: asyncio (FastAPI)

Request/Response Logs:

Attempt 1: Using is_async=True with sync interceptors

INFO - Created GoogleAdsClient (access_token passed, cached)
ERROR - ValueError: Interceptor <google.ads.googleads.interceptors.metadata_interceptor.MetadataInterceptor object at 0x...> 
must be UnaryUnaryClientInterceptor or UnaryStreamClientInterceptor or StreamUnaryClientInterceptor or StreamStreamClientInterceptor.

Traceback (most recent call last):
  File "google/ads/googleads/client.py", line 444, in get_service
    service_transport = service_transport_class(channel=channel, ...)
  File "google/ads/googleads/v22/.../grpc_asyncio.py", line 265, in __init__
    raise ValueError(f"Interceptor {interceptor} must be...")
ValueError: Interceptor must be [async types]

Attempt 2: Using is_async=True without custom interceptors (sync transport created)

INFO - Listing accessible customers (async service, auto-refresh enabled)
ERROR - Unexpected error: object ListAccessibleCustomersResponse can't be used in 'await' expression

Traceback (most recent call last):
  File "adapters/google_ads_adapter.py", line 162, in list_accessible_customers
    response = await customer_service.list_accessible_customers()
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "google/ads/googleads/v22/.../async_client.py", line 456, in list_accessible_customers
    response = await rpc(...)
               ^^^^^^^^^^
TypeError: object ListAccessibleCustomersResponse can't be used in 'await' expression

The issue is that CustomerServiceAsyncClient._client uses sync CustomerServiceClient which has sync transport (CustomerServiceGrpcTransport), not async (CustomerServiceGrpcAsyncIOTransport).

Related Issues:

• Cannot retrieve async services via GoogleAdsClient.get_service ? #1024 - Async mode discussion (claimed to work, but doesn't)
• Related to gRPC async interceptor support

Additional Context:

The library has all necessary components for async:

• ✅ CustomerServiceGrpcAsyncIOTransport exists
• ✅ grpc.aio support in transports
• ✅ Internal _LoggingClientAIOInterceptor implementation
• ❌ But GoogleAdsClient.get_service() doesn't wire them correctly
• ❌ No public async interceptors

This makes is_async=True parameter misleading - it creates async client wrapper but with sync internals.