feat(testing): add fixed tool simulations

This commit is contained in:
Xin Wang
2026-08-26 09:36:37 +08:00
parent 26582bfb87
commit 239b9027ca
16 changed files with 1525 additions and 237 deletions
+41 -4
View File
@@ -22,7 +22,7 @@ from models import (
RuntimeToolBinding,
)
from services.node_specs import copy_graph, graph_references
from sqlalchemy import select
from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession
@@ -135,24 +135,56 @@ def _runtime_tool_ids(
)
def _implicit_startup_tool_names(startup: dict | None) -> list[str]:
"""Return platform client tools used by built-in startup experiences.
A confirmation opening is configured directly on the assistant instead of
through an AssistantToolBinding, but text tests still need to simulate the
underlying client interaction explicitly.
"""
value = startup or {}
opening_mode = value.get("opening_mode") or value.get("openingMode")
opening_message = value.get("opening_message") or value.get("openingMessage")
if opening_mode == "confirmation" and isinstance(opening_message, dict):
return ["show_message"]
return []
async def _tools_for(
session: AsyncSession,
assistant: Assistant,
tool_ids: list[str],
implicit_function_names: list[str] | None = None,
) -> list[RuntimeTool]:
if assistant.type not in {"prompt", "workflow"} or not tool_ids:
function_names = list(dict.fromkeys(implicit_function_names or []))
if assistant.type not in {"prompt", "workflow"} or not (
tool_ids or function_names
):
return []
identity_filters = []
if tool_ids:
identity_filters.append(Tool.id.in_(tool_ids))
if function_names:
identity_filters.append(Tool.function_name.in_(function_names))
rows = (
await session.execute(
select(Tool)
.where(
Tool.id.in_(tool_ids),
or_(*identity_filters),
Tool.status == "active",
)
)
).scalars().all()
tool_by_id = {tool.id: tool for tool in rows}
tool_by_function = {tool.function_name: tool for tool in rows}
tools = [tool_by_id[tool_id] for tool_id in tool_ids if tool_id in tool_by_id]
seen_tool_ids = {tool.id for tool in tools}
for function_name in function_names:
tool = tool_by_function.get(function_name)
if tool is not None and tool.id not in seen_tool_ids:
tools.append(tool)
seen_tool_ids.add(tool.id)
server_ids = {
str(tool.mcp_server_id)
for tool in tools
@@ -268,7 +300,12 @@ async def resolve_runtime_config(
prompt_tool_bindings = await _bound_tools(session, assistant.id)
runtime_tool_ids = _runtime_tool_ids(prompt_tool_bindings, assistant.startup or {})
runtime_tools = await _tools_for(session, assistant, runtime_tool_ids)
runtime_tools = await _tools_for(
session,
assistant,
runtime_tool_ids,
_implicit_startup_tool_names(assistant.startup or {}),
)
return AssistantConfig(
name=assistant.name,