---
sidebar:
  hidden: true
title: roboto.domain.triggers.trigger
---
## Module Contents

### Trigger

```python
class roboto.domain.triggers.trigger.Trigger(
    record: roboto.domain.triggers.record.TriggerRecord,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
)
```

`from roboto.domain.triggers import Trigger`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L45-L387)

A trigger: a firing source (event subscriptions or a schedule), an optional condition, and targets.

Triggers created through the older `causes`/`for_each` API ([`roboto.domain.actions.Trigger`](/reference/python-sdk/roboto/domain/actions/trigger#roboto.domain.actions.trigger.Trigger)) are listed and loaded here too, projected onto this shape and marked `engine == "v1"`; they accept a narrower set of edits (see [`update()`](/reference/python-sdk/roboto/domain/triggers/trigger#roboto.domain.triggers.trigger.Trigger.update)).

**Usage**

Start an agent whenever a dataset gains a `ready` tag:

```python
from roboto.domain.platform_events import OncePer, PlatformEventType
from roboto.domain.triggers import StartAgentTarget, Trigger
trigger = Trigger.create(
    name="analyze-ready-datasets",
    targets=[StartAgentTarget(target_id="analyze", agent_id="ag_abc123")],
    events=[PlatformEventType.DatasetTagAdded],
    once_per=OncePer.Occurrence,
)
```

Post to Slack every Monday at 09:00 UTC:

```python
from roboto.domain.triggers import SendSlackMessageTarget
weekly = Trigger.create(
    name="weekly-status",
    targets=[SendSlackMessageTarget(target_id="post", channel_id="C0123", text="Weekly check-in")],
    schedule="0 9 * * 1",
)
```

**Parameters**

- **record** (`roboto.domain.triggers.record.TriggerRecord`)
- **roboto_client** (`Optional[roboto.http.RobotoClient]`)

**Properties**

- **Trigger.condition** (`roboto.query.ConditionType | None`): Return type: `Optional[roboto.query.ConditionType]`

- **Trigger.enabled** (`bool`)

- **Trigger.engine** (`str`): Which trigger shape this trigger was created in: `"v2"`, or `"v1"` for a trigger created through the older `causes`/`for_each` API.

- **Trigger.events** (`List[roboto.domain.platform_events.PlatformEventType] | None`): The subscribed event types, or `None` for a schedule-fired trigger.

  Return type: `Optional[List[roboto.domain.platform_events.PlatformEventType]]`

- **Trigger.fires_on** (`roboto.domain.triggers.sources.TriggerSource`): What makes the trigger fire: an event subscription or a schedule.

- **Trigger.name** (`str`)

- **Trigger.once_per** (`roboto.domain.platform_events.OncePer`): What the trigger fires at most once per; always `occurrence` (one firing per minute) for a schedule.

- **Trigger.org_id** (`str`)

- **Trigger.record** (`roboto.domain.triggers.record.TriggerRecord`)

- **Trigger.schedule** (`str | None`): The cron expression (UTC) the trigger fires on, or `None` for an event-fired trigger.

  Return type: `Optional[str]`

- **Trigger.targets** (`List[roboto.domain.triggers.targets.TriggerTargetSpec]`)

- **Trigger.trigger_id** (`str`)

#### Trigger.create()

```python
@classmethod
def create(
    name: str,
    targets: list[roboto.domain.triggers.targets.TriggerTargetSpec],
    *,
    events: Optional[list[roboto.domain.platform_events.PlatformEventType]] = None,
    once_per: Optional[roboto.domain.platform_events.OncePer] = None,
    schedule: Optional[str] = None,
    fires_on: Optional[roboto.domain.triggers.sources.TriggerSource] = None,
    condition: Optional[roboto.query.ConditionType] = None,
    enabled: bool = True,
    caller_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L79-L125)

Create a trigger in the caller's org.

**Parameters**

- **name** (`str`): Trigger name, unique within the org.
- **targets** (`list[roboto.domain.triggers.targets.TriggerTargetSpec]`): What to dispatch on a match. At least one.
- **events** (`Optional[list[roboto.domain.platform_events.PlatformEventType]]`): Platform event types to subscribe to (with `once_per`). At least one.
- **once_per** (`Optional[roboto.domain.platform_events.OncePer]`): What the trigger fires at most once per; must be legal for every subscribed event.
- **schedule** (`Optional[str]`): A cron expression (UTC) to fire on instead of events.
- **fires_on** (`Optional[roboto.domain.triggers.sources.TriggerSource]`): The firing source itself, as an alternative to the `events`/`once_per` or `schedule` shorthands.
- **condition** (`Optional[roboto.query.ConditionType]`): Optional predicate over the firing's namespace.
- **enabled** (`bool`): Whether the trigger is active immediately.
- **caller_org_id** (`Optional[str]`): Org to create the trigger in. Defaults to the caller's org.
- **roboto_client** (`Optional[roboto.http.RobotoClient]`): Roboto client instance. Uses the default if not provided.

**Returns**

- `Trigger`: The created trigger.

**Raises**

- [`RobotoInvalidRequestException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoInvalidRequestException): A cross-field validation rule fails (e.g. the condition references a root not exposed by every subscribed event).
- [`RobotoConflictException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoConflictException): The name is already taken in the org.

#### Trigger.delete()

```python
def delete() -> None
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L250-L252)

Delete this trigger. Idempotent.

**Returns**

- `None`

#### Trigger.disable()

```python
def disable() -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L254-L256)

Disable this trigger.

**Returns**

- `Trigger`

#### Trigger.dispatches()

```python
def dispatches(
    limit: int = 100,
) -> collections.abc.Generator[roboto.domain.triggers.dispatch.TriggerDispatchRecord, None, None]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L258-L282)

Yield this trigger's dispatch history, newest first.

A dispatch is one attempt to run one target for one matched event; only matches are recorded, so an empty history means the trigger never fired.

**Parameters**

- **limit** (`int`): Page size for the underlying requests.

**Returns**

- `collections.abc.Generator[roboto.domain.triggers.dispatch.TriggerDispatchRecord, None, None]`

#### Trigger.dry_run()

```python
def dry_run(
    event: Optional[roboto.domain.platform_events.PlatformEvent] = None,
    dataset_id: Optional[str] = None,
    file_id: Optional[str] = None,
    invocation_id: Optional[str] = None,
    event_id: Optional[str] = None,
    event_type: Optional[roboto.domain.platform_events.PlatformEventType] = None,
    scheduled_for: Optional[datetime.datetime] = None,
) -> roboto.domain.triggers.dry_run.TriggerDryRunResponse
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L284-L320)

Ask "would this trigger fire?" without dispatching anything.

Provide either a full `event` or exactly one entity reference, from which the server synthesizes an event (`event_type` optionally picks which kind). A schedule-fired trigger takes no reference: pass `scheduled_for` to pick the minute, or nothing for the schedule's next occurrence. The response is the evaluator's gate-by-gate trace: subscribed, enabled, condition (with per-leaf actual values), target prefilter, already fired.

**Usage**

```python
trace = trigger.dry_run(dataset_id="ds_abc123")
print(trace.verdict)
```

**Parameters**

- **event** (`Optional[roboto.domain.platform_events.PlatformEvent]`)
- **dataset_id** (`Optional[str]`)
- **file_id** (`Optional[str]`)
- **invocation_id** (`Optional[str]`)
- **event_id** (`Optional[str]`)
- **event_type** (`Optional[roboto.domain.platform_events.PlatformEventType]`)
- **scheduled_for** (`Optional[datetime.datetime]`)

**Returns**

- `roboto.domain.triggers.dry_run.TriggerDryRunResponse`

#### Trigger.enable()

```python
def enable() -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L322-L324)

Enable this trigger.

**Returns**

- `Trigger`

#### Trigger.from_id()

```python
@classmethod
def from_id(
    trigger_id: str,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L128-L136)

Load the trigger with the given id, whichever API created it.

**Parameters**

- **trigger_id** (`str`)
- **roboto_client** (`Optional[roboto.http.RobotoClient]`)

**Returns**

- `Trigger`

#### Trigger.from_name()

```python
@classmethod
def from_name(
    name: str,
    owner_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L139-L148)

Load the trigger with the given name, whichever API created it. Names are unique within an org.

**Parameters**

- **name** (`str`)
- **owner_org_id** (`Optional[str]`)
- **roboto_client** (`Optional[roboto.http.RobotoClient]`)

**Returns**

- `Trigger`

#### Trigger.list()

```python
@classmethod
def list(
    owner_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> collections.abc.Generator[Trigger, None, None]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L172-L181)

Yield every trigger in the org, event-fired and scheduled, newest first, whichever API created it.

**Parameters**

- **owner_org_id** (`Optional[str]`)
- **roboto_client** (`Optional[roboto.http.RobotoClient]`)

**Returns**

- `collections.abc.Generator[Trigger, None, None]`

#### Trigger.platform_event_samples()

```python
@classmethod
def platform_event_samples(
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> dict[roboto.domain.platform_events.PlatformEventType, roboto.domain.triggers.samples.PlatformEventSample]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L151-L169)

Fetch a realistic sample of every event type, dereferenced.

Each sample carries the event envelope, every namespace root the type exposes with a full record under it, and the `root.path` list a condition field or `{{ }}` placeholder may name. Use it to see what a template will resolve to before writing one.

**Usage**

```python
samples = Trigger.platform_event_samples()
samples[PlatformEventType.FileUploaded].paths[:3]
# ['envelope.id', 'envelope.type', 'envelope.time']
```

**Parameters**

- **roboto_client** (`Optional[roboto.http.RobotoClient]`)

**Returns**

- `dict[roboto.domain.platform_events.PlatformEventType, roboto.domain.triggers.samples.PlatformEventSample]`

#### Trigger.set_enabled()

```python
def set_enabled(enabled: bool) -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L326-L328)

Enable or disable this trigger.

**Parameters**

- **enabled** (`bool`)

**Returns**

- `Trigger`

#### Trigger.to_dict()

```python
def to_dict() -> dict[str, Any]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L330-L332)

Return this trigger's record as a JSON-able dict.

**Returns**

- `dict[str, Any]`

#### Trigger.update()

```python
def update(
    *,
    fires_on: Union[roboto.domain.triggers.sources.TriggerSource, roboto.sentinels.NotSetType] = NotSet,
    events: Union[List[roboto.domain.platform_events.PlatformEventType], roboto.sentinels.NotSetType] = NotSet,
    once_per: Union[roboto.domain.platform_events.OncePer, roboto.sentinels.NotSetType] = NotSet,
    schedule: Union[str, roboto.sentinels.NotSetType] = NotSet,
    condition: Optional[Union[roboto.query.ConditionType, roboto.sentinels.NotSetType]] = NotSet,
    targets: Union[List[roboto.domain.triggers.targets.TriggerTargetSpec], roboto.sentinels.NotSetType] = NotSet,
    enabled: Union[bool, roboto.sentinels.NotSetType] = NotSet,
) -> Trigger
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/triggers/trigger.py#L334-L387)

Apply a partial update to this trigger and refresh this instance.

Only provided fields change; `condition=None` clears the condition. The firing source is replaced whole: pass `fires_on`, or the `events` / `once_per` / `schedule` shorthands, which are merged over the current source before being sent. A trigger with `engine == "v1"` accepts only edits its older shape can express — a single invoke-action target, event types that map onto it, `once_per` of `file` or `dataset` — and rejects the rest with a message naming what it cannot store.

**Parameters**

- **fires_on** (`Union[roboto.domain.triggers.sources.TriggerSource, roboto.sentinels.NotSetType]`)
- **events** (`Union[List[roboto.domain.platform_events.PlatformEventType], roboto.sentinels.NotSetType]`)
- **once_per** (`Union[roboto.domain.platform_events.OncePer, roboto.sentinels.NotSetType]`)
- **schedule** (`Union[str, roboto.sentinels.NotSetType]`)
- **condition** (`Optional[Union[roboto.query.ConditionType, roboto.sentinels.NotSetType]]`)
- **targets** (`Union[List[roboto.domain.triggers.targets.TriggerTargetSpec], roboto.sentinels.NotSetType]`)
- **enabled** (`Union[bool, roboto.sentinels.NotSetType]`)

**Returns**

- `Trigger`