Webhook Configuration and Usage

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

Navigate to Integrations > Webhooks
Click Add Webhook
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

Always use HTTPS endpoints
Enable certificate validation for public endpoints
Use shared keys for sensitive integrations
Rotate webhook secrets periodically

Performance

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

Reliability

Handle webhook failures gracefully
Implement idempotency in receivers
Validate webhook signatures
Log all webhook deliveries

Troubleshooting

Webhook Not Firing

Check webhook is active: true
Verify event type matches
Check branch scope filter
Review node kind filter
Check automation exists in Prefect

Delivery Failures

Verify URL is accessible
Check SSL certificate validity
Review receiving endpoint logs
Check for rate limiting
Verify payload format

Transform Errors

Test transform independently
Check repository accessibility
Verify Python syntax
Review transform execution logs
Check timeout settings

Documentation Index

​Webhook Types

​Standard Webhook

​Custom Webhook

​Transform Webhook

​Creating Webhooks

​Using the UI

​Using the SDK

​Event Types

​Node Events

​Branch Events

​Repository Events

​Wildcard

​Branch Scope

​Webhook Payload

​Standard Payload Structure

​Event Context

​Webhook Signing

​Configure Shared Key

​Signature Headers

​Verify Signature

​Transform Webhooks

​Create Transform

​Register Transform

​Create Transform Webhook

​Webhook Automation

​Automation Creation

​Auto-Configuration

​Webhook Processing

​Workflow Execution

​Sending Webhook

​Certificate Validation

​Webhook Models

​Webhook Base Class

​TransformWebhook

​Real-World Examples

​Slack Notifications

​CI/CD Integration

​Custom Monitoring

​Best Practices

​Security

​Performance

​Reliability

​Troubleshooting

​Webhook Not Firing

​Delivery Failures

​Transform Errors