Skip to content

events

Class info

Classes

Name Children Inherits
ConnectionContentCondition
llmling_agent.models.events
Simple content matching for connection events.
    ConnectionEventCondition
    llmling_agent.models.events
    Base conditions specifically for connection events.
    ConnectionJinja2Condition
    llmling_agent.models.events
    Flexible Jinja2 condition for connection events.
      ConnectionTriggerConfig
      llmling_agent.models.events
      Trigger config specifically for connection events.
        EmailConfig
        llmling_agent.models.events
        Email event source configuration.
          EventSourceConfig
          llmling_agent.models.events
          Base configuration for event sources.
          FileWatchConfig
          llmling_agent.models.events
          File watching event source.
            TimeEventConfig
            llmling_agent.models.events
            Time-based event source configuration.
              WebhookConfig
              llmling_agent.models.events
              Webhook event source.

                🛈 DocStrings

                Event sources for LLMling agent.

                ConnectionContentCondition

                Bases: ConnectionEventCondition

                Simple content matching for connection events.

                Source code in src/llmling_agent/models/events.py
                218
                219
                220
                221
                222
                223
                224
                225
                226
                227
                228
                229
                230
                231
                232
                233
                234
                class ConnectionContentCondition(ConnectionEventCondition):
                    """Simple content matching for connection events."""
                
                    type: Literal["content"] = Field("content", init=False)
                    """Content-based trigger."""
                
                    words: list[str]
                    """List of words to match."""
                
                    mode: Literal["any", "all"] = "any"
                    """Matching mode."""
                
                    async def check(self, event: ConnectionEventData[Any]) -> bool:
                        if not event.message:
                            return False
                        text = str(event.message.content)
                        return any(word in text for word in self.words)
                

                mode class-attribute instance-attribute

                mode: Literal['any', 'all'] = 'any'
                

                Matching mode.

                type class-attribute instance-attribute

                type: Literal['content'] = Field('content', init=False)
                

                Content-based trigger.

                words instance-attribute

                words: list[str]
                

                List of words to match.

                ConnectionEventCondition

                Bases: BaseModel

                Base conditions specifically for connection events.

                Source code in src/llmling_agent/models/events.py
                209
                210
                211
                212
                213
                214
                215
                class ConnectionEventCondition(BaseModel):
                    """Base conditions specifically for connection events."""
                
                    type: str = Field(init=False)
                
                    async def check(self, event: ConnectionEventData[Any]) -> bool:
                        raise NotImplementedError
                

                ConnectionJinja2Condition

                Bases: ConnectionEventCondition

                Flexible Jinja2 condition for connection events.

                Source code in src/llmling_agent/models/events.py
                237
                238
                239
                240
                241
                242
                243
                244
                245
                246
                247
                248
                249
                250
                251
                252
                class ConnectionJinja2Condition(ConnectionEventCondition):
                    """Flexible Jinja2 condition for connection events."""
                
                    type: Literal["jinja2"] = Field("jinja2", init=False)
                    """Jinja2-based trigger configuration."""
                
                    template: str
                    """Jinja2-Template (needs to return a "boolean" string)."""
                
                    async def check(self, event: ConnectionEventData[Any]) -> bool:
                        from jinja2 import Environment
                
                        env = Environment()
                        template = env.from_string(self.template)
                        result = template.render(event=event)
                        return result.strip().lower() == "true"
                

                template instance-attribute

                template: str
                

                Jinja2-Template (needs to return a "boolean" string).

                type class-attribute instance-attribute

                type: Literal['jinja2'] = Field('jinja2', init=False)
                

                Jinja2-based trigger configuration.

                ConnectionTriggerConfig

                Bases: EventSourceConfig

                Trigger config specifically for connection events.

                Source code in src/llmling_agent/models/events.py
                160
                161
                162
                163
                164
                165
                166
                167
                168
                169
                170
                171
                172
                173
                174
                175
                176
                177
                178
                179
                180
                181
                182
                183
                184
                185
                186
                187
                188
                189
                190
                191
                192
                193
                194
                195
                196
                class ConnectionTriggerConfig(EventSourceConfig):
                    """Trigger config specifically for connection events."""
                
                    type: Literal["connection"] = Field("connection", init=False)
                    """Connection event trigger."""
                
                    source: str | None = None
                    """Connection source name."""
                
                    target: str | None = None
                    """Connection to trigger."""
                
                    event: ConnectionEventType
                    """Event type to trigger on."""
                
                    condition: ConnectionEventConditionType | None = None
                    """Condition-based filter for the event."""
                
                    async def matches_event(self, event: ConnectionEventData[Any]) -> bool:
                        """Check if this trigger matches the event."""
                        # First check event type
                        if event.event_type != self.event:
                            return False
                
                        # Check source/target filters
                        if self.source and event.connection.source.name != self.source:
                            return False
                        if self.target and not any(
                            t.name == self.target for t in event.connection.targets
                        ):
                            return False
                
                        # Check condition if any
                        if self.condition:
                            return await self.condition.check(event)
                
                        return True
                

                condition class-attribute instance-attribute

                condition: ConnectionEventConditionType | None = None
                

                Condition-based filter for the event.

                event instance-attribute

                event: ConnectionEventType
                

                Event type to trigger on.

                source class-attribute instance-attribute

                source: str | None = None
                

                Connection source name.

                target class-attribute instance-attribute

                target: str | None = None
                

                Connection to trigger.

                type class-attribute instance-attribute

                type: Literal['connection'] = Field('connection', init=False)
                

                Connection event trigger.

                matches_event async

                matches_event(event: ConnectionEventData[Any]) -> bool
                

                Check if this trigger matches the event.

                Source code in src/llmling_agent/models/events.py
                178
                179
                180
                181
                182
                183
                184
                185
                186
                187
                188
                189
                190
                191
                192
                193
                194
                195
                196
                async def matches_event(self, event: ConnectionEventData[Any]) -> bool:
                    """Check if this trigger matches the event."""
                    # First check event type
                    if event.event_type != self.event:
                        return False
                
                    # Check source/target filters
                    if self.source and event.connection.source.name != self.source:
                        return False
                    if self.target and not any(
                        t.name == self.target for t in event.connection.targets
                    ):
                        return False
                
                    # Check condition if any
                    if self.condition:
                        return await self.condition.check(event)
                
                    return True
                

                EmailConfig

                Bases: EventSourceConfig

                Email event source configuration.

                Monitors an email inbox for new messages and converts them to events.

                Source code in src/llmling_agent/models/events.py
                116
                117
                118
                119
                120
                121
                122
                123
                124
                125
                126
                127
                128
                129
                130
                131
                132
                133
                134
                135
                136
                137
                138
                139
                140
                141
                142
                143
                144
                145
                146
                147
                148
                149
                150
                151
                152
                153
                154
                155
                156
                157
                class EmailConfig(EventSourceConfig):
                    """Email event source configuration.
                
                    Monitors an email inbox for new messages and converts them to events.
                    """
                
                    type: Literal["email"] = Field("email", init=False)
                    """Email event."""
                
                    host: str = Field(description="IMAP server hostname")
                    """IMAP server hostname (e.g. 'imap.gmail.com')"""
                
                    port: int = Field(default=993)
                    """Server port (defaults to 993 for IMAP SSL)"""
                
                    username: str
                    """Email account username/address"""
                
                    password: SecretStr
                    """Account password or app-specific password"""
                
                    folder: str = Field(default="INBOX")
                    """Folder/mailbox to monitor"""
                
                    ssl: bool = Field(default=True)
                    """Whether to use SSL/TLS connection"""
                
                    check_interval: int = Field(
                        default=60, gt=0, description="Seconds between inbox checks"
                    )
                    """How often to check for new emails (in seconds)"""
                
                    mark_seen: bool = Field(default=True)
                    """Whether to mark processed emails as seen"""
                
                    filters: dict[str, str] = Field(
                        default_factory=dict, description="Email filtering criteria"
                    )
                    """Filtering rules for emails (subject, from, etc)"""
                
                    max_size: int | None = Field(default=None, description="Maximum email size in bytes")
                    """Size limit for processed emails"""
                

                check_interval class-attribute instance-attribute

                check_interval: int = Field(default=60, gt=0, description='Seconds between inbox checks')
                

                How often to check for new emails (in seconds)

                filters class-attribute instance-attribute

                filters: dict[str, str] = Field(
                    default_factory=dict, description="Email filtering criteria"
                )
                

                Filtering rules for emails (subject, from, etc)

                folder class-attribute instance-attribute

                folder: str = Field(default='INBOX')
                

                Folder/mailbox to monitor

                host class-attribute instance-attribute

                host: str = Field(description='IMAP server hostname')
                

                IMAP server hostname (e.g. 'imap.gmail.com')

                mark_seen class-attribute instance-attribute

                mark_seen: bool = Field(default=True)
                

                Whether to mark processed emails as seen

                max_size class-attribute instance-attribute

                max_size: int | None = Field(default=None, description='Maximum email size in bytes')
                

                Size limit for processed emails

                password instance-attribute

                password: SecretStr
                

                Account password or app-specific password

                port class-attribute instance-attribute

                port: int = Field(default=993)
                

                Server port (defaults to 993 for IMAP SSL)

                ssl class-attribute instance-attribute

                ssl: bool = Field(default=True)
                

                Whether to use SSL/TLS connection

                type class-attribute instance-attribute

                type: Literal['email'] = Field('email', init=False)
                

                Email event.

                username instance-attribute

                username: str
                

                Email account username/address

                EventSourceConfig

                Bases: BaseModel

                Base configuration for event sources.

                Source code in src/llmling_agent/models/events.py
                35
                36
                37
                38
                39
                40
                41
                42
                43
                44
                45
                46
                47
                48
                49
                50
                51
                52
                53
                54
                55
                56
                class EventSourceConfig(BaseModel):
                    """Base configuration for event sources."""
                
                    type: str = Field(init=False)
                    """Discriminator field for event source types."""
                
                    name: str
                    """Unique identifier for this event source."""
                
                    enabled: bool = True
                    """Whether this event source is active."""
                
                    template: str = DEFAULT_TEMPLATE
                    """Jinja2 template for formatting events."""
                
                    include_metadata: bool = True
                    """Control metadata visibility in template."""
                
                    include_timestamp: bool = True
                    """Control timestamp visibility in template."""
                
                    model_config = ConfigDict(frozen=True, use_attribute_docstrings=True, extra="forbid")
                

                enabled class-attribute instance-attribute

                enabled: bool = True
                

                Whether this event source is active.

                include_metadata class-attribute instance-attribute

                include_metadata: bool = True
                

                Control metadata visibility in template.

                include_timestamp class-attribute instance-attribute

                include_timestamp: bool = True
                

                Control timestamp visibility in template.

                name instance-attribute

                name: str
                

                Unique identifier for this event source.

                template class-attribute instance-attribute

                template: str = DEFAULT_TEMPLATE
                

                Jinja2 template for formatting events.

                type class-attribute instance-attribute

                type: str = Field(init=False)
                

                Discriminator field for event source types.

                FileWatchConfig

                Bases: EventSourceConfig

                File watching event source.

                Source code in src/llmling_agent/models/events.py
                59
                60
                61
                62
                63
                64
                65
                66
                67
                68
                69
                70
                71
                72
                73
                74
                75
                76
                77
                78
                class FileWatchConfig(EventSourceConfig):
                    """File watching event source."""
                
                    type: Literal["file"] = Field("file", init=False)
                    """File / folder content change events."""
                
                    paths: list[str]
                    """Paths or patterns to watch for changes."""
                
                    extensions: list[str] | None = None
                    """File extensions to monitor (e.g. ['.py', '.md'])."""
                
                    ignore_paths: list[str] | None = None
                    """Paths or patterns to ignore."""
                
                    recursive: bool = True
                    """Whether to watch subdirectories."""
                
                    debounce: int = 1600
                    """Minimum time (ms) between trigger events."""
                

                debounce class-attribute instance-attribute

                debounce: int = 1600
                

                Minimum time (ms) between trigger events.

                extensions class-attribute instance-attribute

                extensions: list[str] | None = None
                

                File extensions to monitor (e.g. ['.py', '.md']).

                ignore_paths class-attribute instance-attribute

                ignore_paths: list[str] | None = None
                

                Paths or patterns to ignore.

                paths instance-attribute

                paths: list[str]
                

                Paths or patterns to watch for changes.

                recursive class-attribute instance-attribute

                recursive: bool = True
                

                Whether to watch subdirectories.

                type class-attribute instance-attribute

                type: Literal['file'] = Field('file', init=False)
                

                File / folder content change events.

                TimeEventConfig

                Bases: EventSourceConfig

                Time-based event source configuration.

                Source code in src/llmling_agent/models/events.py
                 97
                 98
                 99
                100
                101
                102
                103
                104
                105
                106
                107
                108
                109
                110
                111
                112
                113
                class TimeEventConfig(EventSourceConfig):
                    """Time-based event source configuration."""
                
                    type: Literal["time"] = Field("time", init=False)
                    """Time event."""
                
                    schedule: str
                    """Cron expression for scheduling (e.g. '0 9 * * 1-5' for weekdays at 9am)"""
                
                    prompt: str
                    """Prompt to send to the agent when the schedule triggers."""
                
                    timezone: str | None = None
                    """Timezone for schedule (defaults to system timezone)"""
                
                    skip_missed: bool = False
                    """Whether to skip executions missed while agent was inactive"""
                

                prompt instance-attribute

                prompt: str
                

                Prompt to send to the agent when the schedule triggers.

                schedule instance-attribute

                schedule: str
                

                Cron expression for scheduling (e.g. '0 9 * * 1-5' for weekdays at 9am)

                skip_missed class-attribute instance-attribute

                skip_missed: bool = False
                

                Whether to skip executions missed while agent was inactive

                timezone class-attribute instance-attribute

                timezone: str | None = None
                

                Timezone for schedule (defaults to system timezone)

                type class-attribute instance-attribute

                type: Literal['time'] = Field('time', init=False)
                

                Time event.

                WebhookConfig

                Bases: EventSourceConfig

                Webhook event source.

                Source code in src/llmling_agent/models/events.py
                81
                82
                83
                84
                85
                86
                87
                88
                89
                90
                91
                92
                93
                94
                class WebhookConfig(EventSourceConfig):
                    """Webhook event source."""
                
                    type: Literal["webhook"] = Field("webhook", init=False)
                    """webhook-based event."""
                
                    port: int
                    """Port to listen on."""
                
                    path: str
                    """URL path to handle requests."""
                
                    secret: str | None = None
                    """Optional secret for request validation."""
                

                path instance-attribute

                path: str
                

                URL path to handle requests.

                port instance-attribute

                port: int
                

                Port to listen on.

                secret class-attribute instance-attribute

                secret: str | None = None
                

                Optional secret for request validation.

                type class-attribute instance-attribute

                type: Literal['webhook'] = Field('webhook', init=False)
                

                webhook-based event.