gateforge-sdk 0.2.5

Privacy-first LLMOps SDK — Auto-init, tool/agent decorators, session management, nested conversation support

Gateforge SDK (Python)

Privacy-first LLMOps SDK — transparent client wrapping with automatic PII masking, cost tracking, A/B prompt testing, guardrails, and agent trace support.

What it does

Gateforge wraps your existing provider client (OpenAI, Anthropic, Gemini) with a transparent proxy. Your code is unchanged — the SDK intercepts each call to run the full pipeline locally before anything reaches the provider:

Your code
    │
    ▼
pre-call  → A/B variant selection → system prompt injection
          → PII anonymize (local)
          → input guardrail check (local rules)
    │
    ▼
LLM provider (sees masked content only)
    │
    ▼
post-call → PII rehydrate (local)
          → output guardrail check
          → cost + latency compute
          → span emission (fire-and-forget, metadata only)
    │
    ▼
Your code receives: original PII restored, guardrails applied

Content never leaves your environment. Only metadata (tokens, cost, latency, PII types) is sent to Gateforge.

---

Installation

pip install gateforge-sdk

With provider extras:

pip install gateforge-sdk[openai]      # OpenAI only
pip install gateforge-sdk[anthropic]   # Anthropic only
pip install gateforge-sdk[gemini]      # Google Gemini only
pip install gateforge-sdk[all]         # All providers

Get your API key at https://app.gateforge.dev/dashboard/keys.

---

Quick Start

1. Initialize once at startup

import gateforge

gateforge.init(
    api_key="gf-live-YOUR_GATEFORGE_KEY",
    # Optional: provider keys if you use gateforge.chat() high-level API
    # openai_key="sk-...",
)
# Downloads config (PII patterns, guardrail rules, A/B experiments)
# and starts a background refresh thread every 5 minutes

2. Wrap your provider client

import gateforge
from openai import OpenAI

gateforge.init(api_key="gf-live-...")

client = gateforge.wrap_openai(OpenAI(api_key="sk-..."))

# Use exactly as before — the pipeline runs automatically
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "My SSN is 123-45-6789. Explain diabetes."}],
)
print(response.choices[0].message.content)
# → Provider saw "<SSN_1>". Response rehydrated with original value.

Or use auto-detection to wrap any supported client:

client = gateforge.wrap(OpenAI(api_key="sk-..."))          # detects OpenAI
client = gateforge.wrap(Anthropic(api_key="sk-ant-..."))   # detects Anthropic
client = gateforge.wrap(genai.Client(api_key="AIza-..."))  # detects Gemini

---

Provider Wrappers

All wrappers are transparent — input params and return types are identical to the underlying provider SDK.

OpenAI

import gateforge
from openai import OpenAI, AsyncOpenAI

gateforge.init(api_key="gf-live-...")

# Sync
client = gateforge.wrap_openai(OpenAI(api_key="sk-..."))
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)

# Async
async_client = gateforge.wrap_openai(AsyncOpenAI(api_key="sk-..."))
response = await async_client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello!"}],
)

# Streaming — metrics emitted after the stream ends
for chunk in client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
):
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

Anthropic

import gateforge
from anthropic import Anthropic, AsyncAnthropic

gateforge.init(api_key="gf-live-...")

client = gateforge.wrap_anthropic(Anthropic(api_key="sk-ant-..."))
response = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "My phone is [phone]. Help me."}],
)
print(response.content[0].text)

# Async
async_client = gateforge.wrap_anthropic(AsyncAnthropic(api_key="sk-ant-..."))
response = await async_client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)

Gemini

import gateforge
from google import genai

gateforge.init(api_key="gf-live-...")

client = gateforge.wrap_gemini(genai.Client(api_key="AIza-..."))
response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=[{"role": "user", "parts": [{"text": "Hello!"}]}],
)
print(response.text)

---

Agent Trace Support

For multi-step agent loops, wrap the entire run in gateforge.trace(). The SDK automatically assigns step numbers and groups all spans — LLM calls and tool calls — under the same conversation ID.

Basic multi-step trace

import gateforge
from openai import OpenAI

gateforge.init(api_key="gf-live-...")
client = gateforge.wrap_openai(OpenAI(api_key="sk-..."))

with gateforge.trace(conversation_id="conv_abc123"):
    # Step 1 — each LLM call inside the block gets an auto-incremented step
    r1 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "What tools do I need to book a flight?"}],
    )

    # Step 2
    r2 = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": "What tools do I need to book a flight?"},
            {"role": "assistant", "content": r1.choices[0].message.content},
            {"role": "user", "content": "Search for flights to Paris on June 10."},
        ],
    )
# → Dashboard shows: conv_abc123 | step 1 (LLM, gpt-4o-mini) → step 2 (LLM, gpt-4o-mini)

Agent with tool calls

import time
import gateforge
from openai import OpenAI

gateforge.init(api_key="gf-live-...")
client = gateforge.wrap_openai(OpenAI(api_key="sk-..."))

def search_flights(destination: str, date: str) -> list[dict]:
    ...

def book_flight(flight_id: str, passenger: str) -> str:
    ...

with gateforge.trace() as t:
    print(f"Trace ID: {t.conversation_id}")  # auto-generated UUID

    # Step 1 — router decides what to do
    plan = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Book me a flight to Paris for June 10"}],
    )

    # Step 2 — tool call (manually recorded so it appears in the waterfall)
    t0 = time.perf_counter()
    flights = search_flights("Paris", "2026-06-10")
    gateforge.record_tool_call(
        "search_flights",
        latency_ms=(time.perf_counter() - t0) * 1000,
        metadata={"destination": "Paris", "date": "2026-06-10", "results": len(flights)},
    )

    # Step 3 — synthesize results
    selection = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": "Book me a flight to Paris for June 10"},
            {"role": "assistant", "content": plan.choices[0].message.content},
            {"role": "user", "content": f"Available flights: {flights}. Pick the best one."},
        ],
    )

    # Step 4 — confirm booking
    t0 = time.perf_counter()
    confirmation = book_flight(flights[0]["id"], "John Doe")
    gateforge.record_tool_result(
        "book_flight",
        latency_ms=(time.perf_counter() - t0) * 1000,
        metadata={"confirmation_code": confirmation},
    )

# Dashboard waterfall:
#   conv_xyz | user: "Book me a flight to Paris for June 10"
#   ├─ Step 1  LLM  gpt-4o-mini  340 tok  $0.002  820ms
#   ├─ Step 2  tool search_flights                  210ms
#   ├─ Step 3  LLM  gpt-4o-mini  980 tok  $0.008  1400ms
#   └─ Step 4  tool book_flight                     95ms
#   Total: 1320 tokens | $0.010 | 2.5s

Multi-agent (multiple wrapped clients in one trace)

import gateforge
from openai import OpenAI
from anthropic import Anthropic

gateforge.init(api_key="gf-live-...")
openai_client = gateforge.wrap_openai(OpenAI(api_key="sk-..."))
claude_client = gateforge.wrap_anthropic(Anthropic(api_key="sk-ant-..."))

with gateforge.trace(conversation_id="conv_multiagent"):
    # Step 1 — GPT as router
    routing = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Summarize and translate this document."}],
    )

    # Step 2 — Claude as specialist
    summary = claude_client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=512,
        messages=[{"role": "user", "content": "Summarize: " + document_text}],
    )

    # Step 3 — GPT for final output
    result = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": f"Translate to Spanish: {summary.content[0].text}"}
        ],
    )
# All three steps appear in the same trace waterfall, regardless of provider

Async agent

import asyncio
import gateforge
from openai import AsyncOpenAI

gateforge.init(api_key="gf-live-...")

async def run_agent(user_message: str, conversation_id: str):
    client = gateforge.wrap_openai(AsyncOpenAI(api_key="sk-..."))

    with gateforge.trace(conversation_id=conversation_id):
        r1 = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": user_message}],
        )
        r2 = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "user", "content": user_message},
                {"role": "assistant", "content": r1.choices[0].message.content},
                {"role": "user", "content": "Now summarize in one sentence."},
            ],
        )
    return r2.choices[0].message.content

asyncio.run(run_agent("Explain quantum entanglement", "conv_001"))

Per-call conversation ID (without context manager)

When you can't use a context manager (e.g., in a framework that manages request scope), pass CallOptions directly:

from gateforge import CallOptions

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello!"}],
    gateforge_options=CallOptions(conversation_id="conv_xyz", trace=True),
)

---

A/B Prompt Testing

Create experiments in the dashboard, then the SDK assigns variants deterministically at call time — no network round-trip, zero added latency.

from gateforge import CallOptions

# Pin to a specific experiment; assign variant based on user session
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Help me write an email"}],
    gateforge_options=CallOptions(
        experiment_id="exp_email_v2",
        session_id="user_123",   # same user always gets the same variant
    ),
)
# Variant A or B system prompt injected automatically.
# experiment_id and variant are included in telemetry for dashboard comparison.

Variant assignment is a deterministic hash of (experiment_id, session_id) — no network call at call time.

Inside a trace, A/B telemetry is automatically attached to the correct step:

with gateforge.trace(conversation_id="conv_xyz"):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Draft a subject line"}],
        gateforge_options=CallOptions(experiment_id="exp_subject_lines", session_id="user_123"),
    )

---

Guardrails

Define guardrail rules in the dashboard. The SDK evaluates them locally against downloaded rules — no added latency, works offline. Violations are reported to the dashboard asynchronously.

from gateforge import CallOptions, GuardrailBlocked

# Guardrails run automatically when guardrails_enabled is set in your config.
# You can override per-call:
try:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Tell me how to pick a lock"}],
        gateforge_options=CallOptions(guardrails=True),
    )
except GuardrailBlocked as e:
    print(f"Blocked by rule: {e.rule_id}")
    response_text = "I can't help with that."

Configure on_fail behavior per rule in the dashboard:

on_fail	Behavior
block	Raises GuardrailBlocked
warn	Returns response with .warnings attached
retry	Re-calls LLM up to N times with a modified prompt
fallback	Returns a static fallback response

Input guardrails run before the LLM call. Output guardrails run after.

---

PII Protection

What gets detected

• Personal: names, emails, phone numbers, addresses
• Financial: credit cards, bank accounts, SSN, tax IDs
• Healthcare: medical record numbers, symptoms, diagnoses
• Technical: IP addresses, URLs, API keys
• Custom: your own regex patterns (configured in the dashboard)

Domain-specific detection

from gateforge import CallOptions

# Override PII domain per call
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{
        "role": "user",
        "content": "Patient John Doe, SSN 123-45-6789, has hypertension."
    }],
    gateforge_options=CallOptions(pii_domain="healthcare"),
)
# Detected: PERSON, SSN, MEDICAL_CONDITION

Direct anonymization

result = gateforge.anonymize("My email is [e-mail]")
print(result["sanitized"])  # "My email is [EMAIL_001]"
print(result["entities"])   # ["EMAIL"]

original = gateforge.rehydrate("[EMAIL_001] confirmed", context=result["context"])
print(original)  # "[e-mail] confirmed"

---

CallOptions Reference

Pass gateforge_options=CallOptions(...) to any wrapped .create() call to override defaults for that call only:

from gateforge import CallOptions

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    gateforge_options=CallOptions(
        # Trace grouping
        conversation_id="conv_abc",   # attach this call to a conversation
        trace=True,                   # enable trace events for this call

        # A/B testing
        experiment_id="exp_abc",      # target a specific experiment
        session_id="user_123",        # deterministic variant assignment

        # Feature overrides (None = use global config)
        pii=True,                     # force PII on for this call
        guardrails=True,              # force guardrails on
        track_cost=False,             # skip cost tracking for this call
        track_latency=True,
        track_tokens=True,
        ab=False,                     # skip A/B for this call
    ),
)

---

Simple API (high-level)

For quick prototyping. Returns a GatforgeResponse rather than the native provider response.

import gateforge

gateforge.init(
    api_key="gf-live-...",
    openai_key="sk-...",
    anthropic_key="sk-ant-...",
    gemini_key="AIza-...",
)

response = gateforge.chat(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "My email is [e-mail], help me."}],
)

print(response.content)           # "Hello! I'd be happy to help..."
print(response.pii_detected)      # ['EMAIL']
print(response.cost_usd)          # 0.000023
print(response.latency_ms)        # 412.0
print(response.prompt_tokens)     # 18
print(response.completion_tokens) # 27

For production use, prefer gateforge.wrap_openai() / wrap_anthropic() / wrap_gemini() — they return the native provider response object with all fields intact.

---

OTel Integration

If your application already has an OpenTelemetry pipeline, Gateforge emits gen_ai.* spans into it automatically — no extra config needed. To set up OTel from scratch:

pip install gateforge-sdk[otel]

import gateforge
from gateforge import configure_otel

configure_otel(
    endpoint="http://localhost:4317",   # your OTLP collector
    service_name="my-llm-app",
)

gateforge.init(api_key="gf-live-...")
client = gateforge.wrap_openai(OpenAI(api_key="sk-..."))

# Every LLM call now emits a gen_ai span into your OTel pipeline
response = client.chat.completions.create(model="gpt-4o-mini", messages=[...])

---

Supported Models

OpenAI

• GPT-4o, GPT-4o-mini
• GPT-4.1, GPT-4.1-mini, GPT-4.1-nano

Anthropic

• Claude Haiku 4-5
• Claude Sonnet 4-5
• Claude Opus 4

Google Gemini

• Gemini 2.5 Flash
• Gemini 2.5 Pro

---

Dashboard & Monitoring

https://app.gateforge.dev/dashboard

• Request volume and trends
• Cost breakdown by model and provider
• Latency analytics
• PII detection statistics
• A/B experiment results (variant comparison, significance indicator)
• Guardrail violation alerts
• Agent waterfall traces (per-step cost, latency, tool calls)
• API key management

---

Troubleshooting

ImportError: No module named 'gateforge'

pip install gateforge-sdk

RuntimeError: Call gateforge.init() first

gateforge.init(api_key="gf-live-...")
# Call this once before any wrap_*() or chat() calls

PII not detected

1. Check the domain setting matches your data (healthcare, finance, legal, generic)
2. presidio backend is most accurate; regex is fastest
3. Add custom patterns in the dashboard for domain-specific entities

Steps not appearing in trace waterfall

1. Make sure tracing_enabled=True in your feature flags (set in dashboard or override with FeatureFlags)
2. Confirm a conversation_id is active — either via gateforge.trace() or CallOptions(conversation_id=...)

---

Links

• Website
• Dashboard
• API Docs
• Issues

License

MIT — see LICENSE. The SDK is open source; the Gateforge service is commercial with a free tier (1,000 requests/month).