Skip to main content

QWED Open Responses

Verify AI agent outputs before execution. PyPI version Tests License

What is QWED Open Responses?

QWED Open Responses provides deterministic verification guards for AI agent outputs. It works with:
  • OpenAI Responses API
  • LangChain agents
  • LlamaIndex
  • Any AI framework
Use it when you need runtime tool call verification, AI safety guardrails, and policy enforcement before an agent executes an action.

The Problem

When AI agents execute tools or generate structured outputs, they can:
  • 🔧 Call dangerous functions - rm -rf /, DROP TABLE
  • 🧮 Produce incorrect calculations - Financial errors, wrong totals
  • 📋 Violate business rules - Invalid state transitions
  • 🔐 Leak sensitive data - PII, API keys in responses
  • 💰 Exceed budgets - Unlimited API calls

The Solution

QWED Open Responses intercepts and verifies every agent output before execution:
AI Agent Output → Guards → Verified? → Execute

                    YES ───┘
                    NO ────→ Block + Error

How It Works

┌─────────────────────────────────────────────────────────────────┐
│                    AI Agent (GPT, Claude, etc.)                  │
│                                                                  │
│  "Call calculator with x=150, y=10, result=1600"                 │
└──────────────────────────┬───────────────────────────────────────┘

                           │ Tool Call / Structured Output

┌─────────────────────────────────────────────────────────────────┐
│                QWED Open Responses Verifier                      │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌─────────────┐  ┌─────────────┐  ┌───────────────────────────┐│
│  │ SchemaGuard │  │  ToolGuard  │  │       MathGuard          ││
│  │  JSON Valid │  │  Blocklist  │  │  150 × 10 ≠ 1600 ❌      ││
│  └─────────────┘  └─────────────┘  └───────────────────────────┘│
│                                                                  │
│  ┌─────────────┐  ┌─────────────┐  ┌───────────────────────────┐│
│  │ StateGuard  │  │ArgumentGuard│  │      SafetyGuard         ││
│  │ Transitions │  │ Type Check  │  │  PII, Injection, Budget  ││
│  └─────────────┘  └─────────────┘  └───────────────────────────┘│
│                                                                  │
│               MathGuard Failed: 150 × 10 = 1500, not 1600       │
│                                                                  │
└──────────────────────────┬───────────────────────────────────────┘


                  ┌─────────────────┐
                  │  ❌ BLOCKED     │
                  │  Return error   │
                  └─────────────────┘

The 6 Guards

GuardWhat It VerifiesExample Catch
SchemaGuardJSON Schema complianceMissing required field
ToolGuardBlock dangerous tool callsexecute_shell blocked
MathGuardVerify calculations150 × 10 ≠ 1600
StateGuardValid state transitionscompleted → pending invalid
ArgumentGuardTool argument validationamount: "abc" not a number
SafetyGuardPII, injection, budgetSSN detected in output

Installation

Basic

pip install qwed-open-responses

With Framework Integrations

# OpenAI
pip install qwed-open-responses[openai]

# LangChain
pip install qwed-open-responses[langchain]

# All integrations
pip install qwed-open-responses[all]

Quick Start

Basic Verification

from qwed_open_responses import ResponseVerifier
from qwed_open_responses.guards import ToolGuard, MathGuard, SafetyGuard

verifier = ResponseVerifier()

# Verify a tool call before execution
result = verifier.verify_tool_call(
    tool_name="calculator",
    arguments={
        "operation": "multiply",
        "x": 150,
        "y": 10,
        "result": 1500  # Correct!
    },
    guards=[ToolGuard(), MathGuard(), SafetyGuard()]
)

if result.verified:
    print("✅ Safe to execute")
    execute_tool(result.tool_name, result.arguments)
else:
    print(f"❌ Blocked: {result.block_reason}")
    print(f"   Failed guard: {result.failed_guard}")

Verify Structured Output

from qwed_open_responses.guards import SchemaGuard

# Define expected schema
order_schema = {
    "type": "object",
    "required": ["order_id", "total", "items"],
    "properties": {
        "order_id": {"type": "string"},
        "total": {"type": "number", "minimum": 0},
        "items": {"type": "array", "minItems": 1}
    }
}

result = verifier.verify_structured_output(
    output={
        "order_id": "ORD-123",
        "total": 99.99,
        "items": [{"name": "Widget", "price": 99.99}]
    },
    guards=[SchemaGuard(schema=order_schema)]
)

Framework Integration

LangChain

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from qwed_open_responses.middleware.langchain import QWEDCallbackHandler

# Create callback with guards
callback = QWEDCallbackHandler(
    guards=[ToolGuard(), SafetyGuard()],
    block_on_failure=True,  # Stop execution if guard fails
)

# Add to agent
llm = ChatOpenAI(model="gpt-4")
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[callback]
)

# Every tool call is now verified!
result = executor.invoke({"input": "Calculate 25% of 500"})

OpenAI SDK

from qwed_open_responses.middleware.openai_sdk import VerifiedOpenAI
from qwed_open_responses.guards import SchemaGuard, SafetyGuard

# Create verified client
client = VerifiedOpenAI(
    api_key="sk-...",
    guards=[
        SchemaGuard(schema=my_schema),
        SafetyGuard(block_pii=True)
    ]
)

# Use normally - verification is automatic
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Generate an order"}],
    tools=my_tools
)

# Tool calls are verified before returning
for tool_call in response.choices[0].message.tool_calls:
    print(f"Verified tool call: {tool_call.function.name}")

Streaming middleware

The OpenResponsesMiddleware intercepts streaming events from the Open Responses protocol, verifying tool_call and function_call items through your guard stack before they are yielded to the consumer.

Usage

from qwed_open_responses.middleware.streaming_interceptor import OpenResponsesMiddleware
from qwed_open_responses.guards import ToolGuard, SafetyGuard

middleware = OpenResponsesMiddleware(
    guards=[ToolGuard(), SafetyGuard()],
    block_on_failure=True,       # Replace blocked items with system_intervention
    on_blocked=my_callback,      # Optional callback when an item is blocked
)

# Wrap any async stream of Open Responses items
async for item in middleware.verify_stream(response_stream):
    process(item)

# Check runtime stats
print(middleware.get_stats())
# {"total": 12, "verified": 10, "blocked": 2}

Parameters

ParameterTypeDefaultDescription
guardslist[BaseGuard][]Guards to run against each tool-call item
block_on_failureboolTrueWhen True, blocked items are replaced with a system_intervention item. When False, blocked items pass through unmodified (warn-only)
on_blockedcallableNoneOptional callback invoked with (item, result) when a tool call is blocked

How it works

  1. Non-tool items (text, metadata) pass through unchanged
  2. Items with type of tool_call or function_call are verified against all guards
  3. If verification passes, the original item is yielded
  4. If verification fails and block_on_failure is True, a system_intervention item is yielded instead:
{
  "type": "system_intervention",
  "status": "blocked",
  "tool_name": "execute_shell",
  "reason": "QWED blocked execute_shell: dangerous tool call",
  "verification": {
    "guards_passed": ["SafetyGuard"],
    "guards_failed": ["ToolGuard"],
    "mechanism": "QWED Open Responses Streaming Interceptor"
  }
}

Stats methods

MethodDescription
get_stats()Returns {"total": int, "verified": int, "blocked": int}
reset_stats()Resets all counters to zero

Why QWED Open Responses?

Security Comparison

ThreatWithout VerificationWith QWED
Agent calls rm -rf /💀 System destroyedBLOCKED
SQL injection in query💀 Data breachBLOCKED
Wrong calculation💸 Financial lossCAUGHT
PII in API response📋 Compliance violationDETECTED
Infinite tool loop💰 $10,000 API billBUDGET GUARD

Real-World Impact

  • Finance: Prevent wrong calculations in trading bots
  • Healthcare: Block PII leaks in patient summaries
  • E-commerce: Verify order totals before payment
  • DevOps: Prevent dangerous shell commands

Configuration

Environment Variables

VariableDescriptionDefault
QWED_OR_LOG_LEVELLogging levelINFO
QWED_OR_STRICTFail on any guard failuretrue
QWED_OR_MAX_BUDGETMaximum API cost allowed100.0

Custom Guard Configuration

from qwed_open_responses.guards import ToolGuard, SafetyGuard

# Custom tool blocklist
tool_guard = ToolGuard(
    blocklist=["execute_shell", "delete_database", "send_email"],
    allow_unknown=False  # Block tools not in whitelist
)

# Custom safety settings
safety_guard = SafetyGuard(
    block_pii=True,
    block_injection=True,
    max_budget=50.0,  # $50 limit
    harmful_patterns=["password", "secret", "token"]
)

verifier = ResponseVerifier(guards=[tool_guard, safety_guard])

Next Steps