diff --git a/src/openrouter_agent/model_result.py b/src/openrouter_agent/model_result.py index 11016c0..09224b2 100644 --- a/src/openrouter_agent/model_result.py +++ b/src/openrouter_agent/model_result.py @@ -29,6 +29,7 @@ StepResult, Tool, UnsentToolResult, + assert_unique_client_tool_names, get_tool_function, is_auto_resolvable_tool, is_client_tool, @@ -100,6 +101,7 @@ def _find_latest_user_string_index(items: Sequence[Any]) -> int: class ModelResult: def __init__(self, options: Mapping[str, Any]) -> None: + assert_unique_client_tool_names(options.get("tools")) self.options = dict(options) self._run_task: Optional[asyncio.Task[Any]] = None self._final_response: Any = None diff --git a/src/openrouter_agent/tool_types.py b/src/openrouter_agent/tool_types.py index f6db4b9..0fa52d7 100644 --- a/src/openrouter_agent/tool_types.py +++ b/src/openrouter_agent/tool_types.py @@ -2,7 +2,7 @@ from dataclasses import dataclass from enum import Enum -from typing import Any, Callable, Dict, List, Mapping, Optional, Protocol, Sequence, Union +from typing import Any, Callable, Dict, List, Mapping, Optional, Protocol, Sequence, Set, Union from typing_extensions import Literal, TypeAlias, TypedDict @@ -275,6 +275,36 @@ def get_tool_function(tool: Mapping[str, Any]) -> Dict[str, Any]: return value if isinstance(value, dict) else {} +def find_duplicate_client_tool_name(tools: Sequence[Mapping[str, Any]]) -> Optional[str]: + """Return the first client tool name that appears more than once, or `None`. + Server tools have no client name and are skipped.""" + seen: Set[str] = set() + for candidate in tools: + if is_server_tool(candidate): + continue + name = get_tool_function(candidate).get("name") + if not isinstance(name, str): + continue + if name in seen: + return name + seen.add(name) + return None + + +def assert_unique_client_tool_names(tools: Optional[Sequence[Mapping[str, Any]]]) -> None: + """Raise when two client tools share a name. Calls are dispatched by name, so a + duplicate would let whichever tool is listed first receive arguments the model + (or an approver) meant for the other.""" + if not tools: + return + name = find_duplicate_client_tool_name(tools) + if name is not None: + raise ValueError( + f'Duplicate tool name "{name}": every tool passed to call_model must have a unique name. ' + "Rename one, or prefix the names of MCP tools when combining them with local tools." + ) + + def has_execute_function(tool: Mapping[str, Any]) -> bool: fn = get_tool_function(tool) return is_client_tool(tool) and callable(fn.get("execute")) diff --git a/tests/unit/test_duplicate_tool_names.py b/tests/unit/test_duplicate_tool_names.py new file mode 100644 index 0000000..d6f694b --- /dev/null +++ b/tests/unit/test_duplicate_tool_names.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from typing import Any, Dict, List + +import pytest + +from openrouter_agent import call_model, tool +from openrouter_agent.model_result import ModelResult +from tests._fixtures import MemoryStateAccessor, QueuedClient, text_response, tool_call_response + + +def _tool(name: str, hits: List[Dict[str, Any]], **kwargs: Any) -> Any: + def execute(params: Dict[str, Any], ctx: Any) -> Dict[str, Any]: + hits.append(params) + return {"ok": True} + + return tool(name=name, input_schema=dict, output_schema=dict, execute=execute, **kwargs) + + +def test_rejects_two_client_tools_with_the_same_name() -> None: + first = _tool("read_value", []) + second = _tool("read_value", []) + + with pytest.raises(ValueError, match='Duplicate tool name "read_value"'): + ModelResult({"model": "test", "input": "hi", "tools": [first, second]}) + with pytest.raises(ValueError, match='Duplicate tool name "read_value"'): + call_model(QueuedClient([]), {"model": "test", "input": "hi", "tools": [first, second]}) + + +def test_allows_distinct_names() -> None: + ModelResult({"model": "test", "input": "hi", "tools": [_tool("a", []), _tool("b", [])]}) + + +async def test_approved_call_does_not_run_on_a_same_named_tool_added_before_resume() -> None: + state = MemoryStateAccessor() + approved_hits: List[Dict[str, Any]] = [] + other_hits: List[Dict[str, Any]] = [] + approved = _tool("read_value", approved_hits, require_approval=True) + other = _tool("read_value", other_hits) + + first = call_model( + QueuedClient([tool_call_response("resp_1", "read_value", call_id="call_1")]), + {"model": "test", "input": "read it", "tools": [approved], "state": state}, + ) + await first.get_response() + assert state.stored is not None + assert state.stored.status == "awaiting_approval" + + with pytest.raises(ValueError, match="Duplicate tool name"): + call_model( + QueuedClient([text_response("resp_done", "done")]), + { + "model": "test", + "input": [], + "tools": [other, approved], + "state": state, + "approve_tool_calls": ["call_1"], + }, + ) + + assert other_hits == [] + assert approved_hits == [] + assert state.stored.status == "awaiting_approval"