Creating Webhooks With the Canvas SDK
Webhooks are user-defined callbacks that make a request to an API. You may want to create a webhook to notify a system you control that some event has occurred in Canvas. This guide shows how to create a webhook that sends an API request containing the ID of a Task upon its creation.
Initialize a new plugin #
The Canvas CLI gives you a great head start when creating a plugin. Simply run canvas init, and answer the prompt to name your plugin.
$ canvas init
[1/1] project_name (My Cool Plugin): Task Webhook
Project created in /Users/andrew/src/canvas-plugins/task_webhook
This output shows the location of our freshly generated plugin.
Edit the plugin code #
The default content of this file shows you the information you have available to you in the comments. I’ve stripped it down to almost nothing so we can layer in the functionality step by step.
Log a message when a task is created. #
The code below listens for the TASK_CREATED event and logs the string “A Task was created!”.
from canvas_sdk.events import EventType
from canvas_sdk.protocols import BaseProtocol
from logger import log
class Protocol(BaseProtocol):
"""
When a task is created, log a message
"""
RESPONDS_TO = EventType.Name(EventType.TASK_CREATED)
def compute(self):
"""
Log a message as tasks are created.
"""
log.info("A Task was created!")
return []
You can see this log output by first streaming logs with canvas logs and then creating a task. You can create this task with Canvas Chat, a Task Command, or our FHIR Task Create endpoint.
After you’ve installed your plugin and created a task, you should see this in your log stream:
INFO 2024-09-26 17:04:08,396 Starting server, listening on port 50051
INFO 2024-09-26 17:04:08,396 Loading custom-plugins/task_webhook
INFO 2024-09-26 17:04:08,396 Loading plugin 'task_webhook:task_webhook.protocols.my_protocol:Protocol'
INFO 2024-09-26 17:04:24,410 A Task was created!
INFO 2024-09-26 17:04:24,410 task_webhook:task_webhook.protocols.my_protocol:Protocol.compute() completed (0 ms)
INFO 2024-09-26 17:04:24,411 Responded to Event TASK_CREATED (1 ms)
Awesome! But we’re not here to log, we need to make an API request. To do that, we need to use the HTTP client found in the Canvas SDK’s Utils module
Make an HTTP request when a Task is created #
We can use https://webhook.site/ for a quick way to test our webhook and see the requests it receives. Here is the updated code that uses the HTTP client to make the request:
from canvas_sdk.events import EventType
from canvas_sdk.protocols import BaseProtocol
from canvas_sdk.utils import Http
from logger import log
class Protocol(BaseProtocol):
"""
When a task is created, hit a webhook
"""
RESPONDS_TO = EventType.Name(EventType.TASK_CREATED)
def compute(self):
"""
Notify our server of tasks as they are created.
"""
url = "https://webhook.site/ee7aed78-b652-4d9e-b858-04465c409d15"
payload = {
"message": "A Task was created!"
}
http = Http()
response = http.post(url, json=payload)
if response.ok:
log.info("Successfully notified API of task creation!")
else:
log.info("Notification unsuccessful. =[")
return []
After you’ve installed your updated plugin and created a task, you should see this in your log stream:
INFO 2024-09-26 17:18:23,206 Loading custom-plugins/task_webhook
INFO 2024-09-26 17:18:23,207 Reloading plugin 'task_webhook:task_webhook.protocols.my_protocol:Protocol'
INFO 2024-09-26 17:18:33,850 Successfully notified API of task creation!
INFO 2024-09-26 17:18:33,851 task_webhook:task_webhook.protocols.my_protocol:Protocol.compute() completed (693 ms)
INFO 2024-09-26 17:18:33,851 Responded to Event TASK_CREATED (696 ms)
This log output is a great reminder for me to mention that making HTTP requests to external servers will slow plugin execution while it waits on the external server to respond. It’s a good idea to make sure the servers you’re hitting have a sufficiently quick response time.
Checking in on our webhook.site logs shows it received our request! 
Awesome, but you aren’t sending these requests to a server that allows unauthenticated requests! And this request doesn’t even tell you anything about the task it’s notifying you about. This isn’t useful at all!
Let’s do another iteration, this time using the information provided along with the event so that we can send a usable message, securely. We’re specifically going to incorporate the event’s target and secrets.
Make an authenticated HTTP request that includes the newly created Task’s ID #
Within your Protocol class, you have access to self.target, which represents the ID of the subject of the event. In our case, it will be the Task’s ID. This is the same ID used in the FHIR Task endpoints, so you can use it to make FHIR API requests.
You also have access to self.secrets, which is a python dictionary containing the key-value pairs from your plugins configuration page. You declare the keys in your CANVAS_MANIFEST.json, and can then set the values after the plugin is installed.
We’ll set two secrets, one for the unique id of the webhook, and one for an auth token. Here’s what the manifest file looks like with secrets declared:
{
"sdk_version": "0.1.4",
"plugin_version": "0.0.1",
"name": "task_webhook",
"description": "Webhooks for task creation",
"components": {
"protocols": [
{
"class": "task_webhook.protocols.my_protocol:Protocol",
"description": "Hit an API when a task is created",
"data_access": {
"event": "",
"read": [],
"write": []
}
}
]
},
"secrets": ["WEBHOOK_ID", "AUTH_TOKEN"],
"tags": {},
"license": "",
"readme": "./README.md"
}
The line "secrets": ["WEBHOOK_ID", "AUTH_TOKEN"], declares two secrets, WEBHOOK_ID and AUTH_TOKEN. After we update the plugin, we can set values for these in the plugin configuration page. This allows for different values to be used across different installations.
Here’s how that configuration looks:
With those values set, we can use them in our code:
from canvas_sdk.events import EventType
from canvas_sdk.protocols import BaseProtocol
from canvas_sdk.utils import Http
from logger import log
class Protocol(BaseProtocol):
"""
When a task is created, hit a webhook
"""
RESPONDS_TO = EventType.Name(EventType.TASK_CREATED)
def compute(self):
"""
Notify our server of tasks as they are created.
"""
url = f"https://webhook.site/{self.secrets['WEBHOOK_ID']}"
headers = {
"Authorization": f"Bearer {self.secrets['AUTH_TOKEN']}"
}
payload = {
"message": "A Task was created!",
"resource_id": self.target
}
http = Http()
response = http.post(url, json=payload, headers=headers)
if response.ok:
log.info("Successfully notified API of task creation!")
else:
log.info("Notification unsuccessful. =[")
return []
And checking once more on our webhook.site logs shows it received our updated request, including our AUTH_TOKEN value and the created task’s ID.
Conclusion #
I hope you found this helpful, remember that you can take this example even further by listening for many other events and customizing the request based on which event fired, just set RESPONDS_TO to a list of event types instead of a single type. With this approach, one class could provide webhooks for many different data types. Alternatively, you could include several classes, each resposible for some specific request type. When including several classes in one plugin, they all have access to the same secrets dictionary, you just need to declare each class in the manifest file.
Happy coding, I can’t wait to see what you build!