feat: add content_tokens and argument_properties to agent model

src/uipath/agent/models/agent.py

@@ -92,6 +92,23 @@ class AgentGuardrailActionType(str, Enum):
     UNKNOWN = "unknown"  # fallback branch discriminator
 
 
+class AgentToolArgumentPropertiesVariant(str, Enum):
+    """Agent tool argument properties variant enumeration."""
+
+    DYNAMIC = "dynamic"
+    ARGUMENT = "argument"
+    STATIC = "static"
+    TEXT_BUILDER = "textBuilder"
+
+
+class TextTokenType(str, Enum):
+    """Text token type enumeration."""
+
+    SIMPLE_TEXT = "simpleText"
+    VARIABLE = "variable"
+    EXPRESSION = "expression"
+
+
 class BaseCfg(BaseModel):
     """Base configuration model with common settings."""
 
@@ -108,6 +125,59 @@ class ExampleCall(BaseCfg):
     output: str = Field(..., alias="output")
 
 
+class TextToken(BaseCfg):
+    """Text token model."""
+
+    type: TextTokenType
+    raw_string: str = Field(alias="rawString")
+
+
+class BaseAgentToolArgumentProperties(BaseCfg):
+    """Base tool argument properties model."""
+
+    variant: AgentToolArgumentPropertiesVariant
+    is_sensitive: bool = Field(alias="isSensitive")
+
+
+class AgentToolStaticArgumentProperties(BaseAgentToolArgumentProperties):
+    """Static tool argument properties model."""
+
+    variant: Literal[AgentToolArgumentPropertiesVariant.STATIC] = Field(
+        default=AgentToolArgumentPropertiesVariant.STATIC, frozen=True
+    )
+    value: Optional[Any]
+
+
+class AgentToolArgumentArgumentProperties(BaseAgentToolArgumentProperties):
+    """Agent argument argument properties model."""
+
+    variant: Literal[AgentToolArgumentPropertiesVariant.ARGUMENT] = Field(
+        default=AgentToolArgumentPropertiesVariant.ARGUMENT,
+        frozen=True,
+    )
+    argument_path: str = Field(alias="argumentPath")
+
+
+class AgentToolTextBuilderArgumentProperties(BaseAgentToolArgumentProperties):
+    """Agent text builder argument properties model."""
+
+    variant: Literal[AgentToolArgumentPropertiesVariant.TEXT_BUILDER] = Field(
+        default=AgentToolArgumentPropertiesVariant.TEXT_BUILDER,
+        frozen=True,
+    )
+    tokens: List[TextToken]
+
+
+AgentToolArgumentProperties = Annotated[
+    Union[
+        AgentToolStaticArgumentProperties,
+        AgentToolArgumentArgumentProperties,
+        AgentToolTextBuilderArgumentProperties,
+    ],
+    Field(discriminator="variant"),
+]
+
+
 class BaseResourceProperties(BaseCfg):
     """Base resource properties model."""
 
@@ -216,6 +286,9 @@ class AgentMcpTool(BaseCfg):
     description: str = Field(..., alias="description")
     input_schema: Dict[str, Any] = Field(..., alias="inputSchema")
     output_schema: Optional[Dict[str, Any]] = Field(None, alias="outputSchema")
+    argument_properties: Dict[str, AgentToolArgumentProperties] = Field(
+        {}, alias="argumentProperties"
+    )
 
 
 class AgentMcpResourceConfig(BaseAgentResourceConfig):
@@ -360,6 +433,9 @@ class AgentProcessToolResourceConfig(BaseAgentToolResourceConfig):
     properties: AgentProcessToolProperties
     settings: AgentToolSettings = Field(default_factory=AgentToolSettings)
     arguments: Dict[str, Any] = Field(default_factory=dict)
+    argument_properties: Dict[str, AgentToolArgumentProperties] = Field(
+        {}, alias="argumentProperties"
+    )
 
 
 class AgentIntegrationToolParameter(BaseCfg):
@@ -426,6 +502,9 @@ class AgentInternalToolResourceConfig(BaseAgentToolResourceConfig):
     arguments: Optional[Dict[str, Any]] = Field(default_factory=dict)
     is_enabled: Optional[bool] = Field(None, alias="isEnabled")
     output_schema: Dict[str, Any] = Field(..., alias="outputSchema")
+    argument_properties: Dict[str, AgentToolArgumentProperties] = Field(
+        {}, alias="argumentProperties"
+    )
 
 
 class AgentUnknownToolResourceConfig(BaseAgentToolResourceConfig):
@@ -677,6 +756,7 @@ class AgentMessage(BaseCfg):
 
     role: Literal[AgentMessageRole.SYSTEM, AgentMessageRole.USER]
     content: str
+    content_tokens: Optional[List[TextToken]] = Field(None, alias="contentTokens")
 
     @field_validator("role", mode="before")
     @classmethod

src/uipath/agent/utils/text_tokens.py

@@ -0,0 +1,136 @@
+"""Text token utilities for building prompts from tokenized content."""
+
+import json
+from typing import Any
+
+from uipath.agent.models.agent import TextToken, TextTokenType
+
+
+def build_string_from_tokens(
+    tokens: list[TextToken],
+    input_arguments: dict[str, Any],
+    tool_names: list[str] | None = None,
+    escalation_names: list[str] | None = None,
+    context_names: list[str] | None = None,
+) -> str:
+    """Build a string from text tokens with variable replacement.
+
+    Args:
+        tokens: List of text tokens to join
+        input_arguments: Dictionary of input arguments for variable replacement
+        tool_names: Optional list of tool names for tool.* variable resolution
+        escalation_names: Optional list of escalation names for escalation.* variable resolution
+        context_names: Optional list of context names for context.* variable resolution
+    """
+    parts: list[str] = []
+
+    for token in tokens:
+        if token.type == TextTokenType.SIMPLE_TEXT:
+            parts.append(token.raw_string)
+        elif token.type == TextTokenType.EXPRESSION:
+            parts.append(token.raw_string)
+        elif token.type == TextTokenType.VARIABLE:
+            resolved_value = _process_variable_token(
+                token.raw_string,
+                input_arguments,
+                tool_names,
+                escalation_names,
+                context_names,
+            )
+            parts.append(resolved_value)
+        else:
+            parts.append(token.raw_string)
+
+    return "".join(parts)
+
+
+def _process_variable_token(
+    raw_string: str,
+    input_arguments: dict[str, Any],
+    tool_names: list[str] | None = None,
+    escalation_names: list[str] | None = None,
+    context_names: list[str] | None = None,
+) -> str:
+    """Process a variable token and return its resolved value.
+
+    Returns:
+        The resolved variable value or original string if unresolved
+    """
+    if not raw_string or not raw_string.strip():
+        return raw_string
+
+    if raw_string.lower() == "input":
+        return json.dumps(input_arguments, ensure_ascii=False)
+
+    dot_index = raw_string.find(".")
+    if dot_index < 0:
+        return raw_string
+
+    prefix = raw_string[:dot_index].lower()
+    path = raw_string[dot_index + 1 :]
+
+    if prefix == "input":
+        value = safe_get_nested(input_arguments, path)
+        return serialize_argument(value) if value is not None else raw_string
+    elif prefix == "output":
+        return path
+    elif prefix == "tools":
+        found_name = _find_resource_name(path, tool_names)
+        return found_name if found_name else raw_string
+    elif prefix == "escalations":
+        found_name = _find_resource_name(path, escalation_names)
+        return found_name if found_name else raw_string
+    elif prefix == "contexts":
+        found_name = _find_resource_name(path, context_names)
+        return found_name if found_name else raw_string
+
+    return raw_string
+
+
+def _find_resource_name(name: str, resource_names: list[str] | None) -> str | None:
+    """Find a resource name in the list.
+
+    Args:
+        name: The name to search for
+        resource_names: List of resource names to search in
+
+    Returns:
+        The matching resource name, or None if not found
+    """
+    if not resource_names:
+        return None
+
+    name_lower = name.lower()
+    return next(
+        (
+            resource_name
+            for resource_name in resource_names
+            if resource_name.lower() == name_lower
+        ),
+        None,
+    )
+
+
+def safe_get_nested(data: dict[str, Any], path: str) -> Any:
+    """Get nested dictionary value using dot notation (e.g., "user.email")."""
+    keys = path.split(".")
+    current = data
+
+    for key in keys:
+        if isinstance(current, dict) and key in current:
+            current = current[key]
+        else:
+            return None
+
+    return current
+
+
+def serialize_argument(
+    value: str | int | float | bool | list[Any] | dict[str, Any] | None,
+) -> str:
+    """Serialize value for interpolation: primitives as-is, collections as JSON."""
+    if value is None:
+        return ""
+    if isinstance(value, (list, dict, bool)):
+        return json.dumps(value, ensure_ascii=False)
+    return str(value)

tests/agent/models/test_agent.py

@@ -42,10 +42,41 @@ def test_agent_with_all_tool_types_loads(self):
             "name": "Agent with All Tools",
             "metadata": {"isConversational": False, "storageVersion": "22.0.0"},
             "messages": [
-                {"role": "System", "content": "You are an agentic assistant."},
+                {
+                    "role": "System",
+                    "content": "You are an agentic assistant.",
+                    "contentTokens": [
+                        {
+                            "type": "simpleText",
+                            "rawString": "You are an agentic assistant.",
+                        }
+                    ],
+                },
                 {
                     "role": "User",
                     "content": "Use the provided tools. Execute {{task}} the number of {{times}}.",
+                    "contentTokens": [
+                        {
+                            "type": "simpleText",
+                            "rawString": "Use the provided tools. Execute ",
+                        },
+                        {
+                            "type": "variable",
+                            "rawString": "input.task",
+                        },
+                        {
+                            "type": "simpleText",
+                            "rawString": " the number of ",
+                        },
+                        {
+                            "type": "variable",
+                            "rawString": "input.times",
+                        },
+                        {
+                            "type": "simpleText",
+                            "rawString": ".",
+                        },
+                    ],
                 },
             ],
             "inputSchema": {
@@ -229,6 +260,13 @@ def test_agent_with_all_tool_types_loads(self):
                         "properties": {"output": {"type": "string"}},
                     },
                     "settings": {},
+                    "argumentProperties": {
+                        "task": {
+                            "variant": "argument",
+                            "argumentPath": "$['task']",
+                            "isSensitive": False,
+                        }
+                    },
                     "properties": {
                         "processName": "Basic RPA Process",
                         "folderPath": "TestFolder/Complete Solution 30 Sept",
@@ -274,6 +312,18 @@ def test_agent_with_all_tool_types_loads(self):
                                 },
                                 "required": ["timezone"],
                             },
+                            "argumentProperties": {
+                                "timezone": {
+                                    "variant": "textBuilder",
+                                    "tokens": [
+                                        {
+                                            "type": "simpleText",
+                                            "rawString": "Europe/London",
+                                        },
+                                    ],
+                                    "isSensitive": False,
+                                },
+                            },
                         },
                         {
                             "name": "convert_time",