Skip to main content
Webhooks enable Infrahub to send HTTP POST requests to external systems when events occur, allowing real-time integration with CI/CD pipelines, monitoring systems, and custom automation.

​
Webhook Types

Infrahub supports three webhook types:

​
Standard Webhook

Simple webhook that sends event data as-is:
  • Sends raw event payload
  • No data transformation
  • Minimal processing overhead
  • Best for generic integrations

​
Custom Webhook

Webhook with custom payload structure:
  • Sends curated event data
  • Configurable payload format
  • Supports shared key signing
  • Suitable for specific integrations

​
Transform Webhook

Webhook with Python transformation:
  • Executes Python transform before sending
  • Full control over payload structure
  • Can query Infrahub for additional data
  • Most flexible option

​
Creating Webhooks

​
Using the UI

  1. Navigate to Integrations > Webhooks
  2. Click Add Webhook
  3. Configure:
    • Name: Descriptive identifier
    • URL: Destination endpoint
    • Event Type: Trigger condition
    • Branch Scope: Branch filter
    • Node Kind: Optional node type filter
    • Active: Enable/disable webhook

​
Using the SDK

from infrahub_sdk import InfrahubClient

client = InfrahubClient()

webhook = await client.create(
    kind="CoreStandardWebhook",
    data={
        "name": "slack-notification",
        "url": "https://hooks.slack.com/services/XXX/YYY/ZZZ",
        "event_type": "infrahub.node.created",
        "branch_scope": "all_branches",
        "node_kind": "NetworkDevice",
        "active": True,
        "validate_certificates": True,
    }
)
await webhook.save()

​
Event Types

Webhooks can trigger on specific events:

​
Node Events

  • infrahub.node.created: Node created
  • infrahub.node.updated: Node updated
  • infrahub.node.deleted: Node deleted

​
Branch Events

  • infrahub.branch.created: Branch created
  • infrahub.branch.merged: Branch merged
  • infrahub.branch.deleted: Branch deleted

​
Repository Events

  • infrahub.repository.created: Repository added
  • infrahub.repository.updated: Repository updated

​
Wildcard

  • all: Match all events

​
Branch Scope

Filter events by branch:
  • all_branches: All branches (default)
  • default_branch: Only main branch
  • other_branches: All branches except main

​
Webhook Payload

​
Standard Payload Structure

{
  "data": {
    "node_id": "17a9e3e8-1234-5678-90ab-cdef12345678",
    "kind": "NetworkDevice",
    "action": "created",
    "changelog": {
      "display_label": "device-01",
      "attributes": {
        "name": {
          "name": "name",
          "value": "device-01",
          "kind": "Text"
        }
      },
      "relationships": {}
    }
  },
  "id": "evt_1234567890abcdef",
  "branch": "main",
  "account_id": "user-uuid",
  "occured_at": "2024-03-02T10:30:00Z",
  "event": "infrahub.node.created"
}

​
Event Context

Every webhook includes event context: 
# backend/infrahub/webhook/models.py:91
class EventContext(BaseModel):
    id: str  # Event ID
    branch: str | None  # Branch name
    account_id: str | None  # User ID
    occured_at: str  # ISO timestamp
    event: str  # Event type

​
Webhook Signing

Webhooks can be signed for security:

​
Configure Shared Key

webhook = await client.create(
    kind="CoreCustomWebhook",
    data={
        "name": "secure-webhook",
        "url": "https://api.example.com/webhook",
        "shared_key": "your-secret-key-here",
        "event_type": "all",
        "active": True,
    }
)
await webhook.save()

​
Signature Headers

Signed webhooks include these headers: 
webhook-id: msg_1234567890abcdef
webhook-timestamp: 1709376600
webhook-signature: v1,dGVzdCBzaWduYXR1cmU=

​
Verify Signature

import hmac
import hashlib
import base64

def verify_webhook(payload: str, headers: dict, secret: str) -> bool:
    msg_id = headers["webhook-id"]
    timestamp = headers["webhook-timestamp"]
    signature = headers["webhook-signature"]
    
    # Reconstruct signed data
    unsigned_data = f"{msg_id}.{timestamp}.{payload}".encode()
    
    # Calculate expected signature
    expected = hmac.new(
        key=secret.encode(),
        msg=unsigned_data,
        digestmod=hashlib.sha256
    ).digest()
    
    # Extract actual signature (remove "v1," prefix)
    actual = base64.b64decode(signature.split(",")[1])
    
    return hmac.compare_digest(expected, actual)

​
Transform Webhooks

Transform webhooks use Python to customize payloads:

​
Create Transform

First, create a Python transform in a repository: 
# transforms/slack_transform.py
from infrahub_sdk import InfrahubClient

class SlackTransform:
    async def run(self, data: dict, client: InfrahubClient, **kwargs) -> dict:
        # Extract event data
        event_data = data.get("data", {})
        node_kind = event_data.get("kind", "Unknown")
        action = event_data.get("action", "unknown")
        display_label = event_data.get("changelog", {}).get("display_label", "")
        
        # Build Slack message
        return {
            "text": f"Node {action}: {node_kind} '{display_label}'",
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"*{node_kind}* was *{action}*"
                    }
                },
                {
                    "type": "section",
                    "fields": [
                        {"type": "mrkdwn", "text": f"*Name:*\n{display_label}"},
                        {"type": "mrkdwn", "text": f"*Action:*\n{action}"}
                    ]
                }
            ]
        }

​
Register Transform

transform = await client.create(
    kind="CoreTransformPython",
    data={
        "name": "slack-transform",
        "repository": repository_id,
        "file_path": "transforms/slack_transform.py",
        "class_name": "SlackTransform",
        "timeout": 30,
        "convert_query_response": False,
    }
)
await transform.save()

​
Create Transform Webhook

webhook = await client.create(
    kind="CoreCustomWebhook",
    data={
        "name": "slack-webhook",
        "url": "https://hooks.slack.com/services/XXX/YYY/ZZZ",
        "transformation": transform.id,
        "event_type": "infrahub.node.created",
        "active": True,
    }
)
await webhook.save()

​
Webhook Automation

Webhooks use Prefect automations for event triggering:

​
Automation Creation

# backend/infrahub/webhook/models.py:31
class WebhookTriggerDefinition(TriggerDefinition):
    @classmethod
    def from_object(cls, obj: CoreWebhook) -> Self:
        event_trigger = EventTrigger()
        
        # Configure event matching
        if obj.event_type.value == "all":
            event_trigger.events.add("infrahub.*")
        else:
            event_trigger.events.add(obj.event_type.value)
        
        # Configure branch filtering
        if obj.branch_scope.value == "default_branch":
            event_trigger.match_related = {
                "prefect.resource.role": "infrahub.branch",
                "infrahub.resource.label": registry.default_branch,
            }
        elif obj.branch_scope.value == "other_branches":
            event_trigger.match_related = {
                "prefect.resource.role": "infrahub.branch",
                "infrahub.resource.label": f"!{registry.default_branch}",
            }
        
        # Configure node kind filtering
        if obj.node_kind.value:
            event_trigger.match = {"infrahub.node.kind": obj.node_kind.value}
        
        return cls(...)

​
Auto-Configuration

Webhooks auto-configure when created/updated: 
# backend/infrahub/webhook/triggers.py:5
TRIGGER_WEBHOOK_SETUP_UPDATE = BuiltinTriggerDefinition(
    name="webhook-configure-one",
    trigger=EventTrigger(
        events={"infrahub.node.created", "infrahub.node.updated"},
        match={
            "infrahub.node.kind": [InfrahubKind.CUSTOMWEBHOOK, InfrahubKind.STANDARDWEBHOOK],
        },
    ),
    actions=[
        ExecuteWorkflow(
            workflow=WEBHOOK_CONFIGURE_ONE,
            parameters={...},
        ),
    ],
)

​
Webhook Processing

​
Workflow Execution

# backend/infrahub/webhook/tasks.py:66
@flow(name="webhook-process", flow_run_name="Send webhook for {webhook_name}")
async def webhook_process(
    webhook_id: str,
    webhook_name: str,
    webhook_kind: str,
    event_id: str,
    event_type: str,
    event_occured_at: str,
    event_payload: dict,
    branch_name: str | None = None,
) -> None:
    log = get_run_logger()
    client = get_client()
    cache = await get_cache()
    
    # Load webhook from cache or database
    webhook_data_str = await cache.get(key=f"webhook:{webhook_id}")
    if not webhook_data_str:
        webhook_node = await client.get(kind=webhook_kind, id=webhook_id)
        webhook = await convert_node_to_webhook(webhook_node=webhook_node, client=client)
        await cache.set(key=f"webhook:{webhook_id}", value=ujson.dumps(webhook.to_cache()))
    else:
        webhook_data = ujson.loads(webhook_data_str)
        webhook = WEBHOOK_MAP[webhook_data["webhook_type"]].from_cache(webhook_data)
    
    # Build context
    webhook_context = EventContext.from_event(
        event_id=event_id,
        event_type=event_type,
        event_occured_at=event_occured_at,
        event_payload=event_payload,
    )
    
    # Send webhook
    response = await webhook_send(webhook=webhook, context=webhook_context, event_data=event_payload.get("data", {}))
    log.info(f"Successfully sent webhook to {response.url} with status {response.status_code}")

​
Sending Webhook

# backend/infrahub/webhook/tasks.py:34
@task(name="webhook-send", cache_policy=NONE, retries=3)
async def webhook_send(webhook: Webhook, context: EventContext, event_data: dict) -> Response:
    http_service = get_http()
    client = get_client()
    response = await webhook.send(
        data=event_data,
        context=context,
        http_service=http_service,
        client=client
    )
    response.raise_for_status()
    return response

​
Certificate Validation

Control SSL certificate validation: 
webhook = await client.create(
    kind="CoreStandardWebhook",
    data={
        "name": "internal-webhook",
        "url": "https://internal.example.com/webhook",
        "validate_certificates": False,  # Disable for self-signed certs
        "event_type": "all",
        "active": True,
    }
)
await webhook.save()

​
Webhook Models

​
Webhook Base Class

# backend/infrahub/webhook/models.py:116
class Webhook(BaseModel):
    name: str
    url: str
    event_type: str
    validate_certificates: bool | None
    shared_key: str | None  # For signing
    
    async def send(
        self,
        data: dict[str, Any],
        context: EventContext,
        http_service: InfrahubHTTP,
        client: InfrahubClient,
    ) -> Response:
        await self.prepare(data=data, context=context, client=client)
        return await http_service.post(
            url=self.url,
            json=self.get_payload(),
            headers=self._headers,
            verify=self.validate_certificates
        )

​
TransformWebhook

# backend/infrahub/webhook/models.py:209
class TransformWebhook(Webhook):
    repository_id: str
    repository_name: str
    repository_kind: str
    transform_name: str
    transform_class: str
    transform_file: str
    transform_timeout: int
    convert_query_response: bool
    
    async def _prepare_payload(
        self,
        data: dict[str, Any],
        context: EventContext,
        client: InfrahubClient,
    ) -> None:
        # Load repository
        repo = await InfrahubRepository.init(...)
        
        # Execute transform
        self._payload = await repo.execute_python_transform.with_options(
            timeout_seconds=self.transform_timeout
        )(
            branch_name=branch,
            commit=commit,
            location=f"{self.transform_file}::{self.transform_class}",
            convert_query_response=self.convert_query_response,
            data={"data": {"data": data, **context.model_dump()}},
            client=client,
        )

​
Real-World Examples

​
Slack Notifications

# Simple notification
slack_webhook = await client.create(
    kind="CoreCustomWebhook",
    data={
        "name": "slack-device-alerts",
        "url": "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
        "event_type": "infrahub.node.created",
        "node_kind": "NetworkDevice",
        "branch_scope": "default_branch",
        "active": True,
    }
)
await slack_webhook.save()

​
CI/CD Integration

# Trigger CI pipeline on schema changes
ci_webhook = await client.create(
    kind="CoreStandardWebhook",
    data={
        "name": "gitlab-ci-trigger",
        "url": "https://gitlab.com/api/v4/projects/123/trigger/pipeline",
        "event_type": "infrahub.repository.updated",
        "branch_scope": "default_branch",
        "active": True,
        "shared_key": "gitlab-trigger-token",
    }
)
await ci_webhook.save()

​
Custom Monitoring

# Send metrics to monitoring system
monitoring_webhook = await client.create(
    kind="CoreCustomWebhook",
    data={
        "name": "datadog-metrics",
        "url": "https://api.datadoghq.com/api/v1/events",
        "transformation": datadog_transform_id,
        "event_type": "all",
        "active": True,
    }
)
await monitoring_webhook.save()

​
Best Practices

​
Security

  1. Always use HTTPS endpoints
  2. Enable certificate validation for public endpoints
  3. Use shared keys for sensitive integrations
  4. Rotate webhook secrets periodically

​
Performance

  1. Keep transform execution time short
  2. Use caching for frequently accessed data
  3. Implement retry logic in receiving systems
  4. Monitor webhook delivery success rates

​
Reliability

  1. Handle webhook failures gracefully
  2. Implement idempotency in receivers
  3. Validate webhook signatures
  4. Log all webhook deliveries

​
Troubleshooting

​
Webhook Not Firing

  1. Check webhook is active: true
  2. Verify event type matches
  3. Check branch scope filter
  4. Review node kind filter
  5. Check automation exists in Prefect

​
Delivery Failures

  1. Verify URL is accessible
  2. Check SSL certificate validity
  3. Review receiving endpoint logs
  4. Check for rate limiting
  5. Verify payload format

​
Transform Errors

  1. Test transform independently
  2. Check repository accessibility
  3. Verify Python syntax
  4. Review transform execution logs
  5. Check timeout settings

Build docs developers (and LLMs) love

Get started for freeTalk to us