langflow-ai/langflow · error · HTTPException

Invalid API key

Error message

Invalid API key

What it means

Raised by _enforce_a2a_auth when the x-api-key is presented but either fails check_key (unknown/disabled key) or belongs to a different user than the flow owner. The same message covers both cases intentionally: revealing 'valid key, wrong user' would disclose that the key works elsewhere. The key must be issued by the flow's owner account.

Solutions

  1. Regenerate an API key under the flow owner's account and use that in x-api-key
  2. Verify the key still exists and is active in the owner's API-keys page (it may have been revoked)
  3. Confirm flow ownership — if the flow belongs to user A, only user A's keys work; transfer the flow or use the owner's key

Example fix

# before
headers={"x-api-key": os.environ["ADMIN_API_KEY"]}
# after
headers={"x-api-key": os.environ["FLOW_OWNER_API_KEY"]}
Defensive patterns

Strategy: validation

Validate before calling

import httpx

def key_is_live(base_url: str, api_key: str, auth: str) -> bool:
    r = httpx.get(f"{base_url}/api/v1/api_keys/", headers={"x-api-key": api_key}, auth=auth)
    return r.status_code == 200  # a valid, active key lists fine

Try / catch

try:
    resp = await client.send_message(flow_id, payload)
except A2AClientError as e:
    if "Invalid API key" in str(e):
        key = rotate_to_owner_key(flow_id)   # fetch/regenerate under the flow owner
        client = rebuild_client_with_x_api_key(key)
        resp = await client.send_message(flow_id, payload)
    else:
        raise

Prevention

When it happens

Trigger: POST /api/v1/a2a/{flow_id}/jsonrpc with an x-api-key that is revoked, mistyped, from another user's account, or a key created after the server cached key state.

Common situations: Using an admin/personal key from a different account than the one that owns the agent flow; copying a key from another environment; a key revoked or rotated after the client was configured.

Understand the failure class

Related errors

AI-assisted analysis of langflow-ai/langflow@976ec789d2 (2026-08-14). Data as JSON: /api/errors/c83543be03f3d8e4. Report an issue: GitHub.

Appendix: source

Thrown at src/backend/base/langflow/api/v1/a2a.py:136

    # Short writable session (check_key flushes usage counters), closed before
    # dispatch so no lock is held across the up-to-300s run.
    async with session_scope() as session:
        auth_type = await folder_auth_type(flow, session)
        if auth_type == "none":
            return  # public agent
        if auth_type not in ("apikey", "oauth"):
            # Protected folder with a scheme A2A can't enforce: fail closed, never public.
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"A2A access is disabled for this agent: unsupported folder auth type {auth_type!r}.",
            )
        api_key = request.headers.get(A2A_APIKEY_HEADER)
        if not api_key:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="API key required")
        user = await check_key(session, api_key)
        # Same message for invalid and wrong-owner: don't reveal a key is valid for another user.
        if user is None or user.id != flow.user_id:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid API key")


class _FlowContextBuilder(DefaultServerCallContextBuilder):
    """Carry the per-request flow_id into the shared executor via call-context state."""

    def build(self, request: Request) -> ServerCallContext:
        context = super().build(request)
        # Canonicalize to the same string form the resume guard uses (str(UUID(...))), so the durable
        # store scope (_task_scope / _push_config_scope) and the checkpoint/resume guard agree even for
        # a non-canonical UUID in the path (uppercase or hyphenless, both valid to the UUID route
        # converter). Without this, the same task addressed via two encodings lands in two scopes.
        context.state["flow_id"] = str(UUID(request.path_params["flow_id"]))
        return context


async def _is_public_a2a_flow(flow: Flow) -> bool:
    """Whether an A2A flow admits callers without an API key."""
    async with session_scope_readonly() as session:

View on GitHub (pinned to 976ec789d2)