Python SDK - Gravity

Installation

pip install gravity-sdk

Set GRAVITY_API_KEY in your server environment. Only dependency is httpx.

Quick start

Add a few lines to your existing streaming chat endpoint. The ad request runs in parallel with your LLM call — zero added latency.

+ import asyncio
+ from gravity_sdk import Gravity

+ gravity = Gravity(production=True)

  @app.post("/api/chat")
  async def chat(request: Request):
      body = await request.json()
      messages = body["messages"]

+     ad_task = asyncio.create_task(
+         gravity.get_ads(request, messages, [{"placement": "below_response", "placement_id": "main"}])
+     )

      async def event_stream():
          async for token in stream_your_llm(messages):
              yield f"data: {json.dumps({'type': 'chunk', 'content': token})}\n\n"

-         yield f"data: {json.dumps({'type': 'done'})}\n\n"
+         ad_result = await ad_task
+         ads = [a.to_dict() for a in ad_result.ads]
+         yield f"data: {json.dumps({'type': 'done', 'ads': ads})}\n\n"

      return StreamingResponse(event_stream(), media_type="text/event-stream")

Works with FastAPI, Starlette, Django, and Flask.

Constructor

Gravity(*, api_key=None, api_url=None, timeout=4.0, production=False, relevancy=0.2)

Parameter	Type	Default	Description
`api_key`	`str`	`GRAVITY_API_KEY` env var	Your Gravity API key
`api_url`	`str`	Production URL	Gravity API endpoint
`timeout`	`float`	`4.0`	Request timeout in seconds
`production`	`bool`	`False`	`False` returns test ads (no billing)
`relevancy`	`float`	`0.2`	Minimum relevancy threshold, 0.0-1.0. Lower = more ads with weaker contextual matches

The client reuses its HTTP connection pool across calls. Use async with Gravity() as g: or call await gravity.close() for explicit cleanup.

`get_ads()`

await gravity.get_ads(request, messages, placements, *, production=None, relevancy=None)

Parameter	Type	Default	Description
`request`	framework request	—	Your server’s request object (FastAPI, Django, Flask, etc.)
`messages`	`list[dict]`	—	Conversation messages `[{"role": ..., "content": ...}]`
`placements`	`list[dict]`	—	Ad placements, e.g. `[{"placement": "chat", "placement_id": "main"}]`
`production`	`bool \| None`	`None`	Override the constructor-level `production` flag for this call
`relevancy`	`float \| None`	`None`	Override the constructor-level `relevancy` threshold for this call

get_ads() reads gravity_context (session, user, and device) from the posted request body and adds the end-user IP from the request headers. It does not invent a device — your browser client must call gravityContext() and post the result, and your backend must leave it in the body it forwards.

Forward the client’s device — device.ip and device.ua are required. If the request reaching Gravity has no end-user ua/ip — e.g. a pure server-to-server call that drops the client context — it is rejected with HTTP 400 and no ad is served. See Device signals.

Return types

@dataclass
class AdResult:
    ads: list[AdResponse]   # Parsed ad objects
    status: int             # 200, 204, 0 (error)
    elapsed_ms: str         # e.g. "142"
    request_body: dict | None
    error: str | None

@dataclass
class AdResponse:
    ad_text: str
    title: str | None
    cta: str | None
    brand_name: str | None
    url: str | None
    favicon: str | None
    imp_url: str | None
    click_url: str | None
    placement: str | None
    placement_id: str | None
    campaign_id: str | None
    # Experiment / composition fields (present when an experiment is active)
    variant: str | None
    experiment_id: str | None
    composition_id: str | None
    renderer_key: str | None
    composition_mode: str | None
    composition: dict | None

Both have .to_dict() methods that serialize to the camelCase JSON shape renderers expect.

Message handling

The SDK sends the last 2 conversational messages to the Gravity API for contextual ad matching. Only messages with recognized roles are included:

user, assistant, system, developer, model (Gemini’s alias for assistant)

Messages with other roles (e.g. tool, function, ipython) are filtered out — they typically contain structured data rather than natural language.

gravity.get_ads() never raises. On any failure, it returns AdResult(ads=[]). Safe to fire-and-forget in your stream.

PII hashing

The SDK provides hash_pii() for SHA-256 hashing of emails and phone numbers, matching the normalization used by the Gravity publisher pixel and the advertiser-side conversion pipeline.

from gravity_sdk import hash_pii

hashed = hash_pii(email=user.email, phone=user.phone)
# Include hashed PII in the gravity_context.user sent from your client:
# { "gravity_context": { "user": { "id": "u123", "email_hash": "...", "hashed_phone": "..." } } }

Normalization rules:

Email: strip().lower()
Phone: digits only (e.g. "+1 (555) 123-4567" → "15551234567")

Returns a HashedIdentity dict with only the keys that successfully hashed (email_hash, hashed_phone), so it’s safe to spread into a user dict. email_hash is the canonical email field — it matches the email_hash used by the Index /search API.

Experiments

Experiments are managed server-side. Return the ad normally and render it with the standard SDK component; no experiment-specific branching or tracking code is required.

​Installation

​Quick start

​Constructor

​get_ads()

​Return types

​Message handling

​PII hashing

​Experiments

Installation

Quick start

Constructor

`get_ads()`

Return types

Message handling

PII hashing

Experiments