[Bug]: AuthInterceptor applies only the first scheme of a multi-scheme security requirement (AND semantics broken)

[Gwangil]

What happened?

Summary

AuthInterceptor.before() returns as soon as it successfully applies one credential. When a single SecurityRequirement lists multiple schemes — which, per the OpenAPI security requirement semantics the A2A spec follows, means all of them must be satisfied together (AND) — only one credential ever gets attached to the outgoing request.

In my setup, the agent card declares one requirement containing two api-key-in-header schemes (a gateway auth key + a user id header). The gateway requires both headers, but the interceptor only sends one, so every request is rejected.

Because proto3 map iteration order is not deterministic, which of the schemes gets applied can even differ between runs (see log output below), making the failure intermittent-looking and hard to debug.

Expected behavior

Per OpenAPI Security Requirement semantics (A2A's security model is based on the OpenAPI Security Scheme / Security Requirement objects):

• the outer security_requirements list is OR — satisfying any one requirement is sufficient;
• multiple schemes inside a single requirement are AND — all of them must be applied together.

So for the card below, both AUTH_KEY and USER_ID headers should be attached.

Reproduction

# pip install "a2a-sdk==1.1.1"  (also reproduces on main @ 86c6b0d)
import asyncio

from a2a.client.auth.credentials import CredentialService
from a2a.client.auth.interceptor import AuthInterceptor
from a2a.client.interceptors import BeforeArgs
from a2a.types.a2a_pb2 import (
    AgentCard,
    APIKeySecurityScheme,
    SecurityRequirement,
    SecurityScheme,
    StringList,
)

CREDS = {"AUTH_KEY": "gw-secret-key", "USER_ID": "user-123"}


class DictCredentialService(CredentialService):
    async def get_credentials(self, security_scheme_name, context):
        return CREDS.get(security_scheme_name)


card = AgentCard(
    security_schemes={
        "AUTH_KEY": SecurityScheme(
            api_key_security_scheme=APIKeySecurityScheme(
                location="header", name="AUTH_KEY"
            )
        ),
        "USER_ID": SecurityScheme(
            api_key_security_scheme=APIKeySecurityScheme(
                location="header", name="USER_ID"
            )
        ),
    },
    # ONE requirement containing BOTH schemes -> AND: both headers required
    security_requirements=[
        SecurityRequirement(
            schemes={"AUTH_KEY": StringList(), "USER_ID": StringList()}
        )
    ],
)


async def main():
    args = BeforeArgs(input=None, method="message/send", agent_card=card)
    await AuthInterceptor(DictCredentialService()).before(args)
    # Expected: {'AUTH_KEY': 'gw-secret-key', 'USER_ID': 'user-123'}
    # Actual:   only one of the two headers is present
    print(args.context.service_parameters)


asyncio.run(main())

Note that credentials for both schemes are available from the CredentialService — the missing header is purely due to the interceptor's control flow.

Root cause

In src/a2a/client/auth/interceptor.py, every successful branch (bearer, oauth2/oidc, api-key-in-header) ends with a return inside the inner per-scheme loop, so processing stops after the first applied credential instead of finishing the requirement.

Proposed fix

Move the return so it happens after the inner (per-scheme) loop completes, i.e. once one requirement has been fully satisfied. Simply removing the returns and finishing both loops would over-correct: it would apply credentials from alternative (OR) requirements too, and since the bearer/oauth2/oidc branches all write to the same 'Authorization' key, a later requirement would silently overwrite an earlier one (last-write-wins).

A two-pass approach per requirement keeps AND strict (no partial application when a credential is missing) while preserving OR across requirements:

for requirement in agent_card.security_requirements:
    # pass 1: can EVERY scheme in this requirement be satisfied?
    collected: dict[str, str] = {}
    satisfiable = True
    for scheme_name in requirement.schemes:
        credential = await self._credential_service.get_credentials(
            scheme_name, args.context
        )
        if not credential or scheme_name not in agent_card.security_schemes:
            satisfiable = False
            break
        collected[scheme_name] = credential

    if not satisfiable:
        continue  # OR: try the next requirement

    # ... ensure args.context / service_parameters exist ...

    # pass 2: apply ALL schemes of this requirement
    for scheme_name, credential in collected.items():
        scheme = agent_card.security_schemes[scheme_name]
        # existing per-scheme-type application logic, without early returns

    return  # one requirement fully satisfied -> done

One open design question: should a scheme type the interceptor cannot apply (e.g. API key in query/cookie, currently skipped) make the whole requirement unsatisfiable (strict reading of AND), or be skipped as today? Happy to align with whatever the maintainers prefer.

I'm glad to submit a PR (including unit tests for the multi-scheme AND case and the OR fallback) if this direction sounds right.

Current workaround

We're temporarily carrying a local patch that simply removes the early returns and returns after the loop completes. That happens to be safe for our card because it declares a single requirement, but as explained above it is not a general fix (it would break OR semantics for multi-requirement cards) — hence this issue.

Environment

• a2a-sdk 1.1.1 (latest release) — reproduced by execution
• main @ 86c6b0d — reproduced by execution
• Python 3.12, Linux

Related: #445 (broader client auth ergonomics — this issue is only about the correctness of the existing AuthInterceptor control flow).

Relevant log output

$ python repro.py
{'AUTH_KEY': 'gw-secret-key'}          # USER_ID header missing -> gateway returns 401/403

# proto3 map iteration order is not stable across processes, so WHICH header
# gets applied is nondeterministic. Reading the same card in 3 fresh processes:
$ for i in 1 2 3; do python -c "..."; done
['AUTH_KEY', 'USER_ID']
['AUTH_KEY', 'USER_ID']
['USER_ID', 'AUTH_KEY']                # order flipped -> USER_ID would be the one applied

Code of Conduct

• I agree to follow this project's Code of Conduct