Function webhook

examples/adk-cloud-webhook/webhook.py:92–165

(request: Request)

@app.post("/webhook")
async def webhook(request: Request) -> dict[str, str]:
    body = await request.body()
    # construct_event does parse + HMAC-verify in one call and raises
    # E2AWebhookSignatureError on a bad signature or replay. Anyone can reach a
    # public webhook URL; this verification is what proves the payload came from
    # your e2a relay (the signature binds to the sender claim e2a verified via
    # SPF/DKIM), so trust event.data only *after* it succeeds.
    try:
        event = construct_event(body, request.headers.get("X-E2A-Signature", ""), WEBHOOK_SECRET)
    except E2AWebhookSignatureError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="bad signature")

    # We only act on inbound mail; ignore other event types (sent/approved/…).
    if event.type != "email.received":
        return {"status": "ignored", "type": event.type}

    client: E2AClient = request.app.state.e2a
    data = event.data  # WebhookPayload: message_id, recipient, from, conversation_id, …
    recipient = data["recipient"]
    message_id = data["message_id"]

    # The inbound webhook payload is metadata; fetch the parsed message for the
    # subject + body to feed the agent.
    inbound = await client.messages.get(recipient, message_id)

    # First contact has no conversation_id — mint one so this thread has an
    # ID we can echo back on every reply. The same id becomes the ADK
    # session_id, keying multi-turn memory.
    conversation_id = data.get("conversation_id") or f"conv_{uuid.uuid4().hex[:12]}"

    # user_id scopes a session to a particular human counterpart. Different
    # senders get isolated session histories even on the same agent.
    user_id = data["from"]

    sessions = request.app.state.sessions
    session = await sessions.get_session(
        app_name=APP_NAME, user_id=user_id, session_id=conversation_id
    )
    if session is None:
        session = await sessions.create_session(
            app_name=APP_NAME, user_id=user_id, session_id=conversation_id
        )

    prompt = types.Content(
        role="user",
        parts=[types.Part(text=_format_email_for_agent(inbound))],
    )

    runner: Runner = request.app.state.runner
    reply_text = ""
    async for event in runner.run_async(
        user_id=user_id, session_id=conversation_id, new_message=prompt
    ):
        if event.is_final_response() and event.content and event.content.parts:
            reply_text = event.content.parts[0].text or ""

    if not reply_text:
        # Agent produced no response (rare — usually means a tool call without