🧭 Overview

This document explains how the Tool Manager organizes, exposes, and executes tools for an LLM-based orchestrator. Code for each capability is embedded in the relevant sections, so you can see how it’s implemented where it matters.

Key capabilities: - Load and filter tools from configuration and the users choices - Respects Tool exclusivity, whether it is enabled by the admin or whether the user chose the tool as a must (forced tools) - Expose tool definitions and prompt enhancements for the LLM - Detects if a tool requests a handoff so it “takes control” (e.g., deep research) - Execute selected non-conflicting tools in parallel with tool-call deduplication and max-call limits - it returns the ToolCallResponses of all the tools to the orchestrator for further rounds with the LLM and additional processing like preparation for referencing or collection of debug info.

🚀 Initialization and Tool Loading

The Tool Manager is responsible for initializing and managing the tools available to the agent. It supports "Internal"-tools but also both MCP tools and A2A sub-agents, treating them as tools that can be called directly. Here's a breakdown of its functionality:

Internal Tools

Internal tools are loaded directly in the python code. Like the web-search tool or the internal-search tool.

Agent-to-Agent Protocol (A2A)

The A2A protocol enables communication between agents. During initialization, the Tool Manager: 1. Loads all sub-agents defined for the A2A (Agent to Agent) protocol. 2. These sub-agents are treated as callable tools, making them callable by the LLM.

MCP Tools

The Tool Manager also integrates MCP tools, which are added to the pool of available tools. These tools can be invoked directly, just like sub-agents, and are managed by the MCP Manager.

Tool Discovery and Filtering

The Tool Manager combines tools from three sources: 1. Internal tools: Built from the configuration provided by the admin. 2. MCP tools: Retrieved from the MCP Manager. 3. A2A sub-agents: Loaded via the A2A Manager.

After combining these tools, the manager applies several filters: - Exclusivity: If a tool is marked as exclusive, only that tool is loaded. When a tool is exclusive only that tool can be executed no other (e.g. Deep-research). - Enablement: Disabled tools are excluded. This is done by the admin of the Space to say which ones are available. - User Preferences: Should tools be selected by the end-user in the frontend, they are set as exclusive tools for the first iteration with the model. Then only these can be chosen.

Configuration

The available tools (MCP, sub-agents, and internal tools) are derived directly from the front-end configuration, which is set up by the admin of the space.

---

Code Implementation

Constructor and Initialization

The constructor initializes the Tool Manager with the necessary runtime context and managers:

Bases: _ToolManager[Literal['completions']]

Source code in unique_toolkit/unique_toolkit/agentic/tools/tool_manager.py

class ToolManager(_ToolManager[Literal["completions"]]):
    def __init__(
        self,
        logger: Logger,
        config: ToolManagerConfig,
        event: ChatEvent,
        tool_progress_reporter: ToolProgressReporter,
        mcp_manager: MCPManager,
        a2a_manager: A2AManager,
        chat_service: ChatService | None = None,
        language_model_service: LanguageModelService | None = None,
    ) -> None:
        super().__init__(
            logger=logger,
            config=config,
            event=event,
            tool_progress_reporter=tool_progress_reporter,
            mcp_manager=mcp_manager,
            a2a_manager=a2a_manager,
            api_mode="completions",
            builtin_tool_manager=None,
            chat_service=chat_service,
            language_model_service=language_model_service,
        )

    @classmethod
    def from_run_context(
        cls,
        logger: Logger,
        config: ToolManagerConfig,
        *,
        run_context: ToolRunContext,
        tool_progress_reporter: ToolProgressReporter,
        mcp_manager: MCPManager,
        a2a_manager: A2AManager,
        api_mode: Literal["completions"] = "completions",
        builtin_tool_manager: OpenAIBuiltInToolManager | None = None,
        chat_service: ChatService | None = None,
        language_model_service: LanguageModelService | None = None,
    ) -> Self:
        return super().from_run_context(
            logger=logger,
            config=config,
            run_context=run_context,
            tool_progress_reporter=tool_progress_reporter,
            mcp_manager=mcp_manager,
            a2a_manager=a2a_manager,
            api_mode=api_mode,
            builtin_tool_manager=builtin_tool_manager,
            chat_service=chat_service,
            language_model_service=language_model_service,
        )

add_tool(tool, kind='internal')

add_tool(
    tool: Tool[Any],
    kind: Literal["internal", "mcp"] = "internal",
) -> None

add_tool(
    tool: OpenAIBuiltInTool[Any], kind: Literal["builtin"]
) -> None

add_tool(
    tool: SubAgentTool, kind: Literal["sub_agent"]
) -> None

Inject an externally constructed tool into the manager.

Use this for tools that require custom constructor arguments (e.g. a shared registry) that cannot be built through ToolFactory.

Parameters:

Name	Type	Description	Default
tool	Tool[Any] | OpenAIBuiltInTool[Any] | SubAgentTool	The tool instance to register. Its accepted type is narrowed by kind via the overloads above.	required
kind	Literal['internal', 'builtin', 'mcp', 'sub_agent']	Which category the tool belongs to. Determines the internal tracking list it is added to ("internal", "builtin", "mcp" or "sub_agent"). Defaults to "internal".	'internal'

Source code in unique_toolkit/unique_toolkit/agentic/tools/tool_manager.py

def add_tool(
    self,
    tool: Tool[Any] | OpenAIBuiltInTool[Any] | SubAgentTool,
    kind: Literal["internal", "builtin", "mcp", "sub_agent"] = "internal",
) -> None:
    """Inject an externally constructed tool into the manager.

    Use this for tools that require custom constructor arguments (e.g. a
    shared registry) that cannot be built through ToolFactory.

    Args:
        tool: The tool instance to register. Its accepted type is narrowed
            by ``kind`` via the overloads above.
        kind: Which category the tool belongs to. Determines the internal
            tracking list it is added to ("internal", "builtin", "mcp" or
            "sub_agent"). Defaults to "internal".
    """
    if kind == "builtin":
        self._builtin_tools.append(cast(OpenAIBuiltInTool[Any], tool))
    elif kind == "sub_agent":
        self._sub_agents.append(cast(SubAgentTool, tool))
    elif kind == "mcp":
        self._mcp_tools.append(cast(Tool[Any], tool))
    else:
        self._internal_tools.append(cast(Tool[Any], tool))

    self.available_tools.append(tool)
    self._tools.append(tool)

exclude_tool(name)

Exclude a tool by name from the active tool set.

The tool is removed from all internal tracking lists so it will no longer be offered to the model or executed. Returns True if the tool was present in at least one list.

Source code in unique_toolkit/unique_toolkit/agentic/tools/tool_manager.py

def exclude_tool(self, name: str) -> bool:
    """Exclude a tool by name from the active tool set.

    The tool is removed from all internal tracking lists so it will no
    longer be offered to the model or executed.  Returns True if the tool
    was present in at least one list.
    """
    found = False

    filtered_tools = [tool for tool in self._tools if tool.name != name]
    found = found or len(filtered_tools) != len(self._tools)
    self._tools = filtered_tools

    filtered_internal_tools = [
        tool for tool in self._internal_tools if tool.name != name
    ]
    found = found or len(filtered_internal_tools) != len(self._internal_tools)
    self._internal_tools = filtered_internal_tools

    filtered_mcp_tools = [tool for tool in self._mcp_tools if tool.name != name]
    found = found or len(filtered_mcp_tools) != len(self._mcp_tools)
    self._mcp_tools = filtered_mcp_tools

    filtered_sub_agents = [tool for tool in self._sub_agents if tool.name != name]
    found = found or len(filtered_sub_agents) != len(self._sub_agents)
    self._sub_agents = filtered_sub_agents

    filtered_builtin_tools = [
        tool for tool in self._builtin_tools if tool.name != name
    ]
    found = found or len(filtered_builtin_tools) != len(self._builtin_tools)
    self._builtin_tools = filtered_builtin_tools

    filtered_available_tools = [
        tool for tool in self.available_tools if tool.name != name
    ]
    found = found or len(filtered_available_tools) != len(self.available_tools)
    self.available_tools = filtered_available_tools

    return found

filter_duplicate_tool_calls(tool_calls)

Filter out duplicate tool calls based on name and arguments.

Source code in unique_toolkit/unique_toolkit/agentic/tools/tool_manager.py

def filter_duplicate_tool_calls(
    self,
    tool_calls: list[LanguageModelFunction],
) -> list[LanguageModelFunction]:
    """
    Filter out duplicate tool calls based on name and arguments.
    """

    unique_tool_calls = []

    for call in tool_calls:
        if all(not call == other_call for other_call in unique_tool_calls):
            unique_tool_calls.append(call)

    if len(tool_calls) != len(unique_tool_calls):
        self._logger = getLogger(__name__)
        self._logger.warning(
            f"Filtered out {len(tool_calls) - len(unique_tool_calls)} duplicate tool calls."
        )
    return unique_tool_calls

Tool Initialization

The _init__tools method discovers and filters tools:

Source code in unique_toolkit/unique_toolkit/agentic/tools/tool_manager.py

def _init__tools(self, tool_init_event: ChatEvent | None) -> None:
    tool_choices = self._tool_choices
    tool_configs = self._config.tools
    self._logger.info("Initializing tool definitions...")
    self._logger.info(f"Tool choices: {tool_choices}")

    if tool_init_event is not None:
        tool_configs, sub_agents = self._a2a_manager.get_all_sub_agents(
            tool_configs,
            tool_init_event,
        )
    else:
        tool_configs = [
            tool_config
            for tool_config in tool_configs
            if not tool_config.is_sub_agent
        ]
        sub_agents = []
    self._sub_agents = sub_agents

    registered_tool_names = set(t.name for t in self._sub_agents)

    self._builtin_tools = []
    if self._builtin_tool_manager is not None and self._api_mode == "responses":
        self._builtin_tools = (
            self._builtin_tool_manager.get_all_openai_builtin_tools()
        )

    registered_tool_names.update(t.name for t in self._builtin_tools)

    # Get MCP tools (these are already properly instantiated)
    self._mcp_tools = self._mcp_manager.get_all_mcp_tools()

    registered_tool_names.update(t.name for t in self._mcp_tools)

    # Build internal tools from configurations, skipping disabled and failing tools
    self._internal_tools.clear()
    safe_executor = SafeTaskExecutor(logger=self._logger, log_exceptions=False)
    for t in tool_configs:
        if t.name in registered_tool_names:
            continue
        if t.name in OpenAIBuiltInToolName:
            continue
        if not t.is_enabled:
            self._logger.info("Skipping disabled tool '%s'", t.name)
            continue
        if tool_init_event is None:
            self._logger.info(
                "Skipping internal tool '%s' (requires chat event for initialization)",
                t.name,
            )
            continue
        if (
            self._chat_service is not None
            and self._language_model_service is not None
        ):
            result = safe_executor.execute(
                ToolFactory.build_tool_with_settings,
                t.name,
                t,
                t.configuration,
                tool_init_event,
                tool_progress_reporter=self._tool_progress_reporter,
                chat_service=self._chat_service,
                language_model_service=self._language_model_service,
            )
        else:
            result = safe_executor.execute(
                ToolFactory.build_tool_with_settings,
                t.name,
                t,
                t.configuration,
                tool_init_event,
                tool_progress_reporter=self._tool_progress_reporter,
            )
        if result.success:
            self._internal_tools.append(result.unpack())
        else:
            self._logger.warning(
                "Skipping tool '%s' due to initialization failure.",
                t.name,
                exc_info=result.exception,
            )

    # Activator tools are regular function tools that lazily provision a
    # built-in tool when the model calls them. They are offered up front and
    # swapped for the provisioned built-in in `_activate_deferred_tools`.
    if self._builtin_tool_manager is not None and self._api_mode == "responses":
        self._internal_tools.extend(
            self._builtin_tool_manager.get_activator_tools()
        )

    # Combine all types of tools
    self.available_tools = (
        self._internal_tools
        + self._mcp_tools
        + self._sub_agents
        + self._builtin_tools
    )

    for t in self.available_tools:
        if not t.is_enabled():
            continue
        if t.name in self._disabled_tools:
            continue
        # if tool choices are given, only include those tools
        if len(self._tool_choices) > 0 and t.name not in self._tool_choices:
            continue
        # is the tool exclusive and has been choosen by the user?
        if t.is_exclusive() and len(tool_choices) > 0 and t.name in tool_choices:
            self._tools = [t]  # override all other tools
            self._restore_uploaded_search_for_internal_search_if_available(
                exclusive_tool=t
            )
            break
        # if the tool is exclusive but no tool choices are given, skip it
        if t.is_exclusive():
            continue

        self._tools.append(t)

    # Capability tools bypass tool_choices filtering — they are always
    # included when enabled, regardless of what was force-selected.
    active_names = {t.name for t in self._tools}
    for t in self.available_tools:
        if (
            t.is_enabled()
            and t.name not in self._disabled_tools
            and t.is_capability()
            and t.name not in active_names
        ):
            self._tools.append(t)

---

📣 Exposing Tools to the Orchestrator and LLM

The orchestrator that works with the tool-manager needs three kinds of information: - The actual tool objects (for runtime operations) - Tool “definitions” or schemas consumable by the LLM - Additional tool-specific prompt enhancements/guidance to help the LLM format call the correct tool and format the output of the tools correctly.

Get loaded tools and log them:

Source code in unique_toolkit/unique_toolkit/agentic/tools/tool_manager.py

def log_loaded_tools(self):
    self._logger.info(f"Loaded tools: {[tool.name for tool in self._tools]}")

Source code in unique_toolkit/unique_toolkit/agentic/tools/tool_manager.py

def get_tools(self) -> list[Tool[Any]] | list[Tool[Any] | OpenAIBuiltInTool[Any]]:
    return self._tools.copy()

Source code in unique_toolkit/unique_toolkit/agentic/tools/tool_manager.py

def get_tool_by_name(self, name: str) -> Tool[Any] | OpenAIBuiltInTool[Any] | None:
    for tool in self._tools:
        if tool.name == name:
            return tool
    return None

Expose tool definitions and prompts (prompt enhancements):

#tool-manager-get-definitions-prompts

def get_tool_definitions(
    self,
) -> list[LanguageModelTool | LanguageModelToolDescription]:
    return [tool.tool_description() for tool in self._tools]

def get_tool_prompts(self) -> list[ToolPrompts]:
    return [tool.get_tool_prompts() for tool in self._tools]

Evaluation metrics aggregation:

#tool-manager-get-evaluation-check-list

def get_evaluation_check_list(self) -> list[EvaluationMetricName]:
    return list(self._tool_evaluation_check_list)

🎛️ Forced Tools and Admin/User Constraints

Users can force a subset of tools via the UI. Forced tools are surfaced in an LLM API-compatible structure. So that the orchestrator can hand this information over to the LLM call in the correct format.

Retrieve forced tools and add a forced tool programmatically:

#tool-manager-forced-tools

def get_forced_tools(self) -> list[dict[str, Any]]:
    return [
        self._convert_to_forced_tool(t.name)
        for t in self._tools
        if t.name in self._tool_choices
    ]

def add_forced_tool(self, name):
    tool = self.get_tool_by_name(name)
    if not tool:
        raise ValueError(f"Tool {name} not found")
    self._tools.append(tool)
    self._tool_choices.append(tool.name)

def _convert_to_forced_tool(self, tool_name: str) -> dict[str, Any]:
    return {
        "type": "function",
        "function": {"name": tool_name},
    }

🧠 Control-Taking Tools (e.g., Deep Research)

Some tools request a handover from the main orchestrator so they can “take control” of the session. The orchestrator can check this before deciding whether to yield control or to continue its flow.

Check if any selected call belongs to a control-taking tool:

#tool-manager-does-tool-take-control

def _get_tool_calls_that_take_control(
    self, tool_calls: list[LanguageModelFunction]
) -> list[LanguageModelFunction]:
    tools = []
    for tool_call in tool_calls:
        tool_instance = self.get_tool_by_name(tool_call.name)
        if tool_instance and tool_instance.takes_control():
            tools.append(tool_call)

    return tools

def does_a_tool_take_control(self, tool_calls: list[LanguageModelFunction]) -> bool:
    return len(self._get_tool_calls_that_take_control(tool_calls)) > 0

⚙️ Tool Execution Workflow

The orchestrator receives the information from the LLM on what tools to be executed in order for the LLM to receive the requested information. The Tool Manager handles the execution of selected tools with the following steps:

1. Deduplication: It removes duplicate tool calls from the LLM, ensuring identical calls (e.g., same tool with identical parameters) are executed only once. This prevents redundant processing caused by occasional LLM errors.

2. Call Limit Enforcement: The configured maximum number of tool calls is allowed per execution round. This prevents overloading the system with excessive requests.

3. Take-Control Conflict Guard: If a batch contains more than one call and any call belongs to a tool that takes control, no tools in the batch are executed. The manager returns a conflict response for every call: the take-control tool response explains that it must be called alone, and sibling responses explain which tool(s) caused the batch to be rejected.

4. Parallel Execution: When no take-control conflict exists, tools are executed concurrently to save time, as individual tool calls can be time-intensive.

5. Result Handling: Once the tools return their responses, the Tool Manager:

6. Sends the results back to the orchestrator.
7. Updates the call history.
8. Extracts references and debug information for further use.

This streamlined process ensures efficient, accurate, and manageable tool execution.

#tool-manager-execute-selected-tools

def _take_control_conflict_response(
    self,
    tool_call: LanguageModelFunction,
    take_control_names: list[str],
) -> ToolCallResponse:
    if tool_call.name in take_control_names:
        content = (
            f"ERROR: Tool `{tool_call.name}` directly returns its response to the user and therefore must be called on its own. "
            "None of the tools in this batch were executed. "
            f"You may recover by first calling the other tools, then calling `{tool_call.name}` alone."
        )
    else:
        names = ", ".join(f"`{n}`" for n in take_control_names)
        content = f"ERROR: Not executed because the following tool(s) must be called on their own: {names}"

    return ToolCallResponse(
        id=tool_call.id,
        name=tool_call.name,
        content=content,
    )

async def execute_selected_tools(
    self,
    tool_calls: list[LanguageModelFunction],
) -> list[ToolCallResponse]:
    if len(tool_calls) > 1:
        take_control_tools = [
            t.name for t in self._get_tool_calls_that_take_control(tool_calls)
        ]

        if len(take_control_tools) > 0:
            self._logger.warning(
                "Tool(s) %s take control and cannot be called alongside other tools. "
                "Returning error for all %d tool calls.",
                take_control_tools,
                len(tool_calls),
            )
            return [
                self._take_control_conflict_response(tool_call, take_control_tools)
                for tool_call in tool_calls
            ]

    tool_call_responses = await self._execute_parallelized(tool_calls)
    return tool_call_responses

Parallel execution strategy:

#tool-manager-execute-parallelized

async def _execute_parallelized(
    self,
    tool_calls: list[LanguageModelFunction],
) -> list[ToolCallResponse]:
    self._logger.info("Execute tool calls")

    task_executor = SafeTaskExecutor(
        logger=self._logger,
    )

    # Create tasks for each tool call
    tasks = [
        task_executor.execute_async(
            self.execute_tool_call,
            tool_call=tool_call,
        )
        for tool_call in tool_calls
    ]

    # Wait until all tasks are finished
    tool_call_results = await asyncio.gather(*tasks)
    tool_call_results_unpacked: list[ToolCallResponse] = []
    for i, result in enumerate(tool_call_results):
        unpacked_tool_call_result = self._create_tool_call_response(
            result, tool_calls[i]
        )
        tool_call_results_unpacked.append(unpacked_tool_call_result)

    return tool_call_results_unpacked

Execute a single tool call:

#tool-manager-execute-tool-call

async def execute_tool_call(
    self, tool_call: LanguageModelFunction
) -> ToolCallResponse:
    self._logger.info(f"Processing tool call: {tool_call.name}")

    tool_instance = self.get_tool_by_name(
        tool_call.name
    )  # we need to copy this as it will have problematic interference on multi calls.

    if tool_instance:
        # Execute the tool
        tool_response: ToolCallResponse = await tool_instance.run(
            tool_call=tool_call
        )
        evaluation_checks = tool_instance.evaluation_check_list()
        self._tool_evaluation_check_list.update(evaluation_checks)

        return tool_response

    return ToolCallResponse(
        id=tool_call.id,  # type: ignore
        name=tool_call.name,
        error_message=f"Tool of name {tool_call.name} not found",
    )

Normalize outcomes from the task executor:

#tool-manager-create-tool-call-response

def _create_tool_call_response(
    self, result: Result[ToolCallResponse], tool_call: LanguageModelFunction
) -> ToolCallResponse:
    if not result.success:
        return ToolCallResponse(
            id=tool_call.id or "unknown_id",
            name=tool_call.name,
            error_message=str(result.exception),
        )
    unpacked = result.unpack()
    if not isinstance(unpacked, ToolCallResponse):
        return ToolCallResponse(
            id=tool_call.id or "unknown_id",
            name=tool_call.name,
            error_message="Tool call response is not of type ToolCallResponse",
        )
    return unpacked

🔁 Deduplication and Safety

Before executing, the Tool Manager removes duplicate calls with identical names and arguments to prevent repeated work in the same round.

Deduplicate calls and warn when filtered:

#tool-manager-filter-duplicate-tool-calls

def filter_duplicate_tool_calls(
    self,
    tool_calls: list[LanguageModelFunction],
) -> list[LanguageModelFunction]:
    """
    Filter out duplicate tool calls based on name and arguments.
    """

    unique_tool_calls = []

    for call in tool_calls:
        if all(not call == other_call for other_call in unique_tool_calls):
            unique_tool_calls.append(call)

    if len(tool_calls) != len(unique_tool_calls):
        self._logger = getLogger(__name__)
        self._logger.warning(
            f"Filtered out {len(tool_calls) - len(unique_tool_calls)} duplicate tool calls."
        )
    return unique_tool_calls

🗣️ Enhanced Prompting Guidance for the LLM

To optimize tool selection and minimize formatting errors, the orchestrator should:

1. Incorporate Tool Definitions

2. Use get_tool_definitions() to retrieve the function/tool schema and provide it to the LLM. This ensures the LLM understands the available tools and their parameters.

3. Enhance System Prompts with Tool-Specific Guidance

4. Inject get_tool_prompts() content into the system prompt to:

   • Clearly define when each tool should be used.
   • Specify the expected inputs and outputs.
   • Include argument formatting examples for clarity.

5. Iterative Feedback for Improved Formatting

6. In subsequent interactions, provide explicit formatting guidance based on the tools previously selected. This iterative refinement ensures consistent and accurate tool usage.

Key Mechanism:

The orchestrator retrieves both tool definitions and tool prompts. Tool definitions describe the functionality and parameters of each tool, while tool prompts act as enhancements to the system message. These prompts guide the LLM in selecting the correct tool and formatting its arguments effectively. This process improves robustness, ensures accurate tool selection, and enhances the overall response quality.