From e1fcc2495374e175ff5d4c550b9da0d3bc9ddb64 Mon Sep 17 00:00:00 2001 From: raychen <815315825@qq.com> Date: Tue, 3 Nov 2026 15:53:01 +0800 Subject: [PATCH] =?UTF-8?q?feat(tools):=20=E5=A2=9E=E5=8A=A0=E6=8C=89?= =?UTF-8?q?=E9=9C=80=E5=B7=A5=E5=85=B7=E6=90=9C=E7=B4=A2=E4=B8=8E=E5=8A=A0?= =?UTF-8?q?=E8=BD=BD=E8=83=BD=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 背景: 当 Agent 配置大量工具时,完整工具定义会增加 Token 消耗、降低工具选择准确率,并影响 Prompt Cache 稳定性。 - 方案: 新增 ToolSearchToolSet,支持关键词及语义检索、Native 和 Dispatch 调用模式、Session 状态持久化、权限过滤及动态 MCP 工具加载;补充单测和中英文文档。 - 示例: 新增 Tool Search 示例,演示模型按需发现并加载天气工具,再完成多日天气查询。 --- docs/design/02-tool-search-design.md | 1040 ++++++++++++ docs/mkdocs/en/tool_search.md | 245 +++ docs/mkdocs/zh/tool_search.md | 217 +++ examples/tool_search/.env.example | 5 + examples/tool_search/README.md | 34 + examples/tool_search/agent/__init__.py | 1 + examples/tool_search/agent/agent.py | 34 + examples/tool_search/agent/config.py | 28 + examples/tool_search/agent/prompts.py | 17 + examples/tool_search/agent/tools.py | 82 + examples/tool_search/run_agent.py | 83 + mkdocs.yml | 2 + tests/tools/tool_search/test_tool_search.py | 965 ++++++++++++ .../agents/core/_tools_processor.py | 24 +- trpc_agent_sdk/tools/__init__.py | 24 + trpc_agent_sdk/tools/tool_search/__init__.py | 38 + trpc_agent_sdk/tools/tool_search/_semantic.py | 246 +++ trpc_agent_sdk/tools/tool_search/_toolset.py | 1403 +++++++++++++++++ trpc_agent_sdk/tools/tool_search/_types.py | 92 ++ 19 files changed, 4579 insertions(+), 1 deletion(-) create mode 100644 docs/design/02-tool-search-design.md create mode 100644 docs/mkdocs/en/tool_search.md create mode 100644 docs/mkdocs/zh/tool_search.md create mode 100644 examples/tool_search/.env.example create mode 100644 examples/tool_search/README.md create mode 100644 examples/tool_search/agent/__init__.py create mode 100644 examples/tool_search/agent/agent.py create mode 100644 examples/tool_search/agent/config.py create mode 100644 examples/tool_search/agent/prompts.py create mode 100644 examples/tool_search/agent/tools.py create mode 100644 examples/tool_search/run_agent.py create mode 100644 tests/tools/tool_search/test_tool_search.py create mode 100644 trpc_agent_sdk/tools/tool_search/__init__.py create mode 100644 trpc_agent_sdk/tools/tool_search/_semantic.py create mode 100644 trpc_agent_sdk/tools/tool_search/_toolset.py create mode 100644 trpc_agent_sdk/tools/tool_search/_types.py diff --git a/docs/design/02-tool-search-design.md b/docs/design/02-tool-search-design.md new file mode 100644 index 000000000..accde80bd --- /dev/null +++ b/docs/design/02-tool-search-design.md @@ -0,0 +1,1040 @@ +# trpc-agent-python Tool Search 设计 + +> **结论摘要**:Tool Search 用 `ToolSearchToolSet` 把大规模工具目录拆成“常驻工具 + 搜索入口 + 按需加载工具”。它通过 Python 框架已有的 `ToolSet.get_tools()` 和 `BaseTool.process_request()` 生命周期接入每一轮模型请求,不引入独立 Plugin 系统。工具目录支持静态 Toolbox、动态 MCP ToolSet、关键词检索和可选语义检索;发现结果写入 Session State,并按 ToolSet、Agent/Branch 和工具名隔离。Native 模式在下一轮直接暴露已加载工具,Dispatch 模式始终只暴露 `tool_search` 与 `call_tool`,用于保持模型工具面稳定。 + +## 1. 背景与目标 + +当 Agent 配置几十到几百个工具时,如果每次模型请求都携带全部工具定义,会产生以下问题: + +- 工具 Schema 占用大量输入 Token; +- 模型需要在过大的候选集中选择工具,误选概率增加; +- 不同请求使用的工具不同,模型工具面难以保持稳定; +- 动态 MCP Server 的工具列表可能持续变化; +- 多租户场景下,不同用户能使用的工具集合不同; +- Prompt Cache 容易因为工具定义变化而失效。 + +Tool Search 的核心思路是:**初始请求只提供少量常驻工具和一个 `tool_search` 入口,模型根据目录按需发现工具,后续再调用已发现的工具。** + +本设计参考 +[trpc-agent-go Tool Search Plugin](https://github.com/trpc-group/trpc-agent-go/tree/e73500ecabebfc373221b19c24a84e2158c1797f/plugin/toolsearch) +的功能,但不机械复制 Go Plugin 架构,而是使用 Python SDK 原有的 ToolSet、Tool、InvocationContext 和 Session State 能力实现。 + +### 1.1 目标 + +- **按需暴露工具**:未加载的 Deferred Tool 不进入初始模型工具列表。 +- **保持简单接入**:作为普通 ToolSet 配置到 `LlmAgent.tools`。 +- **支持大规模目录**:通过 Toolbox 命名空间组织工具。 +- **支持动态来源**:每轮刷新 MCP ToolSet,并隔离不同 MCP Server 的工具名。 +- **支持两种调用方式**:兼顾原生 Function Call 和固定工具面的 Dispatch 模式。 +- **支持会话延续**:同一 Session 的后续 Agent Loop 可以继续使用已发现工具。 +- **支持并发搜索**:并行 `tool_search` 调用不会覆盖彼此的发现状态。 +- **支持权限控制**:禁止的工具不进入目录、搜索结果和实际调用。 +- **无强制检索依赖**:默认关键词检索不引入额外包;语义检索按需注入 Embedder。 +- **失败可降级**:Embedding 服务失败时可选择降级到关键词检索。 + +### 1.2 非目标 + +- 不实现新的 Runner Plugin 系统; +- 不替代 Tool 自身的 Filter、Callback 和权限控制; +- 不负责模型 Prompt Cache 的创建、计费和命中统计; +- 不内置持久化向量数据库;需要持久化时通过 `ToolSearchVectorStore` 接口注入; +- 不允许通过 Tool Search 绕过原始工具的参数 Schema、Filter 或 Callback; +- 不保证 Native 模式的工具列表始终固定,固定工具面由 Dispatch 模式提供。 + +## 2. 方案设计 + +### 2.1 设计原则与总体架构 + +**设计原则** + +1. **使用现有扩展点**:动态工具集合由 `ToolSet.get_tools()` 管理,工具目录通过 `Tool.process_request()` 注入模型请求。 +2. **每轮重新解析**:每次模型调用前重新构建只读 Snapshot,避免请求执行期间被动态来源修改。 +3. **目录与执行分离**:Catalog 只告诉模型有哪些 Deferred Tool;工具必须经过 `tool_search` 后才可执行。 +4. **权限前置**:权限过滤发生在目录渲染和搜索候选构建之前,禁止的工具不会泄漏给模型或远程 Embedder。 +5. **Session 状态最小化**:只保存“哪个工具已经发现”,不保存 Tool 对象和 Schema。 +6. **并发写入可交换**:每个已发现工具使用独立状态键,避免并行 Event Delta 相互覆盖。 +7. **动态来源隔离命名**:MCP 工具统一使用 `mcp____`,避免跨 Server 和静态工具重名。 +8. **可配置校验策略**:默认 Fail Fast;动态组装场景可启用 Go 风格容错,跳过或合并可安全恢复的配置。 + +**总体架构** + +```text +┌──────────────────────────── LlmAgent / Agent Loop ───────────────────────────┐ +│ │ +│ ToolsProcessor │ +│ │ │ +│ ▼ │ +│ ToolSearchToolSet.get_tools(ctx) │ +│ │ │ +│ ├─ 静态 Preset / Deferred / Toolbox │ +│ ├─ 动态 MCP ToolSet ──► mcp____ │ +│ ├─ Permission Filter │ +│ ├─ Session State:已发现工具 │ +│ └─ 构建当前请求 Snapshot │ +│ │ │ +│ ├─ Native:preset + tool_search + loaded tools │ +│ └─ Dispatch:preset + tool_search + call_tool │ +│ │ +│ tool_search.process_request() │ +│ └─ Catalog ──► system instruction 或 tool description │ +│ │ +│ 模型调用 tool_search │ +│ ├─ tool_names:精确加载 │ +│ ├─ queries:关键词 / 语义检索 │ +│ └─ namespace:限定 Toolbox │ +│ │ │ +│ ▼ │ +│ InvocationContext.state:记录发现结果 │ +│ │ │ +│ └─ 下一轮 Agent Loop 生效 │ +└──────────────────────────────────────────────────────────────────────────────┘ +``` + +一次 Native 模式调用流程: + +```text +1. ToolsProcessor 调 ToolSearchToolSet.get_tools(ctx) +2. ToolSet 返回 preset tools + tool_search +3. tool_search.process_request() 注入 Deferred Tool Catalog +4. 模型调用 tool_search(tool_names=["get_forecast"]) +5. tool_search 将 get_forecast 写入 InvocationContext.state +6. 工具结果 Event 持久化 State Delta +7. Agent Loop 进入下一轮模型请求 +8. get_tools(ctx) 读取 Session State +9. ToolSet 返回 preset tools + tool_search + get_forecast +10. 模型直接调用 get_forecast +``` + +### 2.2 为什么使用 ToolSet,而不是 Plugin 或 Filter + +Go 实现通过一个 Plugin 注册 `BeforeModel` 和 `BeforeTool`: + +```text +BeforeModel:刷新目录、注入 Catalog、增加已加载工具 +BeforeTool:拦截尚未加载的 Deferred Tool +``` + +Python 同样具备 Model/Tool Filter 的 before/after 生命周期,但动态控制本轮模型可见工具本身是 ToolSet 的职责: + +```text +ToolSet.get_tools() 动态决定本轮有哪些工具 +Tool.process_request() 修改模型请求、注入 Catalog +Tool Filter 已解析工具执行前后的校验和处理 +``` + +未加载的 Native Tool 不会由 `get_tools()` 返回,因此不会进入模型请求的 +Declaration。模型仍然异常调用该名称时,ToolsProcessor 会询问 ToolSet 的 +`get_unresolved_tool()` 回退钩子;Tool Search 返回一个不带 Declaration 的 +Guard Tool,给出“先搜索再重试”的提示。Dispatch 模式则由始终可见的 +`call_tool` 显式检查加载状态。 + +因此本方案不增加 Tool Search 专用 Filter: + +- ToolSet 管理工具可见性; +- 原始 Tool Filter/Callback 继续管理已加载工具的执行; +- Dispatch 的 `call_tool` 管理目标工具的加载状态和权限; +- Native 模式通过 ToolSet 回退钩子提示尚未加载的异常调用。 + +### 2.3 公共 API + +核心入口: + +```python +ToolSearchToolSet( + preset_tools=None, + deferred_tools=None, + toolboxes=None, + mcp_toolboxes=None, + options=None, + semantic_index=None, + name="tool_search_tools", +) +``` + +各参数语义: + +| 参数 | 作用 | +|---|---| +| `preset_tools` | 始终暴露给模型;不参与关键词或语义搜索 | +| `deferred_tools` | 无业务命名空间的延迟工具 | +| `toolboxes` | 按静态命名空间组织的延迟工具 | +| `mcp_toolboxes` | 每轮从动态 ToolSet 获取工具的 MCP 命名空间 | +| `options` | 搜索结果数、权限、Catalog 位置、调用模式和降级策略 | +| `semantic_index` | 可选语义索引;未配置时使用关键词检索 | +| `name` | ToolSet 名称,同时参与 Session State 隔离 | + +`Toolbox`: + +```python +Toolbox( + name="weather", + description="天气查询和预报", + tools=[get_weather, get_forecast], +) +``` + +`MCPToolbox`: + +```python +MCPToolbox( + server_name="github", + description="GitHub 仓库操作", + toolset=github_mcp_toolset, +) +``` + +`ToolSearchOptions`: + +```python +ToolSearchOptions( + max_results=5, + permission_filter=None, + catalog_in_description=False, + invocation_mode=InvocationMode.NATIVE, + embedding_fail_open=False, + state_key_prefix="tool_search", + strict_validation=True, +) +``` + +公开导出同时提供带 Tool Search 前缀的别名,避免与其他模块类型冲突: + +```python +from trpc_agent_sdk.tools import ( + SemanticToolIndex, + ToolSearchInvocationMode, + ToolSearchMCPToolbox, + ToolSearchOptions, + ToolSearchToolbox, + ToolSearchToolSet, +) +``` + +### 2.4 Snapshot 与请求生命周期 + +`ToolSearchToolSet` 不长期维护一个可被所有请求修改的全局工具索引,而是在每次 `get_tools(ctx)` 时构建 `_Snapshot`: + +```python +@dataclass(frozen=True) +class _Snapshot: + entries: dict[str, _ToolEntry] + canonical_names: dict[str, str] + deferred_names: frozenset[str] + namespace_tools: dict[str, frozenset[str]] + namespace_descriptions: dict[str, str] + namespace_order: tuple[str, ...] +``` + +Snapshot 包含: + +- 工具名称到 Tool、Declaration、Namespace 的映射; +- 小写名称到标准名称的映射,用于大小写不敏感的精确加载; +- Deferred Tool 集合; +- 每个 Namespace 的成员集合; +- Toolbox 描述和渲染顺序。 + +Snapshot 只服务当前模型请求和由该请求产生的 `tool_search` / `call_tool`。动态 MCP 下一轮变化时会创建新的 Snapshot,不修改已经进入执行阶段的 Snapshot。 + +这保证: + +- 当前模型看到的目录与当前 `tool_search` 使用的搜索集合一致; +- 当前 `call_tool` 使用的 Schema 与搜索结果一致; +- 动态来源变化不会在单次模型请求中造成目录和执行对象不一致。 + +### 2.5 Preset、Deferred 与 Toolbox + +工具分为三类: + +```text +Preset Tool + ├─ 每轮直接暴露 + ├─ 可以通过精确 tool_names 返回 + └─ 不进入 query / embedding 候选集 + +Default Deferred Tool + ├─ 初始不暴露 + ├─ 没有业务 Namespace + └─ 可通过 tool_names 或全局 queries 加载 + +Toolbox Deferred Tool + ├─ 初始不暴露 + ├─ 属于一个业务 Namespace + └─ 可按 Namespace 限定搜索范围 +``` + +静态工具名在同一个 `ToolSearchToolSet` 内必须全局唯一。Namespace 只是搜索范围,不是静态工具名的一部分: + +```text +正确:tool_names=["get_forecast"], namespace="weather" +不推荐:tool_names=["weather.get_forecast"] +``` + +考虑到模型可能从分组目录错误推断出 `namespace.tool`,实现提供有限容错:仅当 Namespace 和工具归属都匹配时,`weather.get_forecast` 才会被解析为 `get_forecast`;错误的 `network.get_forecast` 不会匹配。 + +### 2.6 Catalog 注入 + +Catalog 默认通过占位符注入系统指令: + +```python +TOOL_SEARCH_PLACEHOLDER = "{deferred_tools_section}" +``` + +处理顺序: + +```text +1. system instruction 包含占位符 + └─ 用 Catalog 替换第一个占位符 + +2. 没有占位符,但 Catalog 非空 + └─ 追加到 system instruction + +3. catalog_in_description=True + ├─ Catalog 写入 tool_search description + └─ 从 system instruction 移除字面占位符 +``` + +Toolbox Catalog 示例: + +```text + +Deferred tools are not callable until loaded with tool_search; +after loading, call it directly. +- namespace=weather (天气查询和预报); + exact tool_names=["get_forecast", "get_weather"] +Namespace filters the search scope and must not be prepended to a tool name. + +``` + +如果只有无 Namespace 的 Deferred Tool,则使用更紧凑的兼容格式: + +```text + +read_file +write_file + +``` + +权限过滤在 Catalog 渲染前完成,禁止的工具不会出现在目录中。 + +### 2.7 搜索路由 + +`tool_search` 接收四个字段: + +```text +tool_names 精确工具名列表 +queries 关键词或语义查询列表 +namespace Toolbox 搜索范围 +max_results 本次最多加载的结果数 +``` + +路由优先级: + +```text +tool_names 非空 + └─ 大小写不敏感的精确加载,跨 Namespace + +namespace 非空且 queries 为空 + └─ 按名称列出该 Namespace 的工具 + +queries 非空且配置 SemanticToolIndex + └─ 语义检索 + +queries 非空且未配置 SemanticToolIndex + └─ 关键词检索 + +其他 + └─ 返回空结果 +``` + +未知 Namespace 返回结构化结果和有效 Namespace Catalog,便于模型重试: + +```json +{ + "status": "unknown_namespace", + "message": "Namespace 'missing' is not in the tool catalog.", + "namespace": "missing", + "catalog": [ + { + "name": "weather", + "description": "天气查询和预报" + } + ] +} +``` + +模型请求的 `max_results` 上限为 10。超过有效上限的匹配不会直接丢弃,而是以名称列表放入 `additional_candidates`,模型可以随后通过 `tool_names` 精确加载。 + +结果中的每个重复匹配会标记 `already_loaded=true`。顶层 `status` 同时区分 +“全部首次加载”“部分已经加载”和“全部已经加载”,并按 Native/Dispatch +模式分别提示直接调用或通过 `call_tool` 调用,减少模型重复搜索。 + +### 2.8 关键词检索 + +关键词检索不依赖外部组件,索引内容包括: + +- Tool 名称; +- CamelCase / 下划线拆分后的名称片段; +- Tool Description; +- Toolbox Description,用于 Namespace Bias。 + +名称解析示例: + +```text +getWeatherForecast → get / weather / forecast +get_weather → get / weather +mcp__github__search → mcp / github / search +``` + +Query 使用 Unicode 感知的分隔规则;空格、逗号、中英文标点都会拆分为独立 Term。 + +匹配分数: + +| 条件 | 分数 | +|---|---:| +| 整个 Query 精确等于工具名 | 1000 | +| 复合 Term 命中工具名 | 10 | +| Term 精确命中名称片段 | 10 | +| Term 命中 Description | 3 | +| 仅命中完整名称文本 | 3 | + +`+term` 表示必须命中: + +```python +queries=["+read content"] +``` + +只有同时满足 `read` 的工具才进入候选,然后再使用全部 Term 排序。包含 `_` 或 `-` 的 Required Term 会继续拆成原子 Term。 + +多个 Query 独立评分,同一工具保留最佳分数;最终排序规则: + +```text +分数降序 + → 名称片段数量升序 + → 工具名字典序 +``` + +未指定 Namespace 时会搜索所有 Deferred Tool,并根据 Toolbox Description 的命中情况增加 Namespace Bias,使业务域描述更相关的候选获得额外分数。 + +### 2.9 语义检索 + +语义检索通过注入 `SemanticToolIndex` 开启: + +```python +tool_search = ToolSearchToolSet( + deferred_tools=tools, + semantic_index=SemanticToolIndex( + embedder, + vector_store=my_vector_store, + ), +) +``` + +Embedder 支持以下任意一组接口: + +```text +异步:aembed_documents / aembed_query +同步:embed_documents / embed_query +``` + +同步方法通过 `asyncio.to_thread()` 执行,避免阻塞 Agent Event Loop。Embedder +可以直接返回向量以保持兼容,也可以返回 `(向量, usage)`: + +```python +return vectors, { + "prompt_tokens": 120, + "total_tokens": 120, +} +``` + +每个工具转换成如下文档: + +```text +Tool: get_forecast +Description: Get a multi-day weather forecast +Declaration: <完整 FunctionDeclaration JSON> +``` + +该文本计算 SHA-256 Fingerprint。`SemanticToolIndex` 保存: + +```text +tool_name → fingerprint +tool_name → vector(由 ToolSearchVectorStore 管理) +``` + +Declaration 未变化时复用向量;名称、描述或 Schema 变化后生成新的 Fingerprint 并重新 Embedding。 + +VectorStore 接口包含 `upsert(name, vector)`、`delete(names)` 和 +`search(query_vector, candidate_names, limit)`,支持同步或异步实现。默认使用 +`InMemoryToolSearchVectorStore`;应用可以注入远程或持久化实现。 + +检索过程: + +```text +权限过滤后的候选工具 + → 补齐缺失向量 + → Embedding Query + → 计算余弦相似度 + → 每个 Query 独立排序 + → 按每个工具的最佳 Rank 合并 + → max_results 切分 +``` + +禁止的工具在构建语义文档前已经移除,因此名称、描述和 Schema 不会发送给远程 Embedding 服务。 + +`embedding_fail_open=True` 时,Embedding 或向量计算失败会记录 Warning 并降级到关键词检索;默认 `False`,失败直接返回给调用方。 + +每次模型请求开始时,`tool_search.process_request()` 都会创建新的 Usage +累计周期;当前轮次内累加 Embedder 返回的 `prompt_tokens` / +`total_tokens`。应用可通过 +`tool_search.get_embedding_usage(ctx, reset=False)` 获取当前轮快照,或传入 +`reset=True` 提前清零。 + +MCP 工具声明变化或下线时主动清理旧向量,并在并发刷新期间校验声明指纹,避免过期 Snapshot 重新写入旧向量。 + +### 2.10 Native 与 Dispatch 模式 + +**Native 模式** + +```python +ToolSearchOptions( + invocation_mode=InvocationMode.NATIVE, +) +``` + +工具面变化: + +```text +初始:preset + tool_search +发现后:preset + tool_search + loaded tool A +继续发现:preset + tool_search + loaded tool A + loaded tool B +``` + +Native 模式使用 `InvocationContext.state` 记录已经发现的工具名称。模型每次 +请求前,`get_tools(ctx)` 都会读取这些名称,在最新 Snapshot 中重新查找工具、 +检查权限,并将仍然有效的完整 Tool Schema 加入本轮 `tools` 列表。State +本身不保存 Tool 对象或 Schema。 + +Session 历史只保存 Function Call 的名称、实参和执行结果,不保存当轮 +`LlmRequest.tools` 中的完整工具声明。因此,即使历史中出现过某个工具, +框架仍需要通过 State 确定本轮应重新声明哪些工具。可以将它理解为: + +```text +Session Events:帮助模型理解“之前调用过什么” +Session State:帮助框架确定“本轮需要声明哪些工具” +``` + +优点: + +- 模型使用标准 Function Call 直接调用真实工具; +- 工具参数 Schema 由框架正常提供; +- 原始 Tool Filter、Callback、Streaming 和 Long Running 行为保持自然。 + +代价: + +- 每次发现新工具后模型工具列表会变化; +- 大量已加载工具仍会逐步增加 Token; +- Prompt Cache 可能因为工具面变化而失效。 + +如果模型在 Native 模式下直接调用尚未发现的 Deferred Tool, +`ToolsProcessor` 会通过 ToolSet 的 `get_unresolved_tool()` 回退钩子取得一个 +不带 Declaration 的 Guard Tool,并返回“先使用 `tool_search` 加载后重试”的 +专用提示。对于已经从 MCP 最新非空列表下线、但模型仍持有旧名称的工具, +框架也会根据 `mcp____` 和已注册 Namespace 返回相同提示。 +Guard Tool 不会进入模型请求的工具声明。 + +**Dispatch 模式** + +```python +ToolSearchOptions( + invocation_mode=InvocationMode.DISPATCH, +) +``` + +工具面保持为: + +```text +preset + tool_search + call_tool +``` + +`tool_search` 结果额外返回标准化后的 `input_schema`: + +```json +{ + "name": "get_forecast", + "description": "Get a multi-day weather forecast", + "input_schema": { + "type": "object", + "properties": { + "city": { + "type": "string" + } + } + } +} +``` + +模型随后调用: + +```json +{ + "tool_name": "get_forecast", + "params": { + "city": "Shenzhen", + "days": 3 + } +} +``` + +`call_tool` 依次检查: + +```text +tool_name 是否为空 + → params 是否为对象 + → 工具是否存在 + → 当前调用方是否有权限 + → 工具是否已通过 tool_search 加载 + → 调用原始 Tool +``` + +Dispatch 模式不会使用 State 向模型注入真实工具 Schema。真实工具的 +`input_schema` 由 `tool_search` 作为 Function Response 返回;模型读取该 +响应后,通过固定声明的 `call_tool(tool_name, params)` 发起调用。 + +该模式下,State 只保存“工具已通过搜索加载”的名称,作为 `call_tool` 的 +搜索前置门禁,确保模型不能仅靠猜测工具名称绕过 Namespace、Top-K 和搜索 +结果直接调用。它不等同于权限:调用时仍会根据最新 Snapshot 单独执行权限 +检查。 + +```text +tool_search 返回 name + description + input_schema + ↓ +State 记录 discovered_tools: = true + ↓ +call_tool 检查权限和 discovered 状态 + ↓ +执行最新 Snapshot 中的真实 Tool +``` + +如果产品希望 `call_tool` 支持按准确名称直接分发,也可以取消该 State 门禁, +只保留工具存在性和权限检查;但这会将 Dispatch 从“搜索后调用”改为“可直接 +调用的统一分发器”,不再保证模型调用前看过目标工具的最新 Schema。 + +Dispatch 的优点是 Deferred Tool 工具面固定,更适合大目录和重视 Prompt Cache 稳定性的场景。 + +Progress Streaming Tool 在 Dispatch 中会被完整消费,并按顺序聚合为 `{"chunks": [...]}` 单次返回;Native 模式仍走框架原生 Progress Streaming 路径。 + +### 2.11 Session State 与并发 + +发现状态写入 `InvocationContext.state`,状态键格式: + +```text +:::discovered_tools: +``` + +示例: + +```text +tool_search:tool_search_tools:weather_agent:discovered_tools:get_forecast = true +``` + +隔离维度: + +| 维度 | 作用 | +|---|---| +| `state_key_prefix` | 避免与其他框架状态冲突 | +| `toolset_name` | 同一 Agent 中多个 Tool Search 实例相互隔离 | +| `branch_or_agent` | 多 Agent / Graph Branch 之间相互隔离 | +| `tool_name` | 每个工具独立写入,支持并发合并 | + +使用“每个工具一个状态键”,而不是一个 JSON 工具列表,是为了让并行搜索的 Event Delta 可交换: + +```text +并行搜索 A:...:tool_a = true +并行搜索 B:...:tool_b = true + +无论 Event 以什么顺序持久化,最终两个键都存在。 +``` + +状态通过正常工具结果 Event 的 State Delta 持久化,不在 Tool Search 内直接调用 Session Service。这样遵循框架统一的 Event/State 提交流程。 + +状态只保存工具名称,不保存 Tool 对象。下一轮 `get_tools()` 会在最新 Snapshot 中重新解析名称: + +- 静态工具仍存在且有权限:继续暴露; +- MCP 工具已从最新非空列表删除:不再暴露; +- 权限被收回:不再暴露; +- 名称不存在:忽略旧状态。 + +### 2.12 权限模型 + +权限回调签名: + +```python +ToolPermissionFilter = Callable[ + [InvocationContext, Sequence[str]], + dict[str, bool], +] +``` + +一次请求对全部 Deferred Tool 批量计算权限: + +```python +def permissions(ctx, names): + return { + name: can_use(ctx.user_id, name) + for name in names + } +``` + +没有明确返回 `True` 的工具按禁止处理。权限作用于: + +```text +Catalog 渲染 + → 精确 tool_names + → 关键词候选 + → 语义检索候选 + → Native 已加载工具暴露 + → Dispatch call_tool 实际执行 +``` + +Preset Tool 不由 Tool Search Permission Filter 管理;如果 Preset Tool 需要权限控制,应使用原始 Tool Filter、Agent Callback 或应用自己的授权层。 + +### 2.13 动态 MCP ToolSet + +MCP Toolbox 每次 `get_tools(ctx)` 都调用内部动态 ToolSet: + +```text +ToolSearchToolSet.get_tools() + → MCPToolbox.toolset.get_tools(ctx) + → 获取当前 MCP 工具 + → 重命名 + → 写入当前 Snapshot +``` + +重命名规则: + +```text +原名称:read_file +Server:github +新名称:mcp__github__read_file +``` + +原因是模型 Function Call 只有工具名,没有独立的 MCP Server 字段。多个 Server 都可能暴露 `read_file` 或 `search`,因此必须形成全局唯一名称。 + +重命名代理保留: + +- 原始 Declaration 和参数 Schema; +- Tool Filter; +- `is_streaming` / `is_progress_streaming`; +- `api_variant`; +- `skip_summarization`; +- Long Running 类型标记; +- 实际调用委托给原始 Tool。 + +每个 MCP Namespace 使用独立 `asyncio.Lock` 串行化“获取列表 + 更新缓存”,避免并发请求以相反顺序提交快照。 + +每次非空列表还会计算工具声明指纹: + +- 新增工具在首次语义搜索时生成向量; +- 声明发生变化时清理旧向量,下一次搜索重新生成; +- 从非空列表中删除的工具同时从当前 Snapshot 和语义索引移除; +- Embedding 完成前再次校验最新指纹,防止并发旧 Snapshot 复活已下线或旧版本工具。 + +容错策略: + +```text +动态列表非空 + └─ 替换最后一次成功快照 + +动态列表为空 + └─ 保留最后一次非空快照 + +动态列表抛异常且已有快照 + └─ 记录 Warning,使用最后快照 + +动态列表抛异常且没有快照 + └─ 向上抛出异常 +``` + +空列表被视为临时不可用,而不是“Server 主动删除全部工具”。这是可用性优先的约定。 + +`ToolSearchToolSet.close()` 会关闭所有不同的动态 ToolSet,同一个实例被多个 MCP Toolbox 引用时只关闭一次。 + +### 2.14 配置校验与生命周期 + +以下名称是保留名称: + +```text +tool_search +call_tool +``` + +默认 `strict_validation=True`,初始化时拒绝: + +- Preset / Deferred / Toolbox 静态工具重名; +- 工具使用保留名称; +- 空 Toolbox 名称; +- 重复 Toolbox Namespace; +- 空 MCP Server Name; +- MCP Namespace 与静态 Toolbox Namespace 冲突; +- `max_results <= 0`。 + +设置 `strict_validation=False` 后采用 Go 风格容错: + +- 空 Toolbox / MCP Server Name、无效工具和空 MCP ToolSet:记录 Warning 并跳过; +- 重复静态 Namespace:按注册顺序合并,首个非空 Description 生效; +- 同一工具注册到多个 Namespace:保留第一次归属并记录 Warning; +- MCP Namespace 与已有 Namespace 冲突:保留已有配置,跳过后者; +- 动态 MCP 列表包含相同声明的同名工具:保留第一项。 + +以下问题即使在容错模式下仍然报错,因为无法安全决定最终行为: + +- 工具使用 `tool_search` / `call_tool` 保留名称; +- 同名工具或动态 MCP 工具的 Declaration 不一致; +- `max_results <= 0`; +- 动态工具与现有全局工具发生无法消解的名称冲突。 + +没有 Function Declaration 的动态工具不会进入 Snapshot,并记录 Warning,因为模型无法获得其调用 Schema。 + +工具与 ToolSet 生命周期: + +```text +ToolSearchToolSet 由 Runner/Agent 生命周期管理 + ├─ 静态 Tool 不由 Tool Search 单独关闭 + └─ MCPToolbox.toolset 在 ToolSearchToolSet.close() 中关闭 +``` + +## 3. 使用方式 + +### 3.1 Native 模式 + +```python +from trpc_agent_sdk.agents import LlmAgent +from trpc_agent_sdk.tools import FunctionTool +from trpc_agent_sdk.tools import ToolSearchOptions +from trpc_agent_sdk.tools import ToolSearchToolbox +from trpc_agent_sdk.tools import ToolSearchToolSet +from trpc_agent_sdk.tools import TOOL_SEARCH_PLACEHOLDER + +tool_search = ToolSearchToolSet( + preset_tools=[ + FunctionTool(calculator), + ], + toolboxes=[ + ToolSearchToolbox( + name="weather", + description="天气查询和预报", + tools=[ + FunctionTool(get_weather), + FunctionTool(get_forecast), + ], + ), + ToolSearchToolbox( + name="billing", + description="账单、支付和退款", + tools=[ + FunctionTool(create_invoice), + FunctionTool(refund_payment), + ], + ), + ], + options=ToolSearchOptions(), +) + +agent = LlmAgent( + name="assistant", + model=model, + instruction=( + "调用延迟工具前先使用 tool_search。" + "namespace 只用于限定搜索范围,不是工具名前缀。\n\n" + f"{TOOL_SEARCH_PLACEHOLDER}" + ), + tools=[tool_search], +) +``` + +用户请求: + +```text +查询深圳未来三天的天气。 +``` + +模型调用流程: + +```json +{ + "name": "tool_search", + "arguments": { + "tool_names": ["get_forecast"] + } +} +``` + +下一轮: + +```json +{ + "name": "get_forecast", + "arguments": { + "city": "深圳", + "days": 3 + } +} +``` + +### 3.2 Dispatch 模式 + +```python +from trpc_agent_sdk.tools import ToolSearchInvocationMode + +tool_search = ToolSearchToolSet( + toolboxes=toolboxes, + options=ToolSearchOptions( + invocation_mode=ToolSearchInvocationMode.DISPATCH, + ), +) +``` + +模型工具面始终保持: + +```text +tool_search +call_tool +``` + +搜索后调用: + +```json +{ + "name": "call_tool", + "arguments": { + "tool_name": "get_forecast", + "params": { + "city": "深圳", + "days": 3 + } + } +} +``` + +### 3.3 语义检索与失败降级 + +```python +tool_search = ToolSearchToolSet( + deferred_tools=tools, + semantic_index=SemanticToolIndex(embedder), + options=ToolSearchOptions( + embedding_fail_open=True, + ), +) +``` + +如果用户查询: + +```text +明天出门是否需要带伞? +``` + +语义检索可以将其匹配到 `get_forecast`。Embedding 服务不可用时,如果 `embedding_fail_open=True`,框架使用原始 Query 降级到关键词搜索;降级后可用性仍在,但召回准确率可能下降。 + +## 4. 测试设计 + +测试文件: + +```text +tests/tools/tool_search/test_tool_search.py +``` + +覆盖范围: + +| 分类 | 断言 | +|---|---| +| 关键词搜索 | Namespace、Required Term、精确名称、大小写、结果溢出、加载状态提示 | +| 名称容错 | 正确 `namespace.tool` 可解析,错误 Namespace 不匹配 | +| Agent Loop | ToolsProcessor 下一轮暴露已加载 Native Tool | +| Native 回退 | 未加载及已下线 MCP 工具返回先搜索再重试的专用提示 | +| Catalog | 占位符替换、准确名称提示、Namespace 说明 | +| 配置容错 | Namespace 合并、首次归属、MCP 冲突跳过、Declaration 冲突报错 | +| 权限 | 禁止工具不出现在搜索和已加载工具中 | +| Preset | 始终可见,不进入 Query 候选 | +| Dispatch | 固定工具面、Schema、未加载拦截、真实调用 | +| Streaming | Dispatch 消费 Progress Stream 并按顺序返回完整 `chunks` | +| MCP | 重命名、空列表缓存、异常缓存、关闭、Long Running 标记 | +| 语义搜索 | 相似度排序、Embedding 失败降级、自定义 VectorStore、Usage 累计 | +| 并发状态 | 并行搜索结果合并 | +| 多 Agent | Branch 状态隔离 | + +相关回归测试还应包含: + +```text +tests/tools/test_tool_adapter.py +tests/tools/test_default_toolset.py +tests/tools/test_base_tool.py +tests/tools/test_streaming_progress_tool.py +tests/agents/core/test_tools_processor.py +``` + +验证命令: + +```bash +python -m yapf -i \ + trpc_agent_sdk/tools/tool_search/*.py \ + tests/tools/tool_search/test_tool_search.py \ + examples/tool_search/**/*.py + +python -m pytest \ + tests/tools/tool_search/test_tool_search.py \ + tests/tools/test_tool_adapter.py \ + tests/tools/test_default_toolset.py \ + tests/tools/test_base_tool.py \ + tests/tools/test_streaming_progress_tool.py \ + tests/agents/core/test_tools_processor.py -q + +python -m flake8 \ + trpc_agent_sdk/tools/tool_search \ + tests/tools/tool_search/test_tool_search.py \ + examples/tool_search +``` + +## 5. 与 Go Plugin 的对应关系 + +Python 和 Go 的用户可见核心行为保持一致,但接入结构不同: + +| Go Plugin | Python 实现 | +|---|---| +| `BeforeModel` 刷新目录 | `ToolSearchToolSet.get_tools()` 构建 Snapshot | +| `BeforeModel` 注入 Catalog | `_ToolSearchTool.process_request()` | +| `BeforeModel` 添加已加载工具 | Native `get_tools()` 返回 Session 中已发现工具 | +| `BeforeTool` 拦截未加载工具 | Native 使用 ToolSet 回退 Guard;Dispatch 在 `call_tool` 中检查 | +| Plugin 内可变全局索引 | 每次模型请求构建只读 Snapshot | +| 一个 JSON 列表保存发现状态 | 每个工具独立 State Key | +| Session 级共享 | ToolSet + Agent/Branch 级隔离 | +| 框架 VectorStore | `ToolSearchVectorStore`,默认内存实现并支持注入 | +| Embedding Usage 统计 | 从 Embedder 返回值累计,可按 InvocationContext 读取 | +| Stream 聚合为 `chunks` | Dispatch 同样按顺序返回完整 `chunks` | + +Python 使用 ToolSet 而不是 Plugin 的原因: + +- Python 已有标准动态 ToolSet 生命周期; +- ToolsProcessor 每轮都会重新解析 ToolSet; +- BaseTool 的 `process_request()` 可以修改模型请求; +- Tool Filter 只在工具已解析并进入 `run_async()` 后执行,不能代替动态工具可见性; +- ToolSet 能同时解决模型声明和后续工具执行解析。 + +因此这是框架风格上的等价实现,不是对 Go Plugin 类型结构的直接翻译。 + +## 6. 已知边界与后续演进 + +当前已知边界: + +1. **Native Prompt Cache 变化**:加载新工具后 Function Declaration 集合变化;需要固定工具面时应使用 Dispatch。 +2. **Embedding Usage 独立统计**:Usage 可通过 Tool Search API 读取,但不会直接并入模型响应的 `usage_metadata`。 +3. **MCP 空列表语义**:空列表按临时故障处理,不能表达“Server 主动清空全部工具”。 + +建议后续按优先级演进: + +```text +P2 为动态 MCP 增加显式“确认空目录”语义 +P3 评估将 Embedding Usage 合并到统一 Event Usage 的方案 +``` + +不建议为形式对齐 Go 而新增独立 Plugin 系统;后续能力应继续围绕 ToolSet、Tool、Filter 和 InvocationContext 的现有边界演进。 diff --git a/docs/mkdocs/en/tool_search.md b/docs/mkdocs/en/tool_search.md new file mode 100644 index 000000000..db7ca2da1 --- /dev/null +++ b/docs/mkdocs/en/tool_search.md @@ -0,0 +1,245 @@ +# Tool Search + +When an Agent has a large tool catalog, including every tool definition in each +model request increases input tokens, makes tool selection harder, and can +reduce prompt-cache hits. `ToolSearchToolSet` initially exposes only preset +tools and `tool_search`, then lets the model discover tools on demand. + +## Core concepts + +- **Tool**: A function the model can call, such as `get_weather`. Use + `FunctionTool(get_weather)` to wrap a regular Python function. +- **ToolSearchToolSet**: A ToolSet that manages the tool catalog, performs + searches, and controls when tools become available to the model. Add it to + `LlmAgent.tools` to enable Tool Search. +- **Preset Tool**: Configured through `preset_tools`. Its complete schema is + always sent directly to the model, so no search is required. This is useful + for a small number of frequently used tools, such as a calculator or current + time tool. `preset_tools` is optional; an always-visible tool may also be + added directly to `LlmAgent.tools`. +- **Deferred Tool**: Configured through `deferred_tools`. It is initially + searchable only through the catalog, and its complete schema is not sent to + the model immediately. "Deferred" means deferred exposure to the model, not + delayed execution or a background task. +- **Toolbox**: A named group of static Deferred Tools configured through + `toolboxes`. Toolboxes organize tools by business domain, such as `weather` + or `filesystem`; their tools remain deferred. +- **Namespace**: The Toolbox name used to limit the search scope and reduce + ambiguity between similar tools in different domains. It is not part of a + static tool's callable name. +- **Catalog**: A lightweight index sent to the model. It primarily contains + namespaces, tool names, and descriptions, rather than complete tool schemas. +- **Discovery/loading**: After `tool_search` finds a Deferred Tool, the + framework records it as discovered. Native mode sends its complete schema in + the next model request; Dispatch mode returns the schema in the search result + and permits `call_tool` to invoke it. +- **Session State**: Session-persistent state containing discovered tool names, + not Tool objects or schemas. It lets later turns on the same Agent branch + reuse tools discovered earlier. + +Both `deferred_tools` and `toolboxes` configure static deferred tools. The only +difference is whether the tools are grouped: + +```python +tool_search = ToolSearchToolSet( + # Ungrouped tools use the default empty namespace. + deferred_tools=[ + FunctionTool(search_document), + ], + # Grouped by domain; search with namespace="weather". + toolboxes=[ + ToolSearchToolbox( + name="weather", + description="weather observations and forecasts", + tools=[ + FunctionTool(get_weather), + FunctionTool(get_forecast), + ], + ), + ], +) +``` + +Use `deferred_tools` for a small catalog that does not need domain-scoped +search. Use `toolboxes` when the catalog is large, spans multiple domains, or +contains similar tools that benefit from namespace routing. + +## Basic usage + +```python +from trpc_agent_sdk.agents import LlmAgent +from trpc_agent_sdk.tools import FunctionTool +from trpc_agent_sdk.tools import ToolSearchToolbox +from trpc_agent_sdk.tools import ToolSearchToolSet +from trpc_agent_sdk.tools.tool_search import TOOL_SEARCH_PLACEHOLDER + +tool_search = ToolSearchToolSet( + preset_tools=[FunctionTool(calculator)], + toolboxes=[ + ToolSearchToolbox( + name="weather", + description="weather observations and forecasts", + tools=[ + FunctionTool(get_weather), + FunctionTool(get_forecast), + ], + ), + ], +) + +agent = LlmAgent( + name="assistant", + model=model, + instruction=( + "Use tool_search before calling a deferred tool.\n\n" + f"{TOOL_SEARCH_PLACEHOLDER}" + ), + tools=[tool_search], +) +``` + +`preset_tools` are always sent directly to the model. Tools in +`deferred_tools` and `toolboxes` become callable only after `tool_search` +discovers them. Discovered names are stored in Session State and remain +available to later turns on the same Agent branch. + +## Invocation modes + +`ToolSearchOptions.invocation_mode` supports two modes: + +- `InvocationMode.NATIVE` (default) adds discovered tool definitions to the + next model request. The model invokes their original names. +- `InvocationMode.DISPATCH` invokes every discovered tool through + `call_tool(tool_name, params)`. This keeps the request's tool definitions + stable for large catalogs and prompt-cache-sensitive workloads. + +Dispatch mode rejects calls to tools that were not loaded through search. +Permission filtering applies consistently to the catalog, search results, and +tool execution. In Native mode, a direct call to an unloaded tool receives +specific guidance to load it with `tool_search` and retry; the declaration-free +guard itself is never advertised to the model. The same guidance is returned +for a stale MCP tool name that has disappeared from the latest listing. + +## Search and catalog injection + +The dependency-free default search matches names, descriptions, and namespaces. +Prefix a keyword with `+` to require it, for example +`["+weather", "forecast"]`. + +The model may pass the following arguments to `tool_search`: + +- `queries`: A list of keywords or natural-language queries, such as + `["weather forecast"]`. +- `tool_names`: Exact tool names to load. Names usually come from the catalog + or `additional_candidates` returned by an earlier search. +- `namespace`: An optional Toolbox namespace that limits the search to one + business domain. +- `max_results`: The maximum matches loaded by this request, capped at 10. + Remaining matches are returned by name in `additional_candidates` so the + model can load one precisely later. + +Use `queries` when the exact tool name is unknown, and `tool_names` when it is +already known. For example: + +```json +{ + "queries": ["future weather forecast"], + "namespace": "weather", + "max_results": 3 +} +``` + +The namespace only limits the search scope; it is not part of a static tool's +name. Copy the exact name shown in the catalog into `tool_names`: use +`"get_forecast"`, not `"weather.get_forecast"`. The latter form is accepted as +a compatibility fallback only when its namespace and tool name both match. + +By default, the catalog replaces `{deferred_tools_section}` in the system +instruction. If the placeholder is absent, the catalog is appended. Set +`catalog_in_description=True` to place it in the `tool_search` description. + +For semantic search, pass `SemanticToolIndex(embedder)` to +`ToolSearchToolSet`. The embedder may implement asynchronous `aembed_query` / +`aembed_documents`, or synchronous `embed_query` / `embed_documents`. With +`embedding_fail_open=True`, embedding failures fall back to keyword search. + +`SemanticToolIndex(embedder, vector_store=...)` accepts a custom +`ToolSearchVectorStore`; the default is `InMemoryToolSearchVectorStore`. A +custom store implements `upsert`, `delete`, and `search`, using either +synchronous or asynchronous methods. + +Embedders may continue returning vectors directly. To report token usage, they +may instead return `(vectors, usage)`, where usage contains +`prompt_tokens` / `total_tokens` or `prompt_token_count` / +`total_token_count`. Each model request starts a new usage period. Read the +current turn with `tool_search.get_embedding_usage(ctx)` and pass `reset=True` +to clear it early. + +## Main options + +- `max_results`: Default number of search results. +- `permission_filter`: Returns whether each deferred tool is visible and + callable for the current `InvocationContext`. Use it for tenant-, user-, or + Agent-level authorization. +- `catalog_in_description`: Places the Catalog in the `tool_search` + description. By default, it is placed in the system instruction. +- `invocation_mode`: Selects Native or Dispatch invocation. +- `embedding_fail_open`: Falls back to keyword search when semantic retrieval + fails. +- `state_key_prefix`: Prefix for discovered-tool entries in Session State, + preventing collisions with other framework state. +- `strict_validation`: Enables strict checks for duplicate namespaces, + duplicate tools, and invalid configuration. + +## Configuration tolerance + +`ToolSearchOptions(strict_validation=True)` is the default: empty namespaces, +duplicate namespaces, and duplicate tools fail during initialization. For +large catalogs assembled dynamically, enable Go-style recovery with: + +```python +ToolSearchOptions(strict_validation=False) +``` + +Tolerant mode logs warnings, merges duplicate static namespaces, keeps a +tool's first namespace owner, and skips empty toolboxes, invalid tools, or +colliding MCP namespaces. Reserved-name conflicts, different declarations for +the same tool name, and an invalid `max_results` still fail because they cannot +be recovered safely. + +## Dynamic MCP tools + +Wrap a dynamic ToolSet with `ToolSearchMCPToolbox`: + +```python +from trpc_agent_sdk.tools import ToolSearchMCPToolbox + +tool_search = ToolSearchToolSet( + mcp_toolboxes=[ + ToolSearchMCPToolbox( + server_name="github", + description="GitHub MCP tools", + toolset=mcp_toolset, + ), + ], +) +``` + +`ToolSearchMCPToolbox` differs from a static `ToolSearchToolbox`: a static +Toolbox owns a fixed list of tools, while an MCP Toolbox owns a dynamic ToolSet +that can retrieve the latest declarations from an MCP service whenever the +framework builds a tool snapshot. + +Dynamic names use `mcp__github__` to avoid collisions across +servers. Closing `ToolSearchToolSet` also closes its dynamic ToolSets. The +framework compares declaration fingerprints: changed or removed tools have +their stale semantic vectors cleared, while unchanged tools keep their existing +index entries. Concurrent refreshes cannot publish vectors from stale +declarations. + +A **declaration fingerprint** is derived from a tool's name, description, and +parameter schema. It detects changes to an MCP tool definition; it is not an +authentication credential and is never sent to the model. + +See the runnable +[`examples/tool_search`](https://github.com/trpc-group/trpc-agent-python/tree/main/examples/tool_search). diff --git a/docs/mkdocs/zh/tool_search.md b/docs/mkdocs/zh/tool_search.md new file mode 100644 index 000000000..cb97bfa3c --- /dev/null +++ b/docs/mkdocs/zh/tool_search.md @@ -0,0 +1,217 @@ +# Tool Search + +当 Agent 拥有大量工具时,把全部工具定义放入每次模型请求会增加输入 +Token、影响工具选择,也会降低模型缓存的命中率。`ToolSearchToolSet` 只暴露 +常驻工具和 `tool_search`,让模型按需发现并加载当前会话需要的工具。 + +## 核心概念 + +- **Tool**:模型可以调用的函数,例如 `get_weather`。通常使用 + `FunctionTool(get_weather)` 包装普通 Python 函数。 +- **ToolSearchToolSet**:管理工具目录、执行搜索并控制工具何时对模型可用的 + ToolSet。将它放入 `LlmAgent.tools` 即可启用 Tool Search。 +- **Preset Tool(常驻工具)**:通过 `preset_tools` 配置,完整 Schema + 始终直接提供给模型,不需要先搜索。适合计算器、当前时间等少量高频工具。 + `preset_tools` 不是必填项;常驻工具也可以直接放入 `LlmAgent.tools`。 +- **Deferred Tool(延迟工具)**:通过 `deferred_tools` 配置,初始只在目录中 + 可被搜索,完整 Schema 不会立即提供给模型。这里的“延迟”是指延迟暴露给 + 模型,不是延迟执行或后台任务。 +- **Toolbox(工具箱)**:通过 `toolboxes` 配置的一组静态 Deferred Tool。 + 每个 Toolbox 有名称和描述,可用来按业务域组织工具,例如 `weather`、 + `filesystem`。Toolbox 中的工具仍然是延迟工具。 +- **Namespace(命名空间)**:Toolbox 的名称,用于限定搜索范围,减少不同 + 业务域中相似工具的干扰。它不是静态工具调用名称的一部分。 +- **Catalog(工具目录)**:提供给模型的轻量工具索引,主要包含 Namespace、 + 工具名称和描述,不等于完整工具 Schema。 +- **发现/加载**:`tool_search` 找到某个 Deferred Tool 后,框架记录该工具 + 已发现。Native 模式下一轮会把它的完整 Schema 提供给模型;Dispatch + 模式会在搜索结果中返回 Schema,并允许 `call_tool` 调用。 +- **Session State**:持久化已发现工具名称的会话状态,不保存 Tool 对象或 + Schema。它使同一 Agent 分支的后续轮次能够复用已经发现的工具。 + +`deferred_tools` 与 `toolboxes` 都用于配置静态延迟工具,区别仅在于是否分组: + +```python +tool_search = ToolSearchToolSet( + # 不分组,属于默认空 Namespace + deferred_tools=[ + FunctionTool(search_document), + ], + # 按业务域分组,可以通过 namespace="weather" 限定搜索 + toolboxes=[ + ToolSearchToolbox( + name="weather", + description="天气查询和预报", + tools=[ + FunctionTool(get_weather), + FunctionTool(get_forecast), + ], + ), + ], +) +``` + +如果工具数量较少且不需要按业务域搜索,使用 `deferred_tools` 即可;工具数量 +较多、存在多个业务域或相似工具时,建议使用 `toolboxes` 分组。 + +## 基本用法 + +```python +from trpc_agent_sdk.agents import LlmAgent +from trpc_agent_sdk.tools import FunctionTool +from trpc_agent_sdk.tools import ToolSearchToolbox +from trpc_agent_sdk.tools import ToolSearchToolSet +from trpc_agent_sdk.tools.tool_search import TOOL_SEARCH_PLACEHOLDER + +tool_search = ToolSearchToolSet( + preset_tools=[FunctionTool(calculator)], + toolboxes=[ + ToolSearchToolbox( + name="weather", + description="天气查询和预报", + tools=[ + FunctionTool(get_weather), + FunctionTool(get_forecast), + ], + ), + ], +) + +agent = LlmAgent( + name="assistant", + model=model, + instruction=( + "调用延迟工具前先使用 tool_search。\n\n" + f"{TOOL_SEARCH_PLACEHOLDER}" + ), + tools=[tool_search], +) +``` + +`preset_tools` 始终直接提供给模型;`deferred_tools` 和 `toolboxes` 中的工具 +只有被 `tool_search` 找到后才可调用。已发现的工具名称保存在 Session +State 中,同一 Agent 分支的后续轮次可以继续使用。 + +## 调用模式 + +`ToolSearchOptions.invocation_mode` 支持两种模式: + +- `InvocationMode.NATIVE`(默认):发现后将工具定义直接加入下一次模型请求, + 模型按原工具名称调用,适合工具数量适中的场景。 +- `InvocationMode.DISPATCH`:模型统一通过 `call_tool(tool_name, params)` 调用 + 已发现工具,模型请求中的工具定义数量保持稳定,适合大规模目录和重视 + Prompt Cache 稳定性的场景。 + +Dispatch 模式不会允许绕过搜索调用未加载的工具。权限过滤也会同时作用于 +目录、搜索结果和实际调用。Native 模式下如果模型直接调用尚未加载的工具, +框架会返回专用提示,要求先通过 `tool_search` 加载再重试;该提示工具本身 +不会暴露给模型。对于已经从 MCP 最新工具列表下线、但模型仍持有旧名称的 +工具,也会返回相同提示。 + +## 检索和目录注入 + +默认检索不需要额外依赖,按名称、描述和命名空间进行关键词匹配。关键词前加 +`+` 表示必须命中,例如 `["+weather", "forecast"]`。 + +模型调用 `tool_search` 时可以使用以下参数: + +- `queries`:关键词或自然语言查询列表,例如 `["天气预报"]`。 +- `tool_names`:需要精确加载的工具名称列表。名称通常来自目录或上一次搜索的 + `additional_candidates`。 +- `namespace`:可选的 Toolbox Namespace,只在指定业务域内搜索。 +- `max_results`:本次最多加载的匹配数量,请求上限为 10;其余匹配会通过 + `additional_candidates` 返回名称,模型可以随后精确加载。 + +`queries` 适合“不知道具体工具名称”的场景,`tool_names` 适合“已经知道准确 +名称”的场景。例如: + +```json +{ + "queries": ["查询未来天气"], + "namespace": "weather", + "max_results": 3 +} +``` + +命名空间只用于限定搜索范围,不是静态工具名称的一部分。调用 `tool_names` +时应复制目录展示的准确名称,例如使用 `"get_forecast"`,不要使用 +`"weather.get_forecast"`。为兼容模型偶发的错误推断,框架仅在命名空间和 +工具名都能准确匹配时容错接受后一种写法。 + +目录默认替换系统指令中的 `{deferred_tools_section}`;没有该占位符时会追加到 +系统指令。设置 `catalog_in_description=True` 可改为写入 `tool_search` 描述。 + +如需语义检索,可向 `ToolSearchToolSet` 传入 +`SemanticToolIndex(embedder)`。Embedder 可实现异步 `aembed_query` / +`aembed_documents`,或同步 `embed_query` / `embed_documents`。设置 +`embedding_fail_open=True` 后,Embedding 服务失败时自动降级到关键词检索。 + +`SemanticToolIndex(embedder, vector_store=...)` 支持注入 +`ToolSearchVectorStore`,默认使用 `InMemoryToolSearchVectorStore`。自定义 +VectorStore 需要实现 `upsert`、`delete` 和 `search`,方法可以是同步或异步。 + +Embedder 可以继续直接返回向量;如果服务提供 Token 用量,也可以返回 +`(向量, usage)`,其中 usage 支持 `prompt_tokens` / `total_tokens` 或 +`prompt_token_count` / `total_token_count`。每次模型请求会自动开始新的统计 +周期;当前轮结果通过 `tool_search.get_embedding_usage(ctx)` 读取,传入 +`reset=True` 可提前清零。 + +## 主要配置项 + +- `max_results`:默认搜索结果数量。 +- `permission_filter`:根据当前 `InvocationContext` 返回每个延迟工具是否 + 可见和可调用,用于租户、用户或 Agent 级权限控制。 +- `catalog_in_description`:是否把 Catalog 放入 `tool_search` 的工具描述; + 默认放入系统指令。 +- `invocation_mode`:选择 Native 或 Dispatch 调用模式。 +- `embedding_fail_open`:语义检索失败时是否降级到关键词检索。 +- `state_key_prefix`:Session State 中已发现工具记录的键前缀,用于避免与 + 其他框架状态冲突。 +- `strict_validation`:是否严格校验重复 Namespace、重复工具和无效配置。 + +## 配置容错 + +默认 `ToolSearchOptions(strict_validation=True)`,空 Namespace、重复 +Namespace 或重复工具会在初始化时直接报错。动态组装大量工具时可以设置: + +```python +ToolSearchOptions(strict_validation=False) +``` + +容错模式会记录 Warning,并合并重复静态 Namespace、保留工具的第一次归属, +跳过空 Toolbox、无效工具和发生冲突的 MCP Namespace。保留名称冲突、 +同名工具 Declaration 不一致以及非法 `max_results` 仍然报错,因为这些问题 +无法安全恢复。 + +## 动态 MCP 工具 + +使用 `ToolSearchMCPToolbox` 包装动态 ToolSet: + +```python +from trpc_agent_sdk.tools import ToolSearchMCPToolbox + +tool_search = ToolSearchToolSet( + mcp_toolboxes=[ + ToolSearchMCPToolbox( + server_name="github", + description="GitHub MCP 工具", + toolset=mcp_toolset, + ), + ], +) +``` + +`ToolSearchMCPToolbox` 与静态 `ToolSearchToolbox` 的区别是:静态 Toolbox +直接持有确定的工具列表;MCP Toolbox 持有一个动态 ToolSet,每次构建工具 +快照时可以从 MCP 服务获取最新工具声明。 + +动态工具会以 `mcp__github__` 暴露,避免多个服务重名。关闭 +`ToolSearchToolSet` 时会同步关闭其动态 ToolSet。框架会比较动态工具的声明 +指纹:声明变化或工具下线时清理旧语义向量,未变化的工具继续复用已有索引; +并发刷新时还会阻止过期声明重新写入索引。 + +这里的**声明指纹**是根据工具名称、描述和参数 Schema 计算的标识,用于判断 +MCP 工具定义是否变化;它不是认证凭证,也不会发送给模型。 + +完整可运行代码见 +[`examples/tool_search`](../../../examples/tool_search/)。 diff --git a/examples/tool_search/.env.example b/examples/tool_search/.env.example new file mode 100644 index 000000000..0a57a1786 --- /dev/null +++ b/examples/tool_search/.env.example @@ -0,0 +1,5 @@ +# Set TRPC_AGENT_API_KEY、TRPC_AGENT_BASE_URL、TRPC_AGENT_MODEL_NAME +TRPC_AGENT_API_KEY=your-api-key +TRPC_AGENT_BASE_URL=your-base-url +TRPC_AGENT_MODEL_NAME=your-model-name + diff --git a/examples/tool_search/README.md b/examples/tool_search/README.md new file mode 100644 index 000000000..bac06bb2e --- /dev/null +++ b/examples/tool_search/README.md @@ -0,0 +1,34 @@ +# Tool Search + +This example keeps domain tools out of the initial model request and lets the +model discover only the tools needed for the current conversation. It contains: + +- `calculator`, a preset tool that is always available; +- deferred `weather` and `billing` toolboxes; +- Native and Dispatch invocation modes; +- session-scoped persistence of discovered tools. + +## Run + +```bash +cd examples/tool_search +cp .env.example .env +# Edit model settings in .env. +uv run python run_agent.py +``` + +For the default request, the model first calls `tool_search` to load +`get_forecast`, then invokes that tool. In `native` mode it calls +`get_forecast` directly. In `dispatch` mode it calls `call_tool` with the +discovered tool name and arguments. + +Change `TOOL_SEARCH_QUERY` to test another namespace, for example: + +```dotenv +TOOL_SEARCH_QUERY=Create a 99.5 dollar invoice for Alice. +``` + +The tool catalog is injected at `{deferred_tools_section}` in the Agent +instruction. If the placeholder is absent, it is appended to the system +instruction. Set `TOOL_SEARCH_CATALOG_IN_DESCRIPTION=true` to put the catalog +in the `tool_search` description instead. diff --git a/examples/tool_search/agent/__init__.py b/examples/tool_search/agent/__init__.py new file mode 100644 index 000000000..60c65a00c --- /dev/null +++ b/examples/tool_search/agent/__init__.py @@ -0,0 +1 @@ +"""Real agent demo for Tool Script Safety Guard.""" diff --git a/examples/tool_search/agent/agent.py b/examples/tool_search/agent/agent.py new file mode 100644 index 000000000..cd391bcc6 --- /dev/null +++ b/examples/tool_search/agent/agent.py @@ -0,0 +1,34 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Real LlmAgent wiring for the tool-safety review demo.""" + +from __future__ import annotations + +from trpc_agent_sdk.agents import LlmAgent +from trpc_agent_sdk.models import LLMModel +from trpc_agent_sdk.models import OpenAIModel +from trpc_agent_sdk.tools import ToolSearchOptions + +from .config import get_model_config +from .prompts import INSTRUCTION +from .tools import create_tool_search_toolset + + +def _create_model() -> LLMModel: + api_key, base_url, model_name = get_model_config() + return OpenAIModel(model_name=model_name, api_key=api_key, base_url=base_url) + + +def create_agent(options: ToolSearchOptions | None = None) -> LlmAgent: + """ Create an agent""" + agent = LlmAgent( + name="tool_search_agent", + description="A tool search agent", + model=_create_model(), # You can change this to your preferred model + instruction=INSTRUCTION, + tools=[create_tool_search_toolset(options)], + ) + return agent diff --git a/examples/tool_search/agent/config.py b/examples/tool_search/agent/config.py new file mode 100644 index 000000000..19f5dec13 --- /dev/null +++ b/examples/tool_search/agent/config.py @@ -0,0 +1,28 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Configuration helpers for the real tool-safety agent demo.""" + +from __future__ import annotations + +import os + + +def get_model_config() -> tuple[str, str, str]: + """Return OpenAI-compatible model configuration from environment variables.""" + api_key = os.environ.get("TRPC_AGENT_API_KEY", "") + base_url = os.environ.get("TRPC_AGENT_BASE_URL", "") + model_name = os.environ.get("TRPC_AGENT_MODEL_NAME", "") + missing = [ + name for name, value in ( + ("TRPC_AGENT_API_KEY", api_key), + ("TRPC_AGENT_BASE_URL", base_url), + ("TRPC_AGENT_MODEL_NAME", model_name), + ) if not value + ] + if missing: + joined = ", ".join(missing) + raise RuntimeError(f"Missing model environment variables: {joined}") + return api_key, base_url, model_name diff --git a/examples/tool_search/agent/prompts.py b/examples/tool_search/agent/prompts.py new file mode 100644 index 000000000..8a911fc73 --- /dev/null +++ b/examples/tool_search/agent/prompts.py @@ -0,0 +1,17 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Prompt for the real tool-safety agent demo.""" + +from trpc_agent_sdk.tools import TOOL_SEARCH_PLACEHOLDER + +INSTRUCTION = f""" +Use tool_search before calling a deferred tool. If the catalog already shows +an exact name, copy that name into tool_names. A namespace only filters the +search scope and is not part of the tool name: use "get_forecast", not +"weather.get_forecast". Otherwise, search the matching namespace. + +{TOOL_SEARCH_PLACEHOLDER} +""" diff --git a/examples/tool_search/agent/tools.py b/examples/tool_search/agent/tools.py new file mode 100644 index 000000000..727ce74fe --- /dev/null +++ b/examples/tool_search/agent/tools.py @@ -0,0 +1,82 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Tool factories for the real tool-safety agent demo.""" + +from __future__ import annotations + +import uuid + +from trpc_agent_sdk.tools import FunctionTool +from trpc_agent_sdk.tools import ToolSearchOptions +from trpc_agent_sdk.tools import ToolSearchToolbox +from trpc_agent_sdk.tools import ToolSearchToolSet + + +async def calculator(expression: str) -> dict: + """Evaluate a basic arithmetic expression.""" + allowed = set("0123456789.+-*/() ") + if not expression or any(char not in allowed for char in expression): + return {"error": "Only basic arithmetic expressions are supported."} + return {"expression": expression, "result": eval(expression, {"__builtins__": {}}, {})} + + +async def get_weather(city: str) -> dict: + """Get the current weather for a city.""" + return {"city": city, "condition": "sunny", "temperature_c": 25} + + +async def get_forecast(city: str, days: int = 3) -> dict: + """Get a multi-day weather forecast for a city.""" + return { + "city": city, + "days": days, + "forecast": ["sunny", "cloudy", "sunny"][:days], + } + + +async def create_invoice(customer: str, amount: float) -> dict: + """Create an invoice for a customer.""" + return { + "invoice_id": f"INV-{uuid.uuid4().hex[:8]}", + "customer": customer, + "amount": amount, + } + + +async def refund_payment(payment_id: str, reason: str) -> dict: + """Refund a payment and record the reason.""" + return { + "payment_id": payment_id, + "reason": reason, + "status": "refunded", + } + + +def create_tool_search_toolset(options: ToolSearchOptions | None = None) -> ToolSearchToolSet: + """Create an Agent whose domain tools are loaded on demand.""" + tool_search = ToolSearchToolSet( + preset_tools=[FunctionTool(calculator)], + toolboxes=[ + ToolSearchToolbox( + name="weather", + description="weather observations and forecasts", + tools=[ + FunctionTool(get_weather), + FunctionTool(get_forecast), + ], + ), + ToolSearchToolbox( + name="billing", + description="invoices, payments, and refunds", + tools=[ + FunctionTool(create_invoice), + FunctionTool(refund_payment), + ], + ), + ], + options=options, + ) + return tool_search diff --git a/examples/tool_search/run_agent.py b/examples/tool_search/run_agent.py new file mode 100644 index 000000000..a97e853c4 --- /dev/null +++ b/examples/tool_search/run_agent.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 + +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Run an Agent with on-demand tool discovery.""" + +from __future__ import annotations + +import asyncio +import uuid +from pathlib import Path + +from dotenv import load_dotenv + +from trpc_agent_sdk.runners import Runner +from trpc_agent_sdk.sessions import InMemorySessionService +from trpc_agent_sdk.tools import ToolSearchOptions +from trpc_agent_sdk.tools import ToolSearchInvocationMode +from trpc_agent_sdk.types import Content +from trpc_agent_sdk.types import Part + +load_dotenv(Path(__file__).with_name(".env")) + + +async def main() -> None: + """Run one request while preserving discovered tools in the session.""" + user_id = "tool_search_user" + app_name = "tool_search_agent_demo" + options = ToolSearchOptions( + invocation_mode=ToolSearchInvocationMode.NATIVE, + catalog_in_description=False, + ) + from agent.agent import create_agent + root_agent = create_agent(options) + session_service = InMemorySessionService() + runner = Runner(app_name=app_name, agent=root_agent, session_service=session_service) + + # Demo query list + demo_queries = [ + "What is the three-day weather forecast for Shenzhen?", + ] + + for query in demo_queries: + # Use a new session for each query + current_session_id = str(uuid.uuid4()) + + print(f"🆔 Session ID: {current_session_id[:8]}...") + print(f"📝 User: {query}") + + user_content = Content(parts=[Part.from_text(text=query)]) + + print("🤖 Assistant: ", end="", flush=True) + async for event in runner.run_async(user_id=user_id, session_id=current_session_id, new_message=user_content): + # Check if event.content exists + if not event.content or not event.content.parts: + continue + + if event.partial: + for part in event.content.parts: + if part.text: + print(part.text, end="", flush=True) + continue + + for part in event.content.parts: + # Skip the reasoning part; the output is already generated when partial=True + if part.thought: + continue + if part.function_call: + print(f"\n🔧 [Invoke Tool: {part.function_call.name}({part.function_call.args})]") + elif part.function_response: + print(f"📊 [Tool Result: {part.function_response.response}]") + # Uncomment to get the full text output of the LLM + # elif part.text: + # print(f"\n✅ {part.text}") + + print("\n" + "-" * 40) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/mkdocs.yml b/mkdocs.yml index 336129b91..38667a175 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -38,6 +38,7 @@ nav: - Goal: en/goal.md - Code Executor: en/code_executor.md - OpenAPI Tools: en/openapi_tools.md + - Tool Search: en/tool_search.md - CodeAct: en/codeact.md - Skills: en/skill.md - Session & Memory: @@ -88,6 +89,7 @@ nav: - Goal 工具: zh/goal.md - 代码执行器: zh/code_executor.md - OpenAPI 工具: zh/openapi_tools.md + - 工具搜索: zh/tool_search.md - CodeAct: zh/codeact.md - Skills: zh/skill.md - Session 与 Memory: diff --git a/tests/tools/tool_search/test_tool_search.py b/tests/tools/tool_search/test_tool_search.py new file mode 100644 index 000000000..ac93215ee --- /dev/null +++ b/tests/tools/tool_search/test_tool_search.py @@ -0,0 +1,965 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Tests for deferred tool discovery and loading.""" + +from __future__ import annotations + +import asyncio +from unittest.mock import patch + +import pytest + +from trpc_agent_sdk.agents._base_agent import BaseAgent +from trpc_agent_sdk.agents.core._tools_processor import ToolsProcessor +from trpc_agent_sdk.context import InvocationContext +from trpc_agent_sdk.context import create_agent_context +from trpc_agent_sdk.models import LlmRequest +from trpc_agent_sdk.sessions import InMemorySessionService +from trpc_agent_sdk.tools import BaseToolSet +from trpc_agent_sdk.tools import FunctionTool +from trpc_agent_sdk.tools import LongRunningFunctionTool +from trpc_agent_sdk.tools import StreamingProgressTool +from trpc_agent_sdk.tools.tool_search import CALL_TOOL_NAME +from trpc_agent_sdk.tools.tool_search import InMemoryToolSearchVectorStore +from trpc_agent_sdk.tools.tool_search import InvocationMode +from trpc_agent_sdk.tools.tool_search import MCPToolbox +from trpc_agent_sdk.tools.tool_search import SemanticToolIndex +from trpc_agent_sdk.tools.tool_search import TOOL_SEARCH_PLACEHOLDER +from trpc_agent_sdk.tools.tool_search import TOOL_SEARCH_TOOL_NAME +from trpc_agent_sdk.tools.tool_search import Toolbox +from trpc_agent_sdk.tools.tool_search import ToolSearchDocument +from trpc_agent_sdk.tools.tool_search import ToolSearchOptions +from trpc_agent_sdk.tools.tool_search import ToolSearchToolSet +from trpc_agent_sdk.types import FunctionCall +from trpc_agent_sdk.types import GenerateContentConfig + + +class _StubAgent(BaseAgent): + + async def _run_async_impl(self, ctx): + yield + + +class _DynamicToolSet(BaseToolSet): + + def __init__(self, tools): + super().__init__(name="dynamic") + self.tools = tools + self.closed = False + self.raise_error = False + + async def get_tools(self, invocation_context=None): + if self.raise_error: + raise RuntimeError("dynamic listing unavailable") + return list(self.tools) + + async def close(self): + self.closed = True + + +def _tool(name: str, description: str): + + async def implementation(value: str = ""): + return { + "tool": name, + "value": value, + } + + implementation.__name__ = name + implementation.__doc__ = description + return FunctionTool(implementation) + + +@pytest.fixture +def context(): + service = InMemorySessionService() + session = asyncio.run(service.create_session( + app_name="test", + user_id="user", + session_id="session", + )) + agent = _StubAgent(name="search_agent") + return InvocationContext( + session_service=service, + invocation_id="invocation", + agent=agent, + agent_context=create_agent_context(), + session=session, + ) + + +async def _search(toolset, context, **args): + tools = await toolset.get_tools(context) + search = next(tool for tool in tools if tool.name == TOOL_SEARCH_TOOL_NAME) + return await search.run_async( + tool_context=context, + args=args, + ) + + +class TestKeywordSearch: + + async def test_loads_deferred_tool_and_exposes_it_next_request( + self, + context, + ): + read = _tool("read_file", "Read a file from disk") + write = _tool("write_file", "Write text to a file") + toolset = ToolSearchToolSet( + toolboxes=[Toolbox( + name="filesystem", + description="file and directory operations", + tools=[read, write], + )], + options=ToolSearchOptions(max_results=1), + ) + + initial = await toolset.get_tools(context) + assert [tool.name for tool in initial] == [TOOL_SEARCH_TOOL_NAME] + + result = await _search( + toolset, + context, + namespace="filesystem", + queries=["write file"], + ) + + assert result["tools"][0]["name"] == "write_file" + loaded = await toolset.get_tools(context) + assert [tool.name for tool in loaded] == [ + TOOL_SEARCH_TOOL_NAME, + "write_file", + ] + + async def test_exact_names_are_case_insensitive_and_cross_namespace( + self, + context, + ): + toolset = ToolSearchToolSet(toolboxes=[ + Toolbox("filesystem", "files", [_tool("read_file", "Read file")]), + Toolbox("network", "network", [_tool("http_get", "HTTP GET")]), + ]) + + result = await _search( + toolset, + context, + tool_names=[ + "HTTP_GET", + "filesystem.read_file", + "http_get", + ], + ) + + assert [item["name"] for item in result["tools"]] == [ + "http_get", + "read_file", + ] + + async def test_search_status_distinguishes_partially_and_fully_loaded( + self, + context, + ): + toolset = ToolSearchToolSet(deferred_tools=[ + _tool("read_file", "Read file"), + _tool("write_file", "Write file"), + ]) + await _search( + toolset, + context, + tool_names=["read_file"], + ) + + partial = await _search( + toolset, + context, + tool_names=["read_file", "write_file"], + ) + assert "already_loaded=true" in partial["status"] + + complete = await _search( + toolset, + context, + tool_names=["read_file", "write_file"], + ) + assert complete["status"].startswith("All matches were loaded earlier", ) + + async def test_qualified_alias_requires_matching_namespace( + self, + context, + ): + toolset = ToolSearchToolSet(toolboxes=[ + Toolbox("filesystem", "files", [_tool("read_file", "Read file")]), + ]) + + result = await _search( + toolset, + context, + tool_names=["network.read_file"], + ) + + assert result["tools"] == [] + + async def test_required_terms_and_overflow(self, context): + toolset = ToolSearchToolSet( + deferred_tools=[ + _tool("read_file", "Read local file content"), + _tool("read_url", "Read remote URL content"), + _tool("write_file", "Write local file content"), + ], + options=ToolSearchOptions(max_results=1), + ) + + result = await _search( + toolset, + context, + queries=["+read content"], + ) + + assert len(result["tools"]) == 1 + assert result["additional_candidates"] == ["read_url"] + + async def test_unknown_namespace_returns_catalog(self, context): + toolset = ToolSearchToolSet(toolboxes=[Toolbox("filesystem", "files", [_tool("read_file", "Read file")])]) + + result = await _search( + toolset, + context, + namespace="missing", + queries=["read"], + ) + + assert result["status"] == "unknown_namespace" + assert result["catalog"] == [{ + "name": "filesystem", + "description": "files", + }] + + async def test_namespace_without_query_lists_tools(self, context): + toolset = ToolSearchToolSet(toolboxes=[ + Toolbox( + "filesystem", + "files", + [ + _tool("write_file", "Write file"), + _tool("read_file", "Read file"), + ], + ) + ]) + + result = await _search( + toolset, + context, + namespace="filesystem", + ) + + assert [item["name"] for item in result["tools"]] == [ + "read_file", + "write_file", + ] + + +class TestCatalogAndPermissions: + + async def test_tools_processor_refreshes_native_tools_after_search( + self, + context, + ): + toolset = ToolSearchToolSet(deferred_tools=[_tool("read_file", "Read file")], ) + processor = ToolsProcessor([toolset]) + first_request = LlmRequest(config=GenerateContentConfig(system_instruction="Use tools."), ) + + await processor.process_llm_request(context, first_request) + assert set(first_request.tools_dict) == {TOOL_SEARCH_TOOL_NAME} + + await _search(toolset, context, tool_names=["read_file"]) + second_request = LlmRequest(config=GenerateContentConfig(system_instruction="Use tools."), ) + await processor.process_llm_request(context, second_request) + + assert set(second_request.tools_dict) == { + TOOL_SEARCH_TOOL_NAME, + "read_file", + } + assert "" in str(second_request.config.system_instruction) + + async def test_catalog_replaces_placeholder(self, context): + toolset = ToolSearchToolSet( + toolboxes=[Toolbox("filesystem", "file operations", [_tool("read_file", "Read file")])]) + search = (await toolset.get_tools(context))[0] + request = LlmRequest(config=GenerateContentConfig( + system_instruction=(f"Use tools.\n{TOOL_SEARCH_PLACEHOLDER}"), ), ) + + await search.process_request( + tool_context=context, + llm_request=request, + ) + + instruction = str(request.config.system_instruction) + assert TOOL_SEARCH_PLACEHOLDER not in instruction + assert "" in instruction + assert "read_file" in instruction + assert 'exact tool_names=["read_file"]' in instruction + assert "must not be prepended" in instruction + + async def test_permission_hides_searches_and_loaded_tools(self, context): + + def permissions(_ctx, names): + return {name: name != "delete_file" for name in names} + + toolset = ToolSearchToolSet( + deferred_tools=[ + _tool("read_file", "Read file"), + _tool("delete_file", "Delete file"), + ], + options=ToolSearchOptions(permission_filter=permissions), + ) + + result = await _search( + toolset, + context, + tool_names=["read_file", "delete_file"], + ) + + assert [item["name"] for item in result["tools"]] == ["read_file"] + assert "delete_file" not in [tool.name for tool in await toolset.get_tools(context)] + + async def test_preset_is_visible_but_not_keyword_searchable(self, context): + preset = _tool("calculator", "Calculate numbers") + toolset = ToolSearchToolSet( + preset_tools=[preset], + deferred_tools=[_tool("weather", "Get weather")], + ) + + assert [tool.name for tool in await toolset.get_tools(context)] == [ + "calculator", + TOOL_SEARCH_TOOL_NAME, + ] + result = await _search( + toolset, + context, + queries=["calculate"], + ) + assert result["tools"] == [] + + +class TestConfigurationTolerance: + + def test_strict_mode_rejects_duplicate_namespace(self): + with pytest.raises(ValueError, match="Duplicate toolbox namespace"): + ToolSearchToolSet(toolboxes=[ + Toolbox("filesystem", "files", []), + Toolbox("filesystem", "more files", []), + ]) + + async def test_tolerant_mode_merges_and_keeps_first_tool_owner( + self, + context, + ): + read = _tool("read_file", "Read file") + write = _tool("write_file", "Write file") + with patch("trpc_agent_sdk.tools.tool_search._toolset.logger.warning", ) as warning: + toolset = ToolSearchToolSet( + toolboxes=[ + Toolbox("filesystem", "", [read]), + Toolbox("filesystem", "file operations", [write]), + Toolbox("document", "document operations", [read]), + Toolbox("", "invalid", [_tool("ignored", "Ignored")]), + ], + options=ToolSearchOptions(strict_validation=False), + ) + + filesystem = await _search( + toolset, + context, + namespace="filesystem", + ) + assert [item["name"] for item in filesystem["tools"]] == [ + "read_file", + "write_file", + ] + + document = await _search( + toolset, + context, + namespace="document", + ) + assert document["tools"] == [] + warning.assert_any_call( + "Merging duplicate toolbox namespace %r", + "filesystem", + ) + warning.assert_any_call( + "Tool %r is already registered in namespace %r; keeping " + "the first owner and ignoring namespace %r", + "read_file", + "filesystem", + "document", + ) + warning.assert_any_call( + "Skipping toolbox with empty name: description=%r", + "invalid", + ) + + def test_tolerant_mode_rejects_conflicting_declarations(self): + with pytest.raises(ValueError, match="Conflicting declarations"): + ToolSearchToolSet( + toolboxes=[ + Toolbox( + "first", + "", + [_tool("shared", "First declaration")], + ), + Toolbox( + "second", + "", + [_tool("shared", "Second declaration")], + ), + ], + options=ToolSearchOptions(strict_validation=False), + ) + + async def test_tolerant_mode_skips_colliding_mcp_namespace( + self, + context, + ): + dynamic = _DynamicToolSet([_tool("read", "Read repository")]) + with patch("trpc_agent_sdk.tools.tool_search._toolset.logger.warning", ) as warning: + toolset = ToolSearchToolSet( + toolboxes=[Toolbox( + "github", + "static GitHub tools", + [_tool("search", "Search repository")], + )], + mcp_toolboxes=[MCPToolbox( + server_name="github", + description="dynamic GitHub tools", + toolset=dynamic, + )], + options=ToolSearchOptions(strict_validation=False), + ) + + result = await _search( + toolset, + context, + namespace="github", + ) + assert [item["name"] for item in result["tools"]] == ["search"] + warning.assert_called_once_with( + "Skipping MCP toolbox %r because its namespace is " + "already registered", + "github", + ) + + async def test_tolerant_mode_deduplicates_dynamic_listing( + self, + context, + ): + read = _tool("read", "Read repository") + dynamic = _DynamicToolSet([read, read]) + toolset = ToolSearchToolSet( + mcp_toolboxes=[MCPToolbox( + server_name="github", + description="GitHub tools", + toolset=dynamic, + )], + options=ToolSearchOptions(strict_validation=False), + ) + + with patch("trpc_agent_sdk.tools.tool_search._toolset.logger.warning", ) as warning: + result = await _search( + toolset, + context, + namespace="github", + ) + assert [item["name"] for item in result["tools"]] == [ + "mcp__github__read", + ] + warning.assert_called_once_with( + "Dynamic tool %r is listed more than once by namespace %r; " + "keeping the first entry", + "mcp__github__read", + "github", + ) + + +class TestInvocationModes: + + async def test_dispatch_keeps_fixed_surface_and_calls_loaded_tool( + self, + context, + ): + toolset = ToolSearchToolSet( + deferred_tools=[_tool("echo", "Echo a value")], + options=ToolSearchOptions(invocation_mode=InvocationMode.DISPATCH, ), + ) + + assert [tool.name for tool in await toolset.get_tools(context)] == [ + TOOL_SEARCH_TOOL_NAME, + CALL_TOOL_NAME, + ] + result = await _search( + toolset, + context, + tool_names=["echo"], + ) + assert result["tools"][0]["input_schema"]["type"] == "object" + + tools = await toolset.get_tools(context) + assert [tool.name for tool in tools] == [ + TOOL_SEARCH_TOOL_NAME, + CALL_TOOL_NAME, + ] + call_tool = next(tool for tool in tools if tool.name == CALL_TOOL_NAME) + called = await call_tool.run_async( + tool_context=context, + args={ + "tool_name": "echo", + "params": { + "value": "hello" + }, + }, + ) + assert called == { + "tool": "echo", + "value": "hello", + } + + async def test_dispatch_rejects_unloaded_tool(self, context): + toolset = ToolSearchToolSet( + deferred_tools=[_tool("echo", "Echo a value")], + options=ToolSearchOptions(invocation_mode=InvocationMode.DISPATCH, ), + ) + call_tool = next(tool for tool in await toolset.get_tools(context) if tool.name == CALL_TOOL_NAME) + + result = await call_tool.run_async( + tool_context=context, + args={ + "tool_name": "echo", + "params": {}, + }, + ) + + assert "not loaded" in result["message"] + + async def test_dispatch_aggregates_progress_streaming_tool( + self, + context, + ): + + async def progress(task: str): + """Report progress for a task.""" + yield {"task": task, "status": "started"} + yield {"task": task, "status": "done"} + + toolset = ToolSearchToolSet( + deferred_tools=[StreamingProgressTool(progress)], + options=ToolSearchOptions(invocation_mode=InvocationMode.DISPATCH, ), + ) + await _search(toolset, context, tool_names=["progress"]) + call_tool = next(tool for tool in await toolset.get_tools(context) if tool.name == CALL_TOOL_NAME) + + result = await call_tool.run_async( + tool_context=context, + args={ + "tool_name": "progress", + "params": { + "task": "test" + }, + }, + ) + + assert result == { + "chunks": [ + { + "task": "test", + "status": "started", + }, + { + "task": "test", + "status": "done", + }, + ], + } + + +class TestDynamicAndSemantic: + + async def test_mcp_toolbox_renames_and_retains_last_nonempty_snapshot( + self, + context, + ): + dynamic = _DynamicToolSet([_tool("read", "Read repository")]) + toolset = ToolSearchToolSet(mcp_toolboxes=[ + MCPToolbox( + server_name="github", + description="GitHub repository operations", + toolset=dynamic, + ) + ]) + + first = await _search( + toolset, + context, + namespace="github", + ) + assert first["tools"][0]["name"] == "mcp__github__read" + + dynamic.tools = [] + names = [tool.name for tool in await toolset.get_tools(context)] + assert "mcp__github__read" in names + + dynamic.raise_error = True + names = [tool.name for tool in await toolset.get_tools(context)] + assert "mcp__github__read" in names + + await toolset.close() + assert dynamic.closed is True + + async def test_mcp_toolbox_preserves_long_running_marker( + self, + context, + ): + + async def import_data(source: str): + """Import data from a source.""" + return {"source": source} + + dynamic = _DynamicToolSet([LongRunningFunctionTool(import_data)], ) + toolset = ToolSearchToolSet( + mcp_toolboxes=[MCPToolbox( + server_name="warehouse", + description="Warehouse operations", + toolset=dynamic, + )]) + + result = await _search( + toolset, + context, + namespace="warehouse", + ) + assert result["tools"][0]["name"] == "mcp__warehouse__import_data" + loaded = await toolset.get_tools(context) + renamed = next(tool for tool in loaded if tool.name == "mcp__warehouse__import_data") + assert isinstance(renamed, LongRunningFunctionTool) + + async def test_native_stale_mcp_call_returns_search_guidance( + self, + context, + ): + dynamic = _DynamicToolSet([_tool("read", "Read repository")]) + toolset = ToolSearchToolSet(mcp_toolboxes=[ + MCPToolbox( + server_name="github", + description="GitHub repository operations", + toolset=dynamic, + ) + ]) + await _search( + toolset, + context, + tool_names=["mcp__github__read"], + ) + + dynamic.tools = [_tool("write", "Write repository")] + processor = ToolsProcessor([toolset]) + events = [ + event async for event in processor.execute_tools_async( + [FunctionCall( + id="call-1", + name="mcp__github__read", + args={}, + )], + context, + ) + ] + + assert len(events) == 1 + assert events[0].error_code is None + response = events[0].content.parts[0].function_response.response + assert "not loaded yet" in str(response) + assert "tool_search" in str(response) + + async def test_semantic_search_ranks_by_similarity(self, context): + + class Embedder: + + async def aembed_documents(self, texts): + return [[1.0, 0.0] if "weather" in text else [0.0, 1.0] for text in texts] + + async def aembed_query(self, text): + return [1.0, 0.0] if "forecast" in text else [0.0, 1.0] + + toolset = ToolSearchToolSet( + deferred_tools=[ + _tool("weather", "Get weather forecast"), + _tool("database", "Run SQL query"), + ], + semantic_index=SemanticToolIndex(Embedder()), + ) + + result = await _search( + toolset, + context, + queries=["tomorrow forecast"], + max_results=1, + ) + + assert result["tools"][0]["name"] == "weather" + + async def test_semantic_failure_can_fall_back_to_keyword( + self, + context, + ): + + class BrokenEmbedder: + + async def aembed_documents(self, texts): + raise RuntimeError("embedding unavailable") + + async def aembed_query(self, text): + raise RuntimeError("embedding unavailable") + + toolset = ToolSearchToolSet( + deferred_tools=[_tool("weather", "Get weather forecast")], + semantic_index=SemanticToolIndex(BrokenEmbedder()), + options=ToolSearchOptions(embedding_fail_open=True), + ) + + result = await _search( + toolset, + context, + queries=["weather"], + ) + + assert result["tools"][0]["name"] == "weather" + + async def test_mcp_forgets_changed_and_removed_semantic_entries( + self, + context, + ): + + class Embedder: + + async def aembed_documents(self, texts): + return [[1.0, 0.0] for _ in texts] + + async def aembed_query(self, text): + return [1.0, 0.0] + + class RecordingSemanticIndex(SemanticToolIndex): + + def __init__(self): + super().__init__(Embedder()) + self.forgotten = [] + + async def forget(self, names): + self.forgotten.extend(names) + await super().forget(names) + + semantic_index = RecordingSemanticIndex() + dynamic = _DynamicToolSet([ + _tool("read", "Read repository"), + _tool("write", "Write repository"), + ]) + toolset = ToolSearchToolSet( + mcp_toolboxes=[ + MCPToolbox( + server_name="github", + description="GitHub repository operations", + toolset=dynamic, + ) + ], + semantic_index=semantic_index, + ) + + await _search( + toolset, + context, + namespace="github", + queries=["repository"], + ) + assert semantic_index.forgotten == [] + + dynamic.tools = [ + _tool("read", "Read repository content"), + _tool("write", "Write repository"), + ] + await toolset.get_tools(context) + assert semantic_index.forgotten == ["mcp__github__read"] + + dynamic.tools = [_tool("write", "Write repository")] + await toolset.get_tools(context) + assert semantic_index.forgotten == [ + "mcp__github__read", + "mcp__github__read", + ] + + async def test_semantic_index_does_not_publish_stale_embedding(self): + embedding_started = asyncio.Event() + release_embedding = asyncio.Event() + + class Embedder: + + async def aembed_documents(self, texts): + embedding_started.set() + await release_embedding.wait() + return [[1.0, 0.0] for _ in texts] + + async def aembed_query(self, text): + return [1.0, 0.0] + + document = ToolSearchDocument( + name="read", + description="Read repository", + namespace="github", + text="Tool: read\nDescription: Read repository", + fingerprint="old", + ) + current = True + index = SemanticToolIndex(Embedder()) + search = asyncio.create_task( + index.search( + "repository", + [document], + 1, + verifier=lambda _name, _fingerprint: current, + )) + + await embedding_started.wait() + current = False + release_embedding.set() + + assert await search == [] + + async def test_custom_vector_store_and_embedding_usage(self, context): + + class RecordingStore(InMemoryToolSearchVectorStore): + + def __init__(self): + super().__init__() + self.upserted = [] + self.searches = 0 + + async def upsert(self, name, vector): + self.upserted.append(name) + super().upsert(name, vector) + + async def search(self, query_vector, candidate_names, limit): + self.searches += 1 + return super().search( + query_vector, + candidate_names, + limit, + ) + + class Embedder: + + async def aembed_documents(self, texts): + return ( + [[1.0, 0.0] for _ in texts], + { + "prompt_tokens": 4, + "total_tokens": 4, + }, + ) + + async def aembed_query(self, text): + return ( + [1.0, 0.0], + { + "prompt_token_count": 2, + "total_token_count": 2, + }, + ) + + store = RecordingStore() + toolset = ToolSearchToolSet( + deferred_tools=[_tool("weather", "Get weather")], + semantic_index=SemanticToolIndex( + Embedder(), + vector_store=store, + ), + ) + processor = ToolsProcessor([toolset]) + await processor.process_llm_request( + context, + LlmRequest(config=GenerateContentConfig(system_instruction="Use tools.", )), + ) + + result = await _search( + toolset, + context, + queries=["forecast"], + ) + + assert result["tools"][0]["name"] == "weather" + assert store.upserted == ["weather"] + assert store.searches == 1 + usage = toolset.get_embedding_usage(context) + assert usage is not None + assert usage.prompt_tokens == 6 + assert usage.total_tokens == 6 + await processor.process_llm_request( + context, + LlmRequest(config=GenerateContentConfig(system_instruction="Use tools.", )), + ) + assert toolset.get_embedding_usage(context).total_tokens == 0 + + async def test_native_unloaded_call_returns_search_guidance( + self, + context, + ): + toolset = ToolSearchToolSet(deferred_tools=[_tool("echo", "Echo a value")], ) + processor = ToolsProcessor([toolset]) + events = [ + event async for event in processor.execute_tools_async( + [FunctionCall( + id="call-1", + name="echo", + args={"value": "hello"}, + )], + context, + ) + ] + + assert len(events) == 1 + assert events[0].error_code is None + response = events[0].content.parts[0].function_response.response + assert "not loaded yet" in str(response) + assert "tool_search" in str(response) + + +class TestState: + + async def test_parallel_searches_merge_state(self, context): + toolset = ToolSearchToolSet(deferred_tools=[ + _tool("read_file", "Read file"), + _tool("write_file", "Write file"), + ]) + + await asyncio.gather( + _search(toolset, context, tool_names=["read_file"]), + _search(toolset, context, tool_names=["write_file"]), + ) + + assert [tool.name for tool in await toolset.get_tools(context)] == [ + TOOL_SEARCH_TOOL_NAME, + "read_file", + "write_file", + ] + discovered_keys = [ + key for key, value in context.event_actions.state_delta.items() + if ":discovered_tools:" in key and value is True + ] + assert len(discovered_keys) == 2 + + async def test_branch_state_isolated(self, context): + toolset = ToolSearchToolSet(deferred_tools=[_tool("read_file", "Read file")], ) + context.branch = "branch-a" + await _search(toolset, context, tool_names=["read_file"]) + assert "read_file" in [tool.name for tool in await toolset.get_tools(context)] + + context.branch = "branch-b" + assert "read_file" not in [tool.name for tool in await toolset.get_tools(context)] diff --git a/trpc_agent_sdk/agents/core/_tools_processor.py b/trpc_agent_sdk/agents/core/_tools_processor.py index 67f839155..45eb38df3 100644 --- a/trpc_agent_sdk/agents/core/_tools_processor.py +++ b/trpc_agent_sdk/agents/core/_tools_processor.py @@ -136,6 +136,8 @@ async def __invoke_tools( ) -> Event: # Find the appropriate tool tool = await self._find_tool(tool_call, resolved_tools) + if tool is None: + tool = await self._find_unresolved_tool(tool_call, context) result_event = None if not tool: logger.warning("No tool found for tool call: %s", tool_call.name) @@ -322,7 +324,10 @@ async def find_tool(self, context: InvocationContext, tool_call: FunctionCall) - resolved_tools = await convert_toolunion_to_tool_list(self.tools, context) # Find the tool using the private method - return await self._find_tool(tool_call, resolved_tools) + tool = await self._find_tool(tool_call, resolved_tools) + if tool is not None: + return tool + return await self._find_unresolved_tool(tool_call, context) async def _find_tool(self, tool_call: FunctionCall, resolved_tools: List[BaseTool]) -> Optional[BaseTool]: """Find the appropriate tool for a tool call. @@ -340,6 +345,23 @@ async def _find_tool(self, tool_call: FunctionCall, resolved_tools: List[BaseToo return tool return None + async def _find_unresolved_tool( + self, + tool_call: FunctionCall, + context: InvocationContext, + ) -> Optional[BaseTool]: + """Ask dynamic toolsets for a declaration-free fallback tool.""" + for configured in self.tools: + if not isinstance(configured, BaseToolSet): + continue + get_unresolved_tool = getattr(configured, "get_unresolved_tool", None) + if get_unresolved_tool is None: + continue + fallback = await get_unresolved_tool(tool_call.name, context) + if fallback is not None: + return fallback + return None + async def _execute_tool(self, tool_call: FunctionCall, tool: BaseTool, context: InvocationContext) -> Event: """Execute a callable tool. diff --git a/trpc_agent_sdk/tools/__init__.py b/trpc_agent_sdk/tools/__init__.py index 8443b0c2a..d33e251da 100644 --- a/trpc_agent_sdk/tools/__init__.py +++ b/trpc_agent_sdk/tools/__init__.py @@ -76,6 +76,18 @@ from .task_tools import TaskUpdateTool from .task_tools import get_task_store from .task_tools import render_task_list +from .tool_search import InvocationMode as ToolSearchInvocationMode +from .tool_search import InMemoryToolSearchVectorStore +from .tool_search import MCPToolbox as ToolSearchMCPToolbox +from .tool_search import SemanticToolIndex +from .tool_search import Toolbox as ToolSearchToolbox +from .tool_search import ToolSearchEmbeddingUsage +from .tool_search import ToolSearchOptions +from .tool_search import ToolSearchToolSet +from .tool_search import ToolSearchVectorStore +from .tool_search import TOOL_SEARCH_PLACEHOLDER +from .tool_search import TOOL_SEARCH_TOOL_NAME +from .tool_search import CALL_TOOL_NAME from .goal_tools import GoalOptions from .goal_tools import GoalRecord from .goal_tools import GoalStatus @@ -183,6 +195,18 @@ "get_task_store", "render_task_list", "DEFAULT_TASK_PROMPT", + "InMemoryToolSearchVectorStore", + "ToolSearchInvocationMode", + "ToolSearchMCPToolbox", + "ToolSearchEmbeddingUsage", + "ToolSearchOptions", + "ToolSearchToolbox", + "ToolSearchToolSet", + "ToolSearchVectorStore", + "SemanticToolIndex", + "TOOL_SEARCH_PLACEHOLDER", + "TOOL_SEARCH_TOOL_NAME", + "CALL_TOOL_NAME", "GoalStatus", "GoalRecord", "GoalToolSet", diff --git a/trpc_agent_sdk/tools/tool_search/__init__.py b/trpc_agent_sdk/tools/tool_search/__init__.py new file mode 100644 index 000000000..2e651d382 --- /dev/null +++ b/trpc_agent_sdk/tools/tool_search/__init__.py @@ -0,0 +1,38 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""On-demand discovery and loading for large tool catalogs.""" + +from ._semantic import InMemoryToolSearchVectorStore +from ._semantic import SemanticToolIndex +from ._semantic import ToolSearchVectorStore +from ._toolset import CALL_TOOL_NAME +from ._toolset import TOOL_SEARCH_PLACEHOLDER +from ._toolset import TOOL_SEARCH_TOOL_NAME +from ._toolset import ToolSearchToolSet +from ._types import InvocationMode +from ._types import MCPToolbox +from ._types import Toolbox +from ._types import ToolPermissionFilter +from ._types import ToolSearchDocument +from ._types import ToolSearchEmbeddingUsage +from ._types import ToolSearchOptions + +__all__ = [ + "CALL_TOOL_NAME", + "InMemoryToolSearchVectorStore", + "InvocationMode", + "MCPToolbox", + "SemanticToolIndex", + "TOOL_SEARCH_PLACEHOLDER", + "TOOL_SEARCH_TOOL_NAME", + "Toolbox", + "ToolPermissionFilter", + "ToolSearchDocument", + "ToolSearchEmbeddingUsage", + "ToolSearchOptions", + "ToolSearchToolSet", + "ToolSearchVectorStore", +] diff --git a/trpc_agent_sdk/tools/tool_search/_semantic.py b/trpc_agent_sdk/tools/tool_search/_semantic.py new file mode 100644 index 000000000..e445c17fd --- /dev/null +++ b/trpc_agent_sdk/tools/tool_search/_semantic.py @@ -0,0 +1,246 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Optional semantic index for deferred tools.""" + +from __future__ import annotations + +import asyncio +import inspect +import math +import threading +from collections.abc import Mapping +from typing import Any +from typing import Callable +from typing import Optional +from typing import Protocol +from typing import Sequence + +from ._types import ToolSearchDocument +from ._types import ToolSearchEmbeddingUsage + + +class ToolSearchVectorStore(Protocol): + """Storage contract used by :class:`SemanticToolIndex`. + + Methods may be synchronous or asynchronous. + """ + + def upsert(self, name: str, vector: Sequence[float]) -> Any: + """Insert or replace one named vector.""" + + def delete(self, names: Sequence[str]) -> Any: + """Delete vectors by name.""" + + def search( + self, + query_vector: Sequence[float], + candidate_names: Sequence[str], + limit: int, + ) -> Any: + """Return candidate names ordered by descending similarity.""" + + +class InMemoryToolSearchVectorStore: + """Default process-local vector store.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._vectors: dict[str, list[float]] = {} + + def upsert(self, name: str, vector: Sequence[float]) -> None: + with self._lock: + self._vectors[name] = [float(value) for value in vector] + + def delete(self, names: Sequence[str]) -> None: + with self._lock: + for name in names: + self._vectors.pop(name, None) + + def search( + self, + query_vector: Sequence[float], + candidate_names: Sequence[str], + limit: int, + ) -> list[str]: + allowed = set(candidate_names) + with self._lock: + scored = [( + _cosine_similarity(query_vector, vector), + name, + ) for name, vector in self._vectors.items() if name in allowed] + scored.sort(key=lambda item: (-item[0], item[1])) + return [name for _, name in scored[:limit]] + + +class SemanticToolIndex: + """Rank tool declarations with an injected embedding implementation. + + The embedder may implement LangChain-style ``aembed_documents`` / + ``aembed_query`` methods or their synchronous ``embed_*`` counterparts. + No embedding dependency is required by the core package. + """ + + def __init__( + self, + embedder: Any, + vector_store: Optional[ToolSearchVectorStore] = None, + ) -> None: + if embedder is None: + raise ValueError("embedder must not be None") + self._embedder = embedder + self._vector_store = vector_store or InMemoryToolSearchVectorStore() + self._lock = asyncio.Lock() + self._fingerprints: dict[str, str] = {} + + async def search( + self, + query: str, + candidates: Sequence[ToolSearchDocument], + limit: int, + verifier: Optional[Callable[[str, str], bool]] = None, + usage: Optional[ToolSearchEmbeddingUsage] = None, + ) -> list[str]: + """Return candidate names ordered by descending cosine similarity.""" + if not query.strip() or not candidates or limit <= 0: + return [] + + async with self._lock: + missing = [item for item in candidates if self._fingerprints.get(item.name) != item.fingerprint] + if missing: + vectors, document_usage = await self._embed_documents([item.text for item in missing]) + if len(vectors) != len(missing): + raise ValueError("embedding result count does not match tool document count", ) + if usage is not None: + usage.add(document_usage) + for item, vector in zip(missing, vectors): + if verifier is not None and not verifier( + item.name, + item.fingerprint, + ): + continue + await _maybe_await(self._vector_store.delete([item.name])) + await _maybe_await(self._vector_store.upsert( + item.name, + [float(value) for value in vector], + )) + self._fingerprints[item.name] = item.fingerprint + + query_vector, query_usage = await self._embed_query(query) + if usage is not None: + usage.add(query_usage) + candidate_names = [ + item.name for item in candidates if self._fingerprints.get(item.name) == item.fingerprint + ] + result = await _maybe_await( + self._vector_store.search( + [float(value) for value in query_vector], + candidate_names, + limit, + )) + allowed = set(candidate_names) + return [str(name) for name in result if str(name) in allowed][:limit] + + async def forget(self, names: Sequence[str]) -> None: + """Remove cached vectors for tools that no longer exist.""" + if not names: + return + async with self._lock: + for name in names: + self._fingerprints.pop(name, None) + await _maybe_await(self._vector_store.delete(names)) + + async def _embed_documents( + self, + texts: list[str], + ) -> tuple[list[list[float]], ToolSearchEmbeddingUsage]: + method = getattr(self._embedder, "aembed_documents", None) + if method is not None: + result = method(texts) + result = await result if inspect.isawaitable(result) else result + else: + method = getattr(self._embedder, "embed_documents", None) + if method is None: + raise TypeError("embedder must implement aembed_documents or embed_documents", ) + if inspect.iscoroutinefunction(method): + result = await method(texts) + else: + result = await asyncio.to_thread(method, texts) + values, usage = _split_embedding_result(result) + return values, usage + + async def _embed_query( + self, + text: str, + ) -> tuple[list[float], ToolSearchEmbeddingUsage]: + method = getattr(self._embedder, "aembed_query", None) + if method is not None: + result = method(text) + result = await result if inspect.isawaitable(result) else result + else: + method = getattr(self._embedder, "embed_query", None) + if method is None: + raise TypeError("embedder must implement aembed_query or embed_query", ) + if inspect.iscoroutinefunction(method): + result = await method(text) + else: + result = await asyncio.to_thread(method, text) + values, usage = _split_embedding_result(result) + return values, usage + + +async def _maybe_await(value: Any) -> Any: + return await value if inspect.isawaitable(value) else value + + +def _split_embedding_result(result: Any, ) -> tuple[Any, ToolSearchEmbeddingUsage]: + if (isinstance(result, tuple) and len(result) == 2 and _looks_like_usage(result[1])): + return result[0], _coerce_usage(result[1]) + return result, ToolSearchEmbeddingUsage() + + +def _looks_like_usage(value: Any) -> bool: + if isinstance(value, Mapping): + return True + return any( + hasattr(value, name) for name in ( + "prompt_tokens", + "total_tokens", + "prompt_token_count", + "total_token_count", + )) + + +def _coerce_usage(value: Any) -> ToolSearchEmbeddingUsage: + + def read(*names: str) -> int: + for name in names: + raw = value.get(name) if isinstance(value, Mapping) else getattr( + value, + name, + None, + ) + if raw is not None: + try: + return int(raw) + except (TypeError, ValueError): + return 0 + return 0 + + return ToolSearchEmbeddingUsage( + prompt_tokens=read("prompt_tokens", "prompt_token_count"), + total_tokens=read("total_tokens", "total_token_count"), + ) + + +def _cosine_similarity(left: Sequence[float], right: Sequence[float]) -> float: + if len(left) != len(right) or not left: + return -1.0 + dot = sum(a * b for a, b in zip(left, right)) + left_norm = math.sqrt(sum(value * value for value in left)) + right_norm = math.sqrt(sum(value * value for value in right)) + if left_norm == 0 or right_norm == 0: + return -1.0 + return dot / (left_norm * right_norm) diff --git a/trpc_agent_sdk/tools/tool_search/_toolset.py b/trpc_agent_sdk/tools/tool_search/_toolset.py new file mode 100644 index 000000000..5ad0c189e --- /dev/null +++ b/trpc_agent_sdk/tools/tool_search/_toolset.py @@ -0,0 +1,1403 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Deferred tool discovery and loading.""" + +from __future__ import annotations + +import asyncio +import hashlib +import re +import threading +from dataclasses import dataclass +from typing import Any +from typing import Optional +from typing import Sequence +from typing_extensions import override + +from trpc_agent_sdk.abc import ToolSetABC as BaseToolSet +from trpc_agent_sdk.context import InvocationContext +from trpc_agent_sdk.log import logger +from trpc_agent_sdk.models import LlmRequest +from trpc_agent_sdk.types import FunctionDeclaration +from trpc_agent_sdk.types import Schema +from trpc_agent_sdk.types import Type + +from .._base_tool import BaseTool +from .._long_running_tool import LongRunningFunctionTool +from ._semantic import SemanticToolIndex +from ._types import InvocationMode +from ._types import MCPToolbox +from ._types import Toolbox +from ._types import ToolSearchDocument +from ._types import ToolSearchEmbeddingUsage +from ._types import ToolSearchOptions + +TOOL_SEARCH_PLACEHOLDER = "{deferred_tools_section}" +TOOL_SEARCH_TOOL_NAME = "tool_search" +CALL_TOOL_NAME = "call_tool" + +_DEFAULT_NAMESPACE = "" +_MAX_REQUEST_RESULTS = 10 +_EXACT_NAME_SCORE = 1000 +_QUERY_SEPARATOR = re.compile(r"[^\w\-+]+", re.UNICODE) +_CAMEL_BOUNDARY = re.compile(r"(?<=[a-z0-9])(?=[A-Z])") + +_SEARCH_DESCRIPTION = """Load deferred tools from the tool catalog so they become callable. +After loading a tool, call it directly and do not search for it again. + +- Known name: copy the exact name shown in the catalog into tool_names. +- A namespace is only a search filter; never prepend it to a tool name + (use "get_forecast", not "weather.get_forecast"). +- Unknown name: use queries and the matching namespace. +- Leave namespace empty only when no catalog domain fits.""" + +_DISPATCH_SEARCH_DESCRIPTION = """Load deferred tools from the tool catalog, then invoke them +with call_tool. Search results include input_schema for constructing call_tool params. + +- Known name: copy the exact name shown in the catalog into tool_names. +- A namespace is only a search filter; never prepend it to a tool name + (use "get_forecast", not "weather.get_forecast"). +- Unknown name: use queries and the matching namespace. +- Leave namespace empty only when no catalog domain fits.""" + +_CALL_DESCRIPTION = """Invoke a deferred tool previously loaded by tool_search. +Use the exact tool_name returned by tool_search and params matching input_schema.""" + + +@dataclass(frozen=True) +class _ToolEntry: + name: str + description: str + namespace: Optional[str] + tool: BaseTool + deferred: bool + declaration: FunctionDeclaration + + +@dataclass(frozen=True) +class _Snapshot: + entries: dict[str, _ToolEntry] + canonical_names: dict[str, str] + deferred_names: frozenset[str] + namespace_tools: dict[str, frozenset[str]] + namespace_descriptions: dict[str, str] + namespace_order: tuple[str, ...] + + +class _RenamedTool(BaseTool): + """Expose a dynamic tool under an MCP-style globally unique name.""" + + def __init__(self, original: BaseTool, name: str) -> None: + BaseTool.__init__( + self, + name=name, + description=original.description, + filters=list(original.filters), + ) + self._original = original + + @property + @override + def is_streaming(self) -> bool: + return self._original.is_streaming + + @property + @override + def is_progress_streaming(self) -> bool: + return self._original.is_progress_streaming + + @property + @override + def api_variant(self) -> str: + return self._original.api_variant + + @property + def skip_summarization(self) -> bool: + """Preserve execution behavior of wrapped dynamic tools.""" + return bool(getattr(self._original, "skip_summarization", False)) + + @override + def _get_declaration(self) -> Optional[FunctionDeclaration]: + declaration = self._original._get_declaration() # pylint: disable=protected-access + if declaration is None: + return None + return declaration.model_copy( + update={"name": self.name}, + deep=True, + ) + + @override + async def _run_async_impl( + self, + *, + tool_context: InvocationContext, + args: dict[str, Any], + ) -> Any: + return await self._original._run_async_impl( # pylint: disable=protected-access + tool_context=tool_context, + args=args, + ) + + async def run_streaming( + self, + *, + tool_context: InvocationContext, + args: dict[str, Any], + ): + run_streaming = getattr(self._original, "run_streaming", None) + if run_streaming is None: + raise TypeError(f"Tool {self._original.name!r} does not support progress streaming", ) + async for value in run_streaming( + tool_context=tool_context, + args=args, + ): + yield value + + +class _RenamedLongRunningTool(_RenamedTool, LongRunningFunctionTool): + """Preserve the framework's long-running marker after namespace.""" + + +class _ToolSearchTool(BaseTool): + + def __init__( + self, + owner: "ToolSearchToolSet", + snapshot: _Snapshot, + allowed: Optional[dict[str, bool]], + catalog: str, + ) -> None: + description = owner._search_description + if owner.options.catalog_in_description and catalog: + description = f"{description}\n\n{catalog}" + super().__init__( + name=TOOL_SEARCH_TOOL_NAME, + description=description, + ) + self._owner = owner + self._snapshot = snapshot + self._allowed = allowed + self._catalog = catalog + + @override + def _get_declaration(self) -> FunctionDeclaration: + string_array = Schema( + type=Type.ARRAY, + items=Schema(type=Type.STRING), + ) + return FunctionDeclaration( + name=self.name, + description=self.description, + parameters=Schema( + type=Type.OBJECT, + properties={ + "tool_names": + Schema( + type=Type.ARRAY, + items=Schema(type=Type.STRING), + description=("Exact names copied from the catalog. Namespace is " + "not part of a tool name: use 'get_forecast', not " + "'weather.get_forecast'."), + ), + "queries": + string_array.model_copy(update={ + "description": ("Keyword or semantic queries. Prefix a keyword " + "with '+' to require it in keyword mode.") + }, ), + "namespace": + Schema( + type=Type.STRING, + description=("Toolbox namespace. Leave empty to search " + "all deferred tools."), + ), + "max_results": + Schema( + type=Type.INTEGER, + description="Maximum loaded matches (request cap: 10).", + ), + }, + ), + ) + + @override + async def process_request( + self, + *, + tool_context: InvocationContext, + llm_request: LlmRequest, + ) -> None: + self._owner._reset_embedding_usage(tool_context) + await super().process_request( + tool_context=tool_context, + llm_request=llm_request, + ) + if self._owner.options.catalog_in_description: + _replace_catalog_placeholder(llm_request, "") + else: + _replace_catalog_placeholder(llm_request, self._catalog) + + @override + async def _run_async_impl( + self, + *, + tool_context: InvocationContext, + args: dict[str, Any], + ) -> Any: + return await self._owner._search( + tool_context, + self._snapshot, + self._allowed, + args, + ) + + +class _CallTool(BaseTool): + + def __init__( + self, + owner: "ToolSearchToolSet", + snapshot: _Snapshot, + allowed: Optional[dict[str, bool]], + ) -> None: + super().__init__( + name=CALL_TOOL_NAME, + description=_CALL_DESCRIPTION, + ) + self._owner = owner + self._snapshot = snapshot + self._allowed = allowed + + @override + def _get_declaration(self) -> FunctionDeclaration: + return FunctionDeclaration( + name=self.name, + description=self.description, + parameters=Schema( + type=Type.OBJECT, + required=["tool_name", "params"], + properties={ + "tool_name": + Schema( + type=Type.STRING, + description="Exact deferred tool name returned by tool_search.", + ), + "params": + Schema( + type=Type.OBJECT, + description="Arguments matching the returned input_schema.", + additional_properties=True, + ), + }, + ), + ) + + @override + async def _run_async_impl( + self, + *, + tool_context: InvocationContext, + args: dict[str, Any], + ) -> Any: + tool_name = str(args.get("tool_name", "")).strip() + params = args.get("params") + if not tool_name: + return { + "status": "error", + "message": "tool_name is required.", + } + if not isinstance(params, dict): + return { + "status": "error", + "message": "params must be an object.", + } + entry = _entry_by_name(self._snapshot, tool_name) + if entry is None: + return { + "status": "error", + "message": f"Unknown tool {tool_name!r}. Use tool_search to find it.", + } + if entry.deferred: + if self._allowed is not None and not self._allowed.get(entry.name, False): + return { + "status": "error", + "message": f"You do not have permission to use {entry.name!r}.", + } + loaded = self._owner._load_discovered(tool_context) + if entry.name not in loaded: + return { + "status": + "error", + "message": (f"Tool {entry.name!r} is not loaded yet. Call tool_search " + f"with tool_names=[{entry.name!r}] first."), + } + if entry.tool.is_progress_streaming: + chunks: list[Any] = [] + async for value in entry.tool.run_streaming( + tool_context=tool_context, + args=params, + ): + chunks.append(value) + return {"chunks": chunks} + return await entry.tool.run_async( + tool_context=tool_context, + args=params, + ) + + +class _UnloadedTool(BaseTool): + """Declaration-free guard for a Native call made before discovery.""" + + def __init__(self, name: str) -> None: + super().__init__( + name=name, + description="Deferred tool loading guard.", + ) + + @override + async def _run_async_impl( + self, + *, + tool_context: InvocationContext, + args: dict[str, Any], + ) -> str: + del tool_context, args + return (f"Tool {self.name!r} is not loaded yet. Call tool_search " + f"with tool_names=[{self.name!r}] first, then retry this call.") + + +class ToolSearchToolSet(BaseToolSet): + """Expose a large tool catalog through on-demand discovery. + + Preset tools are always visible. Deferred tools are searchable but become + visible only after ``tool_search`` records them in session state. Dispatch + mode keeps the model-facing surface fixed at ``tool_search`` and + ``call_tool``. + """ + + def __init__( + self, + *, + preset_tools: Optional[Sequence[BaseTool]] = None, + deferred_tools: Optional[Sequence[BaseTool]] = None, + toolboxes: Optional[Sequence[Toolbox]] = None, + mcp_toolboxes: Optional[Sequence[MCPToolbox]] = None, + options: Optional[ToolSearchOptions] = None, + semantic_index: Optional[SemanticToolIndex] = None, + name: str = "tool_search_tools", + ) -> None: + super().__init__(name=name) + self.options = options or ToolSearchOptions() + if self.options.max_results <= 0: + raise ValueError("max_results must be greater than zero") + self._semantic_index = semantic_index + self._preset_tools = tuple(preset_tools or ()) + self._deferred_tools = tuple(deferred_tools or ()) + self._toolboxes = tuple(toolboxes or ()) + self._mcp_toolboxes = tuple(mcp_toolboxes or ()) + self._normalize_configuration() + self._dynamic_cache: dict[str, tuple[BaseTool, ...]] = {} + self._dynamic_fingerprint_cache: dict[str, dict[str, str]] = {} + self._dynamic_tool_names: set[str] = set() + self._dynamic_cache_lock = threading.Lock() + self._usage_lock = threading.Lock() + self._usage_metadata_key = f"_tool_search_embedding_usage:{self.name}:{id(self)}" + self._dynamic_refresh_locks = {box.server_name.strip(): asyncio.Lock() for box in self._mcp_toolboxes} + + @property + def _search_description(self) -> str: + if self.options.invocation_mode == InvocationMode.DISPATCH: + return _DISPATCH_SEARCH_DESCRIPTION + return _SEARCH_DESCRIPTION + + @override + async def get_tools( + self, + invocation_context: Optional[InvocationContext] = None, + ) -> list[BaseTool]: + snapshot = await self._build_snapshot(invocation_context) + allowed = self._permissions(invocation_context, snapshot) + catalog = _render_catalog(snapshot, allowed, self.options.invocation_mode) + tools: list[BaseTool] = [entry.tool for entry in snapshot.entries.values() if not entry.deferred] + tools.append(_ToolSearchTool(self, snapshot, allowed, catalog)) + + if self.options.invocation_mode == InvocationMode.DISPATCH: + tools.append(_CallTool(self, snapshot, allowed)) + return _deduplicate_tools(tools) + + if invocation_context is not None: + for name in self._load_discovered(invocation_context): + entry = snapshot.entries.get(name) + if entry is None or not entry.deferred: + continue + if allowed is not None and not allowed.get(name, False): + continue + tools.append(entry.tool) + return _deduplicate_tools(tools) + + @override + async def get_unresolved_tool( + self, + name: str, + invocation_context: Optional[InvocationContext] = None, + ) -> Optional[BaseTool]: + if (self.options.invocation_mode != InvocationMode.NATIVE or invocation_context is None): + return None + snapshot = await self._build_snapshot(invocation_context) + entry = _entry_by_name(snapshot, name) + if entry is None: + namespace = _mcp_namespace(name) + if namespace is not None and any(box.server_name.strip() == namespace for box in self._mcp_toolboxes): + return _UnloadedTool(name) + return None + if not entry.deferred: + return None + allowed = self._permissions(invocation_context, snapshot) + if allowed is not None and not allowed.get(entry.name, False): + return None + if entry.name in self._load_discovered(invocation_context): + return None + return _UnloadedTool(entry.name) + + @override + async def close(self) -> None: + closed: set[int] = set() + for box in self._mcp_toolboxes: + if id(box.toolset) in closed: + continue + closed.add(id(box.toolset)) + await box.toolset.close() + + async def _build_snapshot( + self, + ctx: Optional[InvocationContext], + ) -> _Snapshot: + entries: dict[str, _ToolEntry] = {} + namespace_tools: dict[str, set[str]] = {} + descriptions: dict[str, str] = {} + order: list[str] = [] + + for tool in self._preset_tools: + _add_entry(entries, tool, namespace=None, deferred=False) + + if self._deferred_tools: + namespace_tools[_DEFAULT_NAMESPACE] = set() + order.append(_DEFAULT_NAMESPACE) + for tool in self._deferred_tools: + entry = _add_entry( + entries, + tool, + namespace=_DEFAULT_NAMESPACE, + deferred=True, + ) + if entry is not None: + namespace_tools[_DEFAULT_NAMESPACE].add(entry.name) + + for box in self._toolboxes: + namespace = box.name.strip() + descriptions[namespace] = box.description.strip() + namespace_tools[namespace] = set() + order.append(namespace) + for tool in box.tools: + entry = _add_entry( + entries, + tool, + namespace=namespace, + deferred=True, + ) + if entry is not None: + namespace_tools[namespace].add(entry.name) + + for box in self._mcp_toolboxes: + namespace = box.server_name.strip() + descriptions[namespace] = box.description.strip() + namespace_tools[namespace] = set() + order.append(namespace) + cache_key = f"mcp:{namespace}" + forget_names: tuple[str, ...] = () + async with self._dynamic_refresh_locks[namespace]: + try: + dynamic = await box.toolset.get_tools(ctx) + except Exception: # pylint: disable=broad-except + with self._dynamic_cache_lock: + wrapped = self._dynamic_cache.get(cache_key, ()) + if not wrapped: + raise + logger.warning( + "Dynamic tool listing failed for %s; using the last " + "nonempty snapshot", + namespace, + exc_info=True, + ) + else: + if dynamic: + wrapped = self._normalize_dynamic_tools( + dynamic, + namespace, + ) + fresh_fingerprints = {tool.name: _tool_fingerprint(tool) for tool in wrapped} + with self._dynamic_cache_lock: + previous_fingerprints = self._dynamic_fingerprint_cache.get( + cache_key, + {}, + ) + changed = { + name + for name, fingerprint in fresh_fingerprints.items() + if (name in previous_fingerprints and previous_fingerprints[name] != fingerprint) + } + removed = set(previous_fingerprints) - set(fresh_fingerprints) + self._dynamic_cache[cache_key] = wrapped + self._dynamic_fingerprint_cache[cache_key] = fresh_fingerprints + self._dynamic_tool_names.update(previous_fingerprints) + self._dynamic_tool_names.update(fresh_fingerprints) + forget_names = tuple(sorted(changed | removed)) + else: + with self._dynamic_cache_lock: + wrapped = self._dynamic_cache.get(cache_key, ()) + if self._semantic_index is not None and forget_names: + await self._semantic_index.forget(forget_names) + for tool in wrapped: + entry = _add_entry( + entries, + tool, + namespace=namespace, + deferred=True, + ) + if entry is not None: + namespace_tools[namespace].add(entry.name) + + return _Snapshot( + entries=entries, + canonical_names={name.lower(): name + for name in entries}, + deferred_names=frozenset(name for name, entry in entries.items() if entry.deferred), + namespace_tools={ + name: frozenset(tools) + for name, tools in namespace_tools.items() + }, + namespace_descriptions=descriptions, + namespace_order=tuple(order), + ) + + def _normalize_configuration(self) -> None: + reserved = {TOOL_SEARCH_TOOL_NAME, CALL_TOOL_NAME} + strict = self.options.strict_validation + seen_tools: dict[str, tuple[BaseTool, str, str]] = {} + + def accept_tool( + tool: Any, + owner: str, + ) -> bool: + if not isinstance(tool, BaseTool): + if strict: + raise ValueError(f"Invalid tool in {owner}: {tool!r}") + logger.warning( + "Skipping invalid tool in namespace %r: %r", + owner, + tool, + ) + return False + if tool.name in reserved: + raise ValueError(f"Tool name {tool.name!r} is reserved") + declaration = tool._get_declaration() # pylint: disable=protected-access + if declaration is None: + logger.warning( + "Skipping tool %r in namespace %r because it has no declaration", + tool.name, + owner, + ) + return False + fingerprint = _tool_fingerprint(tool) + previous = seen_tools.get(tool.name) + if previous is None: + seen_tools[tool.name] = (tool, fingerprint, owner) + return True + if strict: + raise ValueError(f"Duplicate tool name: {tool.name!r}") + _, previous_fingerprint, previous_owner = previous + if previous_fingerprint != fingerprint: + raise ValueError( + f"Conflicting declarations for tool {tool.name!r} " + f"in {previous_owner!r} and {owner!r}", ) + logger.warning( + "Tool %r is already registered in namespace %r; keeping " + "the first owner and ignoring namespace %r", + tool.name, + previous_owner, + owner, + ) + return False + + preset_tools = tuple(tool for tool in self._preset_tools if accept_tool(tool, "preset")) + deferred_tools = tuple(tool for tool in self._deferred_tools if accept_tool(tool, _DEFAULT_NAMESPACE)) + + toolbox_data: dict[str, tuple[str, list[BaseTool]]] = {} + toolbox_order: list[str] = [] + for box in self._toolboxes: + namespace = box.name.strip() + if not namespace: + if strict: + raise ValueError("Toolbox name must not be empty") + logger.warning( + "Skipping toolbox with empty name: description=%r", + box.description, + ) + continue + existing = toolbox_data.get(namespace) + if existing is not None and strict: + raise ValueError(f"Duplicate toolbox namespace: {namespace!r}", ) + if existing is None: + description = box.description.strip() + tools: list[BaseTool] = [] + toolbox_data[namespace] = (description, tools) + toolbox_order.append(namespace) + else: + description, tools = existing + incoming_description = box.description.strip() + if not description and incoming_description: + toolbox_data[namespace] = ( + incoming_description, + tools, + ) + logger.warning( + "Merging duplicate toolbox namespace %r", + namespace, + ) + for tool in box.tools: + if accept_tool(tool, namespace): + tools.append(tool) + + normalized_toolboxes = tuple( + Toolbox( + name=namespace, + description=toolbox_data[namespace][0], + tools=tuple(toolbox_data[namespace][1]), + ) for namespace in toolbox_order) + + namespaces = set(toolbox_order) + normalized_mcp: list[MCPToolbox] = [] + for box in self._mcp_toolboxes: + namespace = box.server_name.strip() + if not namespace: + if strict: + raise ValueError("MCP server_name must not be empty") + logger.warning( + "Skipping MCP toolbox with empty server_name: description=%r", + box.description, + ) + continue + if box.toolset is None: + if strict: + raise ValueError(f"MCP toolbox {namespace!r} must have a toolset", ) + logger.warning( + "Skipping MCP toolbox %r because toolset is None", + namespace, + ) + continue + if namespace in namespaces: + if strict: + raise ValueError(f"Duplicate toolbox namespace: {namespace!r}", ) + logger.warning( + "Skipping MCP toolbox %r because its namespace is " + "already registered", + namespace, + ) + continue + namespaces.add(namespace) + normalized_mcp.append( + MCPToolbox( + server_name=namespace, + description=box.description.strip(), + toolset=box.toolset, + )) + + self._preset_tools = preset_tools + self._deferred_tools = deferred_tools + self._toolboxes = normalized_toolboxes + self._mcp_toolboxes = tuple(normalized_mcp) + + def _normalize_dynamic_tools( + self, + tools: Sequence[Any], + namespace: str, + ) -> tuple[BaseTool, ...]: + wrapped_by_name: dict[str, BaseTool] = {} + fingerprints: dict[str, str] = {} + for tool in tools: + if not isinstance(tool, BaseTool): + if self.options.strict_validation: + raise ValueError(f"Invalid dynamic tool in {namespace!r}: {tool!r}", ) + logger.warning( + "Skipping invalid dynamic tool in namespace %r: %r", + namespace, + tool, + ) + continue + declaration = tool._get_declaration() # pylint: disable=protected-access + if declaration is None: + logger.warning( + "Skipping dynamic tool %r in namespace %r because it " + "has no declaration", + tool.name, + namespace, + ) + continue + wrapped = _rename_dynamic_tool( + tool, + f"mcp__{namespace}__{tool.name}", + ) + fingerprint = _tool_fingerprint(wrapped) + previous = wrapped_by_name.get(wrapped.name) + if previous is None: + wrapped_by_name[wrapped.name] = wrapped + fingerprints[wrapped.name] = fingerprint + continue + if self.options.strict_validation: + raise ValueError(f"Duplicate dynamic tool name: {wrapped.name!r}", ) + if fingerprints[wrapped.name] != fingerprint: + raise ValueError(f"Conflicting declarations for dynamic tool " + f"{wrapped.name!r}", ) + logger.warning( + "Dynamic tool %r is listed more than once by namespace %r; " + "keeping the first entry", + wrapped.name, + namespace, + ) + return tuple(wrapped_by_name.values()) + + def _permissions( + self, + ctx: Optional[InvocationContext], + snapshot: _Snapshot, + ) -> Optional[dict[str, bool]]: + permission_filter = self.options.permission_filter + if permission_filter is None or ctx is None: + return None + names = sorted(snapshot.deferred_names) + result = permission_filter(ctx, names) + return {name: bool(result.get(name, False)) for name in names} + + async def _search( + self, + ctx: InvocationContext, + snapshot: _Snapshot, + allowed: Optional[dict[str, bool]], + args: dict[str, Any], + ) -> dict[str, Any]: + tool_names = _string_list(args.get("tool_names")) + queries = _string_list(args.get("queries")) + namespace = str(args.get("namespace", "")).strip() + requested_max = _positive_int(args.get("max_results")) + max_results = min(requested_max, _MAX_REQUEST_RESULTS) if requested_max else self.options.max_results + + if tool_names: + selected = _select_by_name(snapshot, tool_names, allowed) + overflow: list[str] = [] + else: + error = _validate_namespace(snapshot, namespace) + if error is not None: + return error + if namespace and not queries: + ranked = _namespace_names(snapshot, namespace, allowed) + selected, overflow = _split_results(ranked, max_results) + elif queries: + try: + selected, overflow = await self._search_queries( + ctx, + snapshot, + allowed, + namespace, + queries, + max_results, + ) + except Exception as exc: # pylint: disable=broad-except + if not self.options.embedding_fail_open or self._semantic_index is None: + raise + logger.warning( + "Semantic tool search failed; falling back to keyword search: %s", + exc, + ) + ranked = _keyword_search( + snapshot, + namespace, + queries, + allowed, + ) + selected, overflow = _split_results(ranked, max_results) + else: + selected, overflow = [], [] + + previously_loaded = set(self._load_discovered(ctx)) + deferred_selected = [name for name in selected if name in snapshot.deferred_names] + self._append_discovered(ctx, deferred_selected) + return _format_result( + snapshot, + selected, + overflow, + previously_loaded, + self.options.invocation_mode, + ) + + async def _search_queries( + self, + ctx: InvocationContext, + snapshot: _Snapshot, + allowed: Optional[dict[str, bool]], + namespace: str, + queries: list[str], + max_results: int, + ) -> tuple[list[str], list[str]]: + if self._semantic_index is None: + ranked = _keyword_search( + snapshot, + namespace, + queries, + allowed, + ) + return _split_results(ranked, max_results) + + candidates = _candidate_names( + snapshot, + namespace, + allowed, + ) + documents = [_document_for(snapshot.entries[name]) for name in candidates] + best_rank: dict[str, int] = {} + usage = ToolSearchEmbeddingUsage() + try: + for query in queries: + ranked = await self._semantic_index.search( + query, + documents, + len(documents), + verifier=self._semantic_candidate_is_current, + usage=usage, + ) + for rank, name in enumerate(ranked): + if name not in best_rank or rank < best_rank[name]: + best_rank[name] = rank + finally: + self._record_embedding_usage(ctx, usage) + merged = sorted(best_rank, key=lambda name: (best_rank[name], name)) + return _split_results(merged, max_results) + + def get_embedding_usage( + self, + ctx: InvocationContext, + *, + reset: bool = False, + ) -> Optional[ToolSearchEmbeddingUsage]: + """Return embedding usage accumulated for this invocation context.""" + if self._semantic_index is None: + return None + with self._usage_lock: + usage = ctx.agent_context.get_metadata(self._usage_metadata_key) + if not isinstance(usage, ToolSearchEmbeddingUsage): + usage = ToolSearchEmbeddingUsage() + ctx.agent_context.with_metadata(self._usage_metadata_key, usage) + snapshot = usage.copy() + if reset: + ctx.agent_context.with_metadata( + self._usage_metadata_key, + ToolSearchEmbeddingUsage(), + ) + return snapshot + + def _record_embedding_usage( + self, + ctx: InvocationContext, + usage: ToolSearchEmbeddingUsage, + ) -> None: + if usage.prompt_tokens == 0 and usage.total_tokens == 0: + return + with self._usage_lock: + accumulated = ctx.agent_context.get_metadata(self._usage_metadata_key) + if not isinstance(accumulated, ToolSearchEmbeddingUsage): + accumulated = ToolSearchEmbeddingUsage() + ctx.agent_context.with_metadata( + self._usage_metadata_key, + accumulated, + ) + accumulated.add(usage) + + def _reset_embedding_usage( + self, + ctx: InvocationContext, + ) -> None: + if self._semantic_index is None: + return + with self._usage_lock: + ctx.agent_context.with_metadata( + self._usage_metadata_key, + ToolSearchEmbeddingUsage(), + ) + + def _semantic_candidate_is_current( + self, + name: str, + fingerprint: str, + ) -> bool: + with self._dynamic_cache_lock: + if name not in self._dynamic_tool_names: + return True + return any( + fingerprints.get(name) == fingerprint for fingerprints in self._dynamic_fingerprint_cache.values()) + + def _state_prefix(self, ctx: InvocationContext) -> str: + scope = ctx.branch or ctx.agent_name + return (f"{self.options.state_key_prefix}:{self.name}:{scope}:" + "discovered_tools:") + + def _load_discovered(self, ctx: InvocationContext) -> list[str]: + prefix = self._state_prefix(ctx) + return sorted(key[len(prefix):] for key, value in ctx.state.to_dict().items() + if key.startswith(prefix) and value is True) + + def _append_discovered( + self, + ctx: InvocationContext, + names: Sequence[str], + ) -> None: + prefix = self._state_prefix(ctx) + # One state key per tool makes concurrent Event deltas commutative: + # persisting one search cannot overwrite tools loaded by another. + for name in names: + ctx.state[f"{prefix}{name}"] = True + + +def _rename_dynamic_tool(tool: BaseTool, name: str) -> BaseTool: + if isinstance(tool, LongRunningFunctionTool): + return _RenamedLongRunningTool(tool, name) + return _RenamedTool(tool, name) + + +def _add_entry( + entries: dict[str, _ToolEntry], + tool: BaseTool, + *, + namespace: Optional[str], + deferred: bool, +) -> Optional[_ToolEntry]: + declaration = tool._get_declaration() # pylint: disable=protected-access + if declaration is None: + logger.warning("Skipping tool %r because it has no declaration", tool.name) + return None + if tool.name in entries: + raise ValueError(f"Duplicate tool name: {tool.name!r}") + entry = _ToolEntry( + name=tool.name, + description=declaration.description or tool.description or "", + namespace=namespace, + tool=tool, + deferred=deferred, + declaration=declaration, + ) + entries[tool.name] = entry + return entry + + +def _entry_by_name(snapshot: _Snapshot, name: str) -> Optional[_ToolEntry]: + canonical = snapshot.canonical_names.get(name.lower()) + return snapshot.entries.get(canonical) if canonical else None + + +def _mcp_namespace(name: str) -> Optional[str]: + if not name.startswith("mcp__"): + return None + namespace, separator, tool_name = name[len("mcp__"):].partition("__") + if not separator or not namespace or not tool_name: + return None + return namespace + + +def _string_list(value: Any) -> list[str]: + if not isinstance(value, (list, tuple)): + return [] + return [str(item).strip() for item in value if str(item).strip()] + + +def _positive_int(value: Any) -> int: + try: + parsed = int(value) + except (TypeError, ValueError): + return 0 + return parsed if parsed > 0 else 0 + + +def _select_by_name( + snapshot: _Snapshot, + names: Sequence[str], + allowed: Optional[dict[str, bool]], +) -> list[str]: + selected: list[str] = [] + seen: set[str] = set() + for name in names: + entry = _entry_by_name(snapshot, name) + if entry is None: + # Models sometimes infer ``namespace.tool_name`` from a grouped + # catalog even though namespace is only a search filter. Accept + # that form only when both parts match the real catalog entry. + namespace, separator, candidate = name.rpartition(".") + if separator and namespace and candidate: + qualified_entry = _entry_by_name(snapshot, candidate) + if (qualified_entry is not None and qualified_entry.namespace is not None + and qualified_entry.namespace.lower() == namespace.lower()): + entry = qualified_entry + if entry is None or entry.name in seen: + continue + if entry.deferred and allowed is not None and not allowed.get(entry.name, False): + continue + selected.append(entry.name) + seen.add(entry.name) + return selected + + +def _validate_namespace( + snapshot: _Snapshot, + namespace: str, +) -> Optional[dict[str, Any]]: + if not namespace or namespace in snapshot.namespace_tools: + return None + return { + "status": + "unknown_namespace", + "message": + f"Namespace {namespace!r} is not in the tool catalog.", + "namespace": + namespace, + "catalog": [{ + "name": name, + "description": snapshot.namespace_descriptions.get(name, ""), + } for name in snapshot.namespace_order if name], + } + + +def _namespace_names( + snapshot: _Snapshot, + namespace: str, + allowed: Optional[dict[str, bool]], +) -> list[str]: + return sorted(name for name in snapshot.namespace_tools.get(namespace, ()) + if allowed is None or allowed.get(name, False)) + + +def _candidate_names( + snapshot: _Snapshot, + namespace: str, + allowed: Optional[dict[str, bool]], +) -> list[str]: + names = (snapshot.namespace_tools.get(namespace, ()) if namespace else snapshot.deferred_names) + return sorted(name for name in names if allowed is None or allowed.get(name, False)) + + +def _keyword_search( + snapshot: _Snapshot, + namespace: str, + queries: Sequence[str], + allowed: Optional[dict[str, bool]], +) -> list[str]: + candidates = _candidate_names(snapshot, namespace, allowed) + best: dict[str, int] = {} + namespace_bias = (_namespace_bias(snapshot, queries) if not namespace else {}) + for query in queries: + query = query.strip() + if not query: + continue + exact = snapshot.canonical_names.get(query.lower()) + if exact in candidates: + best[exact] = max(best.get(exact, 0), _EXACT_NAME_SCORE) + continue + terms = _tokenize(query) + required = [term[1:] for term in terms if term.startswith("+")] + optional = [term for term in terms if not term.startswith("+")] + scoring_terms = [part for term in required for part in _split_compound(term)] + optional + for name in candidates: + entry = snapshot.entries[name] + metadata = _search_metadata(entry) + if required and not all( + all(_has_term(metadata, part) for part in _split_compound(term)) for term in required): + continue + score = _score_metadata(metadata, scoring_terms) + if score > 0: + score += namespace_bias.get(entry.namespace or "", 0) + best[name] = max(best.get(name, 0), score) + return sorted( + best, + key=lambda name: ( + -best[name], + len(_search_metadata(snapshot.entries[name])["parts"]), + name, + ), + ) + + +def _tokenize(text: str) -> list[str]: + return [part for part in _QUERY_SEPARATOR.split(text.lower()) if part] + + +def _split_compound(term: str) -> list[str]: + parts = [part for part in re.split(r"[_-]+", term.lower()) if part] + return parts or [term.lower()] + + +def _search_metadata(entry: _ToolEntry) -> dict[str, Any]: + expanded = _CAMEL_BOUNDARY.sub(" ", entry.name).replace("_", " ") + parts = [part for part in expanded.lower().split() if part] + return { + "name": entry.name.lower(), + "parts": parts, + "full": " ".join(parts), + "description": entry.description.lower(), + } + + +def _has_term(metadata: dict[str, Any], term: str) -> bool: + term = term.lower() + return term in metadata["parts"] or _text_has_term(metadata["description"], term) + + +def _score_metadata(metadata: dict[str, Any], terms: Sequence[str]) -> int: + score = 0 + for term in terms: + term = term.lower() + if ("_" in term or "-" in term) and term in metadata["name"]: + score += 10 + if term in metadata["parts"]: + score += 10 + if _text_has_term(metadata["description"], term): + score += 3 + if score == 0: + score = sum(3 for term in terms if _text_has_term(metadata["full"], term.lower())) + return score + + +def _text_has_term(text: str, term: str) -> bool: + if term.isascii(): + return re.search( + rf"(? dict[str, int]: + terms = {term.removeprefix("+") for query in queries for term in _tokenize(query) if term.removeprefix("+")} + scores: dict[str, int] = {} + for namespace, description in snapshot.namespace_descriptions.items(): + hits = sum(1 for term in terms if _text_has_term(description.lower(), term)) + if hits: + scores[namespace] = hits + return scores + + +def _split_results( + names: Sequence[str], + max_results: int, +) -> tuple[list[str], list[str]]: + if max_results > 0 and len(names) > max_results: + return list(names[:max_results]), list(names[max_results:]) + return list(names), [] + + +def _document_for(entry: _ToolEntry) -> ToolSearchDocument: + text = _document_text( + entry.name, + entry.description, + entry.declaration, + ) + return ToolSearchDocument( + name=entry.name, + description=entry.description, + namespace=entry.namespace or "", + text=text, + fingerprint=hashlib.sha256(text.encode("utf-8")).hexdigest(), + ) + + +def _tool_fingerprint(tool: BaseTool) -> str: + declaration = tool._get_declaration() # pylint: disable=protected-access + if declaration is None: + return "" + text = _document_text( + tool.name, + declaration.description or tool.description or "", + declaration, + ) + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _document_text( + name: str, + description: str, + declaration: FunctionDeclaration, +) -> str: + declaration_json = declaration.model_dump_json( + exclude_none=True, + by_alias=True, + ) + return (f"Tool: {name}\n" + f"Description: {description}\n" + f"Declaration: {declaration_json}") + + +def _format_result( + snapshot: _Snapshot, + selected: Sequence[str], + overflow: Sequence[str], + previously_loaded: set[str], + mode: InvocationMode, +) -> dict[str, Any]: + tools: list[dict[str, Any]] = [] + any_already_loaded = False + all_already_loaded = bool(selected) + for name in selected: + entry = snapshot.entries[name] + item: dict[str, Any] = { + "name": name, + "description": entry.description, + } + if name in previously_loaded: + item["already_loaded"] = True + any_already_loaded = True + else: + all_already_loaded = False + if mode == InvocationMode.DISPATCH: + item["input_schema"] = _input_schema(entry.declaration) + tools.append(item) + + if not tools: + status = "No matches. Try different queries, namespace, or tool_names." + elif mode == InvocationMode.DISPATCH: + if all_already_loaded: + status = ("All matches were loaded earlier. Invoke them with " + "call_tool using input_schema.") + elif any_already_loaded: + status = ("Tools loaded. Invoke them with call_tool using " + "input_schema; entries with already_loaded=true were " + "loaded earlier.") + else: + status = "Tools loaded. Invoke them with call_tool using input_schema." + else: + if all_already_loaded: + status = "All matches were loaded earlier. Call them directly." + elif any_already_loaded: + status = ("Tools loaded. Entries with already_loaded=true were " + "loaded earlier. Call them directly.") + else: + status = "Tools loaded. Call them directly." + if overflow: + status += (" More matches are listed in additional_candidates; load one " + "with tool_names instead of searching again.") + result: dict[str, Any] = { + "status": status, + "tools": tools, + } + if overflow: + result["additional_candidates"] = list(overflow) + return result + + +def _input_schema(declaration: FunctionDeclaration) -> dict[str, Any]: + if declaration.parameters_json_schema: + return _normalize_json_schema(dict(declaration.parameters_json_schema)) + if declaration.parameters is None: + return {} + return _normalize_json_schema(declaration.parameters.model_dump( + mode="json", + exclude_none=True, + by_alias=True, + )) + + +def _normalize_json_schema(value: Any) -> Any: + """Normalize Google schema enum values to JSON Schema type names.""" + if isinstance(value, dict): + normalized = {} + for key, item in value.items(): + if key == "type" and isinstance(item, str) and item in { + "ARRAY", + "BOOLEAN", + "INTEGER", + "NUMBER", + "OBJECT", + "STRING", + }: + normalized[key] = item.lower() + else: + normalized[key] = _normalize_json_schema(item) + return normalized + if isinstance(value, list): + return [_normalize_json_schema(item) for item in value] + return value + + +def _render_catalog( + snapshot: _Snapshot, + allowed: Optional[dict[str, bool]], + mode: InvocationMode, +) -> str: + rows: list[tuple[str, str, list[str]]] = [] + for namespace in snapshot.namespace_order: + names = sorted(name for name in snapshot.namespace_tools.get(namespace, ()) + if allowed is None or allowed.get(name, False)) + if names: + rows.append(( + namespace, + snapshot.namespace_descriptions.get(namespace, ""), + names, + )) + if not rows: + return "" + if len(rows) == 1 and rows[0][0] == _DEFAULT_NAMESPACE: + return ("\n" + "\n".join(rows[0][2]) + "\n") + + invocation = ("invoke it with call_tool" if mode == InvocationMode.DISPATCH else "call it directly") + lines = [ + "", + ("Deferred tools are not callable until loaded with tool_search; " + f"after loading, {invocation}."), + ] + for namespace, description, names in rows: + label = namespace or "(no namespace)" + domain = f" ({description})" if description else "" + exact_names = ", ".join(f'"{name}"' for name in names) + lines.append(f"- namespace={label}{domain}; exact tool_names=[{exact_names}]") + lines.append("Namespace filters the search scope and must not be prepended to a tool name.") + lines.append("") + return "\n".join(lines) + + +def _replace_catalog_placeholder( + request: LlmRequest, + catalog: str, +) -> None: + if request.config is None: + request.append_instructions([catalog] if catalog else []) + return + current = request.config.system_instruction + if current is not None and TOOL_SEARCH_PLACEHOLDER in str(current): + request.config.system_instruction = str(current).replace( + TOOL_SEARCH_PLACEHOLDER, + catalog, + 1, + ) + elif catalog: + request.append_instructions([catalog]) + + +def _deduplicate_tools(tools: Sequence[BaseTool]) -> list[BaseTool]: + result: list[BaseTool] = [] + seen: set[str] = set() + for tool in tools: + if tool.name in seen: + continue + seen.add(tool.name) + result.append(tool) + return result diff --git a/trpc_agent_sdk/tools/tool_search/_types.py b/trpc_agent_sdk/tools/tool_search/_types.py new file mode 100644 index 000000000..383eed539 --- /dev/null +++ b/trpc_agent_sdk/tools/tool_search/_types.py @@ -0,0 +1,92 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Public types for deferred tool search.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Callable +from typing import Optional +from typing import Sequence + +from trpc_agent_sdk.abc import ToolSetABC as BaseToolSet +from trpc_agent_sdk.context import InvocationContext + +from .._base_tool import BaseTool + + +class InvocationMode(str, Enum): + """How deferred tools are invoked after discovery.""" + + NATIVE = "native" + DISPATCH = "dispatch" + + +@dataclass(frozen=True) +class Toolbox: + """A named group of static deferred tools.""" + + name: str + description: str + tools: Sequence[BaseTool] + + +@dataclass(frozen=True) +class MCPToolbox: + """A dynamically listed toolset exposed under an MCP-style namespace.""" + + server_name: str + description: str + toolset: BaseToolSet + + +@dataclass(frozen=True) +class ToolSearchDocument: + """Searchable representation of one deferred tool.""" + + name: str + description: str + namespace: str + text: str + fingerprint: str + + +@dataclass +class ToolSearchEmbeddingUsage: + """Embedding token usage accumulated by Tool Search.""" + + prompt_tokens: int = 0 + total_tokens: int = 0 + + def add(self, other: "ToolSearchEmbeddingUsage") -> None: + """Merge another usage value into this accumulator.""" + self.prompt_tokens += other.prompt_tokens + self.total_tokens += other.total_tokens + + def copy(self) -> "ToolSearchEmbeddingUsage": + """Return a detached usage snapshot.""" + return ToolSearchEmbeddingUsage( + prompt_tokens=self.prompt_tokens, + total_tokens=self.total_tokens, + ) + + +ToolPermissionFilter = Callable[[InvocationContext, Sequence[str]], dict[str, bool]] +"""Return per-tool permissions for the active invocation.""" + + +@dataclass(frozen=True) +class ToolSearchOptions: + """Configuration for :class:`ToolSearchToolSet`.""" + + max_results: int = 5 + permission_filter: Optional[ToolPermissionFilter] = None + catalog_in_description: bool = False + invocation_mode: InvocationMode = InvocationMode.NATIVE + embedding_fail_open: bool = False + state_key_prefix: str = "tool_search" + strict_validation: bool = True