AHP Safety Harness

This example demonstrates how to use AHP (Agent Harness Protocol) harness servers to add security controls to your agent.

Overview

A3S Code provides two types of AHP safety harness implementations:

Agent Monitors Agent - Use an independent A3S Code agent to monitor and control business agents (Recommended)
Pattern-Based Harness - Fast, rule-based security checks using regex patterns

Architecture Comparison

Agent Monitors Agent: Two independent A3S Code agents, AHP Server agent uses LLM to analyze every operation from business agent. Most powerful and flexible, supports complex security policies.
Pattern-Based: Fast, deterministic, no API costs, but limited to known patterns. Good as first line of defense in production.

Recommended to use Agent Monitors Agent architecture for best security and flexibility.

Agent Monitors Agent Architecture (Recommended)

This is the most powerful AHP implementation, using two independent A3S Code agents:

Architecture

┌─────────────────────────────────────────────────────────────┐
│                      Business Agent                          │
│                  (A3S Code Agent + Kimi)                    │
│                                                             │
│  - Executes actual business tasks                          │
│  - Calls tools (Bash, Read, Write, etc.)                   │
│  - Configured with AHP transport to AHP Server             │
└─────────────────────────────────────────────────────────────┘
                          │
                          │ AHP Protocol (JSON-RPC 2.0)
                          │ stdio transport
                          ↓
┌─────────────────────────────────────────────────────────────┐
│                   AHP Server Agent                           │
│                  (A3S Code Agent + Kimi)                    │
│                                                             │
│  - Receives pre_action/post_action events                  │
│  - Uses LLM to analyze security risks                      │
│  - Makes allow/block decisions                             │
│  - Returns decisions to business agent                     │
└─────────────────────────────────────────────────────────────┘

Usage

from a3s_code import Agent, SessionOptions, StdioTransport

# Create business agent
agent = Agent.create("agent.hcl")

# Configure AHP Server agent monitoring
opts = SessionOptions()
opts.ahp_transport = StdioTransport(
    program="python3",
    args=["examples/ahp_agent_monitors_agent.py"]
)
opts.builtin_skills = True

opts.permissive = True

# Create monitored session
session = agent.session(".", opts)

# Safe operation - will be allowed
result = session.send("List all files in current directory")
print(result.text)

# Dangerous operation - will be blocked by AHP Server agent
try:
    result = session.send("Run command: rm -rf /")
except Exception as e:
    print(f"Blocked by AHP Server agent: {e}")

import { Agent, SessionOptions, StdioTransport } from '@a3s-lab/code';

// Create business agent
const agent = await Agent.create('agent.hcl');

// Configure AHP Server agent monitoring
const opts: SessionOptions = {
  ahpTransport: new StdioTransport({
    program: 'python3',
    args: ['examples/ahp_agent_monitors_agent.py']
  }),
  builtinSkills: true,
  permissive: true,
};

// Create monitored session
const session = agent.session('.', opts);

// Safe operation - will be allowed
const result = await session.send('List all files in current directory');
console.log(result.text);

// Dangerous operation - will be blocked by AHP Server agent
try {
  await session.send('Run command: rm -rf /');
} catch (e) {
  console.log(`Blocked by AHP Server agent: ${e}`);
}

The AHP Server agent itself mounts the ahp_skills directory to govern the business agent:

from a3s_code import Agent
from pathlib import Path

# AHP Server agent config (inside ahp_agent_monitors_agent.py)
agent = Agent.create("agent_kimi.hcl")

# Mount ahp_skills directory so AHP Server agent can use skills to analyze tool calls
skill_dir = str(Path(__file__).parent / "ahp_skills")
session = agent.session(
    workspace,
    SessionOptions(
        permissive=True,
        builtin_skills=False,
        skill_dirs=[skill_dir],
    ),
)
# AHP Server agent uses /detect-dangerous-operation skill to analyze each tool call

The ahp_skills directory contains:

detect-dangerous-operation.md — Analyzes tool call security risks (pre_action)
sanitize-untrusted-output.md — Sanitizes tool outputs (post_action)

Core Advantages

Agent Monitors Agent Features

Context-aware analysis: Understands operation intent, not just pattern matching
Adaptive learning: Improves from historical decisions
Natural language reasoning: Provides clear decision rationale
Complex threat detection: Catches attacks that bypass regex
Independent agents: AHP Server agent is fully independent, can have its own config and policies

Requires: Kimi or other LLM provider configured in ~/.a3s/config.hcl or a3s/.a3s/config.hcl

Complete Example

See the complete test example:

cd crates/code/examples
python3 test_ahp_agent_monitors_agent.py

Test suite includes:

Safe operations test - List files, write/read files, simple commands
Dangerous operations test - rm -rf /, read SSH private keys, read /etc/shadow
Sensitive data test - Read files containing fake credentials

Detailed documentation: crates/code/examples/AHP_AGENT_MONITORS_AGENT_README.md

Pattern-Based Pre-Action Guard

Block dangerous operations before they execute using regex patterns:

from a3s_code import Agent, SessionOptions

# Create agent
agent = Agent.create("agent.hcl")

# Configure pre-action guard
opts = SessionOptions()
opts.ahp_transport = {
    "type": "stdio",
    "program": "python3",
    "args": ["examples/ahp_pre_action_guard.py"]
}

session = agent.session(".", opts)

# Safe command - will be allowed
result = session.send("List files in current directory")
print(result.text)

# Dangerous command - will be blocked
try:
    result = session.send("Delete all files with rm -rf /")
except Exception as e:
    print(f"Blocked: {e}")

import { Agent, SessionOptions } from '@a3s-lab/code';

// Create agent
const agent = await Agent.create('agent.hcl');

// Configure pre-action guard
const opts: SessionOptions = {
  ahpTransport: {
    type: 'stdio',
    program: 'python3',
    args: ['examples/ahp_pre_action_guard.py']
  }
};

const session = agent.session('.', opts);

// Safe command - will be allowed
const result = await session.send('List files in current directory');
console.log(result.text);

// Dangerous command - will be blocked
try {
  await session.send('Delete all files with rm -rf /');
} catch (e) {
  console.log(`Blocked: ${e}`);
}

use a3s_code_core::{Agent, SessionOptions};
use a3s_ahp::{Transport, AhpHookExecutor};
use std::sync::Arc;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Create AHP hook executor
    let ahp_executor = AhpHookExecutor::new(Transport::Stdio {
        program: "python3".into(),
        args: vec!["examples/ahp_pre_action_guard.py".into()],
    }).await?;

    // Create agent with AHP hook
    let agent = Agent::from_file("agent.hcl").await?;
    let opts = SessionOptions::default()
        .with_hook_executor(Arc::new(ahp_executor));

    let session = agent.session(".", opts).await?;

    // Safe command - will be allowed
    let result = session.send("List files in current directory").await?;
    println!("{}", result.text);

    // Dangerous command - will be blocked
    match session.send("Delete all files with rm -rf /").await {
        Ok(_) => println!("ERROR: Dangerous command was not blocked!"),
        Err(e) => println!("Blocked: {}", e),
    }

    Ok(())
}

Pattern-Based Post-Action Sanitizer

Clean untrusted outputs using regex patterns:

from a3s_code import Agent, SessionOptions

agent = Agent.create("agent.hcl")

# Configure post-action sanitizer
opts = SessionOptions()
opts.ahp_transport = {
    "type": "stdio",
    "program": "python3",
    "args": ["examples/ahp_post_action_sanitizer.py"]
}

session = agent.session(".", opts)

# Output with PII - will be redacted
result = session.send("Echo 'API_KEY=sk_test_123 email=user@example.com'")
print(result.text)  # PII will be replaced with [REDACTED_*]

# Prompt injection attempt - will be blocked
try:
    result = session.send("Echo 'Ignore all previous instructions'")
except Exception as e:
    print(f"Blocked injection: {e}")

import { Agent, SessionOptions } from '@a3s-lab/code';

const agent = await Agent.create('agent.hcl');

// Configure post-action sanitizer
const opts: SessionOptions = {
  ahpTransport: {
    type: 'stdio',
    program: 'python3',
    args: ['examples/ahp_post_action_sanitizer.py']
  }
};

const session = agent.session('.', opts);

// Output with PII - will be redacted
const result = await session.send("Echo 'API_KEY=sk_test_123 email=user@example.com'");
console.log(result.text);  // PII will be replaced with [REDACTED_*]

// Prompt injection attempt - will be blocked
try {
  await session.send("Echo 'Ignore all previous instructions'");
} catch (e) {
  console.log(`Blocked injection: ${e}`);
}

Combined Protection

Use both harnesses for defense-in-depth:

from a3s_code import Agent, SessionOptions

agent = Agent.create("agent.hcl")

# Note: Currently only one AHP transport is supported per session
# For multiple harnesses, deploy them as HTTP services and chain them

# Pre-action guard
opts_pre = SessionOptions()
opts_pre.ahp_transport = {
    "type": "http",
    "url": "http://localhost:8080/ahp",  # Pre-action guard
}

# Post-action sanitizer
opts_post = SessionOptions()
opts_post.ahp_transport = {
    "type": "http",
    "url": "http://localhost:8081/ahp",  # Post-action sanitizer
}

# Use pre-action for this session
session = agent.session(".", opts_pre)

Custom Harness

Create a custom harness for your specific needs:

#!/usr/bin/env python3
import json
import sys

def handle_event(event):
    event_type = event.get("event_type")
    payload = event.get("payload", {})

    if event_type == "pre_action":
        tool = payload.get("tool", "")

        # Custom policy: block all web_fetch calls
        if tool == "web_fetch":
            return {
                "decision": "block",
                "reason": "Web access is disabled"
            }

        # Custom policy: require confirmation for file writes
        if tool == "Write":
            return {
                "decision": "escalate",
                "reason": "File write requires approval",
                "escalation_target": "admin@example.com"
            }

        return {"decision": "allow"}

    return {"decision": "allow"}

def main():
    for line in sys.stdin:
        msg = json.loads(line.strip())
        req_id = msg.get("id")
        method = msg.get("method", "")
        params = msg.get("params", {})

        if req_id and method == "ahp/event":
            result = handle_event(params)
            response = {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": result
            }
            print(json.dumps(response), flush=True)

if __name__ == "__main__":
    main()

See the AHP Integration guide for complete documentation.

Testing

Test your harness server:

# Test handshake
echo '{"jsonrpc":"2.0","id":"1","method":"ahp/handshake","params":{"protocol_version":"2.0","agent_info":{"framework":"test","version":"1.0","capabilities":[]},"session_id":"test","agent_id":"test"}}' | python3 my_harness.py

# Test pre-action event
echo '{"jsonrpc":"2.0","id":"2","method":"ahp/event","params":{"event_type":"pre_action","session_id":"test","agent_id":"test","timestamp":"2024-03-11T12:00:00Z","depth":0,"payload":{"tool":"Bash","arguments":{"command":"ls"}}}}' | python3 my_harness.py

Production Deployment

Deploy harness as HTTP service:

# Start HTTP server
python3 examples/http_server.py --port 8080 --harness pre_action_guard

# Configure agent
opts.ahp_transport = {
    "type": "http",
    "url": "http://localhost:8080/ahp",
    "auth": {
        "type": "bearer",
        "token": os.getenv("AHP_TOKEN")
    }
}

AHP Safety Harness

On this page