# Escalation Configuration

### Overview

Escalation allows a **Digital Employee** to route a task to a **human supervisor** (see [Digital Employee Supervisor](https://gdplabs.gitbook.io/catapa/developer-documentation/digital-employee-de-pipeline/guides/advanced-examples/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`](https://github.com/GDP-ADMIN/CATAPA-SDK/blob/main/python/digital-employee-core/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

{% code lineNumbers="true" %}

```python
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()
```

{% endcode %}

#### Step 2: Set up escalation configuration

{% code lineNumbers="true" %}

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

{% endcode %}

{% hint style="warning" %}
**Important:** Use the term "escalation protocol" in the Digital Employee instructions whenever we need to refer to or trigger the escalation flow.
{% endhint %}

#### Step 3: Build the Digital Employee with escalation

{% code lineNumbers="true" %}

```python
# 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("ops.assistant@example.com")
    .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()
)
```

{% endcode %}

{% hint style="info" %}
**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.
{% endhint %}

#### Step 4: Run the Digital Employee

{% code lineNumbers="true" %}

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

{% endcode %}

{% hint style="info" %}
**Note:** The new pattern uses async/await. Use `employee.invoke(...)` instead of the old `employee.run(...)` method.
{% endhint %}

### Custom Escalation Channels

Escalation channels are extensible. To create your own channel, implement `EscalationChannel` (see [`digital_employee_core/escalation/base_escalation_channel.py`](https://github.com/GDP-ADMIN/CATAPA-SDK/blob/main/python/digital-employee-core/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:**

{% code lineNumbers="true" %}

```python
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."
        )
```

{% endcode %}

For a reference implementation, see [`digital_employee_core/escalation/channels/google_mail_mcp_channel.py`](https://github.com/GDP-ADMIN/CATAPA-SDK/blob/main/python/digital-employee-core/digital_employee_core/escalation/channels/google_mail_mcp_channel.py).

#### 2) Configure the channel

{% code lineNumbers="true" %}

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

agent_builder = DigitalEmployeeAgentBuilder().set_escalation_config(escalation_config)
```

{% endcode %}

#### 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](https://gdplabs.gitbook.io/catapa/developer-documentation/digital-employee-de-pipeline/guides/advanced-examples/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](https://gdplabs.gitbook.io/catapa/developer-documentation/digital-employee-de-pipeline/guides/advanced-examples/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.