# 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: {% code lineNumbers="true" %} ```python from digital_employee_core import DigitalEmployeeIdentity class ExtendedDigitalEmployeeIdentity(DigitalEmployeeIdentity): """Extended Digital Employee Identity with additional attributes.""" employee_id: str ``` {% endcode %} ### 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** {% code lineNumbers="true" %} ```python from pydantic import BaseModel, Field class GreetingInput(BaseModel): user_name: str = Field(default="World", description="Name of the person to greet.") ``` {% endcode %} **Tool Config** {% code lineNumbers="true" %} ```python 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.") ``` {% endcode %} **Tool Implementation** {% code lineNumbers="true" %} ```python 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}" ``` {% endcode %} To configure the tools, we need to create a `config_templates/tool_configs.yaml` file. Here is an example of `tool_configs.yaml` : ```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` : ```yaml HELLO_GREETING_PREFIX: Hello HELLO_GREETING_SUFFIX: Welcome aboard! ``` #### **MCPs** Here is how we can create our own MCP: {% code lineNumbers="true" %} ```python 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"}, ) ``` {% endcode %} 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`: ```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: {% code lineNumbers="true" %} ```python 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 ``` {% endcode %} #### 2. Extend the Agent Class {% code lineNumbers="true" %} ```python 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) ``` {% endcode %} #### 3. Create a Custom Builder {% code lineNumbers="true" %} ```python 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, ) ``` {% endcode %} #### 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** {% code lineNumbers="true" %} ```python 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, ) ``` {% endcode %} **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: {% code lineNumbers="true" %} ```python # Step 1: Create identity first (required for build_step()) identity = DigitalEmployeeIdentity( name="Extended Assistant", email="extended.assistant@example.com", 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() ) ``` {% endcode %} {% hint style="success" %} **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 {% endhint %} {% hint style="info" %} **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 {% endhint %} {% hint style="warning" %} **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 {% endhint %} {% hint style="info" %} **Working Example:** See [examples/extend\_example\_with\_pipe\_operator\_v2.py](https://github.com/GDP-ADMIN/CATAPA-SDK/tree/main/python/digital-employee-core/examples/extend_example_with_pipe_operator_v2.py) for a complete working example that demonstrates this approach. {% endhint %} {% hint style="info" %} 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. {% endhint %} {% hint style="info" %} 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. {% endhint %} **Run the Extended Digital Employee** {% code lineNumbers="true" %} ```python # 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].", ) ) ``` {% endcode %} {% hint style="info" %} **Note:** The new pattern uses async/await. Use `employee.invoke(...)` instead of the old `employee.run(...)` method. {% endhint %} {% hint style="warning" %} **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. {% endhint %}
🔑 Get GL Connectors x-api-key 1. Open , 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]`.
3. If your Gmail account has not been integrated yet, continue with the steps below. 4. Under **Available Modules** section, find the **Google\_mail** integration and click **Add New Integration** button.
5. An authorization URL will appear. Click or copy the URL, then authenticate using your Gmail account.
6. Below is an example of a successfully integrated Gmail account.
{% hint style="info" %} To see more examples, please check [Digital Employee Example](https://github.com/GDP-ADMIN/CATAPA-SDK/tree/main/python/digital-employee-core/examples). {% endhint %}