gravity-sdk 0.1.0
Python server SDK for fetching Gravity ads
Requires Python: >=3.10
Keywords: advertising, ai-ads, contextual-advertising, gravity, llm, sdk

# gravity-py

Python SDK for requesting ads from the [Gravity](https://trygravity.ai) API.

```python
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](https://www.python-httpx.org/).

## Install

```bash
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.

```diff
+ 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

```python
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

```python
@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

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

## License

MIT