Triggers

Overview

Triggers act as a notification system for your AI applications, enabling your agents to respond dynamically to external events occurring within your integrations.

When these events take place, triggers capture relevant information and deliver structured payloads directly to your system, facilitating timely and context-aware responses.

For instance, imagine building a Slack bot designed to generate humorous responses to messages from your co-workers. To achieve this, your application needs to receive notifications whenever someone posts a new message in a specific Slack channel. Triggers fulfill this role by listening for these events and promptly notifying your system, allowing your bot to respond appropriately.

Composio supports two primary methods for delivering these payloads:

• Webhooks: HTTP POST requests sent to a publicly accessible URL that you configure. Webhooks are ideal for scenarios where your application needs to handle events asynchronously and independently from the event source.

• Websockets: Persistent, real-time connections that push event data directly to your application. Websockets are suitable for applications requiring immediate, continuous, and low-latency communication.

Managing triggers

Before proceeding, ensure you’ve created an integration and established a connection to your external account (e.g., Slack, GitHub).

Enable the Trigger

Enable the “New Message Received” trigger for your Slack app through the dashboard, CLI, or code.

1
from composio_openai import ComposioToolSet
2
3
toolset = ComposioToolSet()
4
5
user_id = "default" # User ID referencing an entity retrieved from application logic
6
entity = toolset.get_entity(id=user_id)
7
triggers = toolset.get_trigger("SLACK_RECEIVE_MESSAGE")
8
9
res = entity.enable_trigger(
10
    app=App.SLACK,
11
    trigger_name="SLACK_RECEIVE_MESSAGE",
12
    config={}
13
)
14
15
print(res["status"])

1
# Using same imports as above
2
trigger = toolset.get_trigger("GITHUB_STAR_ADDED_EVENT")
3
print(trigger.config.model_dump_json(indent=4))

1
{
2
    "properties": {
3
        "owner": {
4
            "description": "Owner of the repository",
5
            "title": "Owner",
6
            "default": null,
7
            "type": "string"
8
        },
9
        "repo": {
10
            "description": "Repository name",
11
            "title": "Repo",
12
            "default": null,
13
            "type": "string"
14
        }
15
    },
16
    "title": "WebhookConfigSchema",
17
    "type": "object",
18
    "required": [
19
        "owner",
20
        "repo"
21
    ]
22
}

1
response = entity.enable_trigger(
2
    app=App.GITHUB,
3
    trigger_name="GITHUB_PULL_REQUEST_EVENT",
4
    config={"owner": "composiohq", "repo": "composio"},
5
)

Listeners

Once you have the triggers set up, you can specify listener functions using websockets through the SDK or webhooks.

Specifying Listeners through Websockets

We create a listener and then define a callback function that executes when a listener recieves a payload.

1
from composio_openai import ComposioToolSet, App, Action
2
3
toolset = ComposioToolSet()
4
entity = toolset.get_entity(id="default")
5
6
listener = toolset.create_trigger_listener()
7
@listener.callback(
8
    filters={
9
        "trigger_name": "SLACK_RECEIVE_MESSAGE",
10
    }
11
)
12
def handle_slack_message(event):
13
    print(event)
14
15
listener.wait_forever()

Specifying Listeners through Webhooks

Assuming you’ve already set up a trigger as discussed in previous steps, here’s how you can use webhooks instead to listen in on new events happening in an app.

1
from fastapi import FastAPI, Request
2
from typing import Dict, Any
3
import uvicorn
4
import json
5
6
app = FastAPI(title="Webhook Demo")
7
8
@app.post("/webhook")
9
async def webhook_handler(request: Request):
10
    # Get the raw payload
11
    payload = await request.json()
12
    
13
    # Log the received webhook data
14
    print("Received webhook payload:")
15
    print(json.dumps(payload, indent=2))
16
    
17
    # Return a success response
18
    return {"status": "success", "message": "Webhook received"}
19
20
if __name__ == "__main__":
21
    uvicorn.run(app, host="0.0.0.0", port=8000)

Demo: Roast Slack Messages

Let’s build a fun bot that generates snarky greentext responses to Slack messages using gpt-4.5.

1
from fastapi import FastAPI, Request
2
from openai import OpenAI
3
from composio_openai import ComposioToolSet, App, Action
4
from dotenv import load_dotenv
5
import uvicorn
6
7
load_dotenv()
8
app = FastAPI()
9
client = OpenAI()
10
toolset = ComposioToolSet()
11
entity = toolset.get_entity(id="default")

1
# Set to keep track of threads we've already responded to
2
responded_threads = set()

1
async def generate_response(payload: Dict[str, Any]):
2
    ts = payload.get("data", {}).get("ts", "")
3
    thread_ts = payload.get("data", {}).get("thread_ts", ts)
4
    channel = payload.get("data", {}).get("channel", "")
5
    
6
    # Skip if already responded
7
    if thread_ts in responded_threads:
8
        return
9
    
10
    responded_threads.add(thread_ts)
11
    
12
    # Preprocessor to automatically inject Slack-specific parameters
13
    def slack_send_message_preprocessor(inputs: Dict[str, Any]) -> Dict[str, Any]:
14
        inputs["thread_ts"] = ts          # Ensure reply goes to the correct thread
15
        inputs["channel"] = channel       # Target the specific channel
16
        inputs["mrkdwn"] = False         # Disable markdown for greentext formatting
17
        return inputs

1
# Configure tools with the preprocessor to handle Slack-specific parameters
2
tools = toolset.get_tools(
3
    [Action.SLACK_SENDS_A_MESSAGE_TO_A_SLACK_CHANNEL],
4
    processors={
5
        "pre": {
6
            Action.SLACK_SENDS_A_MESSAGE_TO_A_SLACK_CHANNEL: slack_send_message_preprocessor
7
        }
8
    }
9
)
10
11
response = client.chat.completions.create(
12
    model="gpt-4.5-preview",
13
    messages=[
14
        {"role": "system", "content": "Given a slack text. Generate a snarky greentext response mocking the user. Render the response in ``` codeblocks"},
15
        {"role": "user", "content": payload.get("data", {}).get("text")}
16
    ],
17
    tools=tools,
18
    tool_choice="required"
19
)
20
toolset.handle_tool_calls(response, entity_id="default")

1
@app.post("/webhook")
2
async def webhook_handler(request: Request):
3
    payload = await request.json()
4
    if payload.get("type") == "slack_receive_message":
5
        channel = payload.get("data", {}).get("channel")
6
        if channel == "YOUR_CHANNEL_ID":  # Replace with your channel ID
7
            await generate_response(payload)
8
    return {"status": "success", "message": "Webhook received"}
9
10
uvicorn.run(app, host="0.0.0.0", port=8000)

$
# Start your FastAPI server
>
python webhook.py
>
>
# In another terminal, start ngrok
>
ngrok http 8000

Troubleshooting

If you encounter issues with triggers or webhook listeners, you can use the Composio dashboard to inspect detailed trigger logs. The dashboard allows you to review event payloads, identify errors, and manually resend events for testing purposes.

Access the trigger logs here