Skip to content

argilla.webhooks

Webhooks are a way for web applications to notify each other when something happens. For example, you might want to be notified when a new dataset is created in Argilla.

Usage Examples

To listen for incoming webhooks, you can use the webhook_listener decorator function to register a function to be called when a webhook is received:

from argilla.webhooks import webhook_listener

@webhook_listener(events="dataset.created")
async def my_webhook_listener(dataset):
    print(dataset)

To manually create a new webhook, instantiate the Webhook object with the client and the name:

webhook = rg.Webhook(
    url="https://somehost.com/webhook",
    events=["dataset.created"],
    description="My webhook"
)
webhook.create()

To retrieve a list of existing webhooks, use the client.webhooks attribute:

for webhook in client.webhooks():
    print(webhook)

Webhook

Bases: Resource

The Webhook resource. It represents a webhook that can be used to receive events from the Argilla Server.

Parameters:

Name Type Description Default
url str

The URL of the webhook endpoint.

 required
events List[EventType]

The events that the webhook is subscribed to.

 required
description Optional[str]

The description of the webhook.

 None
_client Argilla

The client used to interact with the Argilla Server.

 None
Source code in src/argilla/webhooks/_resource.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class Webhook(Resource):
    """
    The `Webhook` resource. It represents a webhook that can be used to receive events from the Argilla Server.

    Args:
        url (str): The URL of the webhook endpoint.
        events (List[EventType]): The events that the webhook is subscribed to.
        description (Optional[str]): The description of the webhook.
        _client (Argilla): The client used to interact with the Argilla Server.

    """

    _model: WebhookModel
    _api: WebhooksAPI

    def __init__(self, url: str, events: List[EventType], description: Optional[str] = None, _client: Argilla = None):
        client = _client or Argilla._get_default()
        api = client.api.webhooks
        events = events or []

        super().__init__(api=api, client=client)

        self._model = WebhookModel(url=url, events=list(events), description=description)

    @property
    def url(self) -> str:
        """The URL of the webhook."""
        return self._model.url

    @url.setter
    def url(self, value: str):
        self._model.url = value

    @property
    def events(self) -> List[EventType]:
        """The events that the webhook is subscribed to."""
        return self._model.events

    @events.setter
    def events(self, value: List[EventType]):
        self._model.events = value

    @property
    def enabled(self) -> bool:
        """Whether the webhook is enabled."""
        return self._model.enabled

    @enabled.setter
    def enabled(self, value: bool):
        self._model.enabled = value

    @property
    def description(self) -> Optional[str]:
        """The description of the webhook."""
        return self._model.description

    @description.setter
    def description(self, value: Optional[str]):
        self._model.description = value

    @property
    def secret(self) -> str:
        """The secret of the webhook."""
        return self._model.secret

    @classmethod
    def from_model(cls, model: WebhookModel, client: Optional["Argilla"] = None) -> "Webhook":
        instance = cls(url=model.url, events=model.events, _client=client)
        instance._model = model

        return instance

    def _with_client(self, client: "Argilla") -> "Webhook":
        self._client = client
        self._api = client.api.webhooks

        return self

url: str property writable

The URL of the webhook.

events: List[EventType] property writable

The events that the webhook is subscribed to.

enabled: bool property writable

Whether the webhook is enabled.

description: Optional[str] property writable

The description of the webhook.

secret: str property

The secret of the webhook.

webhook_listener(events, description=None, client=None, server=None, raw_event=False)

Decorator to create a webhook listener for a function.

Parameters:

Name Type Description Default
events Union[str, List[str]]

The events to listen to.

 required
description Optional[str]

The description of the webhook.

 None
client Optional[Argilla]

The Argilla client to use. Defaults to the default client.

 None
server Optional[FastAPI]

The FastAPI server to use. Defaults to the default server.

 None
raw_event bool

Whether to pass the raw event to the function. Defaults to False.

 False

Returns:

Name Type Description
Callable Callable

The decorated function.
Source code in src/argilla/webhooks/_helpers.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def webhook_listener(
    events: Union[str, List[str]],
    description: Optional[str] = None,
    client: Optional["Argilla"] = None,
    server: Optional["FastAPI"] = None,
    raw_event: bool = False,
) -> Callable:
    """
    Decorator to create a webhook listener for a function.

    Parameters:
        events (Union[str, List[str]]): The events to listen to.
        description (Optional[str]): The description of the webhook.
        client (Optional[Argilla]): The Argilla client to use. Defaults to the default client.
        server (Optional[FastAPI]): The FastAPI server to use. Defaults to the default server.
        raw_event (bool): Whether to pass the raw event to the function. Defaults to False.

    Returns:
        Callable: The decorated function.

    """

    client = client or rg.Argilla._get_default()
    server = server or get_webhook_server()

    if isinstance(events, str):
        events = [events]

    def wrapper(func: Callable) -> Callable:
        webhook_url = _webhook_url_for_func(func)

        webhook = None
        for argilla_webhook in client.webhooks:
            if argilla_webhook.url == webhook_url and argilla_webhook.events == events:
                warnings.warn(f"Found existing webhook with for URL {argilla_webhook.url}: {argilla_webhook}")
                webhook = argilla_webhook
                webhook.description = description or webhook.description
                webhook.enabled = True
                webhook.update()
                break

        if not webhook:
            webhook = Webhook(
                url=webhook_url,
                events=events,
                description=description or f"Webhook for {func.__name__}",
            ).create()

        request_handler = WebhookHandler(webhook).handle(func, raw_event)
        server.post(f"/{func.__name__}", tags=["Argilla Webhooks"])(request_handler)

        return request_handler

    return wrapper

get_webhook_server()

Get the current webhook server. If it does not exist, it will create one.

Returns:

Name Type Description
FastAPI FastAPI

The webhook server.
Source code in src/argilla/webhooks/_helpers.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def get_webhook_server() -> "FastAPI":
    """
    Get the current webhook server. If it does not exist, it will create one.

    Returns:
        FastAPI: The webhook server.

    """
    from fastapi import FastAPI

    global _server
    if not _server:
        _server = FastAPI()
    return _server

set_webhook_server(app)

Set the webhook server. This should only be called once.

Parameters:

Name Type Description Default
app FastAPI

The webhook server.

 required
Source code in src/argilla/webhooks/_helpers.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def set_webhook_server(app: "FastAPI"):
    """
    Set the webhook server. This should only be called once.

    Parameters:
        app (FastAPI): The webhook server.

    """
    global _server

    if _server:
        raise ValueError("Server already set")

    _server = app

WebhookHandler

The WebhookHandler class is used to handle incoming webhook requests. This class handles the request verification and event object creation.

Attributes:

Name Type Description
webhook Webhook

The webhook object.
Source code in src/argilla/webhooks/_handler.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class WebhookHandler:
    """
    The `WebhookHandler` class is used to handle incoming webhook requests. This class handles the
    request verification and event object creation.

    Attributes:
        webhook (Webhook): The webhook object.
    """

    def __init__(self, webhook: "Webhook"):
        self.webhook = webhook

    def handle(self, func: Callable, raw_event: bool = False) -> Callable:
        """
        This method handles the incoming webhook requests and calls the provided function.

        Parameters:
            func (Callable): The function to be called when a webhook event is received.
            raw_event (bool): Whether to pass the raw event object to the function.

        Returns:

        """
        from fastapi import Request

        async def request_handler(request: Request):
            event = await self._verify_request(request)
            if event.type not in self.webhook.events:
                return

            if raw_event:
                return await func(event)

            return await func(**event.parsed(self.webhook._client).model_dump())

        return request_handler

    async def _verify_request(self, request: "Request") -> WebhookEvent:
        """
        Verify the request signature and return the event object.

        Arguments:
            request (Request): The request object.

        Returns:
            WebhookEvent: The event object.
        """

        from standardwebhooks.webhooks import Webhook

        body = await request.body()
        headers = dict(request.headers)

        json = Webhook(whsecret=self.webhook.secret).verify(body, headers)
        return WebhookEvent.model_validate(json)

handle(func, raw_event=False)

This method handles the incoming webhook requests and calls the provided function.

Parameters:

Name Type Description Default
func Callable

The function to be called when a webhook event is received.

 required
raw_event bool

Whether to pass the raw event object to the function.

 False

Returns:

Source code in src/argilla/webhooks/_handler.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def handle(self, func: Callable, raw_event: bool = False) -> Callable:
    """
    This method handles the incoming webhook requests and calls the provided function.

    Parameters:
        func (Callable): The function to be called when a webhook event is received.
        raw_event (bool): Whether to pass the raw event object to the function.

    Returns:

    """
    from fastapi import Request

    async def request_handler(request: Request):
        event = await self._verify_request(request)
        if event.type not in self.webhook.events:
            return

        if raw_event:
            return await func(event)

        return await func(**event.parsed(self.webhook._client).model_dump())

    return request_handler

WebhookEvent

Bases: BaseModel

A webhook event.

Attributes:

Name Type Description
type EventType

The type of the event.
timestamp datetime

The timestamp of the event.
data dict

The data of the event.
Source code in src/argilla/webhooks/_event.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class WebhookEvent(BaseModel):
    """
    A webhook event.

    Attributes:
        type (EventType): The type of the event.
        timestamp (datetime): The timestamp of the event.
        data (dict): The data of the event.
    """

    type: EventType
    timestamp: datetime
    data: dict

    def parsed(self, client: "Argilla") -> Union[RecordEvent, DatasetEvent, UserResponseEvent, "WebhookEvent"]:
        """
        Parse the webhook event.

        Args:
            client: The Argilla client.

        Returns:
            Event: The parsed event.

        """
        resource = self.type.resource
        data = self.data or {}

        if resource == "dataset":
            dataset = self._parse_dataset_from_webhook_data(data, client)
            return DatasetEvent(
                type=self.type,
                timestamp=self.timestamp,
                dataset=dataset,
            )

        elif resource == "record":
            record = self._parse_record_from_webhook_data(data, client)
            return RecordEvent(
                type=self.type,
                timestamp=self.timestamp,
                record=record,
            )

        elif resource == "response":
            user_response = self._parse_response_from_webhook_data(data, client)
            return UserResponseEvent(
                type=self.type,
                timestamp=self.timestamp,
                response=user_response,
            )

        return self

    @classmethod
    def _parse_dataset_from_webhook_data(cls, data: dict, client: "Argilla") -> Dataset:
        workspace = Workspace.from_model(WorkspaceModel.model_validate(data["workspace"]), client=client)
        # TODO: Parse settings from the data
        # settings = Settings._from_dict(data)

        dataset = Dataset(name=data["name"], workspace=workspace, client=client)
        dataset.id = UUID(data["id"])

        try:
            dataset.get()
        except ArgillaAPIError as _:
            # TODO: Show notification
            pass
        finally:
            return dataset

    @classmethod
    def _parse_record_from_webhook_data(cls, data: dict, client: "Argilla") -> Record:
        dataset = cls._parse_dataset_from_webhook_data(data["dataset"], client)

        record = Record.from_model(RecordModel.model_validate(data), dataset=dataset)
        try:
            record.get()
        except ArgillaAPIError as _:
            # TODO: Show notification
            pass
        finally:
            return record

    @classmethod
    def _parse_response_from_webhook_data(cls, data: dict, client: "Argilla") -> UserResponse:
        record = cls._parse_record_from_webhook_data(data["record"], client)

        # TODO: Link the user resource to the response
        user_response = UserResponse.from_model(
            model=UserResponseModel(**data, user_id=data["user"]["id"]),
            record=record,
        )

        return user_response

parsed(client)

Parse the webhook event.

Parameters:

Name Type Description Default
client Argilla

The Argilla client.

 required

Returns:

Name Type Description
Event Union[RecordEvent, DatasetEvent, UserResponseEvent, WebhookEvent]

The parsed event.
Source code in src/argilla/webhooks/_event.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def parsed(self, client: "Argilla") -> Union[RecordEvent, DatasetEvent, UserResponseEvent, "WebhookEvent"]:
    """
    Parse the webhook event.

    Args:
        client: The Argilla client.

    Returns:
        Event: The parsed event.

    """
    resource = self.type.resource
    data = self.data or {}

    if resource == "dataset":
        dataset = self._parse_dataset_from_webhook_data(data, client)
        return DatasetEvent(
            type=self.type,
            timestamp=self.timestamp,
            dataset=dataset,
        )

    elif resource == "record":
        record = self._parse_record_from_webhook_data(data, client)
        return RecordEvent(
            type=self.type,
            timestamp=self.timestamp,
            record=record,
        )

    elif resource == "response":
        user_response = self._parse_response_from_webhook_data(data, client)
        return UserResponseEvent(
            type=self.type,
            timestamp=self.timestamp,
            response=user_response,
        )

    return self

DatasetEvent

Bases: BaseModel

A parsed dataset event.

Attributes:

Name Type Description
type EventType

The type of the event.
timestamp datetime

The timestamp of the event.
dataset Dataset

The dataset of the event.
Source code in src/argilla/webhooks/_event.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class DatasetEvent(BaseModel):
    """
    A parsed dataset event.

    Attributes:
        type (EventType): The type of the event.
        timestamp (datetime): The timestamp of the event.
        dataset (Dataset): The dataset of the event.
    """

    type: EventType
    timestamp: datetime
    dataset: Dataset

    model_config = ConfigDict(arbitrary_types_allowed=True)

RecordEvent

Bases: BaseModel

A parsed record event.

Attributes:

Name Type Description
type EventType

The type of the event.
timestamp datetime

The timestamp of the event.
record Record

The record of the event.
Source code in src/argilla/webhooks/_event.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class RecordEvent(BaseModel):
    """
    A parsed record event.

    Attributes:
        type (EventType): The type of the event.
        timestamp (datetime): The timestamp of the event.
        record (Record): The record of the event.
    """

    type: EventType
    timestamp: datetime
    record: Record

    model_config = ConfigDict(arbitrary_types_allowed=True)

UserResponseEvent

Bases: BaseModel

A parsed user response event.

Attributes:

Name Type Description
type EventType

The type of the event.
timestamp datetime

The timestamp of the event.
response UserResponse

The user response of the event.
Source code in src/argilla/webhooks/_event.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class UserResponseEvent(BaseModel):
    """
    A parsed user response event.

    Attributes:
        type (EventType): The type of the event.
        timestamp (datetime): The timestamp of the event.
        response (UserResponse): The user response of the event.
    """

    type: EventType
    timestamp: datetime
    response: UserResponse

    model_config = ConfigDict(arbitrary_types_allowed=True)