TorkTORK
Docs/AutoGen Guide
Microsoft AutoGen Integration

AutoGen + Tork Governance

Add PII guardrails to Microsoft AutoGen conversations. Govern the messages that enter and leave your agents on-device, scan tool results, and keep a local receipt for every decision.

Message Governance

PII redacted on wrapped agents

Local Receipts

A receipt for every governed message

On-Device

No network calls without an api_key

Tool Result Scanning

PII and injection checks on results

Installation

Install Tork with AutoGen dependencies.

bash
pip install tork-governance pyautogen

AutoGen adapters are in tork_governance.adapters.autogen. Governance runs on-device: PII detection and the allow / redact / deny decision are computed locally.

TorkAutoGenMiddleware

Central middleware for governing AutoGen agent conversations.

The middleware wraps an AutoGen agent so that the messages it sends, receives, initiates and generates through the wrapper are governed for PII. Use wrap_agent()to wrap any agent, or govern_message() /process_message() to govern text directly.

pythonmiddleware_example.py
from autogen import AssistantAgent, UserProxyAgent
from tork_governance.adapters.autogen import TorkAutoGenMiddleware

# On-device governance. No api_key = zero network calls.
# Pass api_key="tork_live_..." to also report metadata-only attestations.
middleware = TorkAutoGenMiddleware(agent_id="autogen-conversation")

# Create AutoGen agents
assistant = AssistantAgent(
    name="assistant",
    llm_config={"model": "gpt-4"}
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config={"use_docker": False}
)

# Wrap the agent that starts the conversation
governed_user_proxy = middleware.wrap_agent(user_proxy)

# The initial message is governed (PII redacted, or ValueError when the
# configured action is DENY) before AutoGen sees it.
governed_user_proxy.initiate_chat(
    assistant,
    message="Write a Python function to calculate fibonacci numbers"
)

GovernedAutoGenAgent

The wrapper wrap_agent() returns; construct it directly for explicit control.

GovernedAutoGenAgent governs send(),receive(), initiate_chat() andgenerate_reply(). Denied outgoing messages raise a plainValueError; incoming messages and replies are redacted, never raised on. Every other attribute is delegated to the wrapped agent.

pythongoverned_agent.py
from autogen import AssistantAgent, UserProxyAgent
from tork_governance import Tork, GovernanceAction
from tork_governance.adapters.autogen import TorkAutoGenMiddleware, GovernedAutoGenAgent

# Create standard AutoGen agents
assistant = AssistantAgent(
    name="code_assistant",
    system_message="You are a helpful coding assistant.",
    llm_config={"model": "gpt-4"}
)
user_proxy = UserProxyAgent(name="user", human_input_mode="NEVER")

# DENY raises on blocked messages; the default REDACT replaces PII silently
middleware = TorkAutoGenMiddleware(
    tork=Tork(default_action=GovernanceAction.DENY),
    agent_id="code-assistant"
)
governed_assistant = GovernedAutoGenAgent(assistant, middleware)

try:
    # send() and initiate_chat() govern the outgoing message and raise
    # ValueError when it is denied
    governed_assistant.send("Help me write a web scraper", user_proxy)

    # receive() governs the incoming message (redacts; never raises)
    governed_assistant.receive("My SSN is 123-45-6789", user_proxy)

    # generate_reply() governs the reply before returning it
    reply = governed_assistant.generate_reply(
        messages=[{"content": "Write hello world", "role": "user"}]
    )
except ValueError as e:
    # "Message blocked by governance: <receipt_id>"
    print(f"Blocked: {e}")

Governed GroupChat

Govern multi-agent team conversations.

There is no group-chat wrapper in the adapter. You can govern the seed messages withcreate_message_filter() and the opening message by wrapping the agent that starts the chat. Messages the agents exchange inside the group are not intercepted.

pythongroup_chat.py
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from tork_governance.adapters.autogen import TorkAutoGenMiddleware

middleware = TorkAutoGenMiddleware(agent_id="team-chat")

# Create team of agents
researcher = AssistantAgent(
    name="researcher",
    system_message="You research topics and gather information.",
    llm_config={"model": "gpt-4"}
)

writer = AssistantAgent(
    name="writer",
    system_message="You write content based on research.",
    llm_config={"model": "gpt-4"}
)

user_proxy = UserProxyAgent(
    name="user",
    human_input_mode="NEVER"
)

# There is no group-chat wrapper. Two things ARE available:

# 1. Govern the seed messages with the middleware's message filter
#    (a function that redacts or blocks the "content" of a message dict)
govern_message = middleware.create_message_filter()
seed_messages = [govern_message({"role": "user", "content": "Kick-off notes"})]

group = GroupChat(
    agents=[researcher, writer, user_proxy],
    messages=seed_messages,
    max_round=10
)
manager = GroupChatManager(groupchat=group, llm_config={"model": "gpt-4"})

# 2. Govern the message that starts the chat by wrapping the initiator
governed_user = middleware.wrap_agent(user_proxy)
governed_user.initiate_chat(
    manager,
    message="Write a blog post about AI safety"
)

# Messages the agents exchange inside the group chat are NOT intercepted.
# Every governed message leaves a receipt summary in middleware.receipts.

Code Execution

What the adapter does and does not do for executed code.

AutoGen can execute code generated by agents. The adapter does not inspect or block that code: there is no import allowlist, no dangerous-operation policy, and dashboard policies are not applied on-device. What it does is scan the text that passes through a wrapped agent for PII. Keep AutoGen's own code_execution_config (Docker, work_dir) as your execution sandbox.

Tool Arguments and Results

Scan tool arguments and tool results for PII and prompt injection.

There is no validate_tool_call() and no SQL or recipient policy. What exists: govern serialised tool arguments with process_message() (PII detection), and scan a tool's result with Tork.scan_tool_result()for PII and prompt-injection text before it is appended to the model context.

pythontool_results.py
import json
from tork_governance.adapters.autogen import TorkAutoGenMiddleware

middleware = TorkAutoGenMiddleware(agent_id="sql-agent")

def execute_sql(query: str) -> list[dict]:
    """Your real tool."""
    ...

def governed_execute_sql(query: str) -> list[dict]:
    # 1. Govern the arguments before the tool runs: they are scanned for
    #    PII like any other text. (This is PII detection, not a SQL policy.)
    check = middleware.process_message(json.dumps({"query": query}), "tool_args")
    if check.pii.has_pii:
        print(f"PII in tool arguments: {check.pii.types} ({check.receipt.receipt_id})")

    rows = execute_sql(query)

    # 2. Scan the tool RESULT for PII and prompt-injection text before it is
    #    appended to the model context. Returns sanitized/findings/blocked
    #    and a receipt; never the payload itself.
    scan = middleware.tork.scan_tool_result(
        "execute_sql",
        rows,
        block_on_injection=True,
    )
    if scan.blocked:
        raise ValueError(f"Tool result blocked: {scan.reason}")
    for finding in scan.findings:
        print(f"{finding.kind}:{finding.type} x{finding.count} at {finding.location}")
    return scan.sanitized  # PII masked in place

Error Handling

Handle governance violations gracefully.

The adapter raises a plain ValueError fromsend() and initiate_chat() when the outgoing message is denied, and only when the engine's action is DENY. With the default REDACT action nothing is raised. There are no custom exception classes and no jailbreak detection in this adapter.

pythonerror_handling.py
from autogen import AssistantAgent, UserProxyAgent
from tork_governance import Tork, GovernanceAction
from tork_governance.adapters.autogen import TorkAutoGenMiddleware

# Only a DENY action raises. With the default REDACT action, PII is
# replaced in place and no exception is thrown.
middleware = TorkAutoGenMiddleware(
    tork=Tork(default_action=GovernanceAction.DENY),
    agent_id="safe-chat"
)

def safe_conversation(user_message: str) -> str:
    """Run a governed conversation with error handling."""

    assistant = AssistantAgent(
        name="assistant",
        llm_config={"model": "gpt-4"}
    )

    user_proxy = UserProxyAgent(
        name="user",
        human_input_mode="NEVER",
        max_consecutive_auto_reply=1
    )

    governed_user_proxy = middleware.wrap_agent(user_proxy)

    try:
        governed_user_proxy.initiate_chat(
            assistant,
            message=user_message
        )
        return assistant.last_message()["content"]

    except ValueError as e:
        # The initial message was denied (PII detected, action=DENY).
        # str(e) is "Initial message blocked: <receipt_id>".
        return f"I cannot process that request: {e}"

# Usage
response = safe_conversation("What is machine learning?")  # Allowed
response = safe_conversation("My SSN is 123-45-6789")      # ValueError: PII denied

Multi-Agent Workflow

Complete example of a governed customer service team.

A customer service team with specialized agents. The comments in the sample state exactly which messages are governed: the opening message and any text you pass throughgovern_message(), not the traffic inside the group chat.

pythoncustomer_service.py
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from tork_governance.adapters.autogen import TorkAutoGenMiddleware

middleware = TorkAutoGenMiddleware(agent_id="customer-service")

# Define specialized agents
greeter = AssistantAgent(
    name="greeter",
    system_message="""You greet customers and understand their needs.
    Route to appropriate specialist.""",
    llm_config={"model": "gpt-4"}
)

tech_support = AssistantAgent(
    name="tech_support",
    system_message="""You handle technical issues and troubleshooting.
    Never share internal system details or credentials.""",
    llm_config={"model": "gpt-4"}
)

billing = AssistantAgent(
    name="billing",
    system_message="""You handle billing questions and refunds.
    Never expose full credit card numbers or account details.""",
    llm_config={"model": "gpt-4"}
)

customer = UserProxyAgent(
    name="customer",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=5
)

group = GroupChat(
    agents=[greeter, tech_support, billing, customer],
    messages=[],
    max_round=20
)
manager = GroupChatManager(groupchat=group, llm_config={"model": "gpt-4"})

# What Tork governs here, on-device:
# - the customer's opening message (PII redacted before AutoGen sees it)
# - any text you pass through middleware.govern_message() yourself
# It does NOT intercept the messages exchanged inside the group chat and
# it does not enforce the instructions in the system messages above.
governed_customer = middleware.wrap_agent(customer)
governed_customer.initiate_chat(
    manager,
    message="Hi, my card 4111 1111 1111 1111 was charged twice"
)

# Govern the final answer before showing it to the customer
final = manager.last_message()["content"]
print(middleware.govern_message(final))

Advanced Patterns

Local receipts and async usage

python
import json
from dataclasses import asdict
from autogen import AssistantAgent, UserProxyAgent
from tork_governance.adapters.autogen import TorkAutoGenMiddleware

middleware = TorkAutoGenMiddleware(agent_id="audited-chat")

assistant = AssistantAgent(name="assistant", llm_config={"model": "gpt-4"})
user_proxy = UserProxyAgent(name="user", human_input_mode="NEVER")

governed_user_proxy = middleware.wrap_agent(user_proxy)
governed_user_proxy.initiate_chat(
    assistant,
    message="Summarize our Q4 performance"
)

# Every governed message appends a summary to middleware.receipts:
# {'type', 'agent_id', 'receipt_id', 'action', 'has_pii'}
for entry in middleware.receipts:
    print(f"{entry['type']}: {entry['action']} pii={entry['has_pii']} ({entry['receipt_id']})")

# For the full receipt, govern text directly. Receipts are minted
# on-device from SHA-256 hashes and never contain the text.
result = middleware.process_message("Contact jane@example.com about Q4", "output")
receipt = result.receipt
print(receipt.receipt_id)     # "rcpt_..."
print(receipt.timestamp)
print(receipt.action.value)   # "redact"
print(receipt.input_hash)     # sha256 of the input
print(receipt.output_hash)    # sha256 of the redacted output
print(receipt.pii_types)      # ['email']

# Export for an audit file (dataclass -> dict; the enum is a str)
record = asdict(receipt)
record["action"] = receipt.action.value
with open("autogen_audit_trail.json", "w") as f:
    json.dump(record, f, indent=2)

# With an api_key, each decision is also reported to tork.network as a
# METADATA-ONLY client attestation (never the text). Check the outcome:
reporting = TorkAutoGenMiddleware(api_key="tork_live_...", agent_id="audited-chat")
res = reporting.process_message("Hello", "input")
if res.report is not None:
    res.report.wait()
    print(res.report.attempted, res.report.succeeded, res.report.reason)

Best Practices

Govern at the boundaries

Wrap the agent that starts a chat and govern final answers with govern_message(); the adapter does not intercept intra-group traffic.

Sandbox code execution with AutoGen

Use code_execution_config (Docker, work_dir) for isolation; the adapter does not inspect generated code.

Scan tool results before they reach the model

Use Tork.scan_tool_result() on every external result; set block_on_injection=True for untrusted sources.

Handle ValueError when using DENY

With Tork(default_action=GovernanceAction.DENY) the adapter raises ValueError on denied messages; catch it to respond gracefully.

Keep the local receipts

Every governed message appends a summary to middleware.receipts; the full on-device receipt is on each GovernanceResult.

Imports Reference

python
from tork_governance import Tork, TorkConfig, GovernanceAction, GovernanceResult
from tork_governance.adapters.autogen import (
    TorkAutoGenMiddleware,      # Central middleware
    GovernedAutoGenAgent,       # Wrapped agent (what wrap_agent() returns)
)
# Denied messages raise the built-in ValueError; there are no custom exceptions.

Next Steps

Configure policies in the dashboard and explore other framework integrations.

Documentation

Learn to integrate TORK

Upgrade Plan

Current: free

Support

Get help from our team