Anthropic Computer Use + AxonFlow Integration

Prerequisites: Python 3.10+, AxonFlow running (Getting Started), pip install axonflow anthropic

---

Quick Start

from axonflow import AxonFlow
from axonflow.adapters import ComputerUseGovernor

async with AxonFlow(
    endpoint="http://localhost:8080",
    client_id="desktop-agent",
    client_secret="your-secret",
) as client:
    governor = ComputerUseGovernor(client)

    result = await governor.check_tool_use({
        "name": "bash",
        "input": {"command": "ls -la"},
    })
    if not result.allowed:
        print(f"Blocked: {result.block_reason}")

---

What Computer Use Is

Anthropic Computer Use lets Claude control a desktop environment. Claude sees screenshots, decides what to do, and emits tool_use blocks that your code executes. The pattern is called a sampling loop:

1. You send a screenshot (or initial prompt) to Claude
2. Claude returns one or more tool_use blocks (click, type, bash command, etc.)
3. Your code executes each tool action on the actual machine
4. You capture the result (new screenshot, command output) and send it back
5. Repeat until Claude returns end_turn

Three built-in tools ship with Computer Use:

Tool	What It Does
computer	Mouse clicks, keyboard input, screenshots
bash	Executes shell commands
text_editor	Views and edits files

The loop runs until Claude decides the task is complete. Your code is the executor. Claude is the decision-maker.

---

The Governance Gap

Computer Use has no policy layer. Claude generates tool actions, and your sampling loop executes them. There is no built-in mechanism to:

• Block dangerous commands before they run (rm -rf /, curl | bash, credential access)
• Detect PII in command output before feeding it back to Claude
• Audit tool actions for compliance (what was executed, when, by whom)
• Enforce per-user policies on what operations are allowed
• Rate-limit destructive operations

The Anthropic documentation states that operators are responsible for implementing safety controls. Computer Use is explicitly a beta feature with no built-in guardrails beyond Claude's own judgment.

This means the sampling loop is the only place to insert governance. Every tool action passes through your code. That is where AxonFlow plugs in.

---

ComputerUseGovernor

The ComputerUseGovernor adapter wraps AxonFlow's MCP policy-check endpoints for the Computer Use sampling loop. It takes an AxonFlow client instance and provides two async methods:

• check_tool_use(block) — evaluate a tool_use block before execution
• check_result(tool_name, result) — scan tool output before feeding it back to Claude

Initialization

The governor takes an initialized AxonFlow client. All connection details (endpoint, credentials) are configured on the client, not on the governor itself.

from axonflow import AxonFlow
from axonflow.adapters import ComputerUseGovernor

async with AxonFlow(
    endpoint="http://localhost:8080",
    client_id="desktop-agent",
    client_secret="your-secret",
) as client:
    # Default blocked bash patterns
    governor = ComputerUseGovernor(client)

    # Or override with custom patterns
    governor = ComputerUseGovernor(
        client,
        blocked_bash_patterns=["my_custom_pattern", r"DROP\s+TABLE"],
    )

CheckResult

Both methods return a CheckResult (a frozen dataclass):

from axonflow.adapters import CheckResult

# CheckResult fields:
result.allowed              # bool — True if the action is permitted
result.block_reason         # str | None — reason for blocking (None when allowed)
result.redacted_result      # str | None — redacted output (only from check_result)
result.policies_evaluated   # int — number of policies evaluated

The redacted_result field is only populated by check_result. When PII or sensitive data is detected in tool output, redacted_result contains the sanitized version. For check_tool_use, this field is always None.

check_tool_use

Call this before executing any tool_use block from Claude's response. Both check_tool_use and check_result are async methods and must be awaited:

for block in response.content:
    if block.type == "tool_use":
        result = await governor.check_tool_use({
            "name": block.name,
            "input": block.input,
        })
        if not result.allowed:
            # Feed a blocked result back to Claude instead of executing
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": f"BLOCKED by policy: {result.block_reason}",
                "is_error": True,
            })
            continue

        # Safe to execute
        output = execute_tool(block)
        tool_results.append({
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": output,
        })

For bash tool actions, check_tool_use runs the command against the blocked patterns list before calling AxonFlow. Pattern matches are rejected immediately without a network call.

For computer and text_editor actions, the full tool input is sent to AxonFlow's check-input endpoint for policy evaluation.

check_result

Call this before feeding tool output back to Claude:

output = execute_tool(block)

scan = await governor.check_result(block.name, output)
if not scan.allowed:
    # Hard block: policy violation
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": f"Output blocked: {scan.block_reason}",
        "is_error": True,
    })
elif scan.redacted_result is not None:
    # PII detected and redacted (PII_ACTION=redact mode)
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": scan.redacted_result,
    })
else:
    # Clean output, no PII found
    tool_results.append({
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": output,
    })

This catches PII, credentials, and other sensitive content in command output or file contents before they enter the conversation context. When PII is detected, redacted_result contains the sanitized version of the output with sensitive values masked.

---

Default Blocked Bash Patterns

When check_tool_use receives a bash tool action, it matches the command against these 10 patterns before making any API call:

Pattern	Rationale
rm -rf /	Filesystem destruction
dd if=	Raw disk write, can destroy partitions
mkfs	Filesystem format
curl|bash	Remote code execution via pipe
wget|bash	Remote code execution via pipe
cat ~/.ssh/	SSH key exfiltration
cat ~/.aws/	AWS credential exfiltration
cat .env	Environment variable/secret exfiltration
chmod 777	Overly permissive file permissions
>/dev/sd	Direct disk write

Override the defaults by passing blocked_bash_patterns during initialization:

from axonflow import AxonFlow
from axonflow.adapters import ComputerUseGovernor

async with AxonFlow(
    endpoint="http://localhost:8080",
    client_id="desktop-agent",
    client_secret="your-secret",
) as client:
    governor = ComputerUseGovernor(
        client,
        blocked_bash_patterns=[
            "rm -rf /",
            "curl|bash",
            "cat ~/.ssh/",
            r"DROP\s+TABLE",        # custom: block SQL drops from bash
            r"docker\s+rm\s+-f",    # custom: block force container removal
        ],
    )

Patterns are matched against the full command string.

---

Connector Type Derivation

AxonFlow uses the connector_type field to identify the source of a tool action. The governor derives the connector type from the tool name and (for the computer tool) the action:

Tool	Action	connector_type
computer	left_click	computer_use.left_click
computer	screenshot	computer_use.screenshot
computer	type	computer_use.type
computer	key	computer_use.key
bash	(any)	computer_use.bash
text_editor	(any)	computer_use.text_editor

For the computer tool, the connector type includes the specific action from the input (e.g., computer_use.left_click, computer_use.screenshot). For bash and text_editor, the connector type is always the same regardless of the command or operation.

This lets you write AxonFlow policies that target specific Computer Use actions. For example, a policy that blocks all computer_use.bash actions for a specific tenant while allowing computer_use.left_click and computer_use.screenshot.

---

Full Sampling Loop Integration

A complete sampling loop with governance, using the Anthropic Python SDK:

import anthropic
from axonflow import AxonFlow
from axonflow.adapters import ComputerUseGovernor

TOOLS = [
    {
        "type": "computer_20241022",
        "name": "computer",
        "display_width_px": 1024,
        "display_height_px": 768,
        "display_number": 1,
    },
    {
        "type": "bash_20241022",
        "name": "bash",
    },
    {
        "type": "text_editor_20241022",
        "name": "text_editor",
    },
]


async def run_governed_agent(task: str):
    """Run a Computer Use agent with AxonFlow governance."""
    anthropic_client = anthropic.Anthropic()

    async with AxonFlow(
        endpoint="http://localhost:8080",
        client_id="desktop-agent",
        client_secret="your-secret",
    ) as axonflow_client:
        governor = ComputerUseGovernor(axonflow_client)

        messages = [{"role": "user", "content": task}]

        while True:
            response = anthropic_client.beta.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=4096,
                tools=TOOLS,
                messages=messages,
                betas=["computer-use-2024-10-22"],
            )

            # Check for end of conversation
            if response.stop_reason == "end_turn":
                for block in response.content:
                    if hasattr(block, "text"):
                        return block.text
                return "Task completed."

            # Process tool use blocks
            tool_results = []
            for block in response.content:
                if block.type != "tool_use":
                    continue

                # --- GOVERNANCE: Pre-execution check ---
                check = await governor.check_tool_use({
                    "name": block.name,
                    "input": block.input,
                })
                if not check.allowed:
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": f"BLOCKED: {check.block_reason}",
                        "is_error": True,
                    })
                    continue

                # Execute the tool action
                output = execute_tool_action(block.name, block.input)

                # --- GOVERNANCE: Post-execution scan ---
                scan = await governor.check_result(block.name, output)
                if not scan.allowed:
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": f"Output blocked: {scan.block_reason}",
                        "is_error": True,
                    })
                    continue

                # Use redacted result if PII was found (PII_ACTION=redact mode)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": scan.redacted_result or output,
                })

            # Feed results back into the conversation
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})


def execute_tool_action(tool_name: str, tool_input: dict) -> str:
    """Execute a Computer Use tool action. Replace with your implementation."""
    if tool_name == "bash":
        import subprocess
        result = subprocess.run(
            tool_input["command"],
            shell=True,
            capture_output=True,
            text=True,
            timeout=30,
        )
        return result.stdout + result.stderr

    elif tool_name == "computer":
        # Implement mouse/keyboard actions using pyautogui, xdotool, etc.
        return f"Executed {tool_input.get('action', 'unknown')} action"

    elif tool_name == "text_editor":
        # Implement file view/edit
        return f"Executed {tool_input.get('command', 'unknown')} on {tool_input.get('path', '')}"

    return "Unknown tool"


# Run
import asyncio
result = asyncio.run(run_governed_agent(
    task="Open the browser and navigate to example.com",
))
print(result)

What happens when a command is blocked:

Claude receives an error result with the block reason. It will typically acknowledge the restriction and either try an alternative approach or explain that it cannot complete the requested action. The sampling loop continues normally.

---

Raw API Pattern

If you are not using the Python SDK (or prefer direct HTTP calls), you can call AxonFlow's MCP policy-check endpoints directly from any language.

Pre-execution check

Before executing a tool action, send the tool input to check-input:

curl -s -X POST http://localhost:8080/api/v1/mcp/check-input \
  -H "Content-Type: application/json" \
  -H "Authorization: Basic $(echo -n 'desktop-agent:your-secret' | base64)" \
  -d '{
    "connector_type": "computer_use.bash",
    "statement": "find /var/log -name \"*.log\" -mtime -1",
    "operation": "execute",
    "context": {
      "user_id": "operator-jane",
      "tool": "bash",
      "source": "computer_use"
    }
  }'

Response when allowed:

{
  "allowed": true,
  "policies_evaluated": 76,
  "policy_info": {
    "policies_evaluated": 76,
    "blocked": false,
    "redactions_applied": 0,
    "processing_time_ms": 3
  }
}

Response when blocked:

{
  "allowed": false,
  "block_reason": "Command matches blocked pattern: credential exfiltration",
  "policy_info": {
    "policies_evaluated": ["sys_dangerous_command"]
  }
}

Post-execution scan

After executing a tool action, scan the output before feeding it back to Claude:

curl -s -X POST http://localhost:8080/api/v1/mcp/check-output \
  -H "Content-Type: application/json" \
  -H "Authorization: Basic $(echo -n 'desktop-agent:your-secret' | base64)" \
  -d '{
    "connector_type": "computer_use.bash",
    "message": "Command output from find",
    "metadata": {
      "tool": "bash",
      "output_length": 1523
    },
    "response_data": {
      "stdout": "access.log\nerror.log\nauth.log",
      "exit_code": 0
    }
  }'

Audit logging

Record each tool action for compliance:

curl -s -X POST http://localhost:8080/api/v1/audit/tool-call \
  -H "Content-Type: application/json" \
  -H "Authorization: Basic $(echo -n 'desktop-agent:your-secret' | base64)" \
  -d '{
    "tool_name": "bash",
    "tool_type": "computer_use",
    "input": {"command": "find /var/log -name \"*.log\" -mtime -1"},
    "output": {"result": "access.log\nerror.log\nauth.log"},
    "success": true,
    "duration_ms": 150
  }'

This pattern works from any language that can make HTTP requests: Node.js, Go, Java, Rust, or shell scripts.

---

Policy Examples

Block all bash commands for a tenant

Create a tenant policy that blocks bash execution entirely:

{
  "name": "block-bash-execution",
  "description": "Prevent Computer Use agents from running shell commands",
  "connector_type": "computer_use.bash",
  "action": "block",
  "message": "Shell command execution is not permitted for this tenant"
}

Allow only read-only file operations

{
  "name": "read-only-text-editor",
  "description": "Only allow view commands, block insert/replace",
  "connector_type": "computer_use.text_editor",
  "conditions": {
    "input.command": {"not_in": ["create", "insert", "str_replace"]}
  },
  "action": "block",
  "message": "Only file viewing is permitted"
}

Detect PII in command output

AxonFlow's built-in PII detection policies (sys_pii_ssn, sys_pii_credit_card, sys_pii_email, etc.) apply automatically to check_result calls. No additional configuration is needed. When PII is detected in tool output, result.allowed is still True (under the default PII_ACTION=redact mode), but result.redacted_result contains the sanitized output and result.policies_evaluated reflects how many policies matched. Check policies_evaluated to detect PII, not allowed.

---

Related Resources

• MCP Policy Enforcement — How AxonFlow evaluates MCP connector requests
• Gateway Mode — SDK integration patterns for policy checks and audit
• Workflow Control Protocol (WCP) — Multi-step workflow governance
• Per-Tool Governance — Fine-grained tool-level policy enforcement
• Anthropic Computer Use Documentation — Official Computer Use guide

---

Platform Version: v5.4.0 | SDKs: Python v5.4.0, Go/TypeScript/Java v4.3.0