fix(cache): stabilize tool ordering and cache markers for MCP

This commit is contained in:
pikaxinge
2026-04-01 17:07:22 +00:00
parent 63d646f731
commit 607fd8fd7e
5 changed files with 234 additions and 10 deletions
@@ -0,0 +1,87 @@
from __future__ import annotations
from typing import Any
from nanobot.providers.anthropic_provider import AnthropicProvider
from nanobot.providers.openai_compat_provider import OpenAICompatProvider
def _openai_tools(*names: str) -> list[dict[str, Any]]:
return [
{
"type": "function",
"function": {
"name": name,
"description": f"{name} tool",
"parameters": {"type": "object", "properties": {}},
},
}
for name in names
]
def _anthropic_tools(*names: str) -> list[dict[str, Any]]:
return [
{
"name": name,
"description": f"{name} tool",
"input_schema": {"type": "object", "properties": {}},
}
for name in names
]
def _marked_openai_tool_names(tools: list[dict[str, Any]] | None) -> list[str]:
if not tools:
return []
marked: list[str] = []
for tool in tools:
if "cache_control" in tool:
marked.append((tool.get("function") or {}).get("name", ""))
return marked
def _marked_anthropic_tool_names(tools: list[dict[str, Any]] | None) -> list[str]:
if not tools:
return []
return [tool.get("name", "") for tool in tools if "cache_control" in tool]
def test_openai_compat_marks_builtin_boundary_and_tail_tool() -> None:
messages = [
{"role": "system", "content": "system"},
{"role": "assistant", "content": "assistant"},
{"role": "user", "content": "user"},
]
_, marked_tools = OpenAICompatProvider._apply_cache_control(
messages,
_openai_tools("read_file", "write_file", "mcp_fs_ls", "mcp_git_status"),
)
assert _marked_openai_tool_names(marked_tools) == ["write_file", "mcp_git_status"]
def test_anthropic_marks_builtin_boundary_and_tail_tool() -> None:
messages = [
{"role": "user", "content": "u1"},
{"role": "assistant", "content": "a1"},
{"role": "user", "content": "u2"},
]
_, _, marked_tools = AnthropicProvider._apply_cache_control(
"system",
messages,
_anthropic_tools("read_file", "write_file", "mcp_fs_ls", "mcp_git_status"),
)
assert _marked_anthropic_tool_names(marked_tools) == ["write_file", "mcp_git_status"]
def test_openai_compat_marks_only_tail_without_mcp() -> None:
messages = [
{"role": "system", "content": "system"},
{"role": "assistant", "content": "assistant"},
{"role": "user", "content": "user"},
]
_, marked_tools = OpenAICompatProvider._apply_cache_control(
messages,
_openai_tools("read_file", "write_file"),
)
assert _marked_openai_tool_names(marked_tools) == ["write_file"]
+49
View File
@@ -0,0 +1,49 @@
from __future__ import annotations
from typing import Any
from nanobot.agent.tools.base import Tool
from nanobot.agent.tools.registry import ToolRegistry
class _FakeTool(Tool):
def __init__(self, name: str):
self._name = name
@property
def name(self) -> str:
return self._name
@property
def description(self) -> str:
return f"{self._name} tool"
@property
def parameters(self) -> dict[str, Any]:
return {"type": "object", "properties": {}}
async def execute(self, **kwargs: Any) -> Any:
return kwargs
def _tool_names(definitions: list[dict[str, Any]]) -> list[str]:
names: list[str] = []
for definition in definitions:
fn = definition.get("function", {})
names.append(fn.get("name", ""))
return names
def test_get_definitions_orders_builtins_then_mcp_tools() -> None:
registry = ToolRegistry()
registry.register(_FakeTool("mcp_git_status"))
registry.register(_FakeTool("write_file"))
registry.register(_FakeTool("mcp_fs_list"))
registry.register(_FakeTool("read_file"))
assert _tool_names(registry.get_definitions()) == [
"read_file",
"write_file",
"mcp_fs_list",
"mcp_git_status",
]