monoai.agents

 222class _AgenticLoop:
 223    """Base class for all agentic loop implementations.
 224    
 225    This class provides the core functionality for all AI agents, including
 226    tool management, message creation, model execution, and streaming support.
 227    It's designed to be extended by specific classes that implement different
 228    agentic approaches and reasoning patterns.
 229    
 230    The base class handles common operations like:
 231    - Tool registration and execution
 232    - Message formatting and conversation management
 233    - Model interaction (both streaming and non-streaming)
 234    - Step-based reasoning format parsing
 235    - Debug output and iteration limits
 236    
 237    Attributes
 238    ----------
 239    _model : Any
 240        AI model instance used for execution
 241    _agentic_prompt : str, Prompt, optional
 242        Custom prompt for the agent (None for default). Can be a string, 
 243        a Prompt object, or a path to a .prompt file
 244    _debug : bool
 245        Flag to enable debug output
 246    _max_iter : Optional[int]
 247        Maximum number of iterations allowed (None for unlimited)
 248    _stream_callback : Optional[Callable[[str], None]]
 249        Callback function for handling streaming content
 250    _tools : Dict[str, Any]
 251        Dictionary of available tools, mapped by name
 252    """
 253
 254    _DEFAULT_PROMPT_PATH = "monoai/agents/prompts/"
 255    
 256    def __init__(self, model: Model, agentic_prompt: str=None, debug: bool=False, max_iter: Optional[int]=None, 
 257                 stream_callback: Optional[Callable[[str], None]]=None, human_feedback: Optional[str]=None) -> None:
 258        """Initialize the agent with model and configuration.
 259        
 260        Parameters
 261        ----------
 262        model : Any
 263            AI model instance to use for execution
 264        agentic_prompt : str, optional
 265            Custom prompt for the agent (None to use default)
 266        debug : bool, default False
 267            Enable debug output and logging
 268        max_iter : Optional[int], default None
 269            Maximum number of iterations allowed (None for unlimited)
 270        stream_callback : Optional[Callable[[str], None]], default None
 271            Callback function for handling streaming content chunks
 272        human_feedback : Optional[str], default None
 273            Human feedback mode for controlling agent execution. Can be:
 274            - None: No human feedback required
 275            - "actions": Pause and request confirmation before executing tool actions
 276            - "all": Pause after every step for human review
 277        """
 278        self._model = model
 279        self._agentic_prompt = agentic_prompt
 280        self._debug = debug
 281        self._max_iter = max_iter
 282        self._stream_callback = stream_callback
 283        self._human_feedback = human_feedback
 284        self._tools = {}
 285        self._mcp_servers = {}
 286
 287    def register_tools(self, tools: List[Any]) -> None:
 288        """Register tools with the agent.
 289        
 290        Parameters
 291        ----------
 292        tools : List[Any]
 293            List of tool functions to register. Each tool must have a
 294            __name__ attribute for identification.
 295        """
 296        for tool in tools:
 297            self._tools[tool.__name__] = tool
 298
 299    
 300    def register_mcp_servers(self, mcp_servers: List[Any]) -> None:
 301        """Register MCP servers with the agent.
 302        
 303        Parameters
 304        ----------
 305        mcp_servers : List[Any]
 306            List of MCP server instances to register.
 307        """
 308        tools = {}
 309        self._mcp_servers = {}
 310        for mcp_server in mcp_servers:
 311                self._debug_print(f"Connecting to MCP server: {mcp_server.name}")
 312                # Use context manager for proper connection handling
 313                self._mcp_servers[mcp_server.name] = mcp_server
 314                server_tools =  mcp_server.get_tools()
 315                self._debug_print(f"Retrieved {len(server_tools)} tools from {mcp_server.name}")
 316                for tool in server_tools:
 317                    # Create a new tool name with MCP prefix
 318                    original_name = tool.name
 319                    tool.name = f"mcp_{mcp_server.name}_{original_name}"
 320                    tools[tool.name] = tool
 321                    self._debug_print(f" - {tool.name}")
 322                
 323        self._tools.update(tools)
 324
 325    
 326    def enable_streaming(self, stream_callback: Optional[Callable[[str], None]] = None) -> None:
 327        """Enable streaming responses for this agent.
 328        
 329        When streaming is enabled, the agent will call the provided callback
 330        function with each content chunk as it's generated, allowing for
 331        real-time processing and display of the AI's response.
 332        
 333        Parameters
 334        ----------
 335        stream_callback : Optional[Callable[[str], None]], default None
 336            Callback function to handle streaming content chunks.
 337            If None, uses a default callback that prints content to console.
 338            The callback receives plain text content strings.
 339        """
 340        if stream_callback is None:
 341            def default_callback(content):
 342                print(content, end='', flush=True)
 343            
 344            self._stream_callback = default_callback
 345        else:
 346            self._stream_callback = stream_callback
 347    
 348    def disable_streaming(self) -> None:
 349        """Disable streaming responses for this agent.
 350        
 351        After calling this method, the agent will use standard (non-streaming)
 352        model execution for all subsequent requests.
 353        """
 354        self._stream_callback = None
 355    
 356    @classmethod
 357    def create_streaming(cls, model: Any, stream_callback: Optional[Callable[[str], None]] = None, 
 358                        **kwargs) -> '_AgenticLoop':
 359        """Create an agent instance with streaming enabled.
 360        
 361        This is a convenience class method that creates an agent instance
 362        with streaming already enabled, avoiding the need to call
 363        enable_streaming() separately.
 364        
 365        Parameters
 366        ----------
 367        model : Any
 368            AI model instance to use
 369        stream_callback : Optional[Callable[[str], None]], default None
 370            Callback function for handling streaming content chunks
 371        **kwargs
 372            Additional parameters to pass to the constructor
 373        
 374        Returns
 375        -------
 376        _AgenticLoop
 377            Agent instance with streaming enabled
 378        """
 379        return cls(model, stream_callback=stream_callback, **kwargs)
 380
 381    def _get_tools(self) -> str:
 382        """Generate a descriptive string of available tools.
 383        
 384        This method creates a formatted string listing all registered tools
 385        with their signatures and documentation. This string is typically
 386        included in prompts to help the AI model understand what tools
 387        are available for use.
 388        
 389        Returns
 390        -------
 391        str
 392            Formatted string with descriptions of all available tools,
 393            one per line with " - " prefix. Returns empty string if no tools.
 394            
 395            Example:
 396             - colab_downloader(colab_url): Downloads a Jupyter notebook from Google Drive and returns its Python code as a string.  
 397                Args:     colab_url (str): The Google Drive URL of the notebook (should contain 'drive/').
 398        """
 399        if not self._tools:
 400            return ""
 401        
 402        tools = []
 403        for tool_name, tool_func in self._tools.items():
 404            if tool_name.startswith("mcp_"):
 405                tools.append(f" - {self._encode_mcp_tool(tool_func)}")
 406            else:
 407                tools.append(f" - {self._encode_tool(tool_func)}")
 408        return "\n".join(tools)
 409
 410    def _get_base_messages(self, agent_type: str, query: str) -> List[Dict[str, Any]]:
 411        """Generate base messages for the specific agent type.
 412        
 413        This method creates the initial message structure for the agent,
 414        including the appropriate prompt template and user query. The
 415        prompt template is selected based on the agent type and includes
 416        information about available tools.
 417        
 418        Parameters
 419        ----------
 420        agent_type : str
 421            Type of agent to determine which prompt template to use
 422        query : str
 423            User query to include in the prompt
 424        
 425        Returns
 426        -------
 427        List[Dict[str, Any]]
 428            List of base messages for the agent, including the prompt and query
 429        """
 430        tools = self._get_tools()
 431        prompt_data = {"available_tools": tools}
 432        
 433        # Handle case where _agentic_prompt is already a Prompt object
 434        if self._agentic_prompt is not None and hasattr(self._agentic_prompt, 'as_dict'):
 435            # It's already a Prompt object, just update the data and return
 436            agentic_prompt = self._agentic_prompt
 437            # Update prompt data if the prompt supports it
 438            if hasattr(agentic_prompt, 'prompt_data'):
 439                agentic_prompt.prompt_data = prompt_data
 440        else:           
 441            # Otherwise, determine prompt configuration and create new Prompt
 442            prompt_config = self._determine_prompt_config(agent_type)
 443            
 444            agentic_prompt = Prompt(
 445                **prompt_config,
 446                is_system=True,
 447                prompt_data=prompt_data
 448            )
 449
 450        messages = [agentic_prompt.as_dict()]
 451
 452        if isinstance(query, Prompt):
 453            messages.append(query.as_dict())
 454        else:
 455            messages.append({"role": "user", "content": query})
 456
 457        return messages
 458    
 459    def _determine_prompt_config(self, agent_type: str) -> Dict[str, Any]:
 460        """Determine the prompt configuration based on agentic_prompt setting.
 461        
 462        Parameters
 463        ----------
 464        agent_type : str
 465            Type of agent for fallback prompt selection
 466            
 467        Returns
 468        -------
 469        Dict[str, Any]
 470            Configuration dictionary for Prompt initialization
 471        """
 472        if self._agentic_prompt is None:
 473            return {
 474                "prompt_id": f"{self._DEFAULT_PROMPT_PATH}{agent_type}.prompt"
 475            }
 476        
 477        if isinstance(self._agentic_prompt, str):
 478            if self._agentic_prompt.endswith(".prompt"):
 479                return {"prompt_id": self._agentic_prompt}
 480            else:
 481                return {"prompt": self._agentic_prompt}
 482        else:
 483            # For non-string types (including Prompt objects), treat as prompt content
 484            return {"prompt": self._agentic_prompt}
 485
 486    def _debug_print(self, content: str) -> None:
 487        """Print debug information if debug mode is enabled.
 488        
 489        Parameters
 490        ----------
 491        content : str
 492            Content to print in debug mode
 493        """
 494        if self._debug:
 495            print(content)
 496            print("-------")
 497    
 498    def _request_human_feedback(self, step_type: str, content: str, action_data: Optional[Dict] = None) -> bool:
 499        """Request human feedback for the current step.
 500        
 501        This method pauses execution and asks the user for confirmation
 502        before proceeding with the current step.
 503        
 504        Parameters
 505        ----------
 506        step_type : str
 507            Type of step being executed (e.g., "thought", "action", "observation")
 508        content : str
 509            Content of the current step
 510        action_data : Optional[Dict], default None
 511            Action data if this is an action step
 512        
 513        Returns
 514        -------
 515        bool
 516            True if the user approves the step, False if they want to stop
 517        """
 518        print(f"\n{'='*60}")
 519        print(f"HUMAN FEEDBACK REQUIRED - {step_type.upper()} STEP")
 520        print(f"{'='*60}")
 521        
 522        if action_data:
 523            print(f"Action to execute: {action_data.get('name', 'Unknown')}")
 524            print(f"Arguments: {action_data.get('arguments', {})}")
 525        else:
 526            print(f"Content: {content}")
 527        
 528        print(f"\nOptions:")
 529        print(f"  [y] Yes - Continue with this step")
 530        print(f"  [n] No - Stop execution")
 531        print(f"  [s] Skip - Skip this step and continue")
 532        
 533        while True:
 534            try:
 535                response = input("\nYour choice (y/n/s): ").lower().strip()
 536                if response in ['y', 'yes']:
 537                    return True
 538                elif response in ['n', 'no']:
 539                    return False
 540                elif response in ['s', 'skip']:
 541                    print("Step skipped.")
 542                    return True  # Continue but skip the current step
 543                else:
 544                    print("Please enter 'y' for yes, 'n' for no, or 's' for skip.")
 545            except KeyboardInterrupt:
 546                print("\nExecution interrupted by user.")
 547                return False
 548    
 549
 550    def _parse_step_format(self, content: str) -> Optional[Dict[str, Any]]:
 551        """Parse the step-based format <STEP_TYPE>: <RESULT>.
 552        
 553        This method parses agent responses that follow the structured step format
 554        used by ReAct and similar reasoning patterns. It extracts the step type,
 555        step number, and content from formatted responses.
 556        
 557        Supported step types:
 558        - Thought: Reasoning and analysis steps
 559        - Action: Tool calls and actions (must be valid JSON)
 560        - Observation: Results from tool executions
 561        - Final answer: Conclusive responses to user queries
 562        
 563        Special handling: 
 564        - If content contains both "Thought" and "Action", the Action will be returned with priority.
 565        - If content contains both "Thought" and "Final answer", the Final answer will be returned with priority.
 566        
 567        Parameters
 568        ----------
 569        content : str
 570            Content to parse in the format <STEP_TYPE>: <RESULT>
 571        
 572        Returns
 573        -------
 574        Optional[Dict[str, Any]]
 575            Dictionary containing:
 576            - step_type: Type of step (thought, action, observation, final answer)
 577            - step_number: Optional step number if present
 578            - content: Step content
 579            - action: Parsed action data (for action steps)
 580            - final_answer: Final answer content (for final answer steps)
 581            Returns None if parsing fails
 582        """
 583        if not content or not isinstance(content, str):
 584            return None
 585        
 586        content = content.strip()
 587        
 588        # Check for multiple step types and prioritize accordingly
 589        lines = content.split('\n')
 590        action_start = None
 591        thought_start = None
 592        final_answer_start = None
 593        
 594        # Find the start lines for Action, Thought, and Final answer
 595        for i, line in enumerate(lines):
 596            if re.match(r'^Action(?:\s+\d+)?\s*:', line, re.IGNORECASE):
 597                action_start = i
 598            elif re.match(r'^Thought(?:\s+\d+)?\s*:', line, re.IGNORECASE):
 599                thought_start = i
 600            elif re.match(r'^Final answer(?:\s+\d+)?\s*:', line, re.IGNORECASE):
 601                final_answer_start = i
 602        
 603        # Priority: Final answer > Action > Thought
 604        if final_answer_start is not None:
 605            # Final answer has highest priority
 606            step_type = "final_answer"
 607            
 608            # Extract step number from Final answer line
 609            final_answer_line = lines[final_answer_start]
 610            step_number_match = re.search(r'Final answer\s+(\d+)\s*:', final_answer_line, re.IGNORECASE)
 611            step_number = step_number_match.group(1) if step_number_match else None
 612            
 613            # Extract content from Final answer (everything after the colon on the first line)
 614            final_answer_content = final_answer_line.split(':', 1)[1].strip()
 615            
 616            # Add all subsequent lines until we hit another step type or end
 617            i = final_answer_start + 1
 618            while i < len(lines):
 619                line = lines[i].strip()
 620                if re.match(r'^(Thought|Action|Observation|Final answer)(?:\s+\d+)?\s*:', line, re.IGNORECASE):
 621                    break
 622                if line:  # Only add non-empty lines
 623                    final_answer_content += '\n' + line
 624                i += 1
 625            
 626            step_content = final_answer_content.strip().replace("```json", "").replace("```", "")
 627            
 628            result = {
 629                "step_type": step_type,
 630                "step_number": step_number,
 631                "content": step_content,
 632                "final_answer": step_content
 633            }
 634            
 635            return result
 636            
 637        elif action_start is not None and thought_start is not None:
 638            # Both Thought and Action present, prioritize Action
 639            step_type = "action"
 640            
 641            # Extract step number from Action line
 642            action_line = lines[action_start]
 643            step_number_match = re.search(r'Action\s+(\d+)\s*:', action_line, re.IGNORECASE)
 644            step_number = step_number_match.group(1) if step_number_match else None
 645            
 646            # Extract content from Action (everything after the colon on the first line)
 647            action_content = action_line.split(':', 1)[1].strip()
 648            
 649            # Add all subsequent lines until we hit another step type or end
 650            i = action_start + 1
 651            while i < len(lines):
 652                line = lines[i].strip()
 653                if re.match(r'^(Thought|Action|Observation|Final answer)(?:\s+\d+)?\s*:', line, re.IGNORECASE):
 654                    break
 655                if line:  # Only add non-empty lines
 656                    action_content += '\n' + line
 657                i += 1
 658            
 659            step_content = action_content.strip().replace("```json", "").replace("```", "")
 660            
 661            result = {
 662                "step_type": step_type,
 663                "step_number": step_number,
 664                "content": step_content
 665            }
 666            
 667            # Parse action data (must be JSON)
 668            try:
 669                action_data = json.loads(step_content)
 670                result["action"] = action_data
 671            except json.JSONDecodeError:
 672                # If not valid JSON, keep content as raw string
 673                result["action"] = {"raw": step_content}
 674            
 675            return result
 676        
 677        # Standard single step parsing
 678        step_pattern = r'^(Thought|Action|Observation|Final answer)(?:\s+(\d+))?\s*:\s*(.*)$'
 679        match = re.match(step_pattern, content, re.IGNORECASE | re.DOTALL)
 680        
 681        if match:
 682            step_type = match.group(1).lower()
 683            step_number = match.group(2) if match.group(2) else None
 684            step_content = match.group(3).strip().replace("```json", "").replace("```", "")
 685            
 686            result = {
 687                "step_type": step_type,
 688                "step_number": step_number,
 689                "content": step_content
 690            }
 691            
 692            # Special handling for Action steps (must be JSON)
 693            if step_type == "action":
 694                try:
 695                    action_data = json.loads(step_content)
 696                    result["action"] = action_data
 697                except json.JSONDecodeError:
 698                    # If not valid JSON, keep content as raw string
 699                    result["action"] = {"raw": step_content}
 700            
 701            # Special handling for Final answer steps
 702            elif step_type == "final_answer":
 703                result["final_answer"] = step_content
 704            
 705            return result
 706        
 707        return None
 708
 709    def _execute_model_step(self, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
 710        """Execute a model step and return the response.
 711        
 712        This method handles both streaming and non-streaming model execution.
 713        If streaming is enabled and not already in progress, it uses the
 714        streaming method. Otherwise, it uses standard model execution.
 715        
 716        Parameters
 717        ----------
 718        messages : List[Dict[str, Any]]
 719            List of messages to send to the model
 720        
 721        Returns
 722        -------
 723        Dict[str, Any]
 724            Model response in standard OpenAI format with usage information
 725        """
 726        resp = self._model._execute(messages)
 727        
 728        # Extract usage information from the response
 729        usage_info = None
 730        if hasattr(resp, 'usage') and resp.usage:
 731            usage_info = {
 732                "total_tokens": resp.usage.total_tokens,
 733                "prompt_tokens": resp.usage.prompt_tokens,
 734                "completion_tokens": resp.usage.completion_tokens
 735            }
 736        
 737        message = resp["choices"][0]["message"]
 738        
 739        # Add usage information to the message
 740        if usage_info:
 741            message["usage"] = usage_info
 742        return message
 743    
 744    def _execute_model_step_stream(self, messages: List[Dict[str, Any]]) -> Dict[str, Any]:
 745        """Execute a model step with streaming and return the complete response.
 746        
 747        This method handles asynchronous streaming of model responses, collecting
 748        all chunks and building the final response. It properly manages async
 749        resources to avoid memory leaks and task warnings.
 750        
 751        Parameters
 752        ----------
 753        messages : List[Dict[str, Any]]
 754            List of messages to send to the model
 755        
 756        Returns
 757        -------
 758        Dict[str, Any]
 759            Complete model response in standard OpenAI format
 760        """
 761
 762        import asyncio
 763        from litellm import stream_chunk_builder
 764        
 765        # Usa asyncio.run per gestire correttamente il loop
 766        async def run_streaming():
 767            chunks = []
 768            stream = None
 769            try:
 770                stream = self._model._execute_stream(messages)
 771                async for chunk in stream:
 772                    chunks.append(chunk)
 773                    content = chunk["choices"][0]["delta"]["content"]
 774
 775                    if content is not None:
 776                        self._stream_callback({"type": "text", "delta": content})
 777
 778            finally:
 779                if stream is not None:
 780                    try:
 781                        await stream.aclose()
 782                    except Exception:
 783                        pass
 784            
 785            return chunks
 786
 787        try:
 788            chunks = asyncio.run(run_streaming())
 789            resp = stream_chunk_builder(chunks)
 790            
 791            # Extract usage information from the response
 792            usage_info = None
 793            if hasattr(resp, 'usage') and resp.usage:
 794                usage_info = {
 795                    "total_tokens": resp.usage.total_tokens,
 796                    "prompt_tokens": resp.usage.prompt_tokens,
 797                    "completion_tokens": resp.usage.completion_tokens
 798                }
 799            
 800            message = resp["choices"][0]["message"]
 801            
 802            # Add usage information to the message
 803            if usage_info:
 804                message["usage"] = usage_info
 805                
 806            return message
 807        except Exception as e:
 808            print(f"Streaming error: {e}, falling back to standard execution")
 809            return self._execute_model_step(messages)
 810            
 811    def _create_base_response(self, query: str) -> Dict[str, Any]:
 812        """Create the base response structure.
 813        
 814        Parameters
 815        ----------
 816        query : str
 817            Original user query
 818        
 819        Returns
 820        -------
 821        Dict[str, Any]
 822            Dictionary with base response structure:
 823            - prompt: Original user query
 824            - iterations: Empty list for iterations
 825            - usage: Dictionary for tracking token usage
 826        """
 827        return {
 828            "prompt": query, 
 829            "iterations": [],
 830            "usage": {
 831                "total_tokens": 0,
 832                "prompt_tokens": 0,
 833                "completion_tokens": 0
 834            }
 835        }
 836
 837    def _update_usage(self, response: Dict[str, Any], iteration_usage: Dict[str, int]) -> None:
 838        """Update the total usage in the response.
 839        
 840        Parameters
 841        ----------
 842        response : Dict[str, Any]
 843            Response dictionary to update
 844        iteration_usage : Dict[str, int]
 845            Usage information from current iteration
 846        """
 847        if hasattr(iteration_usage, "usage"):
 848            response["usage"]["total_tokens"] += iteration_usage["usage"]["total_tokens"]
 849            response["usage"]["prompt_tokens"] += iteration_usage["usage"]["prompt_tokens"]
 850            response["usage"]["completion_tokens"] += iteration_usage["usage"]["completion_tokens"]
 851
 852    def _handle_final_answer(self, iteration: Dict[str, Any], response: Dict[str, Any]) -> bool:
 853        """Handle a final answer, returns True if this is the end.
 854        
 855        This method processes iterations that contain final answers and
 856        updates the response structure accordingly. It supports both
 857        the old format (final_answer key) and new step format.
 858        
 859        Parameters
 860        ----------
 861        iteration : Dict[str, Any]
 862            Current iteration potentially containing a final answer
 863        response : Dict[str, Any]
 864            Response dictionary to update
 865        
 866        Returns
 867        -------
 868        bool
 869            True if a final answer was found and processed, False otherwise
 870        
 871        Notes
 872        -----
 873        This method modifies the response object directly.
 874        """
 875        if "final_answer" in iteration:
 876            response["iterations"].append(iteration)
 877            response["response"] = iteration["final_answer"]
 878            return True
 879        elif iteration.get("step_type") == "final answer":
 880            response["iterations"].append(iteration)
 881            response["response"] = iteration["content"]
 882            return True
 883        return False
 884
 885    def _pngimagefile_to_base64_data_uri(self, img) -> str:
 886        """
 887        Converte un oggetto PIL.PngImageFile in una stringa base64 data URI.
 888        
 889        Args:
 890            img (Image.Image): immagine PIL (PngImageFile o convertita in PNG)
 891        
 892        Returns:
 893            str: stringa "data:image/png;base64,<...>"
 894        """
 895        import io
 896        import base64
 897        buffer = io.BytesIO()
 898        img.save(buffer, format="PNG")
 899        buffer.seek(0)
 900
 901        # codifica in base64
 902        encoded = base64.b64encode(buffer.read()).decode("utf-8")
 903
 904        return f"data:image/png;base64,{encoded}"
 905
 906    def _handle_tool_action(self, iteration: Dict[str, Any], response: Dict[str, Any], messages: List[Dict[str, Any]]) -> None:
 907        """Handle a tool action execution.
 908        
 909        This method processes iterations that contain tool actions, executes
 910        the corresponding tools, and updates the conversation with the results.
 911        It supports both old and new step-based formats.
 912        
 913        Parameters
 914        ----------
 915        iteration : Dict[str, Any]
 916            Current iteration containing the tool action
 917        response : Dict[str, Any]
 918            Response dictionary to update
 919        messages : List[Dict[str, Any]]
 920            Message list to update with observation
 921        
 922        Notes
 923        -----
 924        This method modifies the response and messages objects directly.
 925        """
 926        # Check if human feedback is required for actions
 927        if self._human_feedback == "actions" or self._human_feedback == "all":
 928            action_data = iteration.get("action", {})
 929            if not self._request_human_feedback("action", iteration.get("content", ""), action_data):
 930                print("Execution stopped by user.")
 931                return
 932
 933        if "action" in iteration and iteration["action"].get("name"):
 934            tool_call = iteration["action"]
 935            
 936            if self._stream_callback is not None:
 937                self._stream_callback({"type": "tool_call", "tool_call": tool_call})
 938
 939            if tool_call.get("name").startswith("mcp_"):
 940                tool_result = self._call_mcp_tool(tool_call)
 941            else:
 942                tool_result = self._call_tool(tool_call)
 943
 944            iteration["observation"] = tool_result
 945            response["iterations"].append(iteration)
 946
 947            if isinstance(tool_result, dict):
 948                if "image" in tool_result:
 949                    img = self._pngimagefile_to_base64_data_uri(tool_result["image"])
 950                    msg = [{"type": "image_url", "image_url": {"url": img}}, 
 951                            {"type": "text", "text": "Ecco l'immagine"}]
 952                    messages.append({"role": "user", "content": msg})
 953                else:
 954                    msg = json.dumps({"observation": tool_result})
 955                    messages.append({"role": "user", "content": msg})
 956        elif iteration.get("step_type") == "action" and "action" in iteration:
 957            tool_call = iteration["action"]
 958            tool_result = self._call_tool(tool_call)
 959            iteration["observation"] = tool_result
 960            response["iterations"].append(iteration)
 961            
 962            # Add observation in the new system format
 963            observation_msg = f"Observation {iteration.get('step_number', '')}: {tool_result}".strip()
 964            messages.append({"role": "user", "content": observation_msg})
 965
 966    def _handle_default(self, iteration: Dict[str, Any], response: Dict[str, Any], messages: List[Dict[str, Any]]) -> None:
 967        """Handle default case for unhandled iterations.
 968        
 969        This method processes iterations that don't match specific handlers,
 970        adding them to the response and updating the conversation flow.
 971        It supports both step-based and JSON formats.
 972        
 973        Parameters
 974        ----------
 975        iteration : Dict[str, Any]
 976            Current iteration to handle
 977        response : Dict[str, Any]
 978            Response dictionary to update
 979        messages : List[Dict[str, Any]]
 980            Message list to update
 981        
 982        Notes
 983        -----
 984        This method modifies the response and messages objects directly.
 985        """
 986        response["iterations"].append(iteration)
 987        
 988        # For new format, add content as user message
 989        if iteration.get("step_type") in ["thought", "observation"]:
 990            step_type = iteration["step_type"].capitalize()
 991            step_number = iteration.get("step_number", "")
 992            content = iteration["content"]
 993            message_content = f"{step_type} {step_number}: {content}".strip()
 994            messages.append({"role": "user", "content": message_content})
 995        else:
 996            # Fallback to JSON format for compatibility
 997            messages.append({"role": "user", "content": json.dumps(iteration)})
 998
 999    def start(self, query: str) -> Dict[str, Any]:
1000        """Abstract method to start the agentic loop.
1001        
1002        This method must be implemented by subclasses to define the specific
1003        agentic behavior and reasoning pattern.
1004        
1005        Parameters
1006        ----------
1007        query : str
1008            User query to process
1009        
1010        Returns
1011        -------
1012        Dict[str, Any]
1013            Agent response containing iterations and final result
1014        
1015        Raises
1016        ------
1017        NotImplementedError
1018            This method must be implemented by subclasses
1019        """
1020        raise NotImplementedError

21class _FunctionCallingMixin:
22    """Mixin for handling OpenAI function calling tool execution.
23    
24    This mixin provides methods for managing tool calls in OpenAI's native
25    function calling format, converting tool responses into standardized messages
26    that can be used in the conversation flow.
27    
28    The mixin handles the execution of tools registered with the agent and
29    formats their responses according to OpenAI's message format specification.
30    """
31    
32    def _call_tool(self, tool_call: Any) -> Dict[str, str]:
33        """Execute a tool call and return the formatted response.
34        
35        This method takes a tool call object from the AI model's response,
36        executes the corresponding tool function, and returns a properly
37        formatted message that can be added to the conversation history.
38        
39        Parameters
40        ----------
41        tool_call : Any
42            Tool call object containing function execution details.
43            Must have attributes:
44            - function.name: Name of the function to call
45            - function.arguments: JSON string of function arguments
46            - id: Unique identifier for this tool call
47        
48        Returns
49        -------
50        Dict[str, str]
51            Formatted tool response message containing:
52            - tool_call_id: ID of the original tool call
53            - role: Message role (always "tool")
54            - name: Name of the executed function
55            - content: Tool execution result as string
56        
57        Raises
58        ------
59        KeyError
60            If the tool function is not registered with the agent
61        json.JSONDecodeError
62            If the function arguments are not valid JSON
63        Exception
64            If the tool function execution fails
65        """
66
67        function_name = tool_call.function.name
68        function_to_call = self._tools[function_name]
69        function_args = json.loads(tool_call.function.arguments)                
70        function_response = str(function_to_call(**function_args))
71        return {
72            "tool_call_id": tool_call.id,
73            "role": "tool",
74            "name": function_name,
75            "content": function_response,
76        }
77
78    def _call_mcp_tool(self, tool_call: Dict[str, Any]) -> Any:
79        """Execute a MCP tool call."""
80        tool_name = tool_call.function.name
81        tool_arguments = json.loads(tool_call.function.arguments) if tool_call.function.arguments else {}
82        tool_call_id = tool_call.id
83        
84        # Get server name from tool name (format: mcp_servername_toolname)
85        server_name = tool_name.split("_", 2)[1]
86        if server_name not in self._mcp_servers:
87            raise ValueError(f"MCP server '{server_name}' not found")
88        # Simple approach: just return an error for now since MCP is not working
89        return {
90            "tool_call_id": tool_call_id,
91            "role": "tool",
92            "name": tool_name,
93            "content": str(self._mcp_servers[server_name].call_tool(tool_name, tool_arguments))
94        }

 98class _ReactMixin:
 99    """Mixin for handling ReAct-style tool calls with JSON format.
100    
101    This mixin provides methods for managing tool calls in structured JSON format,
102    typical of ReAct-style approaches for AI agents. It handles the encoding
103    of tool functions and execution of tool calls based on parsed JSON responses.
104    
105    The ReAct (Reasoning and Acting) pattern allows agents to reason about
106    problems step-by-step and take actions using tools when needed.
107    """
108    
109    def _encode_tool(self, func: Any) -> str:
110        """Encode a function into a descriptive string format.
111        
112        This method creates a human-readable description of a tool function
113        that can be included in prompts to help the AI model understand
114        what tools are available and how to use them.
115        
116        Parameters
117        ----------
118        func : Any
119            Function to encode. Must have attributes:
120            - __name__: Function name
121            - __doc__: Function documentation
122        
123        Returns
124        -------
125        str
126            Descriptive string in the format:
127            "function_name(signature): documentation"
128            Newlines are replaced with spaces for single-line format.
129        """
130        sig = inspect.signature(func)
131        doc = inspect.getdoc(func)
132        encoded = func.__name__ + str(sig) + ": " + doc
133        encoded = encoded.replace("\n", " ")
134        return encoded
135    
136    def _encode_mcp_tool(self, tool: dict) -> str:
137        """Converte uno schema tipo JSON Schema in sezione Args Google style."""
138        schema = tool.inputSchema
139        if schema.get("type") != "object":
140            raise ValueError("Lo schema radice deve avere type=object")
141
142        args = []
143        args_doc = ["Args:"]
144        properties = schema.get("properties", {})
145        required = set(schema.get("required", []))
146
147        for name, spec in properties.items():
148            args.append(name)
149            typ = spec.get("type", "Any")
150            desc = spec.get("description", "").strip().replace("\n", " ")
151            title = spec.get("title", "")
152            default_info = "" if name in required else " Defaults to None."
153
154            arg_line = f"    {name} ({typ}): {desc}{default_info}"
155            if title:
156                arg_line = f"    {name} ({typ}): {title}. {desc}{default_info}"
157            args_doc.append(arg_line)
158
159        encoded = tool.name + "(" + ", ".join(args) + "): "
160        encoded += tool.description
161        encoded += ". ".join(args_doc)
162        return encoded
163
164    
165    def _call_tool(self, tool_call: Dict[str, Any]) -> Any:
166        """Execute a tool call in ReAct format.
167        
168        This method executes a tool call based on a structured dictionary
169        containing the tool name and arguments. It's designed to work with
170        the ReAct pattern where tools are called through JSON-formatted
171        action specifications.
172        
173        Parameters
174        ----------
175        tool_call : Dict[str, Any]
176            Tool call specification containing:
177            - name: Name of the tool to call
178            - arguments: Dictionary of tool arguments
179        
180        Returns
181        -------
182        Any
183            Result of the tool execution
184        
185        Raises
186        ------
187        KeyError
188            If the tool is not registered with the agent
189        TypeError
190            If the tool arguments don't match the function signature
191        Exception
192            If the tool execution fails
193        """
194        tool = self._tools[tool_call["name"]]
195        kwargs = list(tool_call["arguments"].values())
196        return tool(*kwargs)
197
198    def _call_mcp_tool(self, tool_call: Dict[str, Any]) -> Any:
199        """Execute a MCP tool call.
200        
201        This method executes a MCP tool call based on a structured dictionary
202        containing the tool name and arguments.
203        """
204        # Extract server name from tool name (format: mcp_servername_toolname)
205        tool_name_parts = tool_call["name"].split("_", 2)
206        if len(tool_name_parts) < 3:
207            raise ValueError(f"Invalid MCP tool name format: {tool_call['name']}")
208        
209        server_name = tool_name_parts[1]
210        tool_name = tool_name_parts[2]
211        if server_name not in self._mcp_servers:
212            raise ValueError(f"MCP server '{server_name}' not found")
213
214        return {
215            "tool_call_id": tool_name,
216            "role": "tool",
217            "name": tool_name,
218            "content": str(self._mcp_servers[server_name].call_tool(tool_name, tool_call["arguments"]))
219        }