OpenAI Agents + Tork Governance

I/O Governance

Validate all inputs and outputs

Tool Safety

Govern tool arguments for PII

Streaming

Works with streaming responses

Compliance

A local receipt per governed call

pip install tork-governance openai-agents

from agents import Agent, Runner
from tork_governance.adapters.openai_agents import TorkOpenAIAgentsMiddleware

# On-device governance: PII detection and the decision run on this machine.
# No API key is needed. agent_id is recorded on the middleware's receipts.
middleware = TorkOpenAIAgentsMiddleware(agent_id="research-assistant")

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

# Govern the input, run the agent with the SDK's own Runner, govern the output
input_result = middleware.process_input("What are the latest developments in AI safety?")
run = Runner.run_sync(agent, input_result.output)
output_result = middleware.process_output(str(run.final_output))

print(output_result.output)                     # governed (redacted if PII) text
print(output_result.receipt.receipt_id)         # local receipt for this decision
print(middleware.receipts)                      # [{type, agent_id, receipt_id, action}, ...]

from agents import Agent, Runner
from tork_governance import Tork, GovernanceAction
from tork_governance.adapters.openai_agents import TorkOpenAIAgentsMiddleware

# By default PII is REDACTED and the text goes through. To block instead,
# build the middleware on a Tork whose default_action is DENY.
strict = Tork(default_action=GovernanceAction.DENY)
middleware = TorkOpenAIAgentsMiddleware(tork=strict, agent_id="customer-service")

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

def safe_run(user_input: str) -> str:
    input_result = middleware.process_input(user_input)
    if input_result.action == GovernanceAction.DENY:
        # PII in the user's message and the policy is DENY
        return "I cannot process that request. Please rephrase without personal information."

    run = Runner.run_sync(agent, input_result.output)

    output_result = middleware.process_output(str(run.final_output))
    if output_result.action == GovernanceAction.DENY:
        return "I apologize, but I cannot provide that information."
    return output_result.output

response = safe_run("How do I reset my password?")   # allowed
response = safe_run("My SSN is 123-45-6789")         # denied: PII detected

from agents import Agent, function_tool
from tork_governance import GovernanceAction
from tork_governance.adapters.openai_agents import TorkOpenAIAgentsMiddleware

middleware = TorkOpenAIAgentsMiddleware(agent_id="tool-agent")

# check_tool_call() governs the string "<tool_name>: <tool_args>" for PII and
# returns a GovernanceResult. It never raises, and it has no notion of a
# "dangerous" tool — that judgement stays in your code.
def tool_args_allowed(tool_name: str, tool_args: dict) -> bool:
    result = middleware.check_tool_call(tool_name, tool_args)
    return result.action != GovernanceAction.DENY

@function_tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to a customer."""
    # Redact PII from the body before it leaves your system
    governed_body = middleware.process_output(body).output
    return f"Email sent to {to}: {governed_body}"

@function_tool
def execute_sql(query: str) -> str:
    """Execute a SQL query on the database."""
    if not tool_args_allowed("execute_sql", {"query": query}):
        return "Blocked: query contained personal data"
    return f"Executed: {query}"

agent = Agent(
    name="tool-agent",
    instructions="Use the tools to help the customer.",
    tools=[send_email, execute_sql],
)

# Every check is receipted on the middleware
for entry in middleware.receipts:
    print(entry["type"], entry.get("tool_name"), entry["action"], entry["receipt_id"])

Tool deny-lists are yours to keep

# There is no built-in list of blocked tool names in tork-governance 0.26.1.
# check_tool_call(tool_name, tool_args) governs the text "<name>: <args>" for
# PII and returns a GovernanceResult; it does not judge the tool itself.
# If you need a deny-list, keep it in your code:

DANGEROUS_TOOLS = {"shell", "exec", "eval", "subprocess", "rm", "drop_table"}

def tool_allowed(tool_name: str) -> bool:
    return tool_name not in DANGEROUS_TOOLS

from tork_governance.adapters.openai_agents import TorkOpenAIAgentsMiddleware

middleware = TorkOpenAIAgentsMiddleware(agent_id="multi-agent-system")

# GovernedRunner.run(agent, text) governs the input, then calls agent.run(text)
# on the object you pass, then governs the output. It therefore only works with
# an agent object that exposes a run() method. The OpenAI Agents SDK's Agent
# does not — Runner.run_sync(agent, text) is the SDK's entry point — so with
# that SDK the runner falls through to a placeholder string, not a model call.
# Use process_input()/process_output() around Runner.run_sync instead, as shown
# in the middleware example above.
runner = middleware.create_governed_runner()

class EchoAgent:
    """Any object with run(text) -> str works with GovernedRunner."""
    def run(self, text: str) -> str:
        return f"echo: {text}"

result = runner.run(EchoAgent(), "Contact me at jane@example.com")
print(result)   # "echo: Contact me at [EMAIL_REDACTED]"

from tork_governance import GovernanceAction
from tork_governance.adapters.openai_agents import TorkOpenAIAgentsMiddleware

middleware = TorkOpenAIAgentsMiddleware(agent_id="manual-governance")

# process_input() / process_output() each return a tork_governance.GovernanceResult:
#   .action   GovernanceAction (ALLOW / REDACT / DENY / ESCALATE)
#   .output   the text to use (PII replaced when action is REDACT)
#   .pii      PIIResult: has_pii, types, count, matches
#   .receipt  Receipt minted on this machine

def process_user_message(message: str) -> str:
    result = middleware.process_input(message)
    if result.action == GovernanceAction.DENY:
        raise ValueError(f"Input blocked: {result.receipt.receipt_id}")
    if result.action == GovernanceAction.REDACT:
        print(f"Redacted PII types: {[t.value for t in result.pii.types]}")
    return result.output

def process_agent_response(response: str) -> dict:
    result = middleware.process_output(response)
    if result.action == GovernanceAction.DENY:
        return {"text": "Response blocked by safety policies.", "blocked": True}
    return {
        "text": result.output,
        "receipt_id": result.receipt.receipt_id,
        "input_hash": result.receipt.input_hash,
        "output_hash": result.receipt.output_hash,
    }

processed = process_user_message("Hello, how are you?")
# ^ ALLOW: returns the original text

processed = process_user_message("My credit card is 4111-1111-1111-1111")
# ^ REDACT (the default): returns "My credit card is [CREDIT_CARD_REDACTED]"

response = process_agent_response("Reach me at jane@example.com")
# ^ REDACT: text has the email replaced; receipt fields are returned

from agents import Agent, Runner
from tork_governance import GovernanceAction
from tork_governance.adapters.openai_agents import TorkOpenAIAgentsMiddleware

middleware = TorkOpenAIAgentsMiddleware(agent_id="streaming-agent")

agent = Agent(name="assistant", instructions="You are a helpful assistant.")

async def stream_with_governance(user_input: str):
    """Govern the input before streaming and the full output after it."""
    input_result = middleware.process_input(user_input)
    if input_result.action == GovernanceAction.DENY:
        yield "Error: input blocked by governance policy"
        return

    # Stream with the SDK's own streaming runner. Tokens reach the caller
    # before the output check runs — the check cannot retract them.
    collected = []
    run = Runner.run_streamed(agent, input_result.output)
    async for event in run.stream_events():
        if event.type == "raw_response_event" and hasattr(event.data, "delta"):
            collected.append(event.data.delta)
            yield event.data.delta

    output_result = middleware.process_output("".join(collected))
    print(f"Receipt: {output_result.receipt.receipt_id} ({output_result.action.value})")

from agents import Agent, Runner
from tork_governance.adapters.openai_agents import TorkOpenAIAgentsMiddleware

# One middleware, one policy version, shared receipts for the whole team
middleware = TorkOpenAIAgentsMiddleware(agent_id="customer-support-team")

triage_agent = Agent(
    name="triage",
    instructions="""You are a triage agent. Reply with exactly one category
    for the customer request: billing, technical, or other.""",
)

billing_agent = Agent(
    name="billing",
    instructions="You handle billing questions. Never share card or bank details.",
)

technical_agent = Agent(
    name="technical",
    instructions="You handle technical support. Never share internal architecture.",
)

def governed(agent: Agent, text: str) -> str:
    """Govern in, run, govern out — the same three lines for every agent."""
    input_result = middleware.process_input(text)
    run = Runner.run_sync(agent, input_result.output)
    return middleware.process_output(str(run.final_output)).output

def handle_customer_request(request: str) -> str:
    category = governed(triage_agent, request).lower().strip()

    if "billing" in category:
        return governed(billing_agent, request)
    if "technical" in category:
        return governed(technical_agent, request)
    return governed(triage_agent, f"Handle directly: {request}")

print(handle_customer_request("I need help with my invoice from last month"))

from tork_governance import Tork, GovernanceAction, GovernanceResult, Receipt
from tork_governance.adapters.openai_agents import (
    TorkOpenAIAgentsMiddleware,  # process_input / process_output / check_tool_call
    GovernedOpenAIAgent,         # Wrapper; needs an agent object with run()
    GovernedRunner,              # Runner; needs an agent object with run()
)