Firewall

Firewall lets you block actions you don't want your agent to take, or tool results you don't want it to receive. For example, you can stop your agent from:

• sharing internal pricing with customers
• posting about unreleased products in public Slack channels
• receiving emails that contain a password or API key
• receiving Salesforce or HubSpot records that contain customers' personal details

You can describe each check in natural language, and Firewall blocks any tool call or result that matches it. Your agent can still use the same tools for everything else.

Firewall also comes with a built-in check that blocks prompt injection in tool results before it reaches your agent.

Custom prompt checks

You can build custom prompt checks with hooks. A hook is attached to a session and runs when your agent calls a tool. It can run before the tool call (before_execute) or after it (after_execute).

To write a custom prompt check, use the condition_check hook. It lets you block a tool call or its result if it matches a condition you write in natural language.

Here's a before_execute hook that stops your agent from sharing internal pricing with customers:

session = composio.create(
    user_id="user_123",
    hooks=[
        {
            "id": "no_pricing_to_customers",
            "type": "before_execute",
            "applies_to": {"toolkits": ["gmail"]},
            "handler": {
                "type": "condition_check",
                "condition": "Sharing internal pricing with anyone outside acme.com",
                "action": "block",
            },
        },
    ],
)

When your agent tries to email internal pricing to someone outside your company, the email isn't sent, and the following is returned when the tool is executed:

result = session.execute(
    "GMAIL_SEND_EMAIL",
    arguments={
        "recipient_email": "jane@customer.com",
        "subject": "Pricing",
        "body": "Our internal floor price is $40 per seat.",
    },
)

print(result.result_type)   # 'blocked'
print(result.error)         # 'Blocked by session policy'
print(result.block_reason)  # {'type': 'request_blocked', 'hook_id': 'no_pricing_to_customers'}
print(result.hook_results)  # [{'id': 'no_pricing_to_customers', 'status': 'evaluated', 'outcome': 'block'}]

If you use the SDK with an agent framework, your agent sees the message in error.

If more than one hook applies to a call, hookResults has an entry for each one, and the call is blocked if any of them blocks it.

Here's an after_execute hook that keeps emails with passwords or API keys away from your agent:

session = composio.create(
    user_id="user_123",
    hooks=[
        {
            "id": "no_secrets_in_results",
            "type": "after_execute",
            "name": "No secrets in emails",
            "applies_to": {"toolkits": ["gmail"]},
            "handler": {
                "type": "condition_check",
                "condition": "An email that contains a password, API key or access token",
                "action": "block",
            },
        },
    ],
)

When Gmail returns an email with a password in it, your agent doesn't receive the email.

Use a before_execute hook to block something your agent is about to do, like emailing internal pricing to a customer. Use an after_execute hook to block something a tool returns before your agent reads it, like an email that contains a password or API key. Note that an after_execute hook can't undo the tool call, since the tool has already run.

Writing a condition

It's important to write a condition as a description of the tool call or result you want blocked, not as an instruction for your agent to follow. A hook blocks a tool call or result when it matches the condition, and an instruction like "Don't share internal pricing with anyone outside acme.com" doesn't describe anything to match. Here are some good and bad ways to write a condition:

Good:

• "Sharing internal pricing with anyone outside acme.com"
• "An email that contains a password, API key or access token"

Bad:

• "Don't share internal pricing with anyone outside acme.com"
• "Hide emails that contain a password, API key or access token"

How we evaluate a condition

We use a classifier to compare each tool call to your condition. It sees which tool your agent is calling and the arguments it passes. In an after_execute hook, it also sees the data the tool returned.

The classifier also takes your agent's earlier activity in the session into account. This is important because a single tool call might seem harmless on its own, but when chained together with earlier tool calls, your agent might be doing something it's not allowed to do. For example, an email to a customer with a few numbers in it looks fine, unless your agent just copied those numbers from your internal pricing sheet.

Prompt injection detection

To block prompt injection, use the prompt_injection_check hook. It screens each tool result and blocks it if it contains a prompt injection.

session = composio.create(
    user_id="user_123",
    hooks=[
        {
            "id": "block_injection",
            "type": "after_execute",
            "applies_to": "all",
            "handler": {"type": "prompt_injection_check", "action": "block"},
        },
    ],
)

prompt_injection_check only works in after_execute hooks, since tool results are where outside content reaches your agent.

Other details

Hook IDs and names

You're required to assign a unique id to each hook in a session. You can also set an optional name. When the hook blocks a call, the name is included in the error message your agent sees.

Which tool calls are checked

A hook runs on calls to the toolkits listed in its appliesTo, or on calls to every toolkit if appliesTo is 'all'. That includes proxy calls to a toolkit's API, and tool calls your agent makes from a Workbench script or with multi-execute.

Keep in mind that hooks don't currently run on Composio meta tools themselves, so a condition like "Running a Workbench script" won't block anything.

Updating checks

To change a session's checks, pass a new hooks list to session.update(). The new list replaces the old one, so include every hook you want to keep. To remove them all, pass null (None in Python).

session.update(hooks=[...])  # every hook you want to keep