Gateforge SDK (Python)

Observability for LLM applications and agents — tracing, cost and latency metrics on every call, tool and agent step, with optional local PII anonymization, guardrails, A/B testing and prompt management on top.

What it does

Gateforge answers the questions you cannot answer by reading your own code: what did this agent actually do, which step was slow, which call cost money, which tool ran with what arguments.

You wrap your existing provider client (OpenAI, Anthropic, Gemini). Nothing else in your code changes. From that point on every call, tool and agent step becomes a span carrying tokens, cost and latency:

import gateforge
from openai import OpenAI

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

with gateforge.trace(conversation_id="conv-1"):
    client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Hello!"}],
    )
# {"model": "gpt-4o-mini", "prompt_tokens": 8, "completion_tokens": 12,
#  "cost_usd": 0.0000084, "latency_ms": 640.2, "conversation_id": "conv-1", ...}

The core, and the modules

Always on once you wrap a client:

Capability	What you get
Tracing	A span per LLM call, tool call and agent step, grouped into conversations and sessions. `@gateforge.tool()`, `@gateforge.agent()`, `with gateforge.trace(...)`.
Metrics	Tokens, USD cost and latency on every call — sync, async, streaming.
Delivery	Console, file, HTTP, OpenTelemetry `gen_ai.*`, your own callback, or several at once. Nothing is transmitted until you name a destination.

Optional, each independently switchable — none required for the core to work:

Module	What it does	Turn it off with
PII anonymization	Detects and masks PII locally, before the provider sees it, and rehydrates the original values in the response your code receives. It is what makes the tracing above usable in regulated environments — a feature of the observability, not the product.	`pii_enabled: False`
Guardrails	Local input/output rules; a blocked call raises `GuardrailBlocked`. Inert until you define rules.	`guardrails_enabled: False`
A/B prompt testing	Deterministic variant assignment per session with system-prompt injection. Inert until you pass an `experiment_id`.	`ab_enabled: False`
Prompt management	Versioned prompts with variables, composition and a memory/file cache.	don't import it

A tracing-only setup:

gateforge.init(config={"features": {
    "pii_enabled": False,
    "guardrails_enabled": False,
    "ab_enabled": False,
}})

The full pipeline

With every module enabled, each intercepted call runs:

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
    │
    ▼
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

# No account and no API key. Telemetry has no default destination:
# without one, the local pipeline runs and nothing is transmitted.
gateforge.auto_init(enable_pii=True, enable_guardrails=False)

# Wrapping is always explicit — an unwrapped client is not protected.
client = gateforge.wrap_openai(OpenAI(api_key="sk-..."))

To send telemetry somewhere, name the destination in the environment:

export GATEFORGE_TELEMETRY_URL=https://telemetry.internal
export GATEFORGE_TELEMETRY_TOKEN=...            # if your collector needs auth
export GATEFORGE_TELEMETRY_AUTH_HEADER=Authorization   # default: X-API-Key

Variable	Purpose
`GATEFORGE_TELEMETRY_URL`	Base URL of your collector. Unset ⇒ nothing is transmitted.
`GATEFORGE_TELEMETRY_TOKEN`	Value of the auth header, if required.
`GATEFORGE_TELEMETRY_AUTH_HEADER`	Header carrying that token (default `X-API-Key`).
`GATEFORGE_TELEMETRY_METRIC_PATH`	Default `/sdk/metrics`.
`GATEFORGE_TELEMETRY_EVENT_PATH`	Default `/api/traces/events`.
`GATEFORGE_TELEMETRY_VIOLATION_PATH`	Default `/api/guardrails/{rule_id}/violations`.
`GATEFORGE_TELEMETRY_FILE`	Append JSON lines to this path.
`GATEFORGE_TELEMETRY_CONSOLE`	`1` to also print records to stderr.

Set several and every destination receives every record.

Using a Gateforge control plane instead

A control plane is a server that stores your settings — PII rules and entity overrides, guardrail rules, running experiments, feature flags — and serves them to the SDK, which refetches them periodically so changes take effect without a redeploy. It is optional: init(config=...) covers the same ground from a local file.

Variable	Purpose
`GATEFORGE_API_KEY`	Authenticates against a control plane. Setting it also makes that control plane the telemetry destination, unless `GATEFORGE_TELEMETRY_URL` says otherwise.
`GATEFORGE_BASE_URL`	Which control plane to talk to. Defaults to `https://api.gateforge.dev` only when an API key is present — point it at your own deployment to self-host.

Precedence, highest first:

An exporter= passed to init().
GATEFORGE_TELEMETRY_* — your own collector.
GATEFORGE_API_KEY (+ GATEFORGE_BASE_URL) — the control plane doubles as the destination.
Nothing configured — the local pipeline runs and nothing is transmitted.

Without an API key the SDK never contacts any host implicitly: config comes from your local settings and telemetry goes only where you send it.

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

Category	Examples
Personal	Names, emails, phones, addresses
Financial	Credit cards, bank accounts, SSN
Healthcare	Medical records, symptoms, diagnoses
Technical	IP addresses, URLs, API keys
Custom	Your own regex patterns

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"

Telemetry: sending it to your own system

The SDK produces three kinds of record — a metric per LLM call (tokens, cost, latency, PII entity types), a trace event per step of an agent waterfall, and a violation per guardrail that fires. Where they go is entirely your choice: an exporter is just an object with three methods, and nothing is transmitted anywhere until you configure one.

import gateforge

gateforge.init(exporter=gateforge.ConsoleExporter())   # JSON lines to stderr

Built-in destinations

Exporter	Use it for
`NullExporter()`	The default. The local pipeline runs; nothing leaves the process.
`ConsoleExporter()`	Seeing exactly what would be transmitted, before wiring anything.
`FileExporter("telemetry.jsonl")`	Appending JSON lines to disk for batch ingestion.
`HTTPExporter(...)`	POSTing to your own collector.
`CallbackExporter(fn)`	Pushing into an SDK you already use (Datadog, Prometheus, OTel…).
`MultiExporter([...])`	Several at once; one failing does not stop the others.

Your own HTTP collector

HTTPExporter works against any server. Paths, headers and auth are all configurable, so your endpoints do not have to look like ours:

gateforge.init(
    config="gateforge.json",
    exporter=gateforge.HTTPExporter(
        base_url="https://telemetry.internal",
        api_key="...",                                  # omit for an unauthenticated collector
        metric_path="/v1/llm/metrics",
        trace_event_path="/v1/llm/events",
        violation_path="/v1/rules/{rule_id}/hits",      # {rule_id} is substituted
        headers={"X-Team": "platform"},
        auth_header="Authorization",                    # defaults to X-API-Key
    ),
)

Each record arrives as a JSON object. A metric looks like this — note that it carries PII types, never values:

{"model": "gpt-4o", "provider": "openai", "prompt_tokens": 42,
 "completion_tokens": 17, "cost_usd": 0.000465, "latency_ms": 812.0,
 "pii_types": ["PERSON"], "pii_count": 1, "conversation_id": "conv-1",
 "ts": "2026-08-29T12:00:00+00:00"}

Into an existing observability stack

CallbackExporter hands you every record as fn(kind, record), where kind is "metric", "trace_event" or "violation":

from prometheus_client import Counter, Histogram

COST = Counter("llm_cost_usd_total", "LLM spend", ["model"])
LATENCY = Histogram("llm_latency_seconds", "LLM latency", ["model"])

def to_prometheus(kind, record):
    if kind == "metric":
        COST.labels(record["model"]).inc(record["cost_usd"])
        LATENCY.labels(record["model"]).observe(record["latency_ms"] / 1000)

gateforge.init(exporter=gateforge.CallbackExporter(to_prometheus))

Writing your own

Any object with these three methods is a valid destination — no base class to inherit, no registration:

class MyWarehouse:
    def export_metric(self, metric): ...
    def export_trace_event(self, event): ...
    def export_violation(self, violation): ...

gateforge.init(exporter=MyWarehouse())

OpenTelemetry

Independently of the exporter, every LLM call emits an OTel span using the gen_ai semantic conventions whenever your application has a tracer provider configured, so existing OTel pipelines pick it up with no extra wiring. If you do not already configure OTel, gateforge.configure_otel(endpoint=...) sets up a provider for you (pip install gateforge-sdk[otel]).

How delivery behaves

Exporters are called on a small pool of background worker threads, never on the request path, so a slow or broken collector cannot add latency to your LLM call or raise into your code. Records are queued with a bounded buffer; if the queue fills because a destination cannot keep up, records are dropped and a warning is logged rather than growing memory without limit. Pending records are flushed at interpreter exit, so short-lived scripts do not lose their last events.

Content-bearing fields pass a PII egress filter on that worker before reaching any destination. If the filter cannot run, the content is dropped rather than sent.

Privacy and data flow

Configuration	Network behavior
Local config, no exporter	No Gateforge telemetry or config traffic
`base_url` + API key	Fetches remote config and exports telemetry to that URL
`store_content=False` (default)	Telemetry omits content previews
`store_content=True`	Telemetry may include anonymized content previews
LLM quality judge enabled	Sends anonymized excerpts to the configured judge provider

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

Function	Description
`gateforge.init(api_key, ...)`	Manual initialization
`gateforge.auto_init()`	Auto from environment

Wrapping

Function	Description
`gateforge.wrap(client)`	Auto-detect provider
`gateforge.wrap_openai(client)`	Wrap OpenAI
`gateforge.wrap_anthropic(client)`	Wrap Anthropic
`gateforge.wrap_gemini(client)`	Wrap Gemini

Tracing

Function	Description
`gateforge.trace(conversation_id)`	Context manager for traces
`gateforge.continue_session(id)`	Resume existing conversation
`gateforge.session(user_id)`	Session context manager
`gateforge.tool()`	Decorator for tool tracing
`gateforge.agent()`	Decorator for agent tracing

Session Management

Function	Description
`SessionManager()`	Create session manager
`manager.create_session()`	Create new session
`manager.get_session(id)`	Get existing session
`manager.list_sessions()`	List with filters
`get_current_conversation_id()`	Get active trace ID
`get_current_trace_info()`	Get full trace info

Prompts

Function	Description
`Prompt(...)`	Create prompt
`PromptBuilder()`	Compose prompts
`PromptCache()`	Multi-level cache
`gateforge.get_prompt(name)`	Get from cache/backend
`gateforge.set_prompt(prompt)`	Set in cache

Utilities

Function	Description
`gateforge.anonymize(text)`	Anonymize PII
`gateforge.rehydrate(text, ctx)`	Restore PII
`gateforge.track_metrics(data)`	Send metadata

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

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 the environment — no API key needed
gateforge.auto_init()

PII not detected

Check the domain setting matches your data
Install the language model: python -m spacy download en_core_web_sm
Add custom patterns via custom_rules in your config

Steps not appearing in trace

Ensure tracing_enabled=True
Confirm conversation_id is active

Links

License

Source-available, not open source: PolyForm Internal Use License 1.0.0 plus an attribution requirement. See LICENSE for the full terms.

You may, free of charge:

Use and run the SDK for the internal business operations of you and your company.
Read, modify and extend the source for that internal use.

You must:

Keep the copyright and license notices intact.
Credit GateForge visibly in the UI, about screen, or documentation of anything that uses it: Powered by GateForge SDK — https://github.com/neretj/gateforge-sdk

You may not, without a commercial license:

Redistribute, resell or sublicense the SDK.
Host it for third parties or offer its functionality as part of a product or service to anyone outside your company.

Commercial licensing: support@gateforge.dev.

The SDK 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; using it is optional and never required to run this library.

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
gateforge		gateforge
tests		tests
.env.example		.env.example
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
MANIFEST.in		MANIFEST.in
README.md		README.md
SECURITY.md		SECURITY.md
pyproject.toml		pyproject.toml

Latest commit

History

Folders and files

Repository files navigation

Gateforge SDK (Python)

What it does

The core, and the modules

The full pipeline

Installation

Quick Start

Option 1: Fully local (recommended for first use)

Option 2: Hosted or self-hosted control plane

Option 3: Provider auto-detection

Key Features

1. Auto-Initialization

Using a Gateforge control plane instead

2. Decorator-Based Tracing

3. Conversation Management

4. Prompt System

Provider Wrappers

OpenAI

Anthropic

Gemini

Agent Tracing

Basic Multi-Step Trace

Agent with Tool Calls

Session Management

A/B Testing

Guardrails

PII Protection

Detected Entities

Direct Anonymization

Telemetry: sending it to your own system

Built-in destinations

Your own HTTP collector

Into an existing observability stack

Writing your own

OpenTelemetry

How delivery behaves

Privacy and data flow

CallOptions Reference

API Reference

Initialization

Wrapping

Tracing

Session Management

Prompts

Utilities

Supported Models

OpenAI

Anthropic

Google Gemini

Dashboard

Changelog

Documentation

Troubleshooting

Links

License

About

Resources

Contributing

Security policy

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages