diff --git a/src/uipath_langchain/agent/tools/a2a/a2a_tool.py b/src/uipath_langchain/agent/tools/a2a/a2a_tool.py index b5cc00784..44b679cfc 100644 --- a/src/uipath_langchain/agent/tools/a2a/a2a_tool.py +++ b/src/uipath_langchain/agent/tools/a2a/a2a_tool.py @@ -38,6 +38,7 @@ from uipath_langchain.agent.tools.base_uipath_structured_tool import ( BaseUiPathStructuredTool, ) +from uipath_langchain.agent.tools.static_args import wrap_tools_with_static_args from uipath_langchain.agent.tools.tool_node import ( ToolWrapperMixin, ToolWrapperReturnType, @@ -346,7 +347,7 @@ def create_a2a_tools_and_clients( tools.append(tool) clients.append(a2a_client) - return tools, clients + return wrap_tools_with_static_args(tools), clients @asynccontextmanager diff --git a/src/uipath_langchain/agent/tools/mcp/mcp_tool.py b/src/uipath_langchain/agent/tools/mcp/mcp_tool.py index 6963f1f70..85f37bb6d 100644 --- a/src/uipath_langchain/agent/tools/mcp/mcp_tool.py +++ b/src/uipath_langchain/agent/tools/mcp/mcp_tool.py @@ -15,6 +15,7 @@ StructuredToolWithArgumentProperties, ) +from ..static_args import wrap_tools_with_static_args from ..utils import sanitize_tool_name from .mcp_client import McpClient, SessionInfoFactory @@ -236,4 +237,4 @@ async def create_mcp_tools_and_clients( f"Created {len(resource_tools)} tools for MCP resource '{resource.name}'" ) - return tools, clients + return wrap_tools_with_static_args(tools), clients diff --git a/src/uipath_langchain/agent/tools/schema_editing.py b/src/uipath_langchain/agent/tools/schema_editing.py index de8a19349..a0c9bc88d 100644 --- a/src/uipath_langchain/agent/tools/schema_editing.py +++ b/src/uipath_langchain/agent/tools/schema_editing.py @@ -298,3 +298,40 @@ def _index_of_non_null_schema( _inline_ref_if_present(schema, container["anyOf"], target_index) return container["anyOf"][target_index] + + +def remove_fields_from_schema( + schema: dict[str, Any], + json_paths: list[str], +) -> set[str]: + """Remove the named fields at ``json_paths`` from the schema (in place). + + For each path, navigate to the parent object and drop the named field from + its ``properties`` and ``required``. Used to hide statically-configured + parameters from the model-facing tool schema. Paths that do not resolve to a + named object field (array elements, or fields absent from the schema) are + skipped. Returns the set of paths that were actually removed. + """ + removed: set[str] = set() + for json_path in json_paths: + try: + segments = parse_jsonpath_segments(json_path) + except Exception: + # A malformed path is simply not removable; leave the field in place. + continue + if not segments or segments[-1] == "*": + continue + field_name = segments[-1] + try: + parent = _navigate_schema_inlining_refs(schema, segments[:-1]) + except SchemaNavigationError: + continue + properties = parent.get("properties") + if not isinstance(properties, dict) or field_name not in properties: + continue + del properties[field_name] + required = parent.get("required") + if isinstance(required, list) and field_name in required: + required.remove(field_name) + removed.add(json_path) + return removed diff --git a/src/uipath_langchain/agent/tools/static_args.py b/src/uipath_langchain/agent/tools/static_args.py index ffa13de3a..3b2a8a47f 100644 --- a/src/uipath_langchain/agent/tools/static_args.py +++ b/src/uipath_langchain/agent/tools/static_args.py @@ -30,6 +30,8 @@ InvalidStaticArgError, SchemaNavigationError, apply_static_value_to_schema, + parse_jsonpath_segments, + remove_fields_from_schema, ) from .utils import sanitize_dict_for_serialization @@ -342,3 +344,127 @@ def apply_to_response(self, tool_calls: list[ToolCall]) -> None: static_values = self._sanitized_static_values.get(tool_call["name"]) if static_values: tool_call["args"] = apply_static_args(static_values, tool_call["args"]) + + +def wrap_tool_with_static_args(tool: BaseTool) -> BaseTool: + """Wrap a tool so its *static* parameters are hidden from the model and + injected just before the underlying tool runs. + + The returned tool exposes only the non-static parameters in its + ``args_schema``; on invocation the configured static values are merged back + into the arguments before the original tool's ``coroutine`` / ``func`` runs. + The hidden static values are appended to the tool ``description`` (sensitive + values redacted) so the model can still distinguish otherwise-identical tools + by their configured values. + + This is a per-tool, agent-state-independent counterpart to + ``StaticArgsHandler``: static values are constants + (``AgentToolStaticArgumentProperties``), so no agent input is needed to + resolve them. Tools without static ``argument_properties`` are returned + unchanged, and the original tool is never mutated. + """ + properties: dict[str, AgentToolArgumentProperties] | None = getattr( + tool, "argument_properties", None + ) + if not isinstance(tool, StructuredTool) or not properties: + return tool + + static_properties = { + json_path: props + for json_path, props in properties.items() + if isinstance(props, AgentToolStaticArgumentProperties) + } + if not static_properties: + return tool + + static_values = { + json_path: static_arg.value + for json_path, static_arg in _resolve_argument_properties( + static_properties, {} + ).items() + } + + args_schema = tool.args_schema + if isinstance(args_schema, dict): + schema = copy.deepcopy(args_schema) + elif args_schema is not None and issubclass(args_schema, BaseModel): + schema = args_schema.model_json_schema() + else: + return tool + + removed = remove_fields_from_schema(schema, list(static_values)) + if not removed: + return tool + + inject_values = { + json_path: value + for json_path, value in static_values.items() + if json_path in removed + } + + # Surface the now-hidden static values in the tool description so the model + # can still tell otherwise-identical tools apart (e.g. one send-email tool + # configured for Bob and another for Alice) without the user having to encode + # it in the tool name. Sensitive values are redacted. + preconfigured_lines: list[str] = [] + for json_path in sorted(removed): + static_prop = static_properties[json_path] + segments = parse_jsonpath_segments(json_path) + name = ".".join(segments) if segments else json_path + if static_prop.is_sensitive: + preconfigured_lines.append(f"- {name}: ") + else: + value_text = str(static_prop.value) + if len(value_text) > 200: + value_text = value_text[:200] + "..." + preconfigured_lines.append(f"- {name}: {value_text}") + + description = tool.description + if preconfigured_lines: + description = ( + f"{tool.description}\n\n" + "Pre-configured parameters (set automatically, do not provide):\n" + + "\n".join(preconfigured_lines) + ) + + update: dict[str, Any] = { + "args_schema": create_model(schema), + "description": description, + # Drop the static argument properties now baked into the wrapper so the + # StaticArgsHandler still running in the ReAct llm_node does not also + # process them; non-static variants are left for it (back-compat). + "argument_properties": { + json_path: props + for json_path, props in properties.items() + if json_path not in removed + }, + } + + coroutine = tool.coroutine + if coroutine is not None: + _coroutine = coroutine + + async def _acall(**kwargs: Any) -> Any: + return await _coroutine(**apply_static_args(inject_values, kwargs)) + + update["coroutine"] = _acall + + func = tool.func + if func is not None: + _func = func + + def _call(**kwargs: Any) -> Any: + return _func(**apply_static_args(inject_values, kwargs)) + + update["func"] = _call + + return tool.model_copy(update=update) + + +def wrap_tools_with_static_args(tools: Sequence[BaseTool]) -> list[BaseTool]: + """Hide and inject each tool's static parameters. + + See :func:`wrap_tool_with_static_args`. Tools without static parameters pass + through unchanged. + """ + return [wrap_tool_with_static_args(tool) for tool in tools] diff --git a/src/uipath_langchain/agent/tools/tool_factory.py b/src/uipath_langchain/agent/tools/tool_factory.py index f6a7fb4b7..dfe12c111 100644 --- a/src/uipath_langchain/agent/tools/tool_factory.py +++ b/src/uipath_langchain/agent/tools/tool_factory.py @@ -27,6 +27,7 @@ from .internal_tools import create_internal_tool from .ixp_escalation_tool import create_ixp_escalation_tool from .process_tool import create_process_tool +from .static_args import wrap_tools_with_static_args logger = getLogger(__name__) @@ -92,7 +93,9 @@ async def create_tools_from_resources( tool.metadata = {} tool.metadata[REQUIRE_CONVERSATIONAL_CONFIRMATION] = True - return tools + # Hide statically-configured tool parameters from the model and inject them + # at call time (e.g. a fixed Integration Service search `provider`). + return wrap_tools_with_static_args(tools) async def _build_tool_for_resource( diff --git a/tests/agent/tools/test_wrap_static_args.py b/tests/agent/tools/test_wrap_static_args.py new file mode 100644 index 000000000..f6a288677 --- /dev/null +++ b/tests/agent/tools/test_wrap_static_args.py @@ -0,0 +1,195 @@ +"""Tests for the static-argument tool-wrapping helpers. + +``wrap_tool_with_static_args`` hides a tool's *static* parameters from the +model-facing schema and injects the configured values just before the +underlying tool runs (a per-tool, state-independent counterpart to +``StaticArgsHandler``). +""" + +from typing import Any + +from pydantic import BaseModel, Field +from uipath.agent.models.agent import ( + AgentToolArgumentArgumentProperties, + AgentToolArgumentProperties, + AgentToolStaticArgumentProperties, +) + +from uipath_langchain.agent.tools.schema_editing import remove_fields_from_schema +from uipath_langchain.agent.tools.static_args import ( + wrap_tool_with_static_args, + wrap_tools_with_static_args, +) +from uipath_langchain.agent.tools.structured_tool_with_argument_properties import ( + StructuredToolWithArgumentProperties, +) + + +class _ToolInput(BaseModel): + host: str + port: int = Field(default=8080) + api_key: str + + +def _recording_tool( + argument_properties: dict[str, AgentToolArgumentProperties], +) -> tuple[StructuredToolWithArgumentProperties, dict[str, Any]]: + """A tool whose coroutine records the kwargs it is ultimately called with.""" + received: dict[str, Any] = {} + + async def tool_fn(**kwargs: Any) -> str: + received.update(kwargs) + return "ok" + + tool = StructuredToolWithArgumentProperties( + name="rec", + description="A recording tool", + args_schema=_ToolInput, + coroutine=tool_fn, + output_type=None, + argument_properties=argument_properties, + ) + return tool, received + + +def _static(value: Any) -> AgentToolStaticArgumentProperties: + return AgentToolStaticArgumentProperties(value=value, is_sensitive=False) + + +def _argument(path: str) -> AgentToolArgumentArgumentProperties: + return AgentToolArgumentArgumentProperties(argument_path=path, is_sensitive=False) + + +async def test_static_arg_hidden_from_schema_and_injected_on_call() -> None: + props: dict[str, AgentToolArgumentProperties] = {"$['host']": _static("localhost")} + tool, received = _recording_tool(props) + wrapped = wrap_tool_with_static_args(tool) + + # 'host' is hidden from the model-facing schema; the rest remain. + args_schema = wrapped.args_schema + assert args_schema is not None and not isinstance(args_schema, dict) + properties = args_schema.model_json_schema()["properties"] + assert "host" not in properties + assert "port" in properties + assert "api_key" in properties + + # Invoking with only the non-static args injects the static value. + await wrapped.ainvoke({"port": 9090, "api_key": "k"}) + assert received == {"host": "localhost", "port": 9090, "api_key": "k"} + + +def test_static_property_dropped_but_non_static_kept() -> None: + props: dict[str, AgentToolArgumentProperties] = { + "$['host']": _static("localhost"), + "$['api_key']": _argument("key"), + } + tool, _ = _recording_tool(props) + wrapped = wrap_tool_with_static_args(tool) + + kept = getattr(wrapped, "argument_properties", {}) + assert "$['host']" not in kept # static is now baked into the wrapper + assert "$['api_key']" in kept # argument-variant left for StaticArgsHandler + + +def test_original_tool_is_not_mutated() -> None: + props: dict[str, AgentToolArgumentProperties] = {"$['host']": _static("localhost")} + tool, _ = _recording_tool(props) + wrap_tool_with_static_args(tool) + + original_schema = tool.args_schema + assert original_schema is not None and not isinstance(original_schema, dict) + assert "host" in original_schema.model_json_schema()["properties"] + assert "$['host']" in tool.argument_properties + + +def test_tool_without_static_args_passthrough() -> None: + props: dict[str, AgentToolArgumentProperties] = {"$['host']": _argument("h")} + tool, _ = _recording_tool(props) + assert wrap_tool_with_static_args(tool) is tool + + +def test_wrap_tools_with_static_args_maps_list() -> None: + static_props: dict[str, AgentToolArgumentProperties] = { + "$['host']": _static("localhost") + } + plain_props: dict[str, AgentToolArgumentProperties] = {"$['host']": _argument("h")} + static_tool, _ = _recording_tool(static_props) + plain_tool, _ = _recording_tool(plain_props) + + wrapped = wrap_tools_with_static_args([static_tool, plain_tool]) + assert len(wrapped) == 2 + assert wrapped[0] is not static_tool # static tool was wrapped + assert wrapped[1] is plain_tool # no static args -> returned unchanged + + +def test_remove_fields_from_schema_top_level() -> None: + schema: dict[str, Any] = { + "type": "object", + "properties": {"a": {"type": "string"}, "b": {"type": "string"}}, + "required": ["a", "b"], + } + removed = remove_fields_from_schema(schema, ["$['a']"]) + assert removed == {"$['a']"} + assert "a" not in schema["properties"] + assert "b" in schema["properties"] + assert schema["required"] == ["b"] + + +def test_remove_fields_from_schema_skips_missing_and_array_paths() -> None: + schema: dict[str, Any] = { + "type": "object", + "properties": {"a": {"type": "string"}}, + "required": ["a"], + } + removed = remove_fields_from_schema(schema, ["$['missing']", "$['a'][*]"]) + assert removed == set() + assert "a" in schema["properties"] + + +def test_wrapped_description_lists_static_params() -> None: + props: dict[str, AgentToolArgumentProperties] = { + "$['host']": _static("db.internal") + } + tool, _ = _recording_tool(props) + wrapped = wrap_tool_with_static_args(tool) + + # Original description is preserved, plus a line naming the hidden static + # parameter and its value so the model can still reason about it. + assert tool.description in wrapped.description + assert "host" in wrapped.description + assert "db.internal" in wrapped.description + + +def test_wrapped_description_redacts_sensitive_static_params() -> None: + props: dict[str, AgentToolArgumentProperties] = { + "$['host']": AgentToolStaticArgumentProperties( + value="super-secret-value", is_sensitive=True + ) + } + tool, _ = _recording_tool(props) + wrapped = wrap_tool_with_static_args(tool) + + assert "host" in wrapped.description # the name is still shown + assert "" in wrapped.description # but the value is redacted + assert "super-secret-value" not in wrapped.description + + +def test_same_named_tools_get_distinct_descriptions() -> None: + """Two identically-named tools differing only by a static value (e.g. a + send-email tool for Bob vs Alice) become distinguishable via the description.""" + bob_props: dict[str, AgentToolArgumentProperties] = { + "$['host']": _static("bob@acme.com") + } + alice_props: dict[str, AgentToolArgumentProperties] = { + "$['host']": _static("alice@acme.com") + } + bob, _ = _recording_tool(bob_props) + alice, _ = _recording_tool(alice_props) + + wrapped_bob = wrap_tool_with_static_args(bob) + wrapped_alice = wrap_tool_with_static_args(alice) + + assert wrapped_bob.name == wrapped_alice.name # same (user-configured) name + assert wrapped_bob.description != wrapped_alice.description + assert "bob@acme.com" in wrapped_bob.description + assert "alice@acme.com" in wrapped_alice.description diff --git a/tests/agent/tools/test_wrap_static_args_tool_types.py b/tests/agent/tools/test_wrap_static_args_tool_types.py new file mode 100644 index 000000000..b750dc70e --- /dev/null +++ b/tests/agent/tools/test_wrap_static_args_tool_types.py @@ -0,0 +1,458 @@ +"""Cross-type matrix for the static-argument tool-wrapping helpers. + +Exercises ``wrap_tool_with_static_args`` / ``wrap_tools_with_static_args`` across +the parameter shapes a configured tool can take: + +* static parameters (hidden from the model, injected at call time) +* parameters that come from agent input (argument-variant: left untouched) +* tools where *every* parameter is static (no model-facing input remains) +* tools with no parameters +* tools where no parameter is static +* tools with very large descriptions +* tools with a large number of parameters + +...crossed with the concrete tool kinds used in production: MCP tools (dict +JSON-schema), agent-resource tools (pydantic schema, optional execution +wrapper), and A2A tools (wrapper-bearing, no static args). +""" + +from typing import Any, Callable + +import pytest +from langchain_core.tools import BaseTool, StructuredTool +from pydantic import BaseModel +from pydantic import create_model as make_model +from uipath.agent.models.agent import ( + AgentToolArgumentArgumentProperties, + AgentToolArgumentProperties, + AgentToolStaticArgumentProperties, +) + +from uipath_langchain.agent.tools.a2a.a2a_tool import ( + A2aStructuredToolWithWrapper, + A2aToolInput, +) +from uipath_langchain.agent.tools.extraction_tool import StructuredToolWithWrapper +from uipath_langchain.agent.tools.static_args import ( + wrap_tool_with_static_args, + wrap_tools_with_static_args, +) +from uipath_langchain.agent.tools.structured_tool_with_argument_properties import ( + StructuredToolWithArgumentProperties, +) + +_JSON_TO_PY: dict[str, type] = { + "string": str, + "integer": int, + "number": float, + "boolean": bool, +} + + +def _p(name: str) -> str: + """Top-level jsonpath for a parameter name, e.g. ``host`` -> ``$['host']``.""" + return f"$['{name}']" + + +def _static(value: Any) -> AgentToolStaticArgumentProperties: + return AgentToolStaticArgumentProperties(value=value, is_sensitive=False) + + +def _sensitive(value: Any) -> AgentToolStaticArgumentProperties: + return AgentToolStaticArgumentProperties(value=value, is_sensitive=True) + + +def _argument(path: str) -> AgentToolArgumentArgumentProperties: + return AgentToolArgumentArgumentProperties(argument_path=path, is_sensitive=False) + + +# --- tool builders, one per production tool kind ------------------------------- +# +# Each builder shares the same signature so scenarios can be parametrized over +# them. They return ``(tool, received)`` where ``received`` records the kwargs +# the underlying callable is ultimately invoked with. + +ToolBuilder = Callable[..., tuple[BaseTool, dict[str, Any]]] + + +def _dict_schema(fields: dict[str, str]) -> dict[str, Any]: + """A raw JSON schema dict (the shape MCP tools carry).""" + return { + "type": "object", + "properties": {name: {"type": typ} for name, typ in fields.items()}, + "required": list(fields), + } + + +def _model_schema(fields: dict[str, str]) -> type[BaseModel]: + """A pydantic model (the shape most agent-resource tools carry).""" + field_defs: dict[str, Any] = { + name: (_JSON_TO_PY[typ], ...) for name, typ in fields.items() + } + return make_model("DynamicInput", **field_defs) + + +def _mcp_tool( + fields: dict[str, str], + argument_properties: dict[str, AgentToolArgumentProperties], + description: str = "An MCP tool", + awrapper: Any | None = None, +) -> tuple[StructuredToolWithArgumentProperties, dict[str, Any]]: + received: dict[str, Any] = {} + + async def tool_fn(**kwargs: Any) -> str: + received.update(kwargs) + return "ok" + + tool = StructuredToolWithArgumentProperties( + name="mcp_tool", + description=description, + args_schema=_dict_schema(fields), # MCP tools use raw dict schemas + coroutine=tool_fn, + output_type=Any, + metadata={"tool_type": "mcp", "display_name": "mcp_tool", "slug": "srv"}, + argument_properties=argument_properties, + ) + if awrapper is not None: + tool.set_tool_wrappers(awrapper=awrapper) + return tool, received + + +def _resource_tool( + fields: dict[str, str], + argument_properties: dict[str, AgentToolArgumentProperties], + description: str = "An integration tool", + awrapper: Any | None = None, +) -> tuple[StructuredToolWithArgumentProperties, dict[str, Any]]: + received: dict[str, Any] = {} + + async def tool_fn(**kwargs: Any) -> str: + received.update(kwargs) + return "ok" + + tool = StructuredToolWithArgumentProperties( + name="resource_tool", + description=description, + args_schema=_model_schema(fields), # resource tools use pydantic models + coroutine=tool_fn, + output_type=None, + metadata={"tool_type": "integration", "display_name": "resource_tool"}, + argument_properties=argument_properties, + ) + if awrapper is not None: + tool.set_tool_wrappers(awrapper=awrapper) + return tool, received + + +BUILDERS = [ + pytest.param(_mcp_tool, id="mcp"), + pytest.param(_resource_tool, id="resource"), +] + + +def _schema_props(tool: BaseTool) -> list[str]: + """Model-facing property names, for either a dict (MCP) or pydantic schema.""" + args_schema = tool.args_schema + assert args_schema is not None + if isinstance(args_schema, dict): + return list(args_schema.get("properties", {})) + return list(args_schema.model_json_schema().get("properties", {})) + + +# --- scenario × tool-kind matrix ----------------------------------------------- + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_static_parameters(build: ToolBuilder) -> None: + """A static parameter is hidden from the schema, surfaced in the + description, and injected at call time; non-static params are untouched.""" + fields = {"host": "string", "port": "integer", "api_key": "string"} + tool, received = build(fields, {_p("host"): _static("localhost")}) + + wrapped = wrap_tool_with_static_args(tool) + + assert "host" not in _schema_props(wrapped) + assert "port" in _schema_props(wrapped) + assert "api_key" in _schema_props(wrapped) + assert "host: localhost" in wrapped.description + assert wrapped.metadata == tool.metadata # metadata preserved + + await wrapped.ainvoke({"port": 9090, "api_key": "k"}) + assert received == {"host": "localhost", "port": 9090, "api_key": "k"} + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_static_and_input_parameters(build: ToolBuilder) -> None: + """Static params are baked in; input (argument-variant) params and plain + params stay model-facing, and input params remain in argument_properties.""" + fields = {"provider": "string", "query": "string", "extra": "string"} + tool, received = build( + fields, + {_p("provider"): _static("svc"), _p("query"): _argument("$.q")}, + ) + + wrapped = wrap_tool_with_static_args(tool) + + # provider (static) hidden + baked out; query (input) + extra (plain) remain. + assert "provider" not in _schema_props(wrapped) + assert "query" in _schema_props(wrapped) + assert "extra" in _schema_props(wrapped) + + kept = getattr(wrapped, "argument_properties", {}) + assert _p("provider") not in kept # static baked into the wrapper + assert _p("query") in kept # input variant left for StaticArgsHandler + + await wrapped.ainvoke({"query": "hi", "extra": "e"}) + assert received == {"provider": "svc", "query": "hi", "extra": "e"} + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_all_parameters_static_no_input(build: ToolBuilder) -> None: + """When every parameter is static the model-facing schema is empty and the + tool can be invoked with no arguments at all.""" + fields = {"a": "string", "b": "integer"} + tool, received = build(fields, {_p("a"): _static("x"), _p("b"): _static(7)}) + + wrapped = wrap_tool_with_static_args(tool) + + assert _schema_props(wrapped) == [] # nothing left for the model to fill + assert getattr(wrapped, "argument_properties", {}) == {} # all baked in + assert "a: x" in wrapped.description + assert "b: 7" in wrapped.description + + await wrapped.ainvoke({}) + assert received == {"a": "x", "b": 7} + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_no_static_parameters_passthrough(build: ToolBuilder) -> None: + """A tool whose only properties come from input (no static) is returned + unchanged (identity), so StaticArgsHandler keeps owning it.""" + tool, _ = build({"query": "string"}, {_p("query"): _argument("$.q")}) + assert wrap_tool_with_static_args(tool) is tool + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_no_parameters_passthrough(build: ToolBuilder) -> None: + """A parameterless tool (no argument_properties) is returned unchanged.""" + tool, _ = build({}, {}) + assert wrap_tool_with_static_args(tool) is tool + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_large_description_is_preserved(build: ToolBuilder) -> None: + """A large description is preserved verbatim, with the static lines appended + (the description itself is never truncated).""" + big = "Lorem ipsum dolor sit amet. " * 400 # ~11k chars + tool, _ = build( + {"host": "string", "port": "integer"}, + {_p("host"): _static("localhost")}, + description=big, + ) + + wrapped = wrap_tool_with_static_args(tool) + + assert big in wrapped.description # original kept in full, not truncated + assert "host: localhost" in wrapped.description + assert len(wrapped.description) > len(big) + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_large_static_value_truncated_in_description_only( + build: ToolBuilder, +) -> None: + """A long static value is truncated in the description (a readability cap) + but injected in full at call time.""" + long_value = "v" * 500 + tool, received = build( + {"token": "string", "port": "integer"}, + {_p("token"): _static(long_value)}, + ) + + wrapped = wrap_tool_with_static_args(tool) + + assert "..." in wrapped.description # value truncated for display + assert long_value not in wrapped.description # full value not shown + assert long_value[:200] in wrapped.description # the cap prefix is shown + + await wrapped.ainvoke({"port": 1}) + assert received["token"] == long_value # ...but injected in full + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_sensitive_static_value_redacted_in_description( + build: ToolBuilder, +) -> None: + """A sensitive static value is redacted in the description but injected.""" + tool, received = build( + {"api_key": "string", "port": "integer"}, + {_p("api_key"): _sensitive("super-secret")}, + ) + + wrapped = wrap_tool_with_static_args(tool) + + assert "api_key: " in wrapped.description + assert "super-secret" not in wrapped.description + + await wrapped.ainvoke({"port": 1}) + assert received["api_key"] == "super-secret" + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_large_number_of_parameters(build: ToolBuilder) -> None: + """With many parameters, every static one is hidden+injected and every + non-static one stays model-facing.""" + static_fields = {f"s{i}": "string" for i in range(20)} + input_fields = {f"in{i}": "string" for i in range(20)} + fields = {**static_fields, **input_fields} + props: dict[str, AgentToolArgumentProperties] = { + _p(name): _static(f"val-{name}") for name in static_fields + } + props.update({_p(name): _argument(f"$.{name}") for name in input_fields}) + + tool, received = build(fields, props) + wrapped = wrap_tool_with_static_args(tool) + + remaining = set(_schema_props(wrapped)) + assert remaining == set(input_fields) # only non-static remain + assert not (remaining & set(static_fields)) + + call_args = {name: f"got-{name}" for name in input_fields} + await wrapped.ainvoke(call_args) + + assert len(received) == len(fields) + for name in static_fields: + assert received[name] == f"val-{name}" + for name in input_fields: + assert received[name] == f"got-{name}" + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_execution_wrapper_preserved_through_static_wrapping( + build: ToolBuilder, +) -> None: + """Tools carry a graph-execution wrapper (set_tool_wrappers); static + wrapping must not drop it.""" + + async def _awrapper(tool: BaseTool, call: Any, state: Any) -> None: + return None + + tool, received = build( + {"host": "string", "port": "integer"}, + {_p("host"): _static("localhost")}, + awrapper=_awrapper, + ) + + wrapped = wrap_tool_with_static_args(tool) + + assert wrapped is not tool + assert getattr(wrapped, "awrapper", None) is _awrapper # survives model_copy + await wrapped.ainvoke({"port": 1}) + assert received["host"] == "localhost" + + +@pytest.mark.parametrize("build", BUILDERS) +async def test_original_tool_not_mutated(build: ToolBuilder) -> None: + """The source tool keeps its full schema, description and properties.""" + fields = {"host": "string", "port": "integer"} + tool, _ = build(fields, {_p("host"): _static("localhost")}, description="orig") + + wrap_tool_with_static_args(tool) + + assert set(_schema_props(tool)) == {"host", "port"} + assert tool.description == "orig" + assert _p("host") in getattr(tool, "argument_properties", {}) + + +# --- A2A and other wrapper-only / no-static-args tool kinds -------------------- + + +def _a2a_tool() -> tuple[A2aStructuredToolWithWrapper, Callable[..., Any]]: + async def _send(message: str) -> str: + return "ok" + + async def _awrapper(tool: BaseTool, call: Any, state: Any) -> None: + return None + + tool = A2aStructuredToolWithWrapper( + name="remote_agent", + description="A remote A2A agent", + coroutine=_send, + args_schema=A2aToolInput, + metadata={"tool_type": "a2a"}, + ) + tool.set_tool_wrappers(awrapper=_awrapper) + return tool, _awrapper + + +async def test_a2a_tool_passthrough_preserves_wrapper() -> None: + """A2A tools carry no static argument_properties, so they pass through + wrapping unchanged with their execution wrapper intact.""" + tool, awrapper = _a2a_tool() + + # A2A tools genuinely have no argument_properties. + assert getattr(tool, "argument_properties", None) is None + + wrapped = wrap_tools_with_static_args([tool]) + assert wrapped[0] is tool # identity passthrough + assert wrapped[0].awrapper is awrapper + + +async def test_wrapper_only_resource_tool_passthrough() -> None: + """Resource tools that use the execution-wrapper class but carry no static + args (e.g. IXP extraction / escalation) pass through unchanged.""" + + class _In(BaseModel): + doc_id: str + + async def _fn(**kwargs: Any) -> str: + return "ok" + + tool = StructuredToolWithWrapper( + name="extraction", + description="extract", + args_schema=_In, + coroutine=_fn, + output_type=None, + metadata={"tool_type": "ixp_extraction"}, + ) + assert wrap_tool_with_static_args(tool) is tool + + +async def test_plain_structured_tool_passthrough() -> None: + """A plain LangChain StructuredTool (e.g. a client-side tool) has no + argument_properties and is passed through unchanged.""" + + class _In(BaseModel): + value: str + + async def _fn(**kwargs: Any) -> str: + return "ok" + + tool = StructuredTool( + name="client_side", + description="client side", + args_schema=_In, + coroutine=_fn, + ) + assert wrap_tool_with_static_args(tool) is tool + + +async def test_wrap_tools_with_static_args_mixed_tool_kinds() -> None: + """A heterogeneous list: static tools get wrapped, everything else passes + through, and order/length are preserved.""" + static_mcp, _ = _mcp_tool({"host": "string"}, {_p("host"): _static("h")}) + static_resource, _ = _resource_tool( + {"provider": "string", "q": "string"}, {_p("provider"): _static("svc")} + ) + input_only, _ = _resource_tool({"q": "string"}, {_p("q"): _argument("$.q")}) + a2a, _ = _a2a_tool() + + tools: list[BaseTool] = [static_mcp, static_resource, input_only, a2a] + wrapped = wrap_tools_with_static_args(tools) + + assert len(wrapped) == 4 + assert wrapped[0] is not static_mcp # static -> wrapped + assert wrapped[1] is not static_resource # static -> wrapped + assert wrapped[2] is input_only # input-only -> passthrough + assert wrapped[3] is a2a # a2a -> passthrough