gateplex-python 0.2.0

Official Python SDK for the Gateplex AI observability platform

gateplex-python

Official Python SDK for Gateplex — AI agent observability, guardrails, and monitoring.

---

Installation

pip install gateplex-python

Optional extras for LLM auto-patching:

pip install "gateplex-python[openai]"       # OpenAI support
pip install "gateplex-python[anthropic]"    # Anthropic support
pip install "gateplex-python[all]"          # Everything

---

Quick start

from gateplex import GateplexClient

client = GateplexClient(api_key="gplx_your_key_here")

response = client.log_intercept(
    agent_id="my-agent-uuid",
    event_type="llm_call",       # "llm_call" | "tool_call" | "guardrail_trigger"
    input="What is the capital of France?",
    output="Paris.",
    model="gpt-4o",
    latency_ms=342,
    flagged=False,
    metadata={"user_id": "u_123"},
)

print(response.intercept_id)   # uuid
print(response.flagged)        # bool

---

Features

Automatic latency measurement — context manager

with client.capture(
    agent_id="my-agent",
    event_type="llm_call",
    input=user_prompt,
    model="gpt-4o",
) as ctx:
    ctx.output = call_my_llm(user_prompt)   # latency measured automatically

@monitor decorator

from gateplex import monitor

@monitor(client, agent_id="my-agent", event_type="llm_call", model="gpt-4o")
def call_llm(input: str) -> str:
    return openai_client.chat.completions.create(...)

Background batch logging

Fire-and-forget for high-throughput pipelines:

# Enqueued immediately, sent by a background thread
client.batch_intercept(
    agent_id="my-agent",
    event_type="tool_call",
    input='search("latest AI news")',
    output='["OpenAI launches...", ...]',
    model="gpt-4o",
)

# Force flush before process exit
client.flush()

OpenAI auto-patcher

Automatically intercepts every chat.completions.create call:

import openai
from gateplex import GateplexClient
from gateplex import OpenAIPatcher

gp = GateplexClient(api_key="gplx_...")
oai = openai.OpenAI(api_key="sk-...")

patcher = OpenAIPatcher(gp, agent_id="my-agent", openai_client=oai)
patcher.patch()

# Use oai normally — all calls are logged
response = oai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)

patcher.unpatch()   # remove when done

Anthropic auto-patcher

import anthropic
from gateplex import GateplexClient
from gateplex import AnthropicPatcher

gp = GateplexClient(api_key="gplx_...")
ant = anthropic.Anthropic(api_key="sk-ant-...")

patcher = AnthropicPatcher(gp, agent_id="my-agent", anthropic_client=ant)
patcher.patch()

message = ant.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)

patcher.unpatch()

---

Public REST endpoints

# List intercepts (paginated)
intercepts = client.get_intercepts(limit=20, offset=0, agent_id="my-agent")

# Create an intercept via the public REST API
result = client.create_intercept(
    agent_id="my-agent",
    event_type="llm_call",
    input="Hello",
    output="Hi!",
    model="gpt-4o",
)

# Fetch guardrail definitions
guardrails = client.get_guardrails()
for g in guardrails:
    print(g.name, g.enabled)

---

Configuration

Parameter	Default	Description
api_key	—	Your gplx_... API key (required)
timeout	10.0	HTTP timeout in seconds
max_retries	3	Retries with exponential back-off (on 429/5xx)
batch_size	50	Max events in background queue before forced flush
flush_interval	5.0	Seconds between background flushes
raise_on_error	False	Set True to surface API errors instead of logging them

client = GateplexClient(
    api_key="gplx_...",
    timeout=5.0,
    max_retries=5,
    flush_interval=2.0,
    raise_on_error=True,   # great for dev/test
)

---

Error handling

from gateplex import GateplexAuthError, GateplexAPIError

try:
    client.log_intercept(...)
except GateplexAuthError:
    print("Check your API key")
except GateplexAPIError as e:
    print(f"API error: {e}")

---

License

MIT