# Gateway Administrative Hooks

> This document details the administrative hook points in ContextForge Plugin Framework, covering gateway management operations including server registration, updates, federation, and entity lifecycle management.

- Skill: `tools-only/gateway-administrative-hooks` (Agent Skill, multi-file: 3 files)
- Install (CLI): `npx skillmds@latest add tools-only/gateway-administrative-hooks`
- Raw SKILL.md: https://api.skillmd.com/api/skills/tools-only/gateway-administrative-hooks/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: tools-only (https://skillmd.com/u/tools-only)
- Updated: 2026-09-29
- Page: https://skillmd.com/skills/tools-only/gateway-administrative-hooks

---

# Gateway Administrative Hooks

This document details the administrative hook points in ContextForge Plugin Framework, covering gateway management operations including server registration, updates, federation, and entity lifecycle management.
## Administrative Hook Functions

The framework provides administrative hooks for gateway management operations:

| Hook Function | Description | When It Executes | Primary Use Cases |
|---------------|-------------|-------------------|-------------------|
| [`server_pre_register()`](#server-pre-register-hook) | Process server registration requests before creating server records | Before MCP server is registered in the gateway | Server validation, naming conventions, policy enforcement, auto-configuration |
| [`server_post_register()`](#server-post-register-hook) | Process server registration results after successful creation | After MCP server registration completes | Audit logging, notifications, external integrations, metrics collection |
| [`server_pre_update()`](#server-pre-update-hook) | Process server update requests before applying configuration changes | Before MCP server configuration is modified | Change validation, approval workflows, impact assessment, transformation |
| [`server_post_update()`](#server-post-update-hook) | Process server update results after successful modification | After MCP server updates complete | Change notifications, cache invalidation, discovery updates, audit logging |
| [`server_pre_delete()`](#server-pre-delete-hook) | Process server deletion requests before removing server records | Before MCP server is deleted from the gateway | Access control, dependency checks, data preservation, deletion confirmation |
| [`server_post_delete()`](#server-post-delete-hook) | Process server deletion results after successful removal | After MCP server deletion completes | Resource cleanup, notifications, audit logging, compliance archiving |
| [`server_pre_status_change()`](#server-pre-status-change-hook) | Process server status change requests before activation/deactivation | Before MCP server is activated or deactivated | Access control, dependency validation, impact assessment, quota enforcement |
| [`server_post_status_change()`](#server-post-status-change-hook) | Process server status change results after successful toggle | After MCP server status change completes | Monitoring setup/teardown, notifications, resource management, metrics tracking |
| [`gateway_pre_register()`](#gateway-pre-register-hook) | Process gateway registration requests before creating federation records | Before peer gateway is registered | Gateway validation, federation loop detection, security enforcement, auto-configuration |
| [`gateway_post_register()`](#gateway-post-register-hook) | Process gateway registration results after successful federation | After peer gateway registration completes | Health monitoring setup, federation handshake, discovery updates, capability detection |
| [`gateway_pre_update()`](#gateway-pre-update-hook) | Process gateway update requests before applying federation changes | Before peer gateway configuration is modified | Federation impact assessment, URL validation, authentication changes, confirmation workflows |
| [`gateway_post_update()`](#gateway-post-update-hook) | Process gateway update results after successful modification | After peer gateway updates complete | Federation connection refresh, capability updates, discovery synchronization, monitoring updates |
| [`gateway_pre_delete()`](#gateway-pre-delete-hook) | Process gateway deletion requests before removing federation records | Before peer gateway is removed from federation | Federation dependency checks, resource migration planning, graceful disconnection workflows |
| [`gateway_post_delete()`](#gateway-post-delete-hook) | Process gateway deletion results after successful removal | After peer gateway deletion completes | Federation cleanup, resource deregistration, monitoring teardown, cache invalidation |
| [`gateway_pre_status_change()`](#gateway-pre-status-change-hook) | Process gateway status change requests before enabling/disabling | Before peer gateway is enabled or disabled | Federation impact assessment, dependency validation, connection management |
| [`gateway_post_status_change()`](#gateway-post-status-change-hook) | Process gateway status change results after successful toggle | After peer gateway status change completes | Federation connection activation/deactivation, discovery updates, monitoring adjustments |
## Server Management Hooks

### Server Pre-Register Hook

**Function Signature**: `async def server_pre_register(self, payload: ServerPreOperationPayload, context: PluginContext) -> ServerPreOperationResult`

| Attribute | Type | Description |
|-----------|------|-------------|
| **Hook Name** | `server_pre_register` | Hook identifier for configuration |
| **Execution Point** | Before server registration in gateway | When administrator or API client registers a new MCP server |
| **Purpose** | Server validation, policy enforcement, auto-configuration | Validate and transform server registration data before persistence |

**Payload Structure:**

```python
class ServerInfo(BaseModel):
    """Core server information - modifiable by plugins"""
    id: Optional[str] = Field(None, description="Server UUID identifier")
    name: str = Field(..., description="The server's name")
    description: Optional[str] = Field(None, description="Server description")
    icon: Optional[str] = Field(None, description="URL for the server's icon")
    tags: List[str] = Field(default_factory=list, description="Tags for categorizing the server")

    # Associated entities
    associated_tools: List[str] = Field(default_factory=list, description="Associated tool IDs")
    associated_resources: List[str] = Field(default_factory=list, description="Associated resource IDs")
    associated_prompts: List[str] = Field(default_factory=list, description="Associated prompt IDs")
    associated_a2a_agents: List[str] = Field(default_factory=list, description="Associated A2A agent IDs")

    # Team and organization
    team_id: Optional[str] = Field(None, description="Team ID for resource organization")
    owner_email: Optional[str] = Field(None, description="Email of the server owner")
    visibility: str = Field(default="private", description="Visibility level (private, team, public)")

class ServerAuditInfo(BaseModel):
    """Server audit/operational information - read-only across all server operations"""
    # Operation metadata
    operation_timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    request_id: Optional[str] = None                        # Unique request identifier

    # User and request info
    created_by: Optional[str] = None                        # User performing the operation
    created_from_ip: Optional[str] = None                   # Client IP address
    created_via: Optional[str] = None                       # Operation source ("api", "ui", "bulk_import", "federation")
    created_user_agent: Optional[str] = None                # Client user agent

    # Server state information
    server_id: Optional[str] = None                         # Target server ID (for updates/deletes)
    original_server_info: Optional[ServerInfo] = None       # Original state (for updates/deletes)

    # Database timestamps (populated in post-hooks)
    created_at: Optional[datetime] = None                   # Server creation timestamp
    updated_at: Optional[datetime] = None                   # Server last update timestamp

    # Team/tenant context
    team_id: Optional[str] = None                           # Team performing operation
    tenant_id: Optional[str] = None                         # Tenant context

class ServerPreOperationPayload(BaseModel):
    """Unified payload for server pre-operation hooks (register, update, etc.)"""
    server_info: ServerInfo                                 # Modifiable server information
    headers: HttpHeaderPayload = Field(default_factory=dict) # HTTP headers for passthrough

class ServerPostOperationPayload(BaseModel):
    """Unified payload for server post-operation hooks (register, update, etc.)"""
    server_info: Optional[ServerInfo] = None                # Complete server information (if successful)
    operation_success: bool                                 # Whether operation succeeded
    error_details: Optional[str] = None                     # Error details if operation failed
    headers: HttpHeaderPayload = Field(default_factory=dict) # HTTP headers for passthrough
```

**Payload Attributes (`ServerPreOperationPayload`)**:

| Attribute | Type | Required | Description | Example |
|-----------|------|----------|-------------|---------|
| `server_info` | `ServerInfo` | Yes | Modifiable server information object | See ServerInfo structure above |
| `headers` | `HttpHeaderPayload` |  | HTTP headers for passthrough | `{"Authorization": "Bearer token123"}` |

**Context Information (`ServerAuditInfo`)** - Available in `context.server_audit_info`:

| Attribute | Type | Description | Example |
|-----------|------|-------------|---------|
| `created_by` | `str` | User performing the operation | `"admin@example.com"` |
| `created_from_ip` | `str` | Client IP address | `"192.168.1.100"` |
| `created_via` | `str` | Operation source | `"api"`, `"ui"`, `"bulk_import"`, `"federation"` |
| `created_user_agent` | `str` | Client user agent | `"curl/7.68.0"` |
| `request_id` | `str` | Unique request identifier | `"req-456"` |
| `operation_timestamp` | `datetime` | Operation timestamp | `"2025-01-15T10:30:00Z"` |
| `server_id` | `str` | Target server ID (for updates/deletes) | `"srv-123"` |
| `original_server_info` | `ServerInfo` | Original state (for updates/deletes) | Previous server configuration |
| `team_id` | `str` | Team performing operation | `"team-456"` |

**Return Type (`ServerPreOperationResult`)**:

- Extends `PluginResult[ServerPreOperationPayload]`
- Can modify all payload attributes before server creation
- Can block server registration with violation
- Can request client elicitation for additional information

**Example Use Cases**:

```python
# 1. Server naming convention enforcement
async def server_pre_register(self, payload: ServerPreOperationPayload,
                             context: PluginContext) -> ServerPreOperationResult:
    # Access audit information from context
    audit_info = context.server_audit_info

    # Enforce company naming convention
    if not payload.server_info.name.startswith("company-"):
        payload.server_info.name = f"company-{payload.server_info.name}"

    # Auto-generate description if missing
    if not payload.server_info.description:
        payload.server_info.description = f"Automatically registered server: {payload.server_info.name}"

    # Add mandatory tags based on team from server_info
    if payload.server_info.team_id:
        payload.server_info.tags.append(f"team-{payload.server_info.team_id}")

    # Add creator-based tag from audit info
    if audit_info.created_by:
        user_domain = audit_info.created_by.split("@")[1] if "@" in audit_info.created_by else "unknown"
        payload.server_info.tags.append(f"domain-{user_domain}")

    return ServerPreOperationResult(modified_payload=payload)

# 2. Server validation and security checks
async def server_pre_register(self, payload: ServerPreRegisterPayload,
                             context: PluginContext) -> ServerPreOperationResult:
    # Validate server name against blacklist
    blocked_names = ["admin", "system", "root", "test"]
    if payload.server_info.name.lower() in blocked_names:
        violation = PluginViolation(
            reason="Blocked server name",
            description=f"Server name '{payload.server_info.name}' is not allowed",
            code="BLOCKED_SERVER_NAME"
        )
        return ServerPreOperationResult(continue_processing=False, violation=violation)

    # Check if user has permission to register servers
    user_email = context.server_audit_info.created_by
    if not self._has_server_registration_permission(user_email):
        violation = PluginViolation(
            reason="Insufficient permissions",
            description=f"User {user_email} cannot register servers",
            code="INSUFFICIENT_PERMISSIONS"
        )
        return ServerPreOperationResult(continue_processing=False, violation=violation)

    # Check server registration quota
    current_count = await self._get_user_server_count(user_email)
    max_servers = self._get_user_server_limit(user_email)
    if current_count >= max_servers:
        violation = PluginViolation(
            reason="Server quota exceeded",
            description=f"User has reached maximum of {max_servers} servers",
            code="SERVER_QUOTA_EXCEEDED"
        )
        return ServerPreOperationResult(continue_processing=False, violation=violation)

    return ServerPreOperationResult()

# 3. Auto-configuration and enhancement
async def server_pre_register(self, payload: ServerPreRegisterPayload,
                             context: PluginContext) -> ServerPreOperationResult:
    # Auto-tag based on name patterns
    if "file" in payload.server_info.name.lower():
        payload.server_info.tags.extend(["files", "storage"])
    elif "api" in payload.server_info.name.lower():
        payload.server_info.tags.extend(["api", "integration"])
    elif "db" in payload.server_info.name.lower() or "database" in payload.server_info.name.lower():
        payload.server_info.tags.extend(["database", "data"])

    # Set default icon based on tags
    if not payload.server_info.icon:
        if "files" in payload.server_info.tags:
            payload.server_info.icon = "https://cdn.example.com/icons/file-server.png"
        elif "api" in payload.server_info.tags:
            payload.server_info.icon = "https://cdn.example.com/icons/api-server.png"

    # Add audit headers
    payload.headers["X-Registration-Source"] = context.server_audit_info.created_via
    payload.headers["X-Registration-User"] = context.server_audit_info.created_by

    return ServerPreOperationResult(modified_payload=payload)

# 4. User confirmation for sensitive operations
async def server_pre_register(self, payload: ServerPreRegisterPayload,
                             context: PluginContext) -> ServerPreOperationResult:
    # Check if this is a production-like server name
    production_patterns = ["prod", "production", "live", "main"]
    is_production = any(pattern in payload.server_info.name.lower() for pattern in production_patterns)

    if is_production and not context.elicitation_responses:
        # Request user confirmation for production server
        confirmation_schema = {
            "type": "object",
            "properties": {
                "confirm_production": {
                    "type": "boolean",
                    "description": "Confirm registration of production server"
                },
                "business_justification": {
                    "type": "string",
                    "description": "Business justification for production server",
                    "minLength": 10
                }
            },
            "required": ["confirm_production", "business_justification"]
        }

        elicitation_request = ElicitationRequest(
            message=f"You are registering a production server '{payload.server_info.name}'. Please confirm.",
            schema=confirmation_schema,
            timeout_seconds=300  # 5 minutes
        )

        return ServerPreOperationResult(
            continue_processing=False,
            elicitation_request=elicitation_request
        )

    # Process elicitation response
    if context.elicitation_responses and is_production:
        response = context.elicitation_responses[0]
        if response.action != "accept" or not response.data.get("confirm_production"):
            violation = PluginViolation(
                reason="Production server registration declined",
                description="User declined to register production server",
                code="PRODUCTION_REGISTRATION_DECLINED"
            )
            return ServerPreOperationResult(continue_processing=False, violation=violation)

        # Add justification to server description
        justification = response.data.get("business_justification", "")
        if justification:
            payload.server_info.description = f"{payload.server_info.description or ''}\n\nBusiness Justification: {justification}"

        # Add production tag
        payload.server_info.tags.append("production")

    return ServerPreOperationResult(modified_payload=payload)
```

### Server Post-Register Hook

**Function Signature**: `async def server_post_register(self, payload: ServerPostOperationPayload, context: PluginContext) -> ServerPostOperationResult`

| Attribute | Type | Description |
|-----------|------|-------------|
| **Hook Name** | `server_post_register` | Hook identifier for configuration |
| **Execution Point** | After server registration completes | When MCP server has been successfully created in the gateway |
| **Purpose** | Audit logging, notifications, integrations, metrics | Process successful server registrations and handle follow-up actions |

**Payload Attributes (`ServerPostOperationPayload`)**:

| Attribute | Type | Required | Description | Example |
|-----------|------|----------|-------------|---------|
| `server_info` | `ServerInfo` |  | Complete registered server information (if successful) | Contains all ServerInfo fields |
| `operation_success` | `bool` | Yes | Whether registration succeeded | `true` |
| `error_details` | `str` |  | Error details if registration failed | `"Duplicate server name"` |
| `headers` | `HttpHeaderPayload` |  | HTTP headers for passthrough | `{"Authorization": "Bearer token123"}` |

**Context Information (`ServerAuditInfo`)** - Available in `context.server_audit_info`:

- Same fields as pre-register hook, plus database timestamps
- Contains complete audit trail including `created_at` and `updated_at` timestamps

**Return Type (`ServerPostOperationResult`)**:

- Extends `PluginResult[ServerPostOperationPayload]`
- Cannot modify server data (read-only post-operation hook)
- Can trigger additional actions or external integrations
- Violations in post-hooks log errors but don't affect the operation

**Example Use Cases**:

```python
# 1. Audit logging and compliance
async def server_post_register(self, payload: ServerPostOperationPayload,
                              context: PluginContext) -> ServerPostOperationResult:
    # Access audit information from context
    audit_info = context.server_audit_info

    # Log comprehensive audit record
    audit_record = {
        "event_type": "server_registration",
        "success": payload.operation_success,
        "user": audit_info.created_by,
        "ip_address": audit_info.created_from_ip,
        "user_agent": audit_info.created_user_agent,
        "creation_method": audit_info.created_via,
        "timestamp": audit_info.created_at,
        "request_id": audit_info.request_id
    }

    if payload.operation_success and payload.server_info:
        audit_record.update({
            "server_id": payload.server_info.id,
            "server_name": payload.server_info.name,
            "team_id": payload.server_info.team_id,
            "tags": payload.server_info.tags
        })
    else:
        audit_record["error"] = payload.error_details

    # Send to audit logging system
    await self._send_audit_log(audit_record)

    # Update metrics
    if payload.operation_success:
        await self._increment_metric("servers_registered_total", {
            "method": context.server_audit_info.created_via,
            "team": context.server_audit_info.team_id or "none"
        })
    else:
        await self._increment_metric("server_registration_failures_total", {
            "error_type": "registration_error"
        })

    return ServerPostOperationResult()

# 2. Team notifications and integrations
async def server_post_register(self, payload: ServerPostOperationPayload,
                              context: PluginContext) -> ServerPostOperationResult:
    if payload.operation_success:
        # Send notification to team members
        team_id = context.server_audit_info.team_id
        if team_id:
            team_members = await self._get_team_members(team_id)
            notification = {
                "title": "New MCP Server Registered",
                "message": f"Server '{payload.server_info.name}' has been registered by {context.server_audit_info.created_by}",
                "server_id": payload.server_info.id,
                "registered_by": context.server_audit_info.created_by,
                "timestamp": context.server_audit_info.created_at.isoformat()
            }

            for member in team_members:
                await self._send_notification(member["email"], notification)

        # Integrate with external systems
        await self._sync_to_service_catalog({
            "id": payload.server_info.id,
            "name": payload.server_info.name,
            "owner": context.server_audit_info.created_by,
            "team": team_id,
            "status": "active"
        })

        # Trigger monitoring setup
        await self._setup_server_monitoring(payload.server_info.id, payload.server_info.name)

    return ServerPostOperationResult()

# 3. Error handling and recovery
async def server_post_register(self, payload: ServerPostOperationPayload,
                              context: PluginContext) -> ServerPostOperationResult:
    if not payload.operation_success:
        # Log detailed error for debugging
        error_context = {
            "server_name": payload.server_info.name if payload.server_info else "unknown",
            "error": payload.error_details,
            "user": context.server_audit_info.created_by,
            "request_data": {
                "team_id": context.server_audit_info.team_id,
                "creation_method": context.server_audit_info.created_via,
                "ip_address": context.server_audit_info.created_from_ip
            }
        }

        self.logger.error(f"Server registration failed: {payload.server_info.name if payload.server_info else 'unknown'}",
                         extra=error_context)

        # Send error notification to admin
        if "quota" in payload.error_details.lower():
            await self._notify_admin_quota_exceeded(
                context.server_audit_info.created_by,
                payload.error_details
            )
        elif "permission" in payload.error_details.lower():
            await self._notify_admin_permission_denied(
                context.server_audit_info.created_by,
                payload.server_info.name
            )

        # Update error metrics with classification
        error_type = self._classify_error(payload.error_details)
        await self._increment_metric("server_registration_failures_total", {
            "error_type": error_type,
            "creation_method": context.server_audit_info.created_via
        })

    return ServerPostOperationResult()

# 4. Automated follow-up actions
async def server_post_register(self, payload: ServerPostOperationPayload,
                              context: PluginContext) -> ServerPostOperationResult:
    if payload.operation_success:
        # Auto-create default resources for certain server types
        server_name_lower = payload.server_info.name.lower()

        if "api" in server_name_lower:
            # Create API documentation resource
            await self._create_api_doc_resource(payload.server_info.id, payload.server_info.name)

        elif "database" in server_name_lower or "db" in server_name_lower:
            # Create database schema resource
            await self._create_db_schema_resource(payload.server_info.id, payload.server_info.name)

        elif "file" in server_name_lower:
            # Create file system browser resource
            await self._create_file_browser_resource(payload.server_info.id, payload.server_info.name)

        # Schedule health check for new server
        await self._schedule_server_health_check(payload.server_info.id, delay_minutes=5)

        # Add to server discovery index
        await self._add_to_discovery_index({
            "server_id": payload.server_info.id,
            "name": payload.server_info.name,
            "team": context.server_audit_info.team_id,
            "registered_at": context.server_audit_info.created_at
        })

    return ServerPostOperationResult()
```

### Server Pre-Update Hook

**Function Signature**: `async def server_pre_update(self, payload: ServerPreOperationPayload, context: PluginContext) -> ServerPreOperationResult`

| Attribute | Type | Description |
|-----------|------|-------------|
| **Hook Name** | `server_pre_update` | Hook identifier for configuration |
| **Execution Point** | Before server update applies | When MCP server configuration is being modified |
| **Purpose** | Validation, transformation, access control | Enforce update policies and modify update data |

**Payload Attributes (`ServerPreOperationPayload`)** - Same as pre-register:

| Attribute | Type | Required | Description | Example |
|-----------|------|----------|-------------|---------|
| `server_info` | `ServerInfo` | Yes | Updated server information object | Contains modified server fields |
| `headers` | `HttpHeaderPayload` |  | HTTP headers for passthrough | `{"Authorization": "Bearer token123"}` |

**Context Information (`ServerAuditInfo`)** - Available in `context.server_audit_info`:

- Same fields as register hooks, **plus**:

  - `server_id` - ID of server being updated
  - `original_server_info` - Server state before the update

**Return Type (`ServerPreOperationResult`)**:

- Extends `PluginResult[ServerPreOperationPayload]`
- Can modify server update data before application
- Can block server updates with violation
- Can request client elicitation for change approval

**Example Use Cases**:

```python
# 1. Change validation and approval workflow
async def server_pre_update(self, payload: ServerPreOperationPayload,
                           context: PluginContext) -> ServerPreOperationResult:
    original = context.server_audit_info.original_server_info
    current = payload.server_info

    # Detect critical changes requiring approval
    critical_changes = []
    if original.uri != current.uri:
        critical_changes.append("URI endpoint")
    if original.visibility != current.visibility and current.visibility == "public":
        critical_changes.append("visibility to public")
    if "production" in current.tags and "production" not in original.tags:
        critical_changes.append("production classification")

    # Request approval for critical changes
    if critical_changes and not context.elicitation_responses:
        approval_schema = {
            "type": "object",
            "properties": {
                "approve_changes": {
                    "type": "boolean",
                    "description": f"Approve these critical changes: {', '.join(critical_changes)}"
                },
                "change_justification": {
                    "type": "string",
                    "description": "Business justification for these changes",
                    "minLength": 20
                }
            },
            "required": ["approve_changes", "change_justification"]
        }

        return ServerPreOperationResult(
            continue_processing=False,
            elicitation_request=ElicitationRequest(
                message=f"Critical changes detected for server '{current.name}': {', '.join(critical_changes)}",
                schema=approval_schema,
                timeout_seconds=600
            )
        )

    # Process approval response
    if context.elicitation_responses:
        response = context.elicitation_responses[0]
        if not response.data.get("approve_changes"):
            return ServerPreOperationResult(
                continue_processing=False,
                violation=PluginViolation(
                    reason="Server update declined",
                    description="Critical changes not approved by user",
                    code="UPDATE_NOT_APPROVED"
                )
            )

        # Add justification to update audit
        justification = response.data.get("change_justification", "")
        payload.headers["X-Change-Justification"] = justification

    return ServerPreOperationResult(modified_payload=payload)
```

### Server Post-Update Hook

**Function Signature**: `async def server_post_update(self, payload: ServerPostOperationPayload, context: PluginContext) -> ServerPostOperationResult`

| Attribute | Type | Description |
|-----------|------|-------------|
| **Hook Name** | `server_post_update` | Hook identifier for configuration |
| **Execution Point** | After server update completes | When MCP server has been successfully updated |
| **Purpose** | Audit logging, notifications, cache invalidation | Process successful updates and handle follow-up actions |

**Payload Attributes (`ServerPostOperationPayload`)** - Same as post-register:

| Attribute | Type | Required | Description | Example |
|-----------|------|----------|-------------|---------|
| `server_info` | `ServerInfo` |  | Updated server information (if successful) | Contains all updated ServerInfo fields |
| `operation_success` | `bool` | Yes | Whether update succeeded | `true` |
| `error_details` | `str` |  | Error details if update failed | `"Validation error: Invalid URI"` |
| `headers` | `HttpHeaderPayload` |  | HTTP headers for passthrough | `{"Authorization": "Bearer token123"}` |

**Context Information (`ServerAuditInfo`)** - Available in `context.server_audit_info`:

- Same fields as register hooks, **plus**:

  - `server_id` - ID of server that was updated
  - `original_server_info` - Server state before the update
  - `updated_at` - Database timestamp of the update

**Return Type (`ServerPostOperationResult`)**:

- Extends `PluginResult[ServerPostOperationPayload]`
- Cannot modify server data (read-only post-operation hook)
- Can trigger cache invalidation, notifications, and integrations
- Violations in post-hooks log errors but don't affect the operation

**Example Use Cases**:

```python
# 1. Change notification and cache invalidation
async def server_post_update(self, payload: ServerPostOperationPayload,
                            context: PluginContext) -> ServerPostOperationResult:
    if not payload.operation_success:
        # Log update failure
        self.logger.error(f"Server update failed: {payload.server_info.name if payload.server_info else 'unknown'}",
                         extra={
                             "error": payload.error_details,
                             "user": context.server_audit_info.created_by,
                             "server_id": context.server_audit_info.server_id
                         })
        return ServerPostOperationResult()

    # Calculate changes
    original = context.server_audit_info.original_server_info
    updated = payload.server_info
    changes = []

    if original.name != updated.name:
        changes.append(f"name: '{original.name}' → '{updated.name}'")
    if original.uri != updated.uri:
        changes.append(f"uri: '{original.uri}' → '{updated.uri}'")
    if original.visibility != updated.visibility:
        changes.append(f"visibility: '{original.visibility}' → '{updated.visibility}'")
    if set(original.tags) != set(updated.tags):
        changes.append(f"tags: {original.tags} → {updated.tags}")

    # Send change notifications
    if changes:
        await self._notify_server_changes({
            "server_id": updated.id,
            "server_name": updated.name,
            "changes": changes,
            "updated_by": context.server_audit_info.created_by,
            "timestamp": context.server_audit_info.operation_timestamp.isoformat()
        })

    # Invalidate caches
    await self._invalidate_server_cache(updated.id)

    # Update discovery index
    if original.visibility != updated.visibility or original.tags != updated.tags:
        await self._update_discovery_index({
            "server_id": updated.id,
            "name": updated.name,
            "visibility": updated.visibility,
            "tags": updated.tags
        })

    return ServerPostOperationResult()
```

### Server Pre-Delete Hook

**Function Signature**: `async def server_pre_delete(self, payload: ServerPreOperationPayload, context: PluginContext) -> ServerPreOperationResult`

| Attribute | Type | Description |
|-----------|------|-------------|
| **Hook Name** | `server_pre_delete` | Hook identifier for configuration |
| **Execution Point** | Before server deletion | When MCP server is about to be removed from the gateway |
| **Purpose** | Access control, dependency checks, data preservation | Validate deletion permissions and handle cleanup preparation |

**Payload Attributes (`ServerPreOperationPayload`)** - Same structure as other pre-hooks:

| Attribute | Type | Required | Description | Example |
|-----------|------|----------|-------------|---------|
| `server_info` | `ServerInfo` | Yes | Server information being deleted | Contains server to be removed |
| `headers` | `HttpHeaderPayload` |  | HTTP headers for passthrough | `{"Authorization": "Bearer token123"}` |

**Context Information (`ServerAuditInfo`)** - Available in `context.server_audit_info`:

- Same fields as other operations, **plus**:

  - `server_id` - ID of server being deleted
  - `original_server_info` - Complete server state before deletion (same as `payload.server_info`)

**Return Type (`ServerPreOperationResult`)**:

- Extends `PluginResult[ServerPreOperationPayload]`
- Can modify deletion behavior (e.g., soft delete vs hard delete)
- Can block server deletion with violation
- Can request client elicitation for deletion confirmation

**Example Use Cases**:

```python
# 1. Deletion protection and confirmation
async def server_pre_delete(self, payload: ServerPreOperationPayload,
                           context: PluginContext) -> ServerPreOperationResult:
    server = payload.server_info

    # Protect production servers
    if "production" in server.tags:
        if not context.elicitation_responses:
            confirmation_schema = {
                "type": "object",
                "properties": {
                    "confirm_production_delete": {
                        "type": "boolean",
                        "description": f"Confirm deletion of PRODUCTION server '{server.name}'"
                    },
                    "deletion_reason": {
                        "type": "string",
                        "description": "Reason for deleting this production server",
                        "minLength": 10
                    },
                    "backup_confirmation": {
                        "type": "boolean",
                        "description": "Confirm that data backups have been created"
                    }
                },
                "required": ["confirm_production_delete", "deletion_reason", "backup_confirmation"]
            }

            return ServerPreOperationResult(
                continue_processing=False,
                elicitation_request=ElicitationRequest(
                    message=f"⚠️  PRODUCTION SERVER DELETION\n\nYou are about to delete production server '{server.name}'.\nThis action cannot be undone.",
                    schema=confirmation_schema,
                    timeout_seconds=300
                )
            )

        # Process confirmation response
        response = context.elicitation_responses[0]
        if not response.data.get("confirm_production_delete") or not response.data.get("backup_confirmation"):
            return ServerPreOperationResult(
                continue_processing=False,
                violation=PluginViolation(
                    reason="Production server deletion cancelled",
                    description="User cancelled production server deletion",
                    code="PRODUCTION_DELETE_CANCELLED"
                )
            )

        # Add deletion audit info
        payload.headers["X-Deletion-Reason"] = response.data.get("deletion_reason", "")
        payload.headers["X-Deletion-Confirmed"] = "true"

    # Check for active connections
    active_connections = await self._get_active_connections(server.id)
    if active_connections > 0:
        return ServerPreOperationResult(
            continue_processing=False,
            violation=PluginViolation(
                reason="Server has active connections",
                description=f"Cannot delete server with {active_connections} active connections",
                code="ACTIVE_CONNECTIONS_EXIST"
            )
        )

    return ServerPreOperationResult(modified_payload=payload)

# 2. Dependency validation
async def server_pre_delete(self, payload: ServerPreOperationPayload,
                           context: PluginContext) -> ServerPreOperationResult:
    server = payload.server_info

    # Check for dependent virtual servers
    dependent_servers = await self._find_dependent_servers(server.id)
    if dependent_servers:
        dependent_names = [s.name for s in dependent_servers]
        return ServerPreOperationResult(
            continue_processing=False,
            violation=PluginViolation(
                reason="Server has dependencies",
                description=f"Cannot delete server '{server.name}' - it's used by: {', '.join(dependent_names)}",
                code="DEPENDENCY_VIOLATION"
            )
        )

    # Check for referenced resources
    referenced_resources = await self._find_referencing_resources(server.id)
    if referenced_resources:
        return ServerPreOperationResult(
            continue_processing=False,
            violation=PluginViolation(
                reason="Server has resource dependencies",
                description=f"Server '{server.name}' is referenced by {len(referenced_resources)} resources",
                code="RESOURCE_DEPENDENCY_VIOLATION"
            )
        )

    return ServerPreOperationResult()
```

### Server Post-Delete Hook

**Function Signature**: `async def server_post_delete(self, payload: ServerPostOperationPayload, context: PluginContext) -> ServerPostOperationResult`

| Attribute | Type | Description |
|-----------|------|-------------|
| **Hook Name** | `server_post_delete` | Hook identifier for configuration |
| **Execution Point** | After server deletion completes | When MCP server has been successfully removed |
| **Purpose** | Cleanup, notifications, audit logging | Handle post-deletion cleanup and notifications |

**Payload Attributes (`ServerPostOperationPayload`)** - Same structure as other post-hooks:

| Attribute | Type | Required | Description | Example |
|-----------|------|----------|-------------|---------|
| `server_info` | `ServerInfo` |  | Deleted server information (if successful) | Contains information of deleted server |
| `operation_success` | `bool` | Yes | Whether deletion succeeded | `true` |
| `error_details` | `str` |  | Error details if deletion failed | `"Foreign key constraint violation"` |
| `headers` | `HttpHeaderPayload` |  | HTTP headers for passthrough | `{"Authorization": "Bearer token123"}` |

**Context Information (`ServerAuditInfo`)** - Available in `context.server_audit_info`:

- Same fields as other operations, **plus**:

  - `server_id` - ID of server that was deleted
  - `original_server_info` - Complete server state before deletion
  - Database timestamps reflect the deletion operation

**Return Type (`ServerPostOperationR

…(truncated)