Method __call__

src/tool/context.py:1249–1296

Call a tool by name with optional timeout Args: name: Tool name input: Input for the tool timeout: Timeout in seconds for this specific call (overrides default_timeout if provided) Returns: ToolResponse: Tool resul

(self,
                       name: str,
                       input: Dict[str, Any], 
                       timeout: Optional[float] = None,
                       ctx: SessionContext = None,
                       **kwargs
                       )

            )
    
    async def __call__(self,
                       name: str,
                       input: Dict[str, Any], 
                       timeout: Optional[float] = None,
                       ctx: SessionContext = None,
                       **kwargs
                       ) -> ToolResponse:
        """Call a tool by name with optional timeout
        
        Args:
            name: Tool name
            input: Input for the tool
            timeout: Timeout in seconds for this specific call (overrides default_timeout if provided)
            
        Returns:
            ToolResponse: Tool result
        """
        
        if ctx is None:
            ctx = SessionContext()
        
        tool_info = await self.get_info(name)
        
        version = tool_info.version
        tool_instance = tool_info.instance
        logger.info(f"| ✅ Using tool {name}@{version}")
        
        # Use provided timeout, or fall back to default_timeout
        effective_timeout = timeout if timeout is not None else self.default_timeout
        
        # Other tool args
        tool_kwargs = dict(ctx=ctx)
        
        # If timeout is None (no timeout), call tool directly
        if effective_timeout is None:
            return await tool_instance(**input, **tool_kwargs)
        
        # Otherwise, use asyncio.wait_for to enforce timeout
        try:
            return await asyncio.wait_for(tool_instance(**input, **tool_kwargs), timeout=effective_timeout)
        except asyncio.TimeoutError:
            error_msg = f"Tool '{name}' execution timed out after {effective_timeout} seconds"
            logger.error(f"| ⏱️ {error_msg}")
            return ToolResponse(
                success=False,
                message=error_msg,
                extra=None
            )