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:

• DigitalEmployeeBuilder().set_supervisor(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
import asyncio

from dotenv import load_dotenv

from digital_employee_core import (
    DigitalEmployeeAgentBuilder,
    DigitalEmployeeBuilder,
    DigitalEmployeePipelineBuilder,
)
from digital_employee_core.connectors.mcps import google_docs_mcp
from digital_employee_core.escalation import EscalationConfig
from digital_employee_core.escalation.channels.google_mail_mcp_channel import (
    GoogleMailMCPEscalationChannel,
)

load_dotenv()

Step 2: Set up escalation configuration

escalation_config = EscalationConfig(
    enabled=True,
    channels=[GoogleMailMCPEscalationChannel()],
)

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

Step 3: Build the Digital Employee with escalation

# Create agent builder with escalation configured
agent_builder = (
    DigitalEmployeeAgentBuilder()
    .add_mcps([google_docs_mcp])
    .add_configs_from_env(["GOOGLE_MAIL_MCP_URL", "GOOGLE_DOCS_MCP_URL", "GOOGLE_MCP_X_API_KEY"])
    .set_escalation_config(escalation_config)
)

# Build the Digital Employee
digital_employee = (
    DigitalEmployeeBuilder()
    .set_name("Ops Assistant - Example")
    .set_email("[e-mail]")
    .set_job(
        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."
        ),
    )
    .set_supervisor(
        name=os.getenv("SUPERVISOR_NAME", ""),
        email=os.getenv("SUPERVISOR_EMAIL", ""),
    )
    .add_configs_from_env(["GOOGLE_MAIL_MCP_URL", "GOOGLE_DOCS_MCP_URL", "GOOGLE_MCP_X_API_KEY"])
    .set_pipeline(DigitalEmployeePipelineBuilder().add_step(agent_builder))
    .build()
)

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

• GOOGLE_MAIL_MCP_URL

• GOOGLE_DOCS_MCP_URL

• GOOGLE_MCP_X_API_KEY

• SUPERVISOR_NAME

• SUPERVISOR_EMAIL

These values are loaded via .add_configs_from_env() and become part of the deployed agent configuration. Adjust the environment variables based on your escalation channel.

Step 4: Run the Digital Employee

# Run the digital employee
response = asyncio.run(
    digital_employee.invoke(
        message="Read google docs with document_id='123413'"
    )
)

Note: The new pattern uses async/await. Use employee.invoke(...) instead of the old employee.run(...) method.

Custom Escalation Channels

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

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) Configure the channel

escalation_config = EscalationConfig(
    enabled=True,
    channels=[CustomEscalationChannel()],
)

agent_builder = DigitalEmployeeAgentBuilder().set_escalation_config(escalation_config)

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 building.