# Triggers

Triggers let your workflow server notify the Ebbot platform when something happens in the integrated system. A trigger defines a subscription type in the manifest and registers runtime routes that dispatch events to the workflow engine.

## How Triggers are discovered

The workflow server scans a module (usually named `fns`) for functions decorated with `@workflow_trigger` from `integrations_sdk.triggers`. Each decorated function becomes a trigger in the generated manifest.

## Define a Trigger

Triggers register HTTP routes on the FastAPI app and use `dispatch(subscription_id, payload)` to send events to subscribers.

```python
from fastapi import FastAPI
from pydantic import BaseModel
from integrations_sdk.triggers import workflow_trigger


class HookData(BaseModel):
    messageId: str
    message: str


@workflow_trigger(
    description="Message received",
    result=HookData,
    installInstructions="Docs on point, README magic",
    docs="How the hook_trigger trigger works",
)
def hook_trigger(dispatch, app: FastAPI):
    @app.post("/hook-trigger/{subscriptionId}")
    def hook_trigger(subscriptionId: str, data: HookData):
        dispatch(subscriptionId, data)
```

Key points:

* `description`, `docs`, and `installInstructions` are surfaced in the manifest.
* `result` is the Pydantic model or JSON schema for the dispatched payload.
* The inner FastAPI route is where you parse inbound events and call `dispatch`.

## Attach Trigger metadata

Use `TriggerMetadata` to include a reference to the source item that triggered the event. This is helpful when you need to track which external record a dispatch corresponds to, for example when reconciling usage for billing. You can also pass along any additional metadata that you might want to include as data.

```python
from fastapi import FastAPI
from pydantic import BaseModel
from integrations_sdk.triggers import TriggerMetadata, workflow_trigger


class HookData(BaseModel):
    messageId: str
    message: str


@workflow_trigger(
    description="Message received",
    result=HookData,
    installInstructions="Docs on point, README magic",
    docs="How the hook_trigger trigger works",
)
def hook_trigger(dispatch, app: FastAPI):
    @app.post("/hook-trigger/{subscriptionId}")
    def trigger_endpoint(subscriptionId: str, data: HookData):
        metadata = TriggerMetadata(
            referenceId=data.messageId,
            data={"message": data.message},
        )
        dispatch(subscriptionId, data, metadata)
```

## Connection env and subscription options

Triggers can read env/secrets from the connection or from the subscription itself. This drives the subscription schema in the manifest and ensures the platform collects the right data at install time.

### Connection env and secrets

```python
from pydantic import BaseModel
from integrations_sdk.triggers import GetEnvFn, workflow_trigger


class HookDataWithEnv(BaseModel):
    option: str
    secret: str


@workflow_trigger(
    description="env, secrets from connection",
    result=HookDataWithEnv,
    connectionEnv=["notSecret"],
    connectionSecrets=["secret"],
)
def hook_trigger_env_secret(dispatch, app, getEnv: GetEnvFn):
    @app.post("/hook-trigger-env-secret/{subscriptionId}")
    def hook_trigger_secret_env(subscriptionId: str):
        env = getEnv(subscriptionId)
        dispatch(
            subscriptionId,
            HookDataWithEnv(option=env.info["notSecret"], secret=env.secrets["secret"]),
        )
```

### Subscription options and secrets

```python
from pydantic import BaseModel
from integrations_sdk.triggers import GetEnvFn, workflow_trigger


class TriggerEnv(BaseModel):
    option: str


class SecretEnv(BaseModel):
    secret: str


class HookDataWithEnv(BaseModel):
    option: str
    secret: str


@workflow_trigger(
    description="env, secrets on subscription",
    result=HookDataWithEnv,
    triggerOptions=TriggerEnv,
    triggerSecrets=SecretEnv,
)
def hook_trigger_own_env_secret(dispatch, app, getEnv: GetEnvFn):
    @app.post("/hook-trigger-own-env/{subscriptionId}")
    def hook_trigger_secret_env(subscriptionId: str):
        env = getEnv(subscriptionId)
        dispatch(
            subscriptionId,
            {"option": env.info["option"], "secret": env.secrets["secret"]},
        )
```

Notes:

* `connectionEnv` / `connectionSecrets` pull from the connection.
* `triggerOptions` / `triggerSecrets` are collected per subscription.
* `getEnv(subscription_id)` resolves the correct values for that subscription.

## Trigger lifecycle hooks

Use `TriggerEvents` to run logic when a subscription is created. This is useful for provisioning remote webhooks or storing derived configuration.

```python
from integrations_sdk.triggers import TriggerEvents, workflow_trigger
from integrations_sdk.workflow import Subscription


@workflow_trigger(description="", result=HookData)
def on_created(dispatch, app, events: TriggerEvents):
    def call_me_on_created(subscription: Subscription) -> Subscription:
        subscription.options = {"extra_prop": "Property prop"}
        return subscription

    events.on_created(call_me_on_created)
```

If the hook raises an exception, the subscription creation is rolled back.

## Multiple Triggers on one endpoint

Use `with_triggers` when several triggers share a route and you want to dispatch by trigger name.

```python
from typing import Literal
from pydantic import BaseModel
from integrations_sdk.triggers import GetEnvFn, GetSubscriptionsByNameFn, with_triggers


@with_triggers(["hook_trigger", "hook_trigger_own_env_secret"])
def triggers(dispatch, getEnv: GetEnvFn, app, getSubscriptions: GetSubscriptionsByNameFn):
    class Payload(BaseModel):
        type: Literal["hook_trigger", "hook_trigger_secret"]
        messageId: str
        message: str

    @app.post("/multiple-triggers")
    def shared_fn(payload: Payload):
        if payload.type == "hook_trigger":
            for subscription in getSubscriptions("hook_trigger"):
                return dispatch(
                    subscription.id,
                    HookData(messageId=payload.messageId, message=payload.message),
                )
        for subscription in getSubscriptions("hook_trigger_own_env_secret"):
            env = getEnv(subscription.id)
            dispatch(
                subscription.id,
                {"option": env.info["option"], "secret": env.secrets["secret"]},
            )
```

## Subscription schemas in the Manifest

When you set `connectionEnv` / `connectionSecrets` or `triggerOptions` / `triggerSecrets`, the trigger's `subscriptionSchema` is included in the manifest so the Ebbot platform knows which values are required to install the subscription.

## Runtime behavior

At runtime, the workflow server:

* Resolves the subscription by ID.
* Validates incoming payloads with your Pydantic models.
* Dispatches events to the engine for each subscription.

If required env or secrets are missing, dispatching fails with a validation error rather than sending incomplete data.


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.ebbot.ai/ebbot-docs/developer-resources/ebbot-automations/integrations-sdk/triggers.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.