Extend

Some digital employee core components can be extended, for example, identity, connectors (tools & MCPs), and the digital employee agent.

Extend Use Cases

Identity

Digital employee core can be extended using this code example:

from digital_employee_core import DigitalEmployeeIdentity

class ExtendedDigitalEmployeeIdentity(DigitalEmployeeIdentity):
    """Extended Digital Employee Identity with additional attributes."""

    employee_id: str

Connectors

Tools

We can add our own tools in addition to those already provided by GL Connectors by extending the BaseTool.

Here is the example:

Tool Input

from pydantic import BaseModel, Field


class GreetingInput(BaseModel):
    user_name: str = Field(default="World", description="Name of the person to greet.")

Tool Config

from pydantic import BaseModel, Field


class GreetingConfig(BaseModel):
    greeting_prefix: str = Field(default="Hello", description="The greeting word/phrase.")
    greeting_suffix: str = Field(default="", description="Optional suffix after the greeting.")

Tool Implementation

from typing import Any

from langchain_core.runnables import RunnableConfig
from langchain_core.tools import BaseTool
from pydantic import BaseModel


class GreetingTool(BaseTool):
    """A simple greeting tool with configurable greeting."""

    name: str = "greeting"
    description: str = "Greet someone with a configurable greeting message."
    args_schema: type[BaseModel] = GreetingInput
    tool_config_schema: type[BaseModel] = GreetingConfig

    def _run(self, user_name: str = "World", config: RunnableConfig = None, **_kwargs: Any) -> str:
        try:
            tool_config = self.get_tool_config(config)
        except Exception as e:
            return f"Error: Failed to retrieve tool configuration. Details: {e}"

        greeting = tool_config.greeting_prefix
        suffix = tool_config.greeting_suffix

        return f"{greeting}, {user_name}! {suffix}"

To configure the tools, we need to create a config_templates/tool_configs.yaml file. Here is an example of tool_configs.yaml :

greeting:
  greeting_prefix: ${HELLO_GREETING_PREFIX}
  greeting_suffix: ${HELLO_GREETING_SUFFIX}

To define default value for configs, create a config_templates/defaults.yaml file. Here is an example of defaults.yaml :

HELLO_GREETING_PREFIX: Hello
HELLO_GREETING_SUFFIX: Welcome aboard!

MCPs

Here is how we can create our own MCP:

from glaip_sdk.mcps import MCP

new_google_mail_mcp = MCP(
    name="new_google_mail_mcp",
    description="MCP for Google Mail Operation for DE",
    transport="http",
    config={"url": "https://default.com/google_mail/mcp"},
)

We need to provide the config_templates/mcp_configs.yaml file to be able to use the MCP. Here is an example of mcp_configs.yaml:

new_google_mail_mcp:
  config:
    url: ${NEW_GOOGLE_MAIL_MCP_URL}
  authentication:
    type: api-key
    key: X-API-Key
    value: ${NEW_GOOGLE_MCP_X_API_KEY}

Digital Employee Agent

To create a custom Digital Employee Agent with specialized behavior, you can extend both DigitalEmployeeAgent and its builder. Here's a complete example with multiple pipeline steps using the pipe operator (|):

1. Create Custom Components (Optional)

First, let's create custom components for preprocessing and postprocessing:

from typing import Any
from gllm_core.schema import Component
from gllm_pipeline.steps import step


class PreprocessComponent(Component):
    """Component that preprocesses the input message."""

    def __init__(self) -> None:
        super().__init__()

    async def _run(self, message: str, **kwargs: Any) -> str:
        # Add preprocessing logic here
        return f"[PREPROCESSED] {message}"


class PostprocessComponent(Component):
    """Component that postprocesses the agent response."""

    def __init__(self) -> None:
        super().__init__()

    async def _run(self, agent_response: dict, **kwargs: Any) -> dict:
        # Add postprocessing logic here
        agent_response["processed"] = True
        return agent_response

2. Extend the Agent Class

from pathlib import Path

from glaip_sdk import MCP, Agent, Tool

from digital_employee_core import (
    DEFAULT_MODEL_NAME,
    DigitalEmployeeAgent,
    DigitalEmployeeConfig,
    DigitalEmployeeIdentity,
)
from digital_employee_core.config_templates.loader import ConfigTemplateLoader


class DigitalEmployeeWithConfigurableTool(DigitalEmployeeAgent):
    """Digital Employee Agent with configurable custom tool support.

    This subclass demonstrates how to extend the base DigitalEmployeeAgent
    with additional tool configs loaded from YAML templates.
    """

    def __init__(  # noqa: PLR0913
        self,
        identity: DigitalEmployeeIdentity,
        tools: list[Tool] | None = None,
        sub_agents: list[Agent] | None = None,
        mcps: list[MCP] | None = None,
        configs: list[DigitalEmployeeConfig] | None = None,
        model: str | None = DEFAULT_MODEL_NAME,
    ):
        """Initialize the Digital Employee Agent with configurable tool support.

        Args:
            identity (DigitalEmployeeIdentity): The Digital Employee's identity.
            tools (list[Tool] | None, optional): List of tools the Digital
                Employee can use. Defaults to None.
            sub_agents (list[Agent] | None, optional): List of sub-agents (for
                future use). Defaults to None.
            mcps (list[MCP] | None, optional): List of MCPs the Digital
                Employee can use. Defaults to None.
            configs (list[DigitalEmployeeConfig] | None, optional):
                List of configuration objects for customizing tool
                behavior and settings. Defaults to None.
            model (str | None, optional): Model identifier to use for the
                agent. Defaults to DEFAULT_MODEL_NAME.
        """
        # Initialize parent DigitalEmployeeAgent
        agent_config = {"model": model} if model else {}
        super().__init__(
            identity=identity,
            tools=tools or [],
            sub_agents=sub_agents or [],
            mcps=mcps or [],
            configs=configs or [],
            agent_config=agent_config,
        )

        # Set up config loader for custom tool templates
        config_dir = Path(__file__).parent / "config_templates"
        self.custom_config_loader = ConfigTemplateLoader(template_dir=config_dir)
        self.add_config_loader(self.custom_config_loader)

3. Create a Custom Builder

from digital_employee_core import DigitalEmployeeAgentBuilder


class DigitalEmployeeWithConfigurableToolBuilder(DigitalEmployeeAgentBuilder):
    """Builder for creating DigitalEmployeeWithConfigurableTool instances.

    This custom builder extends the base DigitalEmployeeAgentBuilder to create
    DigitalEmployeeWithConfigurableTool instances with their custom tool logic.

    Note: DigitalEmployeeAgentBuilder now implements StepBuilder, so this
    builder can be used directly with DigitalEmployeePipelineBuilder.add_step().

    Example:
        agent = (
            DigitalEmployeeWithConfigurableToolBuilder()
            .add_tools([Tool.from_langchain(GreetingTool)])
            .add_configs(configs)
            .build(identity)
        )
    """

    def build(self, identity: DigitalEmployeeIdentity) -> DigitalEmployeeWithConfigurableTool:
        """Build the DigitalEmployeeWithConfigurableTool instance.

        Args:
            identity (DigitalEmployeeIdentity): The identity for the agent.

        Returns:
            DigitalEmployeeWithConfigurableTool: The created Digital Employee Agent instance
                with configurable tool support.
        """
        # Create the custom agent with all configured settings
        return DigitalEmployeeWithConfigurableTool(
            identity=identity,
            tools=self._tools,
            sub_agents=self._sub_agents,
            mcps=self._mcps,
            configs=self._configs,
            model=self._model,
        )

4. Instantiate the Extended Digital Employee

Now that we have created the custom agent and builder, let's instantiate the Digital Employee.

Import the Package

import asyncio

from digital_employee_core.digital_employee.state import DigitalEmployeeState
from extended_digital_employee.connectors.mcps import new_google_mail_mcp  # new MCP
from extended_digital_employee.connectors.tools import GreetingTool  # new Tool
from extended_digital_employee.digital_employee import DigitalEmployeeWithConfigurableTool
from extended_digital_employee.digital_employee import DigitalEmployeeWithConfigurableToolBuilder

from glaip_sdk import Tool

from digital_employee_core import (
    DigitalEmployeeBuilder,
    DigitalEmployeePipelineBuilder,
)

Initialize the Extended Digital Employee with Multiple Steps (Improved Approach)

This example demonstrates the improved approach where ALL steps can be combined using the pipe operator (|) by using build_step() from the agent builder:

# Step 1: Create identity first (required for build_step())
identity = DigitalEmployeeIdentity(
    name="Extended Assistant",
    email="[e-mail]",
    job=DigitalEmployeeJob(
        title="Extended Digital Assistant",
        description="A digital employee with custom preprocessing and postprocessing",
        instruction="You are a helpful digital assistant. Answer questions clearly and concisely.",
    ),
)

# Step 2: Create preprocessing and postprocessing steps
preprocess_step = step(
    component=PreprocessComponent(),
    input_state_map={"message": "message"},
    output_state="message",
    name="Preprocess",
)

postprocess_step = step(
    component=PostprocessComponent(),
    input_state_map={"agent_response": "agent_response"},
    output_state="agent_response",
    name="Postprocess",
)

# Step 3: Create agent step using build_step()
agent_builder = DigitalEmployeeWithConfigurableToolBuilder()

# Use build_step() to convert builder to ComponentStep
agent_step = agent_builder.build_step(identity=identity, configs=[])

# Step 4: Combine ALL steps using pipe operator (|)
# Now we can use pipe operator for ALL steps!
full_pipeline = preprocess_step | agent_step | postprocess_step
full_pipeline.state_type = DigitalEmployeeState

# Step 5: Build the Digital Employee with the full pipeline
extended_digital_employee = (
    DigitalEmployeeBuilder()
    .set_name(identity.name)
    .set_email(identity.email)
    .set_job(
        title=identity.job.title,
        description=identity.job.description,
        instruction=identity.job.instruction,
    )
    .set_pipeline(full_pipeline)
    .build()
)

Key Improvement:

By using build_step(identity, configs) on the agent builder, we can convert any StepBuilder to a ComponentStep. This enables using the pipe operator (|) for ALL steps in the pipeline:

• ✓ Works: step1 | agent_step | step3 (all steps combined with pipe)

• ✓ Clean, readable pipeline definition

• ✓ Full flexibility in pipeline composition

Why This Works:

1. build_step(identity, configs) converts a StepBuilder to a ComponentStep

2. ComponentStep has the __or__ method implemented

3. The pipe operator combines ComponentStep instances into a Pipeline

4. The Pipeline can be added to DigitalEmployeePipelineBuilder as a single step

Important Requirement:

To use build_step(), you need to create the identity first. This means:

1. Create DigitalEmployeeIdentity with all required fields (name, email, job)

2. Call build_step(identity=identity, configs=[]) on your builder

3. Use the resulting ComponentStep with the pipe operator

Working Example:

See examples/extend_example_with_pipe_operator_v2.py for a complete working example that demonstrates this approach.

In this example, we use new MCPs (via .add_mcps([new_google_mail_mcp])) and tools (via .add_tools([Tool.from_langchain(GreetingTool)])) that are specifically created for the extended digital employee.

We also use HELLO_GREETING_PREFIX, HELLO_GREETING_SUFFIX, NEW_GOOGLE_MAIL_MCP_URL and NEW_GOOGLE_MCP_X_API_KEY configs, which are defined in the config template of the new tool and MCP.

Run the Extended Digital Employee

# Run the extended digital employee
result = asyncio.run(
    extended_digital_employee.invoke(
        message="My name is [your_name], please greet me and send the greeting to [your_email].",
    )
)

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

Important: Before running the sample code, replace the following placeholders:

1. Replace [gl-connectors-x-api-key] with x-api-key from GL Connectors. See below for one way to do it.

2. (Optional) Replace NEW_GOOGLE_MAIL_MCP_URL if you are using a different GL Connectors server instance.

🔑 Get GL Connectors x-api-key

1. Open https://connectors.gdplabs.id/console, then sign in.

2. In the Credentials section, expand the x-api-key panel and click Copy combined value button. Paste this value to replace [gl-connectors-x-api-key].

1. If your Gmail account has not been integrated yet, continue with the steps below.

2. Under Available Modules section, find the Google_mail integration and click Add New Integration button.

1. An authorization URL will appear. Click or copy the URL, then authenticate using your Gmail account.

1. Below is an example of a successfully integrated Gmail account.

To see more examples, please check Digital Employee Example.