mirror of
https://github.com/ggozad/oterm.git
synced 2026-10-10 17:23:22 +02:00
d077f671f1Tool identity was the bare tool name, so a name exported by two MCP
servers resolved to both toolsets and pydantic-ai rejected the agent.
MCP tools are now identified as `{server}_{tool}` and reach the model
through PrefixedToolset, making per-server selection independent.
Since the server name now reaches the provider, reject names outside
[a-zA-Z0-9_-] when the config loads and skip tools whose qualified name
is over 64 characters.
Tools load before the store so the 0.22.0 upgrade can qualify saved
selections from the connected servers' tool lists.
Point the ty pre-commit hook at the pinned version, matching CI and the
ruff hooks.
Closes #321
104 lines
3.1 KiB
Python
104 lines
3.1 KiB
Python
from pydantic_ai import Tool as PydanticTool
|
|
|
|
from oterm.tools import (
|
|
builtin_tools,
|
|
discover_tools,
|
|
known_tool_names,
|
|
make_tool_def,
|
|
)
|
|
|
|
|
|
def _hello() -> str:
|
|
"""Say hi."""
|
|
return "hi"
|
|
|
|
|
|
class TestMakeToolDef:
|
|
def test_wraps_callable(self):
|
|
tool_def = make_tool_def(_hello)
|
|
assert tool_def["name"] == "_hello"
|
|
assert tool_def["description"] == "Say hi."
|
|
assert isinstance(tool_def["tool"], PydanticTool)
|
|
|
|
def test_missing_docstring_gives_empty_description(self):
|
|
def undocumented():
|
|
return "x"
|
|
|
|
tool_def = make_tool_def(undocumented)
|
|
assert tool_def["description"] == ""
|
|
|
|
|
|
class TestDiscoverTools:
|
|
def test_discovers_registered_entry_points(self):
|
|
tool_defs = discover_tools()
|
|
names = {t["name"] for t in tool_defs}
|
|
assert {"think", "date_time", "shell"}.issubset(names)
|
|
|
|
def test_descriptions_populated_for_builtins(self):
|
|
for tool_def in discover_tools():
|
|
assert tool_def["description"] != ""
|
|
|
|
def test_non_callable_entry_point_is_skipped(self, monkeypatch):
|
|
import oterm.log
|
|
import oterm.tools as tools_mod
|
|
|
|
class _FakeEntryPoint:
|
|
name = "broken"
|
|
value = "oterm.types:ChatModel"
|
|
|
|
def load(self):
|
|
return 42
|
|
|
|
monkeypatch.setattr(
|
|
tools_mod, "entry_points", lambda group=None: [_FakeEntryPoint()]
|
|
)
|
|
before = len(oterm.log.log_lines)
|
|
assert discover_tools() == []
|
|
messages = [msg for _, msg in oterm.log.log_lines[before:]]
|
|
assert any("not callable" in m for m in messages)
|
|
|
|
def test_loader_error_is_logged_and_skipped(self, monkeypatch):
|
|
import oterm.log
|
|
import oterm.tools as tools_mod
|
|
|
|
class _FailingEntryPoint:
|
|
name = "exploder"
|
|
value = "does.not:exist"
|
|
|
|
def load(self):
|
|
raise ImportError("kaboom")
|
|
|
|
monkeypatch.setattr(
|
|
tools_mod, "entry_points", lambda group=None: [_FailingEntryPoint()]
|
|
)
|
|
before = len(oterm.log.log_lines)
|
|
assert discover_tools() == []
|
|
messages = [msg for _, msg in oterm.log.log_lines[before:]]
|
|
assert any("exploder" in m for m in messages)
|
|
|
|
|
|
def test_builtin_tools_is_a_list():
|
|
assert isinstance(builtin_tools, list)
|
|
|
|
|
|
class TestKnownToolNames:
|
|
def test_mcp_names_are_qualified_by_server(self, monkeypatch):
|
|
import oterm.tools as tools_mod
|
|
import oterm.tools.capabilities as capabilities_mod
|
|
import oterm.tools.mcp.setup as mcp_setup_mod
|
|
|
|
monkeypatch.setattr(tools_mod, "builtin_tools", [{"name": "shell"}])
|
|
monkeypatch.setattr(capabilities_mod, "capability_defs", [])
|
|
monkeypatch.setattr(
|
|
mcp_setup_mod,
|
|
"mcp_tool_meta",
|
|
{
|
|
"k8s": [{"name": "query_prometheus", "description": ""}],
|
|
"grafana": [{"name": "query_prometheus", "description": ""}],
|
|
},
|
|
)
|
|
assert known_tool_names() == {
|
|
"shell",
|
|
"k8s_query_prometheus",
|
|
"grafana_query_prometheus",
|
|
}
|