> ## Documentation Index
> Fetch the complete documentation index at: https://ahasend.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

> ## Agent Instructions
> AhaSend stores all message data in the EU. See /facts.
> The API base URL is https://api.ahasend.com. Send the API key in the Authorization: Bearer header. See /api-reference/authentication.
> Use send.ahasend.com for SMTP in the EU, or send-us.ahasend.com as a US connection point forwarding to the EU. Ports 25, 587 and 2525 use STARTTLS. Port 465 is not supported. See /smtp.
> For send-only integrations, create an API Key v2 under Credentials → Add and scope it to messages:send:{your-domain}. See /send-api/credentials and /api-reference/scopes.

# Send Email from Google Cloud Functions (Python)

> Send email from Google Cloud Run functions in Python through AhaSend, with a protected HTTP function, local sandbox tests and clear failure responses.

AhaSend lets you send transactional email from Google Cloud Run functions in Python through the HTTP API; this guide adds a protected function and runs it locally.

Cloud Run functions is the current name for the service previously called Cloud Functions.

## Prerequisites

Use a [verified sending domain](/docs/domains), your account ID, and a domain-scoped [send-only API v2 key](/docs/send-api/credentials). Use [sandbox mode](/docs/send-api/sandbox) for every test. Management tasks need a separate [full API key](/docs/api-reference/authentication).

Use Python 3.13 and the [Python Functions Framework](https://github.com/GoogleCloudPlatform/functions-framework-python). You can complete the local test without a Google Cloud account.

## Install the Dependencies

```bash theme={null}
python3.13 -m venv .venv
source .venv/bin/activate
python -m pip install functions-framework
python -m pip freeze > requirements.txt
```

## Try a Sandbox Send

Set `AHASEND_API_KEY`, `AHASEND_ACCOUNT_ID` and `AHASEND_FROM` in your shell or local secret store. Use a [send-only key](/docs/send-api/credentials) and an address on your [verified domain](/docs/domains). Save this as `quick_send.py`:

```python quick_send.py theme={null}
import json, os, urllib.request
payload = {
    "from": {"email": os.environ["AHASEND_FROM"]},
    "recipients": [{"email": "recipient@example.com"}],
    "subject": "My first sandbox email",
    "text_content": "Hello from AhaSend",
    "sandbox": True,
}
url = f"https://api.ahasend.com/v2/accounts/{os.environ['AHASEND_ACCOUNT_ID']}/messages"
request = urllib.request.Request(url, json.dumps(payload).encode(), headers={
    "Authorization": f"Bearer {os.environ['AHASEND_API_KEY']}",
    "Content-Type": "application/json",
})
with urllib.request.urlopen(request, timeout=10) as response:
    statuses = [item["status"] for item in json.load(response)["data"]]
if not statuses or any(s not in ("queued", "scheduled") for s in statuses):
    raise RuntimeError("A recipient was rejected")
print({"statuses": statuses})
```

Run `python quick_send.py` in the activated environment. It prints acceptance statuses and exits with an error if sending fails. Sandbox acceptance does not deliver mail. The production version below adds input checks and a stable request key.

## Add the Shared Sending Code

Save this as `send_email.py` beside the function file. It uses Python's standard library, so this example needs no SDK package. Set `AHASEND_API_KEY`, `AHASEND_ACCOUNT_ID`, `AHASEND_FROM`, `AHASEND_TO` and `SEND_TOKEN` in the runtime environment. Use a long random `SEND_TOKEN` for requests from your trusted server. For the first run set `AHASEND_TO=recipient@example.com` and `AHASEND_DELIVERY_MODE=sandbox`; keep a sandbox credential in place while testing. Set `live` only when ready to deliver to your own approved recipients.

```python send_email.py theme={null}
import hmac, json, os, re, urllib.error, urllib.request

def send_email(body, authorization):
    token = os.environ.get("SEND_TOKEN", "")
    if not token:
        return 500, {"error": "Missing server configuration"}
    if not hmac.compare_digest(authorization.encode(), ("Bearer " + token).encode()):
        return 401, {"error": "Unauthorized"}
    event_id = body.get("event_id") if isinstance(body, dict) else None
    if not isinstance(event_id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{8,64}", event_id):
        return 400, {"error": "event_id must be 8-64 letters, digits, _ or -"}
    mode = os.environ.get("AHASEND_DELIVERY_MODE", "sandbox")
    required = ("AHASEND_API_KEY", "AHASEND_ACCOUNT_ID", "AHASEND_FROM", "AHASEND_TO")
    if mode not in ("sandbox", "live") or any(not os.environ.get(k) for k in required):
        return 500, {"error": "Missing or invalid server configuration"}
    payload = {
        "from": {"email": os.environ["AHASEND_FROM"]},
        "recipients": [{"email": os.environ["AHASEND_TO"]}],
        "subject": "Welcome",
        "text_content": "Your account is ready.",
        "sandbox": mode == "sandbox",
    }
    url = f"https://api.ahasend.com/v2/accounts/{os.environ['AHASEND_ACCOUNT_ID']}/messages"
    request = urllib.request.Request(url, json.dumps(payload).encode(), headers={
        "Authorization": f"Bearer {os.environ['AHASEND_API_KEY']}",
        "Content-Type": "application/json",
        "Idempotency-Key": f"{mode}-welcome-{event_id}",
    })
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            result = json.load(response)
        statuses = [item["status"] for item in result["data"]]
        if not statuses or any(s not in ("queued", "scheduled") for s in statuses):
            return 422, {"error": "A recipient was rejected"}
        return 202, {"statuses": statuses}
    except (urllib.error.URLError, TimeoutError, ValueError, KeyError, TypeError):
        return 502, {"error": "Email request failed"}
```

Only your server chooses the sender, recipient and message body. Look up a user's verified address from your database before adapting this to a real signup flow. Keep that saved message unchanged when retrying an `event_id`; a new business event needs a new ID. See [idempotency](/docs/api-reference/idempotency) for time limits and uncertain outcomes. Record an internal event ID on failures without logging API keys or message bodies. Add request limits at the hosting layer.

## Create the HTTP Function

Save this as `main.py`:

```python main.py theme={null}
import functions_framework
from send_email import send_email

@functions_framework.http
def send_email_http(request):
    if request.method != "POST":
        return {"error": "Use POST"}, 405
    if len(request.get_data()) > 16384:
        return {"error": "Request too large"}, 413
    if not request.is_json:
        return {"error": "Use application/json"}, 415
    body = request.get_json(silent=True)
    if body is None:
        return {"error": "Invalid JSON"}, 400
    status, result = send_email(body, request.headers.get("authorization", ""))
    return result, status
```

## Run with the Functions Framework

If port 8080 is busy, set `PORT` to an unused local port in both shells. The commands below default to 8080.

Export the shared code's environment variables, activate `.venv`, then start the function:

```bash theme={null}
functions-framework --target=send_email_http --port="${PORT:-8080}"
```

In another shell with `SEND_TOKEN` set, run:

```bash theme={null}
curl --fail-with-body "http://localhost:${PORT:-8080}"   -H "Authorization: Bearer $SEND_TOKEN" -H 'Content-Type: application/json'   -d '{"event_id":"local-google-python-001"}'
```

Expect HTTP 202 with acceptance statuses. A wrong token returns 401, invalid input returns 400 and a failed upstream request returns 502. Use a new event ID for each new test and keep the original one when retrying the same job.

## Prepare the Hosted Function

Deploy with the Python 3.13 runtime and entry point `send_email_http`. Include `main.py`, `send_email.py` and `requirements.txt`. Follow [Google's function deployment steps](https://cloud.google.com/run/docs/deploy-functions), load credentials with [Secret Manager](https://cloud.google.com/run/docs/configuring/services/secrets), and restrict invokers with IAM. With both IAM and the application bearer token enabled, send the Google identity token in `X-Serverless-Authorization` and the app token in `Authorization`; see [service authentication](https://cloud.google.com/run/docs/authenticating/service-to-service).

Use [delivery webhooks](/docs/integrations/webhooks) to learn the final result. Also see [Google functions in Node.js](/docs/guides/google-cloud-functions-nodejs), [Go](/docs/guides/google-cloud-functions) and [Python SMTP](/docs/smtp/python).

## Related Guides

* Before sending: [verify a domain](/docs/domains) and [create a send-only key](/docs/send-api/credentials).
* Other ways to send: [REST API](/docs/send-api/send-email), [SMTP](/docs/smtp), [CLI quickstart](/docs/cli/quickstart), [Node.js SDK](/docs/guides/nodejs-sdk) and [Go SDK](/docs/guides/go-sdk).
* Request rules: [API authentication](/docs/api-reference/authentication), [scopes](/docs/api-reference/scopes), [idempotency](/docs/api-reference/idempotency), [errors](/docs/api-reference/errors) and [rate limits](/docs/api-reference/rate-limits).
* Testing and events: [sandbox mode](/docs/send-api/sandbox), [CLI webhook testing](/docs/cli/webhook-testing), [event payloads](/docs/api-reference/webhooks), [signature verification](/docs/api-reference/webhooks/security) and [delivery retries](/docs/api-reference/webhooks/retry-policy).
* Data and limits: [retention](/docs/retention), [tracking](/docs/tracking/open-tracking) and [plans and feature availability](/docs/facts#plans-and-features).