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:

Server Management Hooks

Server Pre-Register Hook

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

Payload Structure:

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):

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

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:

# 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

Payload Attributes (ServerPostOperationPayload):

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:

# 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

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

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:

# 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

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

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:

# 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

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

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:

# 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

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

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)