OpenAI Agents + Tork Governance

I/O Governance

Validate all inputs and outputs

Tool Safety

Block dangerous tool calls

Streaming

Works with streaming responses

Compliance

Automatic audit receipts

pip install tork-governance openai-agents

from openai_agents import Agent, Runner
from tork.adapters.openai_agents import TorkOpenAIAgentsMiddleware
from tork.core.engine import GovernanceEngine

# Initialize governance engine
engine = GovernanceEngine(api_key="your_tork_api_key")

# Create middleware
middleware = TorkOpenAIAgentsMiddleware(
    engine=engine,
    agent_id="my-openai-agent"
)

# Create an OpenAI Agent
agent = Agent(
    name="research-assistant",
    instructions="You are a helpful research assistant.",
    model="gpt-4"
)

# Wrap with governance
governed_agent = middleware.wrap_agent(agent)

# Run with governance - inputs and outputs are evaluated
result = governed_agent.run("What are the latest developments in AI safety?")

print(result["output"])  # Governed output
print(result["receipt"])  # Compliance receipt

from openai_agents import Agent
from tork.adapters.openai_agents import TorkOpenAIAgentsMiddleware, GovernedOpenAIAgent
from tork.adapters.openai_agents.exceptions import InputBlockedError, OutputBlockedError

# Create middleware
middleware = TorkOpenAIAgentsMiddleware(
    api_key="your_tork_api_key",
    agent_id="customer-service"
)

# Create and wrap agent
agent = Agent(
    name="support-bot",
    instructions="""You are a customer support assistant.
    Help customers with their questions professionally.
    Never share internal company information.""",
    model="gpt-4"
)

governed_agent = GovernedOpenAIAgent(agent, middleware)

# Handle governance violations gracefully
def safe_run(user_input: str) -> str:
    try:
        result = governed_agent.run(user_input)
        return result["output"]

    except InputBlockedError as e:
        # User input violated policy (e.g., PII, jailbreak attempt)
        return "I cannot process that request. Please rephrase without personal information."

    except OutputBlockedError as e:
        # Agent output violated policy
        return "I apologize, but I cannot provide that information."

# Usage
response = safe_run("How do I reset my password?")  # Allowed
response = safe_run("My SSN is 123-45-6789")        # Blocked - PII detected

from openai_agents import Agent, function_tool
from tork.adapters.openai_agents import TorkOpenAIAgentsMiddleware
from tork.adapters.openai_agents.exceptions import ToolCallBlockedError

middleware = TorkOpenAIAgentsMiddleware(
    api_key="your_tork_api_key",
    agent_id="tool-agent"
)

# Define tools for the agent
@function_tool
def search_database(query: str) -> str:
    """Search the customer database."""
    return f"Found results for: {query}"

@function_tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a customer."""
    return f"Email sent to {to}"

@function_tool
def execute_sql(query: str) -> str:
    """Execute a SQL query on the database."""
    # This would run SQL - governance checks the query
    return f"Executed: {query}"

# Validate tool calls before execution
def safe_tool_call(tool_name: str, tool_args: dict) -> dict:
    try:
        # Check if tool call is allowed
        result = middleware.check_tool_call(tool_name, tool_args)
        return {"allowed": True, "args": tool_args}

    except ToolCallBlockedError as e:
        return {"allowed": False, "reason": str(e)}

# Examples
safe_tool_call("search_database", {"query": "customer@email.com"})
# ^ Allowed

safe_tool_call("execute_sql", {"query": "DROP TABLE users;"})
# ^ Blocked - destructive SQL

safe_tool_call("shell", {"command": "rm -rf /"})
# ^ Blocked - dangerous tool name

# Built-in dangerous tools are blocked automatically:
# shell, exec, eval, subprocess, os.system, run_command,
# execute_code, file_delete, rm, drop_table

Blocked Tools (Default)

# Built-in dangerous tools that are automatically blocked:

DANGEROUS_TOOLS = {
    "shell",           # Shell command execution
    "exec",            # Code execution
    "eval",            # Expression evaluation
    "subprocess",      # Subprocess spawning
    "os.system",       # OS command execution
    "run_command",     # Generic command runner
    "execute_code",    # Code execution
    "file_delete",     # File deletion
    "rm",              # Remove files
    "drop_table",      # Database table deletion
}

# These are blocked by default. Configure additional
# restrictions via the Tork dashboard or API.

from openai_agents import Agent
from tork.adapters.openai_agents import TorkOpenAIAgentsMiddleware, GovernedRunner

# Create middleware
middleware = TorkOpenAIAgentsMiddleware(
    api_key="your_tork_api_key",
    agent_id="multi-agent-system"
)

# Create a governed runner for multiple agents
runner = middleware.create_governed_runner()

# Define multiple agents
research_agent = Agent(
    name="researcher",
    instructions="You research topics thoroughly.",
    model="gpt-4"
)

writer_agent = Agent(
    name="writer",
    instructions="You write clear, engaging content.",
    model="gpt-4"
)

editor_agent = Agent(
    name="editor",
    instructions="You review and improve content.",
    model="gpt-4"
)

# Run any agent with governance using the runner
research_result = runner.run(
    research_agent,
    "Research the latest trends in sustainable energy"
)

# Pass research to writer
writer_result = runner.run(
    writer_agent,
    f"Write an article based on this research: {research_result['output']}"
)

# Pass to editor
final_result = runner.run(
    editor_agent,
    f"Edit and improve this article: {writer_result['output']}"
)

# All agent outputs include governance results and receipts
print(final_result["output"])
print(f"Receipts generated: {final_result['receipt'].id}")

from tork.adapters.openai_agents import TorkOpenAIAgentsMiddleware
from tork.adapters.openai_agents.exceptions import InputBlockedError, OutputBlockedError

middleware = TorkOpenAIAgentsMiddleware(
    api_key="your_tork_api_key",
    agent_id="manual-governance"
)

# Manual input processing (for custom agent implementations)
def process_user_message(message: str) -> str:
    """Process user input with governance."""
    try:
        result = middleware.process_input(message)

        # Check what happened
        if result["result"].decision.name == "ALLOW":
            return result["text"]  # Original or unmodified
        elif result["result"].decision.name == "REDACT":
            # PII was redacted
            print(f"Redacted content: {result['result'].violations}")
            return result["text"]  # Modified text with redactions

    except InputBlockedError as e:
        raise ValueError(f"Input blocked: {e}")

# Manual output processing
def process_agent_response(response: str, tool_calls: list = None) -> dict:
    """Process agent output with governance."""
    try:
        result = middleware.process_output(response, tool_calls)

        return {
            "text": result["text"],
            "receipt_id": result["receipt"].id,
            "receipt_hash": result["receipt"].hash,
        }

    except OutputBlockedError as e:
        return {
            "text": "Response blocked by safety policies.",
            "blocked": True,
            "reason": str(e),
        }

# Usage
processed_input = process_user_message("Hello, how are you?")
# ^ Allowed, returns original text

processed_input = process_user_message("My credit card is 4111-1111-1111-1111")
# ^ Redacted, returns "[REDACTED]"

response = process_agent_response("Here is the confidential data...")
# ^ May be blocked or redacted based on policies

from openai_agents import Agent
from tork.adapters.openai_agents import TorkOpenAIAgentsMiddleware
from tork.adapters.openai_agents.exceptions import InputBlockedError

middleware = TorkOpenAIAgentsMiddleware(
    api_key="your_tork_api_key",
    agent_id="streaming-agent"
)

agent = Agent(
    name="assistant",
    instructions="You are a helpful assistant.",
    model="gpt-4"
)

governed_agent = middleware.wrap_agent(agent)

def stream_with_governance(user_input: str):
    """Stream agent response with governance checks."""

    # Pre-validate input before streaming
    try:
        input_result = middleware.process_input(user_input)
        processed_input = input_result["text"]
    except InputBlockedError as e:
        yield f"Error: {e}"
        return

    # Stream the response (using native OpenAI streaming)
    collected_output = []

    # If using OpenAI's streaming API directly:
    for chunk in agent.stream(processed_input):
        collected_output.append(chunk)
        yield chunk

    # Post-validate complete output
    full_output = "".join(collected_output)
    output_result = middleware.process_output(full_output)

    # Log the compliance receipt
    print(f"Receipt: {output_result['receipt'].id}")

# Usage
for chunk in stream_with_governance("Explain quantum computing"):
    print(chunk, end="", flush=True)

from openai_agents import Agent
from tork.adapters.openai_agents import TorkOpenAIAgentsMiddleware

# Shared middleware for all agents
middleware = TorkOpenAIAgentsMiddleware(
    api_key="your_tork_api_key",
    agent_id="customer-support-team"
)

# Create specialized agents
triage_agent = Agent(
    name="triage",
    instructions="""You are a triage agent. Analyze customer requests and
    route them to the appropriate specialist. Categories:
    - billing: Payment and invoice questions
    - technical: Product issues and bugs
    - sales: Pricing and upgrades""",
    model="gpt-4"
)

billing_agent = Agent(
    name="billing",
    instructions="""You handle billing questions. Never share:
    - Full credit card numbers
    - Bank account details
    - Internal pricing formulas""",
    model="gpt-4"
)

technical_agent = Agent(
    name="technical",
    instructions="""You handle technical support. Never share:
    - Internal system architecture
    - Security vulnerabilities
    - Database schemas""",
    model="gpt-4"
)

# Wrap all agents
governed_triage = middleware.wrap_agent(triage_agent)
governed_billing = middleware.wrap_agent(billing_agent)
governed_technical = middleware.wrap_agent(technical_agent)

def handle_customer_request(request: str) -> str:
    """Route and handle customer request with governance."""

    # Triage the request
    triage_result = governed_triage.run(
        f"Analyze this request and respond with just the category: {request}"
    )
    category = triage_result["output"].lower().strip()

    # Route to specialist
    if "billing" in category:
        result = governed_billing.run(request)
    elif "technical" in category:
        result = governed_technical.run(request)
    else:
        result = governed_triage.run(f"Handle directly: {request}")

    return result["output"]

# Usage
response = handle_customer_request("I need help with my invoice from last month")
print(response)

from tork.adapters.openai_agents import (
    TorkOpenAIAgentsMiddleware,  # Central middleware
    GovernedOpenAIAgent,         # Wrapped agent
    GovernedRunner,              # Multi-agent runner
    OpenAIAgentGovernanceError,  # Base exception
    InputBlockedError,           # Input blocked
    OutputBlockedError,          # Output blocked
    ToolCallBlockedError,        # Tool call blocked
)