Method single_call_tool

ms_agent/tools/tool_manager.py:201–244

(self, tool_info: ToolCall)

        return sorted(tools, key=lambda t: (t.get('tool_name', ''), ))

    async def single_call_tool(self, tool_info: ToolCall):
        if self._concurrent_limiter is None:
            if self._init_lock is None:
                self._init_lock = asyncio.Lock()
            async with self._init_lock:
                if self._concurrent_limiter is None:
                    self._concurrent_limiter = asyncio.Semaphore(
                        MAX_CONCURRENT_TOOLS)

        async with self._concurrent_limiter:
            brief_info = json.dumps(tool_info, ensure_ascii=False)
            if len(brief_info) > 1024:
                brief_info = brief_info[:1024] + '...'
            try:
                tool_name = tool_info['tool_name']
                tool_args = tool_info['arguments']
                while isinstance(tool_args, str):
                    try:
                        tool_args = json.loads(tool_args)
                    except Exception:  # noqa
                        return f'The input {tool_args} is not a valid JSON, fix your arguments and try again'
                assert tool_name in self._tool_index, f'Tool name {tool_name} not found'
                tool_ins, server_name, _ = self._tool_index[tool_name]
                call_args = tool_args
                if isinstance(tool_ins, AgentTool):
                    call_args = dict(tool_args or {})
                    call_id = tool_info.get('id') or str(uuid.uuid4())
                    call_args['__call_id'] = call_id
                response = await asyncio.wait_for(
                    tool_ins.call_tool(
                        server_name,
                        tool_name=tool_name.split(self.TOOL_SPLITER)[1],
                        tool_args=call_args),
                    timeout=self.tool_call_timeout)
                return response
            except asyncio.TimeoutError:
                import traceback
                logger.warning(traceback.format_exc())
                # TODO: How to get the information printed by the tool before hanging to return to the model?
                return f'Execute tool call timeout: {brief_info}'
            except Exception as e:
                import traceback
                logger.warning(traceback.format_exc())
                return f'Tool calling failed: {brief_info}, details: {str(e)}'

    async def parallel_call_tool(self, tool_list: List[ToolCall]):
        tasks = [self.single_call_tool(tool) for tool in tool_list]