gravity-sdk 0.1.0

Python server SDK for fetching Gravity ads

gravity-py

Python SDK for requesting ads from the Gravity API.

from gravity_sdk import Gravity

gravity = Gravity(production=True)  # reads GRAVITY_API_KEY from env

result = await gravity.get_ads(request, messages, placements)

Works with FastAPI, Starlette, Django, and Flask. Only dependency is httpx.

Install

pip install gravity-sdk

Set GRAVITY_API_KEY in your server environment.

Integration

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": "chat", "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")

gravity.get_ads() takes your server's request object, the conversation messages, and your ad placements, then calls the Gravity API and returns the ads. Never raises — returns AdResult(ads=[]) on any failure.

Message handling

The SDK sends the last 2 conversational messages to the Gravity API for contextual ad matching. Only messages with recognized conversational 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 because they typically contain structured data rather than natural language.

Constructor

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

Parameter	Type	Description
api_key	str	Gravity API key (default: GRAVITY_API_KEY env var)
api_url	str	Gravity API endpoint URL (default: production)
timeout	float	Request timeout in seconds (default: 3.0)
production	bool	Serve real ads when True. Defaults to False (test ads).
relevancy	float	Minimum relevancy threshold, 0.0–1.0 (default: 0.2). 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.

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

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

Development

uv sync
uv run pytest
uv run ruff check src/ tests/

License

MIT