Refactor static args handling and schema editing utilities

src/uipath_langchain/agent/tools/integration_tool.py

@@ -27,6 +27,7 @@
     ToolWrapperReturnType,
 )
 
+from .schema_editing import strip_matching_enums
 from .structured_tool_with_argument_properties import (
     StructuredToolWithArgumentProperties,
 )
@@ -62,7 +63,7 @@ def convert_integration_parameters_to_argument_properties(
             )
         elif param.field_variant == "argument":
             value_str = str(param.value) if param.value is not None else ""
-            match = re.fullmatch(r"\{\{(.+?)\}\}", value_str)
+            match = _TEMPLATE_PATTERN.match(value_str)
             if not match:
                 raise AgentStartupError(
                     code=AgentStartupErrorCode.INVALID_TOOL_CONFIG,
@@ -81,7 +82,7 @@ def convert_integration_parameters_to_argument_properties(
     return result
 
 
-_TEMPLATE_PATTERN = re.compile(r"^\{\{.*\}\}$")
+_TEMPLATE_PATTERN = re.compile(r"^\{\{(.+?)\}\}$")
 
 
 def _param_name_to_segments(param_name: str) -> list[str]:
@@ -133,52 +134,6 @@ def _is_param_name_to_jsonpath(param_name: str) -> str:
     return "$" + "".join(parts)
 
 
-def _resolve_schema_ref(
-    root_schema: dict[str, Any], node: dict[str, Any]
-) -> dict[str, Any]:
-    """Resolve a $ref pointer in a JSON schema node.
-
-    Returns the referenced definition if $ref is present,
-    otherwise returns the node unchanged.
-    """
-    ref = node.get("$ref")
-    if ref is None:
-        return node
-    parts = ref.lstrip("#/").split("/")
-    current = root_schema
-    for part in parts:
-        current = current[part]
-    return current
-
-
-def _navigate_schema_to_field(
-    root_schema: dict[str, Any], segments: list[str]
-) -> dict[str, Any] | None:
-    """Navigate a JSON schema to a leaf field using parsed path segments.
-
-    Args:
-        root_schema: The root schema (needed for $ref resolution).
-        segments: Path segments from _parse_is_param_name.
-
-    Returns:
-        The schema dict of the leaf field, or None if the path doesn't exist.
-    """
-    current = root_schema
-    for seg in segments:
-        current = _resolve_schema_ref(root_schema, current)
-        if seg == "*":
-            items = current.get("items")
-            if items is None:
-                return None
-            current = items
-        else:
-            props = current.get("properties")
-            if props is None or seg not in props:
-                return None
-            current = props[seg]
-    return _resolve_schema_ref(root_schema, current)
-
-
 def strip_template_enums_from_schema(
     schema: dict[str, Any],
     parameters: list[AgentIntegrationToolParameter],
@@ -206,21 +161,7 @@ def strip_template_enums_from_schema(
             continue
 
         segments = _param_name_to_segments(param.name)
-        field_schema = _navigate_schema_to_field(schema, segments)
-        if field_schema is None:
-            continue
-
-        enum = field_schema.get("enum")
-        if enum is None:
-            continue
-
-        cleaned = [
-            v for v in enum if not (isinstance(v, str) and _TEMPLATE_PATTERN.match(v))
-        ]
-        if not cleaned:
-            del field_schema["enum"]
-        else:
-            field_schema["enum"] = cleaned
+        strip_matching_enums(schema, segments, _TEMPLATE_PATTERN)
 
     return schema
 
@@ -392,7 +333,7 @@ async def integration_tool_wrapper(
         call: ToolCall,
         state: AgentGraphState,
     ) -> ToolWrapperReturnType:
-        call["args"] = handle_static_args(resource, state, call["args"])
+        call["args"] = handle_static_args(tool, state, call["args"])
         return await tool.ainvoke(call)
 
     tool = StructuredToolWithArgumentProperties(

src/uipath_langchain/agent/tools/schema_editing.py

@@ -2,6 +2,7 @@
 
 import copy
 import json
+import re
 from typing import Any
 
 from jsonpath_ng import (  # type: ignore[import-untyped]
@@ -108,6 +109,37 @@ def _navigate_schema_inlining_refs(
     return current
 
 
+def strip_matching_enums(
+    schema: dict[str, Any],
+    path_segments: list[str],
+    pattern: re.Pattern[str],
+) -> None:
+    """Navigate to a field in schema and remove enum values matching the pattern.
+
+    If the path doesn't exist in the schema, this is a no-op.
+    After removing matching values, if the enum becomes empty, the enum key is deleted.
+
+    Args:
+        schema: The root JSON schema (modified in place).
+        path_segments: Path segments to navigate to the field.
+        pattern: Compiled regex pattern to match enum values against.
+    """
+    try:
+        field_schema = _navigate_schema_inlining_refs(schema, path_segments)
+    except SchemaModificationError:
+        return
+
+    enum = field_schema.get("enum")
+    if enum is None:
+        return
+
+    cleaned = [v for v in enum if not (isinstance(v, str) and pattern.match(v))]
+    if not cleaned:
+        del field_schema["enum"]
+    else:
+        field_schema["enum"] = cleaned
+
+
 def _apply_sensitive_schema_modification(
     schema: dict[str, Any],
     path_parts: list[str],

src/uipath_langchain/agent/tools/static_args.py

@@ -8,13 +8,10 @@
 from langchain_core.tools import StructuredTool
 from pydantic import BaseModel
 from uipath.agent.models.agent import (
-    AgentIntegrationToolParameter,
-    AgentIntegrationToolResourceConfig,
     AgentToolArgumentArgumentProperties,
     AgentToolArgumentProperties,
     AgentToolStaticArgumentProperties,
     AgentToolTextBuilderArgumentProperties,
-    BaseAgentResourceConfig,
 )
 from uipath.agent.utils.text_tokens import build_string_from_tokens
 
@@ -137,24 +134,20 @@ def apply_static_argument_properties_to_schema(
 
 
 def resolve_static_args(
-    resource: BaseAgentResourceConfig,
+    resource: Any,
     agent_input: dict[str, Any],
 ) -> dict[str, Any]:
     """Resolves static arguments for a given resource with a given input.
 
     Args:
-        resource: The agent resource configuration.
-        input: The input arguments passed to the agent.
+        resource: The agent resource configuration or tool with argument_properties.
+        agent_input: The input arguments passed to the agent.
 
     Returns:
         A dictionary of expanded arguments to be used in the tool call.
     """
 
-    if isinstance(resource, AgentIntegrationToolResourceConfig):
-        return resolve_integration_static_args(
-            resource.properties.parameters, agent_input
-        )
-    elif hasattr(resource, "argument_properties"):
+    if hasattr(resource, "argument_properties"):
         # TODO: MCP tools don't inherit from BaseAgentResourceConfig; will need to handle separately
         static_arguments = _resolve_argument_properties_to_static_arguments(
             resource.argument_properties, agent_input
@@ -167,47 +160,6 @@ def resolve_static_args(
         return {}  # to be implemented for other resource types in the future
 
 
-def resolve_integration_static_args(
-    parameters: list[AgentIntegrationToolParameter],
-    agent_input: dict[str, Any],
-) -> dict[str, Any]:
-    """Resolves static arguments for an integration tool resource.
-
-    Args:
-        resource: The AgentIntegrationToolResourceConfig instance.
-        input: The input arguments passed to the agent.
-
-    Returns:
-        A dictionary of expanded static arguments for the integration tool.
-    """
-
-    static_args: dict[str, Any] = {}
-    for param in parameters:
-        value = None
-
-        # static parameter, use the defined static value
-        if param.field_variant == "static":
-            value = param.value
-        # argument parameter, extract value from agent input
-        elif param.field_variant == "argument":
-            if (
-                not isinstance(param.value, str)
-                or not param.value.startswith("{{")
-                or not param.value.endswith("}}")
-            ):
-                raise ValueError(
-                    f"Parameter value must be in the format '{{argument_name}}' when field_variant is 'argument', got {param.value}"
-                )
-            arg_name = param.value[2:-2].strip()
-            # currently only support top-level arguments
-            value = agent_input.get(arg_name)
-
-        if value is not None:
-            static_args[param.name] = value
-
-    return static_args
-
-
 def apply_static_args(
     static_args: dict[str, Any],
     kwargs: dict[str, Any],
@@ -244,12 +196,12 @@ def apply_static_args(
 
 
 def handle_static_args(
-    resource: BaseAgentResourceConfig, state: BaseModel, input_args: dict[str, Any]
+    resource: Any, state: BaseModel, input_args: dict[str, Any]
 ) -> dict[str, Any]:
     """Resolves and applies static arguments for a tool call.
     Args:
-        resource: The agent resource configuration.
-        runtime: The tool runtime providing the current state.
+        resource: The agent resource configuration or tool with argument_properties.
+        state: The current agent state.
         input_args: The original input arguments to the tool.
     Returns:
         A dictionary of input arguments with static arguments applied.

tests/agent/tools/test_integration_tool.py

@@ -461,6 +461,31 @@ def test_argument_parameter_with_invalid_template_raises(self):
         with pytest.raises(AgentStartupError):
             convert_integration_parameters_to_argument_properties(params)
 
+    @pytest.mark.parametrize(
+        "invalid_value",
+        [
+            "{missing_closing",
+            "missing_opening}",
+            "{{missing_closing}",
+            "{missing_opening}}",
+            "no_braces_at_all",
+        ],
+    )
+    def test_argument_parameter_with_malformed_braces_raises(self, invalid_value):
+        """Various malformed brace patterns raise AgentStartupError."""
+        params = [
+            AgentIntegrationToolParameter(
+                name="test_param",
+                type="string",
+                value=invalid_value,
+                field_location="body",
+                field_variant="argument",
+            ),
+        ]
+
+        with pytest.raises(AgentStartupError):
+            convert_integration_parameters_to_argument_properties(params)
+
 
 class TestStripTemplateEnumsFromSchema:
     """Test cases for strip_template_enums_from_schema function."""
@@ -783,7 +808,8 @@ def test_handles_ref_resolution(self):
 
         result = strip_template_enums_from_schema(schema, parameters)
 
-        config_props = result["definitions"]["Config"]["properties"]
+        # The $ref is inlined and modified on the inlined copy
+        config_props = result["properties"]["config"]["properties"]
         assert config_props["mode"]["enum"] == ["auto"]
 
     def test_skips_argument_param_when_schema_path_not_found(self):