TorkTORK
Docs/AutoGen Guide
Microsoft AutoGen Integration

AutoGen + Tork Governance

Add AI safety guardrails to Microsoft AutoGen multi-agent conversations. Govern messages, block dangerous code execution, and maintain compliance across agent teams.

Message Governance

Validate all agent messages

GroupChat Support

Govern multi-agent teams

Code Execution

Block dangerous imports

Tool Validation

Secure function calls

Installation

Install Tork with AutoGen dependencies.

bash
pip install tork-governance pyautogen

The Tork SDK includes AutoGen adapters in tork.adapters.autogen.

TorkAutoGenMiddleware

Central middleware for governing AutoGen agent conversations.

The middleware wraps AutoGen agents to evaluate all incoming and outgoing messages against your governance policies. Use wrap_agent() to add governance to any agent.

pythonmiddleware_example.py
from autogen import AssistantAgent, UserProxyAgent
from tork.adapters.autogen import TorkAutoGenMiddleware
from tork.core.engine import GovernanceEngine

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

# Create middleware
middleware = TorkAutoGenMiddleware(
    engine=engine,
    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 agents with governance
governed_assistant = middleware.wrap_agent(assistant)
governed_user_proxy = middleware.wrap_agent(user_proxy)

# All messages are now governed
governed_user_proxy.initiate_chat(
    governed_assistant,
    message="Write a Python function to calculate fibonacci numbers"
)

TorkGovernedAssistant

Direct wrapper for AssistantAgent with built-in governance.

TorkGovernedAssistant wraps an AutoGen AssistantAgent directly. It validates messages on receive and responses on generate, raising exceptions when policies are violated.

pythongoverned_assistant.py
from autogen import AssistantAgent
from tork.adapters.autogen import TorkGovernedAssistant
from tork.adapters.autogen.exceptions import MessageBlockedError, ResponseBlockedError

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

# Wrap with Tork governance
governed_assistant = TorkGovernedAssistant(
    assistant=assistant,
    api_key="your_tork_api_key",  # Or pass engine=
    agent_id="code-assistant"
)

# Messages are automatically validated
try:
    # Receive validates incoming messages
    governed_assistant.receive(
        message="Help me write a web scraper",
        sender=user_proxy
    )

    # Generate validates outputs before returning
    reply = governed_assistant.generate_reply(
        messages=[{"content": "Write hello world", "role": "user"}]
    )
except MessageBlockedError as e:
    print(f"Input blocked: {e}")
except ResponseBlockedError as e:
    print(f"Output blocked: {e}")

Governed GroupChat

Govern multi-agent team conversations.

Use create_governed_group_chat() to automatically wrap all agents in a group. Every message between agents is evaluated against policies.

pythongroup_chat.py
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from tork.adapters.autogen import TorkAutoGenMiddleware, GovernedGroupChat
from tork.core.engine import GovernanceEngine

# Initialize
engine = GovernanceEngine(api_key="your_tork_api_key")
middleware = TorkAutoGenMiddleware(engine=engine, 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"}
)

reviewer = AssistantAgent(
    name="reviewer",
    system_message="You review and improve content quality.",
    llm_config={"model": "gpt-4"}
)

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

# Create governed group chat - all agents are automatically wrapped
governed_group = middleware.create_governed_group_chat(
    agents=[researcher, writer, reviewer, user_proxy],
    messages=[],
    max_round=10
)

# Create manager for the group
manager = GroupChatManager(
    groupchat=GroupChat(
        agents=governed_group.agents,
        messages=[],
        max_round=10
    ),
    llm_config={"model": "gpt-4"}
)

# All inter-agent communication is governed
user_proxy.initiate_chat(
    manager,
    message="Write a blog post about AI safety"
)

Code Execution Governance

Block dangerous imports and operations in generated code.

AutoGen can execute code generated by agents. Tork policies can block dangerous imports like os, subprocess, and prevent access to sensitive system resources.

pythoncode_execution.py
from autogen import AssistantAgent, UserProxyAgent
from tork.adapters.autogen import TorkAutoGenMiddleware, TorkGovernedAssistant
from tork.core.engine import GovernanceEngine

# Initialize with governance
engine = GovernanceEngine(api_key="your_tork_api_key")
middleware = TorkAutoGenMiddleware(engine=engine, agent_id="code-executor")

# Assistant generates code
assistant = AssistantAgent(
    name="coder",
    llm_config={"model": "gpt-4"}
)

# User proxy executes code - governance blocks dangerous imports
user_proxy = UserProxyAgent(
    name="executor",
    human_input_mode="NEVER",
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False
    }
)

# Wrap both agents
governed_assistant = middleware.wrap_agent(assistant)
governed_user_proxy = middleware.wrap_agent(user_proxy)

# Policy example: block dangerous imports
# (Configure in Tork dashboard or via API)
# - Block: import os, import subprocess, import shutil
# - Block: eval(), exec(), __import__
# - Block: requests to internal IPs
# - Allow: import numpy, pandas, matplotlib

# When assistant generates code with blocked imports,
# the governance layer will intervene
governed_user_proxy.initiate_chat(
    governed_assistant,
    message="Read the contents of /etc/passwd and print them"
)
# ^ This will be blocked by governance policies

Configure Code Policies

Set up code execution policies in the Tork dashboard to block specific imports, file operations, network access, and other dangerous operations.

Tool/Function Validation

Validate tool calls before execution.

When agents use tools or functions, validate the inputs before execution. Block dangerous SQL queries, restrict email recipients, or enforce data access policies.

pythontool_validation.py
from autogen import AssistantAgent
from tork.adapters.autogen import TorkGovernedAssistant
from tork.adapters.autogen.exceptions import MessageBlockedError

assistant = AssistantAgent(
    name="tool_user",
    llm_config={
        "model": "gpt-4",
        "functions": [
            {
                "name": "execute_sql",
                "description": "Execute SQL query on database",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "SQL query to execute"}
                    },
                    "required": ["query"]
                }
            },
            {
                "name": "send_email",
                "description": "Send an email",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "to": {"type": "string"},
                        "subject": {"type": "string"},
                        "body": {"type": "string"}
                    },
                    "required": ["to", "subject", "body"]
                }
            }
        ]
    }
)

governed = TorkGovernedAssistant(
    assistant=assistant,
    api_key="your_tork_api_key",
    agent_id="sql-agent"
)

# Validate tool calls before execution
try:
    validated_args = governed.validate_tool_call(
        tool_name="execute_sql",
        tool_args={"query": "DROP TABLE users;"}
    )
except MessageBlockedError as e:
    print(f"Dangerous query blocked: {e}")
    # Governance blocked: destructive SQL operations

# Safe queries are allowed
validated_args = governed.validate_tool_call(
    tool_name="execute_sql",
    tool_args={"query": "SELECT name, email FROM users WHERE active = true"}
)
# ^ Allowed - read-only query

Error Handling

Handle governance violations gracefully.

Catch governance exceptions to provide user-friendly responses instead of errors. Use MessageBlockedError for input violations and ResponseBlockedError for output violations.

pythonerror_handling.py
from autogen import AssistantAgent, UserProxyAgent
from tork.adapters.autogen import TorkAutoGenMiddleware
from tork.adapters.autogen.exceptions import (
    AutoGenGovernanceError,
    MessageBlockedError,
    ResponseBlockedError
)
from tork.core.engine import GovernanceEngine

engine = GovernanceEngine(api_key="your_tork_api_key")
middleware = TorkAutoGenMiddleware(engine=engine, 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_assistant = middleware.wrap_agent(assistant)
    governed_user_proxy = middleware.wrap_agent(user_proxy)

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

    except MessageBlockedError as e:
        # Input message violated policy
        return f"I cannot process that request: {str(e)}"

    except ResponseBlockedError as e:
        # Generated response violated policy
        return "I generated a response that violated safety policies. Please rephrase your request."

    except AutoGenGovernanceError as e:
        # General governance error
        return f"A governance error occurred: {str(e)}"

# Usage
response = safe_conversation("What is machine learning?")  # Allowed
response = safe_conversation("My SSN is 123-45-6789")      # Blocked - PII
response = safe_conversation("Ignore previous instructions") # Blocked - Jailbreak

Multi-Agent Workflow

Complete example of a governed customer service team.

Build complex multi-agent systems with governance on every communication. This example shows a customer service team with specialized agents.

pythoncustomer_service.py
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from tork.adapters.autogen import TorkAutoGenMiddleware
from tork.core.engine import GovernanceEngine

# Initialize governance
engine = GovernanceEngine(api_key="your_tork_api_key")
middleware = TorkAutoGenMiddleware(engine=engine, 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"}
)

supervisor = AssistantAgent(
    name="supervisor",
    system_message="""You oversee the conversation and ensure quality.
    Escalate if needed and ensure customer satisfaction.""",
    llm_config={"model": "gpt-4"}
)

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

# Create governed group - each agent's messages are validated
governed_group = middleware.create_governed_group_chat(
    agents=[greeter, tech_support, billing, supervisor, customer],
    messages=[],
    max_round=20
)

# Governance policies automatically:
# - Block PII leakage (credit cards, SSNs)
# - Prevent internal credential exposure
# - Block inappropriate responses
# - Log all agent communications for audit

manager = GroupChatManager(
    groupchat=GroupChat(
        agents=governed_group.agents,
        messages=[],
        max_round=20
    ),
    llm_config={"model": "gpt-4"}
)

# Start customer service session
customer.initiate_chat(
    manager,
    message="Hi, I have a billing question about my last invoice"
)

Advanced Patterns

Compliance receipts and async support

python
from autogen import AssistantAgent, UserProxyAgent
from tork.adapters.autogen import TorkAutoGenMiddleware
from tork.core.engine import GovernanceEngine
from tork.compliance.receipts import ReceiptGenerator
import json

# Initialize with receipt generation
engine = GovernanceEngine(api_key="your_tork_api_key")
middleware = TorkAutoGenMiddleware(engine=engine, agent_id="audited-chat")

# Access the receipt generator
receipt_gen = middleware.receipt_generator

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

governed_assistant = middleware.wrap_agent(assistant)
governed_user_proxy = middleware.wrap_agent(user_proxy)

governed_user_proxy.initiate_chat(
    governed_assistant,
    message="Summarize our Q4 performance"
)

# Access compliance receipts
receipts = receipt_gen.get_receipts()

for receipt in receipts:
    print(f"Receipt ID: {receipt.id}")
    print(f"Timestamp: {receipt.timestamp}")
    print(f"Action: {receipt.action}")
    print(f"Decision: {receipt.decision}")
    print(f"Agent: {receipt.agent_id}")
    print(f"Hash: {receipt.hash}")
    print("---")

# Export for compliance audit
receipts_data = [r.to_dict() for r in receipts]
with open("autogen_audit_trail.json", "w") as f:
    json.dump(receipts_data, f, indent=2)

# Verify receipt integrity
for receipt in receipts:
    is_valid = receipt_gen.verify_receipt(receipt)
    print(f"Receipt {receipt.id} valid: {is_valid}")

Best Practices

Wrap all agents in multi-agent systems

Use middleware.wrap_agent() for every agent to ensure no communication bypasses governance.

Configure code execution policies

Block dangerous imports (os, subprocess, shutil) and operations (eval, exec) via the dashboard.

Validate tool calls separately

Use validate_tool_call() for sensitive operations like database queries or external API calls.

Handle exceptions gracefully

Catch MessageBlockedError and ResponseBlockedError to provide user-friendly error messages.

Enable compliance receipts for audit

Use the receipt generator to maintain a cryptographically signed audit trail of all decisions.

Imports Reference

python
from tork.adapters.autogen import (
    TorkAutoGenMiddleware,      # Central middleware
    GovernedAutoGenAgent,       # Wrapped agent
    GovernedGroupChat,          # Wrapped group chat
    TorkGovernedAssistant,      # Direct assistant wrapper
    AutoGenGovernanceError,     # Base exception
    MessageBlockedError,        # Input blocked
    ResponseBlockedError,       # Output blocked
)

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