Gateforge SDK (Python)

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

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:

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

PII detection, anonymization and rehydration run in your process. The LLM provider receives the anonymized prompt. Gateforge telemetry is disabled when no exporter or control-plane URL is configured; when enabled it can include metadata and, if store_content is explicitly enabled, anonymized content. The optional LLM quality judge also sends anonymized prompt/response excerpts to the configured judge provider. See Privacy and data flow.

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
pip install gateforge-sdk[dev]         # Development tools

An API key is only needed when using a Gateforge-compatible hosted or self-hosted control plane. Fully local use does not require one.

Quick Start

Option 1: Fully local (recommended for first use)

import gateforge
from openai import OpenAI

gateforge.init(config={
    "pii_domain": "generic",
    "pii_backend": "presidio",
    "features": {"pii_enabled": True, "guardrails_enabled": False},
})

# Wrapping is required. An unwrapped provider client is not protected.
client = gateforge.wrap_openai(OpenAI(api_key="sk-..."))

Option 2: Hosted or self-hosted control plane

import gateforge
from openai import OpenAI

gateforge.init(
    api_key="gf-live-YOUR_KEY",
    base_url="https://your-gateforge-control-plane.example",
)

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

# Use exactly as before — pipeline runs automatically
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello!"}],
)

Option 3: Provider auto-detection

import gateforge
from openai import OpenAI
from anthropic import Anthropic
from google import genai

gateforge.init(config={"features": {"pii_enabled": True}})

# Auto-detects provider
client1 = gateforge.wrap(OpenAI(api_key="sk-..."))
client2 = gateforge.wrap(Anthropic(api_key="sk-ant-..."))
client3 = gateforge.wrap(genai.Client(api_key="AIza-..."))

Key Features

1. Auto-Initialization

import gateforge
from openai import OpenAI

# Requires GATEFORGE_API_KEY and, for remote config/telemetry,
# GATEFORGE_BASE_URL. It does not monkey-patch provider clients.

gateforge.auto_init(enable_pii=True, enable_guardrails=False)
client = gateforge.wrap_openai(OpenAI(api_key="sk-..."))

2. Decorator-Based Tracing

import gateforge

# Tool with automatic tracing
@gateforge.tool()
def get_weather(location: str) -> str:
    import requests
    return requests.get(f"https://wttr.in/{location}?format=3").text

# Agent with automatic tracing
@gateforge.agent()
def weather_agent(message: str) -> str:
    return get_weather("Madrid")

# Session to group multiple calls
with gateforge.session(user_id="user-123"):
    response = weather_agent("What's the weather?")

3. Conversation Management

from gateforge import SessionManager, trace

# Manage sessions
manager = SessionManager()
session = manager.create_session(
    user_id="user-123",
    tags=["weather-chat"],
)

# Use in trace
with trace(conversation_id=session.conversation_id):
    response = run_agent(message)

# Track activity
manager.touch(session.conversation_id)
session.set_metadata("last_model", "gpt-4o-mini")

4. Prompt System

from gateforge import Prompt, PromptBuilder, PromptCache

# Create prompt with variables
prompt = Prompt(
    name="greeting",
    content="Hello, {{name}}! You are {{role}}.",
    variables={"name": "User", "role": "a developer"},
)
rendered = prompt.render(name="Alice")  # "Hello, Alice!..."

# Compose prompts
builder = PromptBuilder()
builder.add_system("You are helpful")
builder.add_user("What's the weather?")
builder.add_variable("location", "Madrid")
prompt = builder.build()

# Cache prompts (memory + file + backend)
cache = PromptCache(memory_ttl=300, file_ttl=3600)
cache.set(prompt)
retrieved = cache.get("greeting")

Provider Wrappers

All wrappers are transparent — input params and return types match the underlying 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!"}],
)

# Async
async_client = gateforge.wrap_openai(AsyncOpenAI(api_key="sk-..."))
response = await async_client.chat.completions.create(...)

# Streaming
for chunk in client.chat.completions.create(..., stream=True):
    print(chunk.choices[0].delta.content, end="")

Anthropic

import gateforge
from anthropic import Anthropic

client = gateforge.wrap_anthropic(Anthropic(api_key="sk-ant-..."))
response = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=512,
    messages=[{"role": "user", "content": "Hello!"}],
)

Gemini

import gateforge
from google import genai

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!"}]}],
)

Agent Tracing

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"):
    # Each LLM call gets auto-incremented step number
    r1 = client.chat.completions.create(...)  # step 1
    r2 = client.chat.completions.create(...)  # step 2

Agent with Tool Calls

import gateforge

@gateforge.tool()
def search_flights(destination: str, date: str) -> list:
    ...

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

@gateforge.agent()
def travel_agent(request: str) -> str:
    flights = search_flights("Paris", "2026-06-10")
    confirmation = book_flight(flights[0]["id"], "John Doe")
    return f"Booked: {confirmation}"

# All tool calls automatically traced
with gateforge.session(user_id="user-123"):
    response = travel_agent("Book me a flight to Paris")

Session Management

from gateforge import SessionManager, get_current_conversation_id

manager = SessionManager()

# Create session
session = manager.create_session(user_id="user-123")

# Get current conversation from active trace
cid = get_current_conversation_id()

# List sessions
sessions = manager.list_sessions(user_id="user-123", limit=10)

# Serialize for persistence
data = manager.to_dict()  # Save to DB
manager2 = SessionManager.from_dict(data)  # Load

A/B Testing

from gateforge import CallOptions

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",  # Deterministic variant
    ),
)
# Variant A or B injected automatically

Guardrails

from gateforge import CallOptions, GuardrailBlocked

try:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "..."}],
        gateforge_options=CallOptions(guardrails=True),
    )
except GuardrailBlocked as e:
    print(f"Blocked by rule: {e.rule_id}")

PII Protection

PII protection only applies to clients returned by gateforge.wrap(...) or a provider-specific wrapper. Creating an ordinary OpenAI, Anthropic or Gemini client after initialization does not intercept its calls.

Detected Entities

Direct Anonymization

import gateforge

result = gateforge.anonymize("My email is john@example.com")
print(result["sanitized"])  # "My email is [EMAIL_001]"
print(result["entities"])   # ["EMAIL"]

original = gateforge.rehydrate("[EMAIL_001]", context=result["context"])
print(original)  # "john@example.com"

Privacy and data flow

The wrapped model provider always receives the anonymized prompt required to produce a response. PII detection is not infallible; validate the selected backend and locale against your data before relying on it for compliance.

CallOptions Reference

from gateforge import CallOptions

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    gateforge_options=CallOptions(
        # Trace grouping
        conversation_id="conv_abc",
        
        # A/B testing
        experiment_id="exp_abc",
        session_id="user_123",
        
        # Feature overrides
        pii=True,
        guardrails=True,
        track_cost=False,
    ),
)

API Reference

Initialization

Wrapping

Tracing

Session Management

Prompts

Utilities

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

https://app.gateforge.dev/dashboard

• Request volume and trends
• Cost breakdown by model/provider
• Latency analytics
• PII detection statistics
• A/B experiment results
• Guardrail violation alerts
• Agent waterfall traces
• API key management

Changelog

See CHANGELOG.md for release notes.

Documentation

• README and API overview
• Changelog
• Contributing
• Security policy

Troubleshooting

ImportError: No module named 'gateforge'

pip install gateforge-sdk

RuntimeError: Call gateforge.init() first

# Fully local
gateforge.init(config={"features": {"pii_enabled": True}})

# Or from GATEFORGE_API_KEY + GATEFORGE_BASE_URL
gateforge.auto_init()

PII not detected

1. Check domain setting matches your data
2. Add custom patterns in dashboard

Steps not appearing in trace

1. Ensure tracing_enabled=True
2. Confirm conversation_id is active

Links

• Website
• Dashboard
• PyPI
• Issues

License

MIT — see LICENSE.

The SDK is free software and has no runtime dependency on any Gateforge service: it runs the full pipeline locally and sends telemetry only to the destination you configure, if any. Gateforge also offers a hosted control plane and dashboard as a commercial product; using it is optional and never required to run this library.