Framework Integration Gallery

Production-ready snippets for the most popular agent frameworks.

LangChain (Python)

Wrap any DynamicTool with AgentSoap verification to ensure every outbound call is scanned.

import hmac
import hashlib
import time
import requests
from langchain.tools import DynamicTool

def verify_with_soap(payload, agent_id="langchain-agent-01"):
    api_key = "your_api_key"
    secret = "your_hmac_secret"
    url = "https://api.your-domain.com/v1/verify/payload"
    
    timestamp = str(int(time.time()))
    # HMAC for Body
    signature_payload = timestamp + '{"text_content":"' + payload + '"}'
    signature = hmac.new(secret.encode(), signature_payload.encode(), hashlib.sha256).hexdigest()
    
    # HMAC for Agent Identity
    agent_token = hmac.new(secret.encode(), agent_id.encode(), hashlib.sha256).hexdigest()
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "X-Soap-Signature": signature,
        "X-Soap-Timestamp": timestamp,
        "X-Soap-Agent-ID": agent_id,
        "X-Soap-Agent-Token": agent_token,
        "Content-Type": "application/json"
    }
    
    response = requests.post(url, json={"text_content": payload}, headers=headers)
    return response.json().get("action_directive") == "ALLOW"

def secure_tool_func(input_str):
    if not verify_with_soap(input_str):
        return "SECURITY ALERT: This action was blocked by AgentSoap."
    # ... actual tool logic ...
    return f"Processed: {input_str}"

secure_tool = DynamicTool(
    name="SecureAction",
    func=secure_tool_func,
    description="Perform a sensitive action, protected by AgentSoap."
)

CrewAI

Force your 'Manager' agent to route all external communications through the AgentSoap middleware.

from crewai import Agent, Crew, Process
from my_tools import AgentSoapTool

# The Soap Agent acts as a mandatory filter
soap_agent = Agent(
    role='Security Gatekeeper',
    goal='Ensure all manager communications are safe and compliant.',
    backstory='You are the AgentSoap middleware. You verify every message before it leaves the system.',
    tools=[AgentSoapTool()],
    verbose=True
)

manager = Agent(
    role='System Manager',
    goal='Coordinate complex tasks and communicate with external APIs.',
    backstory='You lead the team but must pass all outbound requests through the Security Gatekeeper.',
    allow_delegation=True
)

crew = Crew(
    agents=[manager, soap_agent],
    tasks=[...],
    process=Process.sequential # Ensures soap checks happen in order
)

Microsoft AutoGen

Implement the 'Safety Proxy' pattern where one agent acts as the 'Soap' for another agent's outbound messages.

import autogen

def soap_filter(recipient, messages, sender, config):
    # This hook intercepts messages before they are delivered
    last_message = messages[-1].get("content", "")
    
    # Call AgentSoap API
    if not verify_with_soap(last_message):
        return False, "BLOCK: Potential security risk detected."
    
    return True, None

user_proxy = autogen.UserProxyAgent(
    name="UserProxy",
    is_termination_msg=lambda x: x.get("content", "").find("TERMINATE") >= 0,
    human_input_mode="NEVER",
)

assistant = autogen.AssistantAgent(
    name="Assistant",
    llm_config={"config_list": config_list},
)

# Register the Soap filter
assistant.register_reply(
    [autogen.Agent, None],
    reply_func=soap_filter,
    position=0 # Run before any other reply logic
)