Escalation Configuration

Overview

Escalation allows a Digital Employee to route a task to a human supervisor (see Digital Employee Supervisor) when it is blocked (e.g., tool failures, missing permissions, critical ambiguity). Escalation is opt-in and must be explicitly enabled.

In this repository, see examples/escalation_example.py for a working end-to-end example.

Key Concepts

Supervisor

Escalation is designed to reach a supervisor configured on the Digital Employee identity:

• DigitalEmployeeIdentity.supervisor

• DigitalEmployeeSupervisor(name=..., email=...)

If escalation is enabled and a supervisor exists, the escalation protocol is included in the generated prompt.

Escalation Channels

Escalation is delivered via one or more channels (e.g., email).

Channels can introduce additional runtime dependencies (MCP connectors/tools). This matters at deployment time.

In this guide, we will use the GoogleMailMCPEscalationChannel as an example.

Example

Step 1: Import dependencies

import os

from digital_employee_core import (
    DigitalEmployee,
    DigitalEmployeeConfiguration,
    DigitalEmployeeIdentity,
    DigitalEmployeeJob,
    DigitalEmployeeSupervisor,
)
from digital_employee_core.connectors.mcps import google_docs_mcp
from digital_employee_core.escalation.channels.google_mail_mcp_channel import (
    GoogleMailMCPEscalationChannel,
)

Step 2: Define the job with escalation instructions

job = DigitalEmployeeJob(
    title="Operations Assistant",
    description="A digital employee that can escalate to a supervisor when blocked",
    instruction=(
        "You help with operational tasks. "
        "If a tool fails or you encounter a critical blocker, follow the escalation protocol."
    ),
)

Important: Use the term "escalation protocol" in the Digital Employee instructions whenever we need to refer to or trigger the escalation flow.

Step 3: Configure the supervisor

supervisor = DigitalEmployeeSupervisor(
    name=os.getenv("SUPERVISOR_NAME", ""),
    email=os.getenv("SUPERVISOR_EMAIL", ""),
)

Note: The following environment variables are required to enable escalation:

• SUPERVISOR_NAME

• SUPERVISOR_EMAIL

Step 4: Create the identity

identity = DigitalEmployeeIdentity(
    name="Ops Assistant - Example",
    email="[e-mail]",
    job=job,
    supervisor=supervisor,
)

Step 5: Set up configurations

configurations = [
    DigitalEmployeeConfiguration(key="GOOGLE_MAIL_MCP_URL", value=os.getenv("GOOGLE_MAIL_MCP_URL", "")),
    DigitalEmployeeConfiguration(key="GOOGLE_DOCS_MCP_URL", value=os.getenv("GOOGLE_DOCS_MCP_URL", "")),
    DigitalEmployeeConfiguration(key="GOOGLE_MCP_X_API_KEY", value=os.getenv("GOOGLE_MCP_X_API_KEY", "")),
]

Note: The following environment variables are required to enable the Google Mail MCP escalation channel:

• GOOGLE_MAIL_MCP_URL

• GOOGLE_MCP_X_API_KEY

These values are typically passed into DigitalEmployeeConfiguration so they become part of the deployed agent configuration. Adjust the environment variables based on your escalation channel.

Step 6: Initialize the Digital Employee

digital_employee = DigitalEmployee(
    identity=identity,
    mcps=[google_docs_mcp],
    configurations=configurations,
)

Step 7: Enable escalation and add channels

digital_employee.enable_escalation()
digital_employee.add_escalation_channel(GoogleMailMCPEscalationChannel())

Step 8: Deploy and run

digital_employee.deploy()
result = digital_employee.run(message="Read google docs with document_id='123413'")

Deployment Order Matters

• Enable escalation and add channels before calling digital_employee.deploy().

• The deploy step will bundle any required MCP connectors/tools introduced by escalation channels.

Custom Escalation Channels

Escalation channels are extensible. To create your own channel, implement EscalationChannel (see digital_employee_core/escalation/base_escalation_channel.py) and register it via digital_employee.add_escalation_channel(...).

1) Implement the channel

Create a new class that extends EscalationChannel that defines:

• get_required_mcps(): MCP connectors the channel needs at runtime.

• get_required_tools(): Additional tools (if any). Return [] if not needed.

• get_prompt_header(): Short title shown in the escalation protocol.

• get_prompt_body(supervisor, ...): Concrete instructions describing how the agent should escalate using your tools/MCP.

Minimal skeleton:

from typing import Any

from glaip_sdk import MCP, Tool

from digital_employee_core.escalation.base_escalation_channel import EscalationChannel
from digital_employee_core.identity.identity import DigitalEmployeeSupervisor


class CustomEscalationChannel(EscalationChannel):
    def get_required_mcps(self) -> list[MCP]:
        return []

    def get_required_tools(self) -> list[Tool]:
        return []

    def get_prompt_header(self, **kwargs: Any) -> str:
        return "Custom Escalation"

    def get_prompt_body(self, supervisor: DigitalEmployeeSupervisor, **kwargs: Any) -> str:
        return (
            f"When blocked, notify {supervisor.name} ({supervisor.email}) using <YOUR_TOOL>. "
            "Include: timestamp, failed action, what you tried, and what you need from the supervisor."
        )

For a reference implementation, see digital_employee_core/escalation/channels/google_mail_mcp_channel.py.

2) Register the channel before deploy

digital_employee.enable_escalation()
digital_employee.add_escalation_channel(CustomEscalationChannel())
digital_employee.deploy()

Notes

• One instance per channel type: EscalationChannel equality is based on class type, so adding the same channel type multiple times is treated as a duplicate.

• Tool restrictions: If your channel uses MCP tools, ensure those tools are allowed (if you are using whitelisted MCP tools, see MCP Allowed Tools Configuration).

Best Practices

• Define clear escalation triggers in DigitalEmployeeJob.instruction (e.g., tool failures, permission issues, urgent blockers).

• Configure least-privilege tools for MCPs where possible (see MCP Allowed Tools Configuration).

• Validate supervisor identity (non-empty name/email) during development to avoid “silent” non-actionable escalations.

• Test with prompt preview (build_prompt()) before deploying.