docs: add flow plot visualization example to documentation (#2668)

Co-Authored-By: Joe Moura <joao@crewai.com>

.github/workflows/notify-downstream.yml

@@ -0,0 +1,33 @@
+name: Notify Downstream
+
+on:
+  push:
+    branches:
+      - main
+
+permissions:
+  contents: read
+
+jobs:
+  notify-downstream:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Generate GitHub App token
+        id: app-token
+        uses: tibdex/github-app-token@v2
+        with:
+          app_id: ${{ secrets.OSS_SYNC_APP_ID }}
+          private_key: ${{ secrets.OSS_SYNC_APP_PRIVATE_KEY }}
+
+      - name: Notify Repo B
+        uses: peter-evans/repository-dispatch@v3
+        with:
+          token: ${{ steps.app-token.outputs.token }}
+          repository: ${{ secrets.OSS_SYNC_DOWNSTREAM_REPO }}
+          event-type: upstream-commit
+          client-payload: |
+            {
+              "commit_sha": "${{ github.sha }}"
+            }
+

.github/workflows/tests.yml

@@ -12,6 +12,9 @@ jobs:
   tests:
     runs-on: ubuntu-latest
     timeout-minutes: 15
+    strategy:
+      matrix:
+        python-version: ['3.10', '3.11', '3.12']
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
@@ -21,9 +24,8 @@ jobs:
         with:
           enable-cache: true
 
-
+      - name: Set up Python ${{ matrix.python-version }}
-      - name: Set up Python
+        run: uv python install ${{ matrix.python-version }}
-        run: uv python install 3.12.8
 
       - name: Install the project
         run: uv sync --dev --all-extras

.gitignore

@@ -26,3 +26,4 @@ test_flow.html
 crewairules.mdc
 plan.md
 conceptual_plan.md
+build_image

README.md

@@ -257,10 +257,14 @@ reporting_task:
 from crewai import Agent, Crew, Process, Task
 from crewai.project import CrewBase, agent, crew, task
 from crewai_tools import SerperDevTool
+from crewai.agents.agent_builder.base_agent import BaseAgent
+from typing import List
 
 @CrewBase
 class LatestAiDevelopmentCrew():
 	"""LatestAiDevelopment crew"""
+	agents: List[BaseAgent]
+	tasks: List[Task]
 
 	@agent
 	def researcher(self) -> Agent:
@@ -401,11 +405,16 @@ You can test different real life examples of AI crews in the [CrewAI-examples re
 
 ### Using Crews and Flows Together
 
-CrewAI's power truly shines when combining Crews with Flows to create sophisticated automation pipelines. Here's how you can orchestrate multiple Crews within a Flow:
+CrewAI's power truly shines when combining Crews with Flows to create sophisticated automation pipelines.
+CrewAI flows support logical operators like `or_` and `and_` to combine multiple conditions. This can be used with `@start`, `@listen`, or `@router` decorators to create complex triggering conditions.
+- `or_`: Triggers when any of the specified conditions are met. 
+- `and_`Triggers when all of the specified conditions are met.
+
+Here's how you can orchestrate multiple Crews within a Flow:
 
 ```python
-from crewai.flow.flow import Flow, listen, start, router
+from crewai.flow.flow import Flow, listen, start, router, or_
-from crewai import Crew, Agent, Task
+from crewai import Crew, Agent, Task, Process
 from pydantic import BaseModel
 
 # Define structured state for precise control
@@ -479,7 +488,7 @@ class AdvancedAnalysisFlow(Flow[MarketState]):
         )
         return strategy_crew.kickoff()
 
-    @listen("medium_confidence", "low_confidence")
+    @listen(or_("medium_confidence", "low_confidence"))
     def request_additional_analysis(self):
         self.state.recommendations.append("Gather more data")
         return "Additional analysis required"

docs/changelog.mdx

@@ -4,6 +4,36 @@ description: View the latest updates and changes to CrewAI
 icon: timeline
 ---
 
+<Update label="2025-04-07" description="v0.114.0">
+  ## Release Highlights
+  <Frame>
+    <img src="/images/v01140.png" />
+  </Frame>
+
+  **New Features & Enhancements**
+  - Agents as an atomic unit. (`Agent(...).kickoff()`)
+  - Support for [Custom LLM implementations](https://docs.crewai.com/guides/advanced/custom-llm).
+  - Integrated External Memory and [Opik observability](https://docs.crewai.com/how-to/opik-observability).
+  - Enhanced YAML extraction.
+  - Multimodal agent validation.
+  - Added Secure fingerprints for agents and crews.
+
+  **Core Improvements & Fixes**
+  - Improved serialization, agent copying, and Python compatibility.
+  - Added wildcard support to `emit()`
+  - Added support for additional router calls and context window adjustments.
+  - Fixed typing issues, validation, and import statements.
+  - Improved method performance.
+  - Enhanced agent task handling, event emissions, and memory management.
+  - Fixed CLI issues, conditional tasks, cloning behavior, and tool outputs.
+
+  **Documentation & Guides**
+  - Improved documentation structure, theme, and organization.
+  - Added guides for Local NVIDIA NIM with WSL2, W&B Weave, and Arize Phoenix.
+  - Updated tool configuration examples, prompts, and observability docs.
+  - Guide on using singular agents within Flows.
+</Update>
+
 <Update label="2025-03-17" description="v0.108.0">
   **Features**
   - Converted tabs to spaces in `crew.py` template

docs/concepts/agents.mdx

@@ -18,6 +18,18 @@ In the CrewAI framework, an `Agent` is an autonomous unit that can:
 Think of an agent as a specialized team member with specific skills, expertise, and responsibilities. For example, a `Researcher` agent might excel at gathering and analyzing information, while a `Writer` agent might be better at creating content.
 </Tip>
 
+<Note type="info" title="Enterprise Enhancement: Visual Agent Builder">
+CrewAI Enterprise includes a Visual Agent Builder that simplifies agent creation and configuration without writing code. Design your agents visually and test them in real-time.
+
+![Visual Agent Builder Screenshot](../images/enterprise/crew-studio-quickstart)
+
+The Visual Agent Builder enables:
+- Intuitive agent configuration with form-based interfaces
+- Real-time testing and validation
+- Template library with pre-configured agent types
+- Easy customization of agent attributes and behaviors
+</Note>
+
 ## Agent Attributes
 
 | Attribute                               | Parameter                | Type                          | Description                                                                                                          |
@@ -106,7 +118,7 @@ class LatestAiDevelopmentCrew():
   @agent
   def researcher(self) -> Agent:
     return Agent(
-      config=self.agents_config['researcher'],
+      config=self.agents_config['researcher'], # type: ignore[index]
       verbose=True,
       tools=[SerperDevTool()]
     )
@@ -114,7 +126,7 @@ class LatestAiDevelopmentCrew():
   @agent
   def reporting_analyst(self) -> Agent:
     return Agent(
-      config=self.agents_config['reporting_analyst'],
+      config=self.agents_config['reporting_analyst'], # type: ignore[index]
       verbose=True
     )
 ```

docs/concepts/collaboration.mdx

@@ -23,8 +23,7 @@ The `Crew` class has been enriched with several attributes to support advanced f
 | **Process Flow** (`process`)           | Defines execution logic (e.g., sequential, hierarchical) for task distribution.                                                                                                                                                |
 | **Verbose Logging** (`verbose`)        | Provides detailed logging for monitoring and debugging. Accepts integer and boolean values to control verbosity level.                                                                                                       |
 | **Rate Limiting** (`max_rpm`)          | Limits requests per minute to optimize resource usage. Setting guidelines depend on task complexity and load.                                                                                                                 |
-| **Internationalization / Customization** (`language`, `prompt_file`) | Supports prompt customization for global usability. [Example of file](https://github.com/joaomdmoura/crewAI/blob/main/src/crewai/translations/en.json)                                                                                      |
+| **Internationalization / Customization** (`prompt_file`) | Supports prompt customization for global usability. [Example of file](https://github.com/joaomdmoura/crewAI/blob/main/src/crewai/translations/en.json)                                                                                      |
-| **Execution and Output Handling** (`full_output`) | Controls output granularity, distinguishing between full and final outputs.                                                                                                                                                  |
 | **Callback and Telemetry** (`step_callback`, `task_callback`) | Enables step-wise and task-level execution monitoring and telemetry for performance analytics.                                                                                                                               |
 | **Crew Sharing** (`share_crew`)        | Allows sharing crew data with CrewAI for model improvement. Privacy implications and benefits should be considered.                                                                                                          |
 | **Usage Metrics** (`usage_metrics`)    | Logs all LLM usage metrics during task execution for performance insights.                                                                                                                           |

docs/concepts/crews.mdx

@@ -20,13 +20,10 @@ A crew in crewAI represents a collaborative group of agents working together to
 | **Function Calling LLM** _(optional)_ | `function_calling_llm` | If passed, the crew will use this LLM to do function calling for tools for all agents in the crew. Each agent can have its own LLM, which overrides the crew's LLM for function calling.                                                                  |
 | **Config** _(optional)_               | `config`               | Optional configuration settings for the crew, in `Json` or `Dict[str, Any]` format.                                                                                                                                                                       |
 | **Max RPM** _(optional)_              | `max_rpm`              | Maximum requests per minute the crew adheres to during execution. Defaults to `None`.                                                                                                                                                                     |
-| **Language** _(optional)_             | `language`             | Language used for the crew, defaults to English.                                                                                                                                                                                                          |
-| **Language File** _(optional)_        | `language_file`        | Path to the language file to be used for the crew.                                                                                                                                                                                                        |
 | **Memory** _(optional)_               | `memory`               | Utilized for storing execution memories (short-term, long-term, entity memory).                                                                                                                                                                           |
 | **Memory Config** _(optional)_        | `memory_config`        | Configuration for the memory provider to be used by the crew.                                                                                                                                                                                             |
 | **Cache** _(optional)_                | `cache`                | Specifies whether to use a cache for storing the results of tools' execution. Defaults to `True`.                                                                                                                                                         |
 | **Embedder** _(optional)_             | `embedder`             | Configuration for the embedder to be used by the crew. Mostly used by memory for now. Default is `{"provider": "openai"}`.                                                                                                                                |
-| **Full Output** _(optional)_          | `full_output`          | Whether the crew should return the full output with all tasks outputs or just the final output. Defaults to `False`.                                                                                                                                      |
 | **Step Callback** _(optional)_        | `step_callback`        | A function that is called after each step of every agent. This can be used to log the agent's actions or to perform other operations; it won't override the agent-specific `step_callback`.                                                               |
 | **Task Callback** _(optional)_        | `task_callback`        | A function that is called after the completion of each task. Useful for monitoring or additional operations post-task execution.                                                                                                                          |
 | **Share Crew** _(optional)_           | `share_crew`           | Whether you want to share the complete crew information and execution with the crewAI team to make the library better, and allow us to train models.                                                                                                      |
@@ -55,12 +52,16 @@ After creating your CrewAI project as outlined in the [Installation](/installati
 ```python code
 from crewai import Agent, Crew, Task, Process
 from crewai.project import CrewBase, agent, task, crew, before_kickoff, after_kickoff
-
+from crewai.agents.agent_builder.base_agent import BaseAgent
+from typing import List
 
 @CrewBase
 class YourCrewName:
     """Description of your crew"""
 
+    agents: List[BaseAgent]
+    tasks: List[Task]
+
     # Paths to your YAML configuration files
     # To see an example agent and task defined in YAML, checkout the following:
     # - Task: https://docs.crewai.com/concepts/tasks#yaml-configuration-recommended
@@ -83,27 +84,27 @@ class YourCrewName:
     @agent
     def agent_one(self) -> Agent:
         return Agent(
-            config=self.agents_config['agent_one'],
+            config=self.agents_config['agent_one'], # type: ignore[index]
             verbose=True
         )
 
     @agent
     def agent_two(self) -> Agent:
         return Agent(
-            config=self.agents_config['agent_two'],
+            config=self.agents_config['agent_two'], # type: ignore[index]
             verbose=True
         )
 
     @task
     def task_one(self) -> Task:
         return Task(
-            config=self.tasks_config['task_one']
+            config=self.tasks_config['task_one'] # type: ignore[index]
         )
 
     @task
     def task_two(self) -> Task:
         return Task(
-            config=self.tasks_config['task_two']
+            config=self.tasks_config['task_two'] # type: ignore[index]
         )
 
     @crew

docs/concepts/event-listener.mdx

@@ -13,11 +13,25 @@ CrewAI provides a powerful event system that allows you to listen for and react
 CrewAI uses an event bus architecture to emit events throughout the execution lifecycle. The event system is built on the following components:
 
 1. **CrewAIEventsBus**: A singleton event bus that manages event registration and emission
-2. **CrewEvent**: Base class for all events in the system
+2. **BaseEvent**: Base class for all events in the system
 3. **BaseEventListener**: Abstract base class for creating custom event listeners
 
 When specific actions occur in CrewAI (like a Crew starting execution, an Agent completing a task, or a tool being used), the system emits corresponding events. You can register handlers for these events to execute custom code when they occur.
 
+<Note type="info" title="Enterprise Enhancement: Prompt Tracing">
+CrewAI Enterprise provides a built-in Prompt Tracing feature that leverages the event system to track, store, and visualize all prompts, completions, and associated metadata. This provides powerful debugging capabilities and transparency into your agent operations.
+
+![Prompt Tracing Dashboard](../images/enterprise/prompt-tracing.png)
+
+With Prompt Tracing you can:
+- View the complete history of all prompts sent to your LLM
+- Track token usage and costs
+- Debug agent reasoning failures
+- Share prompt sequences with your team
+- Compare different prompt strategies
+- Export traces for compliance and auditing
+</Note>
+
 ## Creating a Custom Event Listener
 
 To create a custom event listener, you need to:
@@ -234,7 +248,7 @@ Each event handler receives two parameters:
 1. **source**: The object that emitted the event
 2. **event**: The event instance, containing event-specific data
 
-The structure of the event object depends on the event type, but all events inherit from `CrewEvent` and include:
+The structure of the event object depends on the event type, but all events inherit from `BaseEvent` and include:
 
 - **timestamp**: The time when the event was emitted
 - **type**: A string identifier for the event type

docs/concepts/flows.mdx

@@ -545,6 +545,119 @@ The `third_method` and `fourth_method` listen to the output of the `second_metho
 
 When you run this Flow, the output will change based on the random boolean value generated by the `start_method`.
 
+## Adding Agents to Flows
+
+Agents can be seamlessly integrated into your flows, providing a lightweight alternative to full Crews when you need simpler, focused task execution. Here's an example of how to use an Agent within a flow to perform market research:
+
+```python
+import asyncio
+from typing import Any, Dict, List
+
+from crewai_tools import SerperDevTool
+from pydantic import BaseModel, Field
+
+from crewai.agent import Agent
+from crewai.flow.flow import Flow, listen, start
+
+
+# Define a structured output format
+class MarketAnalysis(BaseModel):
+    key_trends: List[str] = Field(description="List of identified market trends")
+    market_size: str = Field(description="Estimated market size")
+    competitors: List[str] = Field(description="Major competitors in the space")
+
+
+# Define flow state
+class MarketResearchState(BaseModel):
+    product: str = ""
+    analysis: MarketAnalysis | None = None
+
+
+# Create a flow class
+class MarketResearchFlow(Flow[MarketResearchState]):
+    @start()
+    def initialize_research(self) -> Dict[str, Any]:
+        print(f"Starting market research for {self.state.product}")
+        return {"product": self.state.product}
+
+    @listen(initialize_research)
+    async def analyze_market(self) -> Dict[str, Any]:
+        # Create an Agent for market research
+        analyst = Agent(
+            role="Market Research Analyst",
+            goal=f"Analyze the market for {self.state.product}",
+            backstory="You are an experienced market analyst with expertise in "
+            "identifying market trends and opportunities.",
+            tools=[SerperDevTool()],
+            verbose=True,
+        )
+
+        # Define the research query
+        query = f"""
+        Research the market for {self.state.product}. Include:
+        1. Key market trends
+        2. Market size
+        3. Major competitors
+
+        Format your response according to the specified structure.
+        """
+
+        # Execute the analysis with structured output format
+        result = await analyst.kickoff_async(query, response_format=MarketAnalysis)
+        if result.pydantic:
+            print("result", result.pydantic)
+        else:
+            print("result", result)
+
+        # Return the analysis to update the state
+        return {"analysis": result.pydantic}
+
+    @listen(analyze_market)
+    def present_results(self, analysis) -> None:
+        print("\nMarket Analysis Results")
+        print("=====================")
+
+        if isinstance(analysis, dict):
+            # If we got a dict with 'analysis' key, extract the actual analysis object
+            market_analysis = analysis.get("analysis")
+        else:
+            market_analysis = analysis
+
+        if market_analysis and isinstance(market_analysis, MarketAnalysis):
+            print("\nKey Market Trends:")
+            for trend in market_analysis.key_trends:
+                print(f"- {trend}")
+
+            print(f"\nMarket Size: {market_analysis.market_size}")
+
+            print("\nMajor Competitors:")
+            for competitor in market_analysis.competitors:
+                print(f"- {competitor}")
+        else:
+            print("No structured analysis data available.")
+            print("Raw analysis:", analysis)
+
+
+# Usage example
+async def run_flow():
+    flow = MarketResearchFlow()
+    result = await flow.kickoff_async(inputs={"product": "AI-powered chatbots"})
+    return result
+
+
+# Run the flow
+if __name__ == "__main__":
+    asyncio.run(run_flow())
+```
+
+This example demonstrates several key features of using Agents in flows:
+
+1. **Structured Output**: Using Pydantic models to define the expected output format (`MarketAnalysis`) ensures type safety and structured data throughout the flow.
+
+2. **State Management**: The flow state (`MarketResearchState`) maintains context between steps and stores both inputs and outputs.
+
+3. **Tool Integration**: Agents can use tools (like `WebsiteSearchTool`) to enhance their capabilities.
+
 ## Adding Crews to Flows
 
 Creating a flow with multiple crews in CrewAI is straightforward.
@@ -677,6 +790,9 @@ Visualizing your AI workflows can provide valuable insights into the structure a
 
 Plots in CrewAI are graphical representations of your AI workflows. They display the various tasks, their connections, and the flow of data between them. This visualization helps in understanding the sequence of operations, identifying bottlenecks, and ensuring that the workflow logic aligns with your expectations.
 
+![Example of a Flow Plot](/images/flow_plot_example.png)
+*An example visualization of a simple flow with start method, sequential steps, and directional execution.*
+
 ### How to Generate a Plot
 
 CrewAI provides two convenient methods to generate plots of your flows:

docs/concepts/knowledge.mdx

@@ -42,6 +42,16 @@ CrewAI supports various types of knowledge sources out of the box:
 | `collection_name`          | **str**                              | No       | Name of the collection where the knowledge will be stored. Used to identify different sets of knowledge. Defaults to "knowledge" if not provided.     |
 | `storage`                  | **Optional[KnowledgeStorage]**       | No       | Custom storage configuration for managing how the knowledge is stored and retrieved. If not provided, a default storage will be created.              |
 
+
+<Tip>
+Unlike retrieval from a vector database using a tool, agents preloaded with knowledge will not need a retrieval persona or task.
+Simply add the relevant knowledge sources your agent or crew needs to function.
+
+Knowledge sources can be added at the agent or crew level.
+Crew level knowledge sources will be used by **all agents** in the crew.
+Agent level knowledge sources will be used by the **specific agent** that is preloaded with the knowledge.
+</Tip>
+
 ## Quickstart Example
 
 <Tip>
@@ -146,6 +156,26 @@ result = crew.kickoff(
 )
 ```
 
+## Knowledge Configuration
+
+You can configure the knowledge configuration for the crew or agent.
+
+```python Code
+from crewai.knowledge.knowledge_config import KnowledgeConfig
+
+knowledge_config = KnowledgeConfig(results_limit=10, score_threshold=0.5)
+
+agent = Agent(
+    ...
+    knowledge_config=knowledge_config
+)
+```
+
+<Tip>
+  `results_limit`: is the number of relevant documents to return. Default is 3.
+  `score_threshold`: is the minimum score for a document to be considered relevant. Default is 0.35.
+</Tip>
+
 ## More Examples
 
 Here are examples of how to use different types of knowledge sources:

docs/concepts/llamaindex-tools.mdx

@@ -1,71 +0,0 @@
----
-title: Using LlamaIndex Tools
-description: Learn how to integrate LlamaIndex tools with CrewAI agents to enhance search-based queries and more.
-icon: toolbox
----
-
-## Using LlamaIndex Tools
-
-<Info>
-    CrewAI seamlessly integrates with LlamaIndex’s comprehensive toolkit for RAG (Retrieval-Augmented Generation) and agentic pipelines, enabling advanced search-based queries and more. 
-</Info>
-
-Here are the available built-in tools offered by LlamaIndex.
-
-```python Code
-from crewai import Agent
-from crewai_tools import LlamaIndexTool
-
-# Example 1: Initialize from FunctionTool
-from llama_index.core.tools import FunctionTool
-
-your_python_function = lambda ...: ...
-og_tool = FunctionTool.from_defaults(
-    your_python_function, 
-    name="<name>", 
-    description='<description>'
-)
-tool = LlamaIndexTool.from_tool(og_tool)
-
-# Example 2: Initialize from LlamaHub Tools
-from llama_index.tools.wolfram_alpha import WolframAlphaToolSpec
-wolfram_spec = WolframAlphaToolSpec(app_id="<app_id>")
-wolfram_tools = wolfram_spec.to_tool_list()
-tools = [LlamaIndexTool.from_tool(t) for t in wolfram_tools]
-
-# Example 3: Initialize Tool from a LlamaIndex Query Engine
-query_engine = index.as_query_engine()
-query_tool = LlamaIndexTool.from_query_engine(
-    query_engine,
-    name="Uber 2019 10K Query Tool",
-    description="Use this tool to lookup the 2019 Uber 10K Annual Report"
-)
-
-# Create and assign the tools to an agent
-agent = Agent(
-    role='Research Analyst',
-    goal='Provide up-to-date market analysis',
-    backstory='An expert analyst with a keen eye for market trends.',
-    tools=[tool, *tools, query_tool]
-)
-
-# rest of the code ...
-```
-
-## Steps to Get Started
-
-To effectively use the LlamaIndexTool, follow these steps:
-
-<Steps>
-    <Step title="Package Installation">
-    Make sure that `crewai[tools]` package is installed in your Python environment:
-    <CodeGroup>
-        ```shell Terminal
-        pip install 'crewai[tools]'
-        ```
-    </CodeGroup>
-    </Step>
-    <Step title="Install and Use LlamaIndex">
-    Follow the LlamaIndex documentation [LlamaIndex Documentation](https://docs.llamaindex.ai/) to set up a RAG/agent pipeline.
-    </Step>
-</Steps>

docs/concepts/llms.mdx

@@ -438,7 +438,7 @@ In this section, you'll find detailed examples that help you select, configure,
         @agent
         def researcher(self) -> Agent:
             return Agent(
-                config=self.agents_config['researcher'],
+                config=self.agents_config['researcher'], # type: ignore[index]
                 llm=local_nvidia_nim_llm
             )
         
@@ -535,14 +535,13 @@ In this section, you'll find detailed examples that help you select, configure,
   <Accordion title="Hugging Face">
     Set the following environment variables in your `.env` file:
     ```toml Code
-    HUGGINGFACE_API_KEY=<your-api-key>
+    HF_TOKEN=<your-api-key>
     ```
 
     Example usage in your CrewAI project:
     ```python Code
     llm = LLM(
-        model="huggingface/meta-llama/Meta-Llama-3.1-8B-Instruct",
+        model="huggingface/meta-llama/Meta-Llama-3.1-8B-Instruct"
-        base_url="your_api_endpoint"
     )
     ```
   </Accordion>

docs/concepts/memory.mdx

@@ -18,7 +18,8 @@ reason, and learn from past interactions.
 | **Long-Term Memory** | Preserves valuable insights and learnings from past executions, allowing agents to build and refine their knowledge over time. |
 | **Entity Memory**    | Captures and organizes information about entities (people, places, concepts) encountered during tasks, facilitating deeper understanding and relationship mapping. Uses `RAG` for storing entity information. |
 | **Contextual Memory**| Maintains the context of interactions by combining `ShortTermMemory`, `LongTermMemory`, and `EntityMemory`, aiding in the coherence and relevance of agent responses over a sequence of tasks or a conversation. |
-| **User Memory**      | Stores user-specific information and preferences, enhancing personalization and user experience. |
+| **External Memory**  | Enables integration with external memory systems and providers (like Mem0), allowing for specialized memory storage and retrieval across different applications. Supports custom storage implementations for flexible memory management. |
+| **User Memory**      | ⚠️ **DEPRECATED**: This component is deprecated and will be removed in a future version. Please use [External Memory](#using-external-memory) instead. |
 
 ## How Memory Systems Empower Agents
 
@@ -144,6 +145,7 @@ from crewai.memory import LongTermMemory
 # Simple memory configuration
 crew = Crew(memory=True)  # Uses default storage locations
 ```
+Note that External Memory won’t be defined when `memory=True` is set, as we can’t infer which external memory would be suitable for your case
 
 ### Custom Storage Configuration
 ```python
@@ -164,7 +166,10 @@ crew = Crew(
 
 [Mem0](https://mem0.ai/) is a self-improving memory layer for LLM applications, enabling personalized AI experiences. 
 
-To include user-specific memory you can get your API key [here](https://app.mem0.ai/dashboard/api-keys) and refer the [docs](https://docs.mem0.ai/platform/quickstart#4-1-create-memories) for adding user preferences.
+
+### Using Mem0 API platform
+
+To include user-specific memory you can get your API key [here](https://app.mem0.ai/dashboard/api-keys) and refer the [docs](https://docs.mem0.ai/platform/quickstart#4-1-create-memories) for adding user preferences. In this case `user_memory` is set to `MemoryClient` from mem0.
 
 
 ```python Code
@@ -175,18 +180,7 @@ from mem0 import MemoryClient
 # Set environment variables for Mem0
 os.environ["MEM0_API_KEY"] = "m0-xx"
 
-# Step 1: Record preferences based on past conversation or user input
+# Step 1: Create a Crew with User Memory
-client = MemoryClient()
-messages = [
-    {"role": "user", "content": "Hi there! I'm planning a vacation and could use some advice."},
-    {"role": "assistant", "content": "Hello! I'd be happy to help with your vacation planning. What kind of destination do you prefer?"},
-    {"role": "user", "content": "I am more of a beach person than a mountain person."},
-    {"role": "assistant", "content": "That's interesting. Do you like hotels or Airbnb?"},
-    {"role": "user", "content": "I like Airbnb more."},
-]
-client.add(messages, user_id="john")
-
-# Step 2: Create a Crew with User Memory
 
 crew = Crew(
     agents=[...],
@@ -197,11 +191,12 @@ crew = Crew(
     memory_config={
         "provider": "mem0",
         "config": {"user_id": "john"},
+        "user_memory" : {} #Set user_memory explicitly to a dictionary, we are working on this issue.
     },
 )
 ```
 
-## Memory Configuration Options
+#### Additional Memory Configuration Options
 If you want to access a specific organization and project, you can set the `org_id` and `project_id` parameters in the memory configuration.
 
 ```python Code
@@ -215,10 +210,172 @@ crew = Crew(
     memory_config={
         "provider": "mem0",
         "config": {"user_id": "john", "org_id": "my_org_id", "project_id": "my_project_id"},
+        "user_memory" : {} #Set user_memory explicitly to a dictionary, we are working on this issue.
     },
 )
 ```
 
+### Using Local Mem0 memory
+If you want to use local mem0 memory, with a custom configuration, you can set a parameter `local_mem0_config` in the config itself.
+If both os environment key is set and local_mem0_config is given, the API platform takes higher priority over the local configuration.
+Check [this](https://docs.mem0.ai/open-source/python-quickstart#run-mem0-locally) mem0 local configuration docs for more understanding.
+In this case `user_memory` is set to `Memory` from mem0.
+
+
+```python Code
+from crewai import Crew
+
+
+#local mem0 config
+config = {
+    "vector_store": {
+        "provider": "qdrant",
+        "config": {
+            "host": "localhost",
+            "port": 6333
+        }
+    },
+    "llm": {
+        "provider": "openai",
+        "config": {
+            "api_key": "your-api-key",
+            "model": "gpt-4"
+        }
+    },
+    "embedder": {
+        "provider": "openai",
+        "config": {
+            "api_key": "your-api-key",
+            "model": "text-embedding-3-small"
+        }
+    },
+    "graph_store": {
+        "provider": "neo4j",
+        "config": {
+            "url": "neo4j+s://your-instance",
+            "username": "neo4j",
+            "password": "password"
+        }
+    },
+    "history_db_path": "/path/to/history.db",
+    "version": "v1.1",
+    "custom_fact_extraction_prompt": "Optional custom prompt for fact extraction for memory",
+    "custom_update_memory_prompt": "Optional custom prompt for update memory"
+}
+
+crew = Crew(
+    agents=[...],
+    tasks=[...],
+    verbose=True,
+    memory=True,
+    memory_config={
+        "provider": "mem0",
+        "config": {"user_id": "john", 'local_mem0_config': config},
+        "user_memory" : {} #Set user_memory explicitly to a dictionary, we are working on this issue.
+    },
+)
+```
+
+### Using External Memory
+
+External Memory is a powerful feature that allows you to integrate external memory systems with your CrewAI applications. This is particularly useful when you want to use specialized memory providers or maintain memory across different applications.
+Since it’s an external memory, we’re not able to add a default value to it - unlike with Long Term and Short Term memory. 
+
+#### Basic Usage with Mem0
+
+The most common way to use External Memory is with Mem0 as the provider:
+
+```python
+import os
+from crewai import Agent, Crew, Process, Task
+from crewai.memory.external.external_memory import ExternalMemory
+
+os.environ["MEM0_API_KEY"] = "YOUR-API-KEY"
+
+agent = Agent(
+    role="You are a helpful assistant",
+    goal="Plan a vacation for the user",
+    backstory="You are a helpful assistant that can plan a vacation for the user",
+    verbose=True,
+)
+task = Task(
+    description="Give things related to the user's vacation",
+    expected_output="A plan for the vacation",
+    agent=agent,
+)
+
+crew = Crew(
+    agents=[agent],
+    tasks=[task],
+    verbose=True,
+    process=Process.sequential,
+    external_memory=ExternalMemory(
+        embedder_config={"provider": "mem0", "config": {"user_id": "U-123"}} # you can provide an entire Mem0 configuration
+    ),
+)
+
+crew.kickoff(
+    inputs={"question": "which destination is better for a beach vacation?"}
+)
+```
+
+#### Using External Memory with Custom Storage
+
+You can also create custom storage implementations for External Memory. Here's an example of how to create a custom storage:
+
+```python
+from crewai import Agent, Crew, Process, Task
+from crewai.memory.external.external_memory import ExternalMemory
+from crewai.memory.storage.interface import Storage
+
+
+class CustomStorage(Storage):
+    def __init__(self):
+        self.memories = []
+
+    def save(self, value, metadata=None, agent=None):
+        self.memories.append({"value": value, "metadata": metadata, "agent": agent})
+
+    def search(self, query, limit=10, score_threshold=0.5):
+        # Implement your search logic here
+        return []
+
+    def reset(self):
+        self.memories = []
+
+
+# Create external memory with custom storage
+external_memory = ExternalMemory(
+    storage=CustomStorage(),
+    embedder_config={"provider": "mem0", "config": {"user_id": "U-123"}},
+)
+
+agent = Agent(
+    role="You are a helpful assistant",
+    goal="Plan a vacation for the user",
+    backstory="You are a helpful assistant that can plan a vacation for the user",
+    verbose=True,
+)
+task = Task(
+    description="Give things related to the user's vacation",
+    expected_output="A plan for the vacation",
+    agent=agent,
+)
+
+crew = Crew(
+    agents=[agent],
+    tasks=[task],
+    verbose=True,
+    process=Process.sequential,
+    external_memory=external_memory,
+)
+
+crew.kickoff(
+    inputs={"question": "which destination is better for a beach vacation?"}
+)
+```
+
+
 ## Additional Embedding Providers
 
 ### Using OpenAI embeddings (already default)

docs/concepts/tasks.mdx

@@ -12,6 +12,18 @@ Tasks provide all necessary details for execution, such as a description, the ag
 
 Tasks within CrewAI can be collaborative, requiring multiple agents to work together. This is managed through the task properties and orchestrated by the Crew's process, enhancing teamwork and efficiency.
 
+<Note type="info" title="Enterprise Enhancement: Visual Task Builder">
+CrewAI Enterprise includes a Visual Task Builder in Crew Studio that simplifies complex task creation and chaining. Design your task flows visually and test them in real-time without writing code.
+
+![Task Builder Screenshot](../images/enterprise/crew-studio-quickstart.png)
+
+The Visual Task Builder enables:
+- Drag-and-drop task creation
+- Visual task dependencies and flow
+- Real-time testing and validation
+- Easy sharing and collaboration
+</Note>
+
 ### Task Execution Flow
 
 Tasks can be executed in two ways:
@@ -101,7 +113,7 @@ class LatestAiDevelopmentCrew():
   @agent
   def researcher(self) -> Agent:
     return Agent(
-      config=self.agents_config['researcher'],
+      config=self.agents_config['researcher'], # type: ignore[index]
       verbose=True,
       tools=[SerperDevTool()]
     )
@@ -109,20 +121,20 @@ class LatestAiDevelopmentCrew():
   @agent
   def reporting_analyst(self) -> Agent:
     return Agent(
-      config=self.agents_config['reporting_analyst'],
+      config=self.agents_config['reporting_analyst'], # type: ignore[index]
       verbose=True
     )
 
   @task
   def research_task(self) -> Task:
     return Task(
-      config=self.tasks_config['research_task']
+      config=self.tasks_config['research_task'] # type: ignore[index]
     )
 
   @task
   def reporting_task(self) -> Task:
     return Task(
-      config=self.tasks_config['reporting_task']
+      config=self.tasks_config['reporting_task'] # type: ignore[index]
     )
 
   @crew
@@ -276,26 +288,20 @@ To add a guardrail to a task, provide a validation function through the `guardra
 
 ```python Code
 from typing import Tuple, Union, Dict, Any
+from crewai import TaskOutput
 
-def validate_blog_content(result: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
+def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
     """Validate blog content meets requirements."""
     try:
         # Check word count
         word_count = len(result.split())
         if word_count > 200:
-            return (False, {
+            return (False, "Blog content exceeds 200 words")
-                "error": "Blog content exceeds 200 words",
-                "code": "WORD_COUNT_ERROR",
-                "context": {"word_count": word_count}
-            })
 
         # Additional validation logic here
         return (True, result.strip())
     except Exception as e:
-        return (False, {
+        return (False, "Unexpected error during validation")
-            "error": "Unexpected error during validation",
-            "code": "SYSTEM_ERROR"
-        })
 
 blog_task = Task(
     description="Write a blog post about AI",
@@ -313,29 +319,24 @@ blog_task = Task(
    - Type hints are recommended but optional
 
 2. **Return Values**:
-   - Success: Return `(True, validated_result)`
+   - On success: it returns a tuple of `(bool, Any)`. For example: `(True, validated_result)` 
-   - Failure: Return `(False, error_details)`
+   - On Failure: it returns a tuple of `(bool, str)`. For example: `(False, "Error message explain the failure")`
 
 ### Error Handling Best Practices
 
 1. **Structured Error Responses**:
 ```python Code
-def validate_with_context(result: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
+from crewai import TaskOutput
+
+def validate_with_context(result: TaskOutput) -> Tuple[bool, Any]:
     try:
         # Main validation logic
         validated_data = perform_validation(result)
         return (True, validated_data)
     except ValidationError as e:
-        return (False, {
+        return (False, f"VALIDATION_ERROR: {str(e)}")
-            "error": str(e),
-            "code": "VALIDATION_ERROR",
-            "context": {"input": result}
-        })
     except Exception as e:
-        return (False, {
+        return (False, str(e))
-            "error": "Unexpected error",
-            "code": "SYSTEM_ERROR"
-        })
 ```
 
 2. **Error Categories**:
@@ -346,28 +347,25 @@ def validate_with_context(result: str) -> Tuple[bool, Union[Dict[str, Any], str]
 3. **Validation Chain**:
 ```python Code
 from typing import Any, Dict, List, Tuple, Union
+from crewai import TaskOutput
 
-def complex_validation(result: str) -> Tuple[bool, Union[str, Dict[str, Any]]]:
+def complex_validation(result: TaskOutput) -> Tuple[bool, Any]:
     """Chain multiple validation steps."""
     # Step 1: Basic validation
     if not result:
-        return (False, {"error": "Empty result", "code": "EMPTY_INPUT"})
+        return (False, "Empty result")
 
     # Step 2: Content validation
     try:
         validated = validate_content(result)
         if not validated:
-            return (False, {"error": "Invalid content", "code": "CONTENT_ERROR"})
+            return (False, "Invalid content")
 
         # Step 3: Format validation
         formatted = format_output(validated)
         return (True, formatted)
     except Exception as e:
-        return (False, {
+        return (False, str(e))
-            "error": str(e),
-            "code": "VALIDATION_ERROR",
-            "context": {"step": "content_validation"}
-        })
 ```
 
 ### Handling Guardrail Results
@@ -382,19 +380,16 @@ When a guardrail returns `(False, error)`:
 Example with retry handling:
 ```python Code
 from typing import Optional, Tuple, Union
+from crewai import TaskOutput, Task
 
-def validate_json_output(result: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
+def validate_json_output(result: TaskOutput) -> Tuple[bool, Any]:
     """Validate and parse JSON output."""
     try:
         # Try to parse as JSON
         data = json.loads(result)
         return (True, data)
     except json.JSONDecodeError as e:
-        return (False, {
+        return (False, "Invalid JSON format")
-            "error": "Invalid JSON format",
-            "code": "JSON_ERROR",
-            "context": {"line": e.lineno, "column": e.colno}
-        })
 
 task = Task(
     description="Generate a JSON report",
@@ -414,7 +409,7 @@ It's also important to note that the output of the final task of a crew becomes
 ### Using `output_pydantic`
 The `output_pydantic` property allows you to define a Pydantic model that the task output should conform to. This ensures that the output is not only structured but also validated according to the Pydantic model.
 
-Here’s an example demonstrating how to use output_pydantic:
+Here's an example demonstrating how to use output_pydantic:
 
 ```python Code
 import json
@@ -495,7 +490,7 @@ In this example:
 ### Using `output_json`
 The `output_json` property allows you to define the expected output in JSON format. This ensures that the task's output is a valid JSON structure that can be easily parsed and used in your application.
 
-Here’s an example demonstrating how to use `output_json`:
+Here's an example demonstrating how to use `output_json`:
 
 ```python Code
 import json

docs/concepts/tools.mdx

@@ -15,6 +15,18 @@ A tool in CrewAI is a skill or function that agents can utilize to perform vario
 This includes tools from the [CrewAI Toolkit](https://github.com/joaomdmoura/crewai-tools) and [LangChain Tools](https://python.langchain.com/docs/integrations/tools),
 enabling everything from simple searches to complex interactions and effective teamwork among agents.
 
+<Note type="info" title="Enterprise Enhancement: Tools Repository">
+CrewAI Enterprise provides a comprehensive Tools Repository with pre-built integrations for common business systems and APIs. Deploy agents with enterprise tools in minutes instead of days.
+
+![Tools Repository Screenshot](../images/enterprise/tools-repository.png)
+
+The Enterprise Tools Repository includes:
+- Pre-built connectors for popular enterprise systems
+- Custom tool creation interface
+- Version control and sharing capabilities
+- Security and compliance features
+</Note>
+
 ## Key Characteristics of Tools
 
 - **Utility**: Crafted for tasks such as web searching, data analysis, content generation, and agent collaboration.
@@ -79,7 +91,7 @@ research = Task(
 )
 
 write = Task(
-    description='Write an engaging blog post about the AI industry, based on the research analyst’s summary. Draw inspiration from the latest blog posts in the directory.',
+    description='Write an engaging blog post about the AI industry, based on the research analyst's summary. Draw inspiration from the latest blog posts in the directory.',
     expected_output='A 4-paragraph blog post formatted in markdown with engaging, informative, and accessible content, avoiding complex jargon.',
     agent=writer,
     output_file='blog-posts/new_post.md'  # The final blog post will be saved here
@@ -141,7 +153,7 @@ Here is a list of the available tools and their descriptions:
 ## Creating your own Tools
 
 <Tip>
-  Developers can craft `custom tools` tailored for their agent’s needs or
+  Developers can craft `custom tools` tailored for their agent's needs or
   utilize pre-built options.
 </Tip>
 

docs/docs.json

@@ -8,25 +8,27 @@
     "dark": "#C94C3C"
   },
   "favicon": "favicon.svg",
+  "contextual": {
+    "options": ["copy", "view", "chatgpt", "claude"]
+  },
   "navigation": {
     "tabs": [
       {
-        "tab": "Get Started",
+        "tab": "Documentation",
         "groups": [
           {
             "group": "Get Started",
             "pages": [
               "introduction",
               "installation",
-              "quickstart",
+              "quickstart"
-              "changelog"
             ]
           },
           {
             "group": "Guides",
             "pages": [
               {
-                "group": "Concepts",
+                "group": "Strategy",
                 "pages": [
                   "guides/concepts/evaluating-use-cases"
                 ]
@@ -76,41 +78,7 @@
               "concepts/testing",
               "concepts/cli",
               "concepts/tools",
-              "concepts/event-listener",
+              "concepts/event-listener"
-              "concepts/langchain-tools",
-              "concepts/llamaindex-tools"
-            ]
-          },
-          {
-            "group": "How to Guides",
-            "pages": [
-              "how-to/create-custom-tools",
-              "how-to/sequential-process",
-              "how-to/hierarchical-process",
-              "how-to/custom-manager-agent",
-              "how-to/llm-connections",
-              "how-to/customizing-agents",
-              "how-to/multimodal-agents",
-              "how-to/coding-agents",
-              "how-to/force-tool-output-as-result",
-              "how-to/human-input-on-execution",
-              "how-to/kickoff-async",
-              "how-to/kickoff-for-each",
-              "how-to/replay-tasks-from-latest-crew-kickoff",
-              "how-to/conditional-tasks"
-            ]
-          },
-          {
-            "group": "Agent Monitoring & Observability",
-            "pages": [
-              "how-to/weave-integration",
-              "how-to/agentops-observability",
-              "how-to/langfuse-observability",
-              "how-to/langtrace-observability",
-              "how-to/mlflow-observability",
-              "how-to/openlit-observability",
-              "how-to/opik-observability",
-              "how-to/portkey-observability"
             ]
           },
           {
@@ -140,6 +108,7 @@
               "tools/hyperbrowserloadtool",
               "tools/linkupsearchtool",
               "tools/llamaindextool",
+              "tools/langchaintool",
               "tools/serperdevtool",
               "tools/s3readertool",
               "tools/s3writertool",
@@ -169,6 +138,40 @@
               "tools/youtubevideosearchtool"
             ]
           },
+          {
+            "group": "Agent Monitoring & Observability",
+            "pages": [
+              "how-to/agentops-observability",
+              "how-to/arize-phoenix-observability",
+              "how-to/langfuse-observability",
+              "how-to/langtrace-observability",
+              "how-to/mlflow-observability",
+              "how-to/openlit-observability",
+              "how-to/opik-observability",
+              "how-to/portkey-observability",
+              "how-to/weave-integration"
+            ]
+          },
+          {
+            "group": "Learn",
+            "pages": [
+              "how-to/conditional-tasks",
+              "how-to/coding-agents",
+              "how-to/create-custom-tools",
+              "how-to/custom-llm",
+              "how-to/custom-manager-agent",
+              "how-to/customizing-agents",
+              "how-to/force-tool-output-as-result",
+              "how-to/hierarchical-process",
+              "how-to/human-input-on-execution",
+              "how-to/kickoff-async",
+              "how-to/kickoff-for-each",
+              "how-to/llm-connections",
+              "how-to/multimodal-agents",
+              "how-to/replay-tasks-from-latest-crew-kickoff",
+              "how-to/sequential-process"
+            ]
+          },
           {
             "group": "Telemetry",
             "pages": [
@@ -187,14 +190,35 @@
             ]
           }
         ]
+      },
+      {
+        "tab": "Releases",
+        "groups": [
+          {
+            "group": "Releases",
+            "pages": [
+              "changelog"
+            ]
+          }
+        ]
       }
     ],
     "global": {
       "anchors": [
         {
-          "anchor": "Community",
+          "anchor": "Website",
+          "href": "https://crewai.com",
+          "icon": "globe"
+        },
+        {
+          "anchor": "Forum",
           "href": "https://community.crewai.com",
           "icon": "discourse"
+        },
+        {
+          "anchor": "Get Help",
+          "href": "mailto:support@crewai.com",
+          "icon": "headset"
         }
       ]
     }
@@ -208,6 +232,12 @@
     "strict": false
   },
   "navbar": {
+    "links": [
+      {
+        "label": "Start Free Trial",
+        "href": "https://app.crewai.com"
+      }
+    ],
     "primary": {
       "type": "github",
       "href": "https://github.com/crewAIInc/crewAI"
@@ -217,7 +247,12 @@
     "prompt": "Search CrewAI docs"
   },
   "seo": {
-    "indexing": "navigable"
+    "indexing": "all"
+  },
+  "errors": {
+    "404": {
+      "redirect": true
+    }
   },
   "footer": {
     "socials": {

docs/guides/crews/first-crew.mdx

@@ -185,15 +185,20 @@ Let's modify the `crew.py` file:
 from crewai import Agent, Crew, Process, Task
 from crewai.project import CrewBase, agent, crew, task
 from crewai_tools import SerperDevTool
+from crewai.agents.agent_builder.base_agent import BaseAgent
+from typing import List
 
 @CrewBase
 class ResearchCrew():
     """Research crew for comprehensive topic analysis and reporting"""
 
+    agents: List[BaseAgent]
+    tasks: List[Task]
+
     @agent
     def researcher(self) -> Agent:
         return Agent(
-            config=self.agents_config['researcher'],
+            config=self.agents_config['researcher'], # type: ignore[index]
             verbose=True,
             tools=[SerperDevTool()]
         )
@@ -201,20 +206,20 @@ class ResearchCrew():
     @agent
     def analyst(self) -> Agent:
         return Agent(
-            config=self.agents_config['analyst'],
+            config=self.agents_config['analyst'], # type: ignore[index]
             verbose=True
         )
 
     @task
     def research_task(self) -> Task:
         return Task(
-            config=self.tasks_config['research_task']
+            config=self.tasks_config['research_task'] # type: ignore[index]
         )
 
     @task
     def analysis_task(self) -> Task:
         return Task(
-            config=self.tasks_config['analysis_task'],
+            config=self.tasks_config['analysis_task'], # type: ignore[index]
             output_file='output/report.md'
         )
 

docs/guides/flows/first-flow.mdx

@@ -203,35 +203,40 @@ These task definitions provide detailed instructions to our agents, ensuring the
 # src/guide_creator_flow/crews/content_crew/content_crew.py
 from crewai import Agent, Crew, Process, Task
 from crewai.project import CrewBase, agent, crew, task
+from crewai.agents.agent_builder.base_agent import BaseAgent
+from typing import List
 
 @CrewBase
 class ContentCrew():
     """Content writing crew"""
 
+    agents: List[BaseAgent]
+    tasks: List[Task]
+
     @agent
     def content_writer(self) -> Agent:
         return Agent(
-            config=self.agents_config['content_writer'],
+            config=self.agents_config['content_writer'], # type: ignore[index]
             verbose=True
         )
 
     @agent
     def content_reviewer(self) -> Agent:
         return Agent(
-            config=self.agents_config['content_reviewer'],
+            config=self.agents_config['content_reviewer'], # type: ignore[index]
             verbose=True
         )
 
     @task
     def write_section_task(self) -> Task:
         return Task(
-            config=self.tasks_config['write_section_task']
+            config=self.tasks_config['write_section_task'] # type: ignore[index]
         )
 
     @task
     def review_section_task(self) -> Task:
         return Task(
-            config=self.tasks_config['review_section_task'],
+            config=self.tasks_config['review_section_task'], # type: ignore[index]
             context=[self.write_section_task()]
         )
 
@@ -263,6 +268,7 @@ Let's create our flow in the `main.py` file:
 ```python
 #!/usr/bin/env python
 import json
+import os
 from typing import List, Dict
 from pydantic import BaseModel, Field
 from crewai import LLM
@@ -341,6 +347,9 @@ class GuideCreatorFlow(Flow[GuideCreatorState]):
         outline_dict = json.loads(response)
         self.state.guide_outline = GuideOutline(**outline_dict)
 
+        # Ensure output directory exists before saving
+        os.makedirs("output", exist_ok=True)
+
         # Save the outline to a file
         with open("output/guide_outline.json", "w") as f:
             json.dump(outline_dict, f, indent=2)

docs/how-to/arize-phoenix-observability.mdx

@@ -0,0 +1,145 @@
+---
+title: Arize Phoenix
+description: Arize Phoenix integration for CrewAI with OpenTelemetry and OpenInference
+icon: magnifying-glass-chart
+---
+
+# Arize Phoenix Integration
+
+This guide demonstrates how to integrate **Arize Phoenix** with **CrewAI** using OpenTelemetry via the [OpenInference](https://github.com/openinference/openinference) SDK. By the end of this guide, you will be able to trace your CrewAI agents and easily debug your agents.
+
+> **What is Arize Phoenix?** [Arize Phoenix](https://phoenix.arize.com) is an LLM observability platform that provides tracing and evaluation for AI applications.
+
+[![Watch a Video Demo of Our Integration with Phoenix](https://storage.googleapis.com/arize-assets/fixtures/setup_crewai.png)](https://www.youtube.com/watch?v=Yc5q3l6F7Ww)
+
+## Get Started
+
+We'll walk through a simple example of using CrewAI and integrating it with Arize Phoenix via OpenTelemetry using OpenInference.
+
+You can also access this guide on [Google Colab](https://colab.research.google.com/github/Arize-ai/phoenix/blob/main/tutorials/tracing/crewai_tracing_tutorial.ipynb).
+
+### Step 1: Install Dependencies
+
+```bash
+pip install openinference-instrumentation-crewai crewai crewai-tools arize-phoenix-otel
+```
+
+### Step 2: Set Up Environment Variables
+
+Setup Phoenix Cloud API keys and configure OpenTelemetry to send traces to Phoenix. Phoenix Cloud is a hosted version of Arize Phoenix, but it is not required to use this integration.
+
+You can get your free Serper API key [here](https://serper.dev/).
+
+```python
+import os
+from getpass import getpass
+
+# Get your Phoenix Cloud credentials
+PHOENIX_API_KEY = getpass("🔑 Enter your Phoenix Cloud API Key: ")
+
+# Get API keys for services
+OPENAI_API_KEY = getpass("🔑 Enter your OpenAI API key: ")
+SERPER_API_KEY = getpass("🔑 Enter your Serper API key: ")
+
+# Set environment variables
+os.environ["PHOENIX_CLIENT_HEADERS"] = f"api_key={PHOENIX_API_KEY}"
+os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "https://app.phoenix.arize.com" # Phoenix Cloud, change this to your own endpoint if you are using a self-hosted instance
+os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY
+os.environ["SERPER_API_KEY"] = SERPER_API_KEY
+```
+
+### Step 3: Initialize OpenTelemetry with Phoenix
+
+Initialize the OpenInference OpenTelemetry instrumentation SDK to start capturing traces and send them to Phoenix. 
+
+```python
+from phoenix.otel import register
+
+tracer_provider = register(
+    project_name="crewai-tracing-demo",
+    auto_instrument=True,
+)
+```
+
+### Step 4: Create a CrewAI Application
+
+We'll create a CrewAI application where two agents collaborate to research and write a blog post about AI advancements.
+
+```python
+from crewai import Agent, Crew, Process, Task
+from crewai_tools import SerperDevTool
+
+search_tool = SerperDevTool()
+
+# Define your agents with roles and goals
+researcher = Agent(
+    role="Senior Research Analyst",
+    goal="Uncover cutting-edge developments in AI and data science",
+    backstory="""You work at a leading tech think tank.
+    Your expertise lies in identifying emerging trends.
+    You have a knack for dissecting complex data and presenting actionable insights.""",
+    verbose=True,
+    allow_delegation=False,
+    # You can pass an optional llm attribute specifying what model you wanna use.
+    # llm=ChatOpenAI(model_name="gpt-3.5", temperature=0.7),
+    tools=[search_tool],
+)
+writer = Agent(
+    role="Tech Content Strategist",
+    goal="Craft compelling content on tech advancements",
+    backstory="""You are a renowned Content Strategist, known for your insightful and engaging articles.
+    You transform complex concepts into compelling narratives.""",
+    verbose=True,
+    allow_delegation=True,
+)
+
+# Create tasks for your agents
+task1 = Task(
+    description="""Conduct a comprehensive analysis of the latest advancements in AI in 2024.
+    Identify key trends, breakthrough technologies, and potential industry impacts.""",
+    expected_output="Full analysis report in bullet points",
+    agent=researcher,
+)
+
+task2 = Task(
+    description="""Using the insights provided, develop an engaging blog
+    post that highlights the most significant AI advancements.
+    Your post should be informative yet accessible, catering to a tech-savvy audience.
+    Make it sound cool, avoid complex words so it doesn't sound like AI.""",
+    expected_output="Full blog post of at least 4 paragraphs",
+    agent=writer,
+)
+
+# Instantiate your crew with a sequential process
+crew = Crew(
+    agents=[researcher, writer], tasks=[task1, task2], verbose=1, process=Process.sequential
+)
+
+# Get your crew to work!
+result = crew.kickoff()
+
+print("######################")
+print(result)
+```
+
+### Step 5: View Traces in Phoenix
+
+After running the agent, you can view the traces generated by your CrewAI application in Phoenix. You should see detailed steps of the agent interactions and LLM calls, which can help you debug and optimize your AI agents.
+
+Log into your Phoenix Cloud account and navigate to the project you specified in the `project_name` parameter. You'll see a timeline view of your trace with all the agent interactions, tool usages, and LLM calls.
+
+![Example trace in Phoenix showing agent interactions](https://storage.googleapis.com/arize-assets/fixtures/crewai_traces.png)
+
+
+### Version Compatibility Information
+- Python 3.8+
+- CrewAI >= 0.86.0
+- Arize Phoenix >= 7.0.1
+- OpenTelemetry SDK >= 1.31.0
+
+
+### References
+- [Phoenix Documentation](https://docs.arize.com/phoenix/) - Overview of the Phoenix platform.
+- [CrewAI Documentation](https://docs.crewai.com/) - Overview of the CrewAI framework.
+- [OpenTelemetry Docs](https://opentelemetry.io/docs/) - OpenTelemetry guide
+- [OpenInference GitHub](https://github.com/openinference/openinference) - Source code for OpenInference SDK.

docs/how-to/bring-your-own-agent.mdx

@@ -0,0 +1,443 @@
+---
+title: Bring your own agent
+description: Learn how to bring your own agents that work within a Crew.
+icon: robots
+---
+
+Interoperability is a core concept in CrewAI. This guide will show you how to bring your own agents that work within a Crew.
+
+
+## Adapter Guide for Bringing your own agents (Langgraph Agents, OpenAI Agents, etc...)
+We require 3 adapters to turn any agent from different frameworks to work within crew.
+
+1. BaseAgentAdapter
+2. BaseToolAdapter
+3. BaseConverter
+
+
+## BaseAgentAdapter
+This abstract class defines the common interface and functionality that all
+agent adapters must implement. It extends BaseAgent to maintain compatibility
+with the CrewAI framework while adding adapter-specific requirements.
+
+Required Methods:
+
+1. `def configure_tools`
+2. `def configure_structured_output`
+
+## Creating your own Adapter
+To integrate an agent from a different framework (e.g., LangGraph, Autogen, OpenAI Assistants) into CrewAI, you need to create a custom adapter by inheriting from `BaseAgentAdapter`. This adapter acts as a compatibility layer, translating between the CrewAI interfaces and the specific requirements of your external agent.
+
+Here's how you implement your custom adapter:
+
+1.  **Inherit from `BaseAgentAdapter`**:
+    ```python
+    from crewai.agents.agent_adapters.base_agent_adapter import BaseAgentAdapter
+    from crewai.tools import BaseTool
+    from typing import List, Optional, Any, Dict
+
+    class MyCustomAgentAdapter(BaseAgentAdapter):
+        # ... implementation details ...
+    ```
+
+2.  **Implement `__init__`**:
+    The constructor should call the parent class constructor `super().__init__(**kwargs)` and perform any initialization specific to your external agent. You can use the optional `agent_config` dictionary passed during CrewAI's `Agent` initialization to configure your adapter and the underlying agent.
+
+    ```python
+    def __init__(self, agent_config: Optional[Dict[str, Any]] = None, **kwargs: Any):
+        super().__init__(agent_config=agent_config, **kwargs)
+        # Initialize your external agent here, possibly using agent_config
+        # Example: self.external_agent = initialize_my_agent(agent_config)
+        print(f"Initializing MyCustomAgentAdapter with config: {agent_config}")
+    ```
+
+3.  **Implement `configure_tools`**:
+    This abstract method is crucial. It receives a list of CrewAI `BaseTool` instances. Your implementation must convert or adapt these tools into the format expected by your external agent framework. This might involve wrapping them, extracting specific attributes, or registering them with the external agent instance.
+
+    ```python
+    def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
+        if tools:
+            adapted_tools = []
+            for tool in tools:
+                # Adapt CrewAI BaseTool to the format your agent expects
+                # Example: adapted_tool = adapt_to_my_framework(tool)
+                # adapted_tools.append(adapted_tool)
+                pass # Replace with your actual adaptation logic
+
+            # Configure the external agent with the adapted tools
+            # Example: self.external_agent.set_tools(adapted_tools)
+            print(f"Configuring tools for MyCustomAgentAdapter: {adapted_tools}") # Placeholder
+        else:
+            # Handle the case where no tools are provided
+            # Example: self.external_agent.set_tools([])
+            print("No tools provided for MyCustomAgentAdapter.")
+    ```
+
+4.  **Implement `configure_structured_output`**:
+    This method is called when the CrewAI `Agent` is configured with structured output requirements (e.g., `output_json` or `output_pydantic`). Your adapter needs to ensure the external agent is set up to comply with these requirements. This might involve setting specific parameters on the external agent or ensuring its underlying model supports the requested format. If the external agent doesn't support structured output in a way compatible with CrewAI's expectations, you might need to handle the conversion or raise an appropriate error.
+
+    ```python
+    def configure_structured_output(self, structured_output: Any) -> None:
+        # Configure your external agent to produce output in the specified format
+        # Example: self.external_agent.set_output_format(structured_output)
+        self.adapted_structured_output = True # Signal that structured output is handled
+        print(f"Configuring structured output for MyCustomAgentAdapter: {structured_output}")
+    ```
+
+By implementing these methods, your `MyCustomAgentAdapter` will allow your custom agent implementation to function correctly within a CrewAI crew, interacting with tasks and tools seamlessly. Remember to replace the example comments and print statements with your actual adaptation logic specific to the external agent framework you are integrating.
+
+## BaseToolAdapter implementation
+The `BaseToolAdapter` class is responsible for converting CrewAI's native `BaseTool` objects into a format that your specific external agent framework can understand and utilize. Different agent frameworks (like LangGraph, OpenAI Assistants, etc.) have their own unique ways of defining and handling tools, and the `BaseToolAdapter` acts as the translator.
+
+Here's how you implement your custom tool adapter:
+
+1.  **Inherit from `BaseToolAdapter`**:
+    ```python
+    from crewai.agents.agent_adapters.base_tool_adapter import BaseToolAdapter
+    from crewai.tools import BaseTool
+    from typing import List, Any
+
+    class MyCustomToolAdapter(BaseToolAdapter):
+        # ... implementation details ...
+    ```
+
+2.  **Implement `configure_tools`**:
+    This is the core abstract method you must implement. It receives a list of CrewAI `BaseTool` instances provided to the agent. Your task is to iterate through this list, adapt each `BaseTool` into the format expected by your external framework, and store the converted tools in the `self.converted_tools` list (which is initialized in the base class constructor).
+
+    ```python
+    def configure_tools(self, tools: List[BaseTool]) -> None:
+        """Configure and convert CrewAI tools for the specific implementation."""
+        self.converted_tools = [] # Reset in case it's called multiple times
+        for tool in tools:
+            # Sanitize the tool name if required by the target framework
+            sanitized_name = self.sanitize_tool_name(tool.name)
+
+            # --- Your Conversion Logic Goes Here ---
+            # Example: Convert BaseTool to a dictionary format for LangGraph
+            # converted_tool = {
+            #     "name": sanitized_name,
+            #     "description": tool.description,
+            #     "parameters": tool.args_schema.schema() if tool.args_schema else {},
+            #     # Add any other framework-specific fields
+            # }
+
+            # Example: Convert BaseTool to an OpenAI function definition
+            # converted_tool = {
+            #     "type": "function",
+            #     "function": {
+            #         "name": sanitized_name,
+            #         "description": tool.description,
+            #         "parameters": tool.args_schema.schema() if tool.args_schema else {"type": "object", "properties": {}},
+            #     }
+            # }
+
+            # --- Replace above examples with your actual adaptation ---
+            converted_tool = self.adapt_tool_to_my_framework(tool, sanitized_name) # Placeholder
+
+            self.converted_tools.append(converted_tool)
+            print(f"Adapted tool '{tool.name}' to '{sanitized_name}' for MyCustomToolAdapter") # Placeholder
+
+        print(f"MyCustomToolAdapter finished configuring tools: {len(self.converted_tools)} adapted.") # Placeholder
+
+    # --- Helper method for adaptation (Example) ---
+    def adapt_tool_to_my_framework(self, tool: BaseTool, sanitized_name: str) -> Any:
+        # Replace this with the actual logic to convert a CrewAI BaseTool
+        # to the format needed by your specific external agent framework.
+        # This will vary greatly depending on the target framework.
+        adapted_representation = {
+            "framework_specific_name": sanitized_name,
+            "framework_specific_description": tool.description,
+            "inputs": tool.args_schema.schema() if tool.args_schema else None,
+            "implementation_reference": tool.run # Or however the framework needs to call it
+        }
+        # Also ensure the tool works both sync and async
+        async def async_tool_wrapper(*args, **kwargs):
+            output = tool.run(*args, **kwargs)
+            if inspect.isawaitable(output):
+                return await output
+            else:
+                return output
+
+        adapted_tool = MyFrameworkTool(
+            name=sanitized_name,
+            description=tool.description,
+            inputs=tool.args_schema.schema() if tool.args_schema else None,
+            implementation_reference=async_tool_wrapper
+        )
+        
+        return adapted_representation
+
+    ```
+
+3.  **Using the Adapter**:
+    Typically, you would instantiate your `MyCustomToolAdapter` within your `MyCustomAgentAdapter`'s `configure_tools` method and use it to process the tools before configuring your external agent.
+
+    ```python
+    # Inside MyCustomAgentAdapter.configure_tools
+    def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
+        if tools:
+            tool_adapter = MyCustomToolAdapter() # Instantiate your tool adapter
+            tool_adapter.configure_tools(tools)  # Convert the tools
+            adapted_tools = tool_adapter.tools() # Get the converted tools
+
+            # Now configure your external agent with the adapted_tools
+            # Example: self.external_agent.set_tools(adapted_tools)
+            print(f"Configuring external agent with adapted tools: {adapted_tools}") # Placeholder
+        else:
+            # Handle no tools case
+            print("No tools provided for MyCustomAgentAdapter.")
+    ```
+
+By creating a `BaseToolAdapter`, you decouple the tool conversion logic from the agent adaptation, making the integration cleaner and more modular. Remember to replace the placeholder examples with the actual conversion logic required by your specific external agent framework.
+
+## BaseConverter
+The `BaseConverterAdapter` plays a crucial role when a CrewAI `Task` requires an agent to return its final output in a specific structured format, such as JSON or a Pydantic model. It bridges the gap between CrewAI's structured output requirements and the capabilities of your external agent.
+
+Its primary responsibilities are:
+1.  **Configuring the Agent for Structured Output:** Based on the `Task`'s requirements (`output_json` or `output_pydantic`), it instructs the associated `BaseAgentAdapter` (and indirectly, the external agent) on what format is expected.
+2.  **Enhancing the System Prompt:** It modifies the agent's system prompt to include clear instructions on *how* to generate the output in the required structure.
+3.  **Post-processing the Result:** It takes the raw output from the agent and attempts to parse, validate, and format it according to the required structure, ultimately returning a string representation (e.g., a JSON string).
+
+Here's how you implement your custom converter adapter:
+
+1.  **Inherit from `BaseConverterAdapter`**:
+    ```python
+    from crewai.agents.agent_adapters.base_converter_adapter import BaseConverterAdapter
+    # Assuming you have your MyCustomAgentAdapter defined
+    # from .my_custom_agent_adapter import MyCustomAgentAdapter
+    from crewai.task import Task
+    from typing import Any
+
+    class MyCustomConverterAdapter(BaseConverterAdapter):
+        # Store the expected output type (e.g., 'json', 'pydantic', 'text')
+        _output_type: str = 'text' 
+        _output_schema: Any = None # Store JSON schema or Pydantic model
+
+        # ... implementation details ...
+    ```
+
+2.  **Implement `__init__`**:
+    The constructor must accept the corresponding `agent_adapter` instance it will work with.
+
+    ```python
+    def __init__(self, agent_adapter: Any): # Use your specific AgentAdapter type hint
+        self.agent_adapter = agent_adapter
+        print(f"Initializing MyCustomConverterAdapter for agent adapter: {type(agent_adapter).__name__}")
+    ```
+
+3.  **Implement `configure_structured_output`**:
+    This method receives the CrewAI `Task` object. You need to check the task's `output_json` and `output_pydantic` attributes to determine the required output structure. Store this information (e.g., in `_output_type` and `_output_schema`) and potentially call configuration methods on your `self.agent_adapter` if the external agent needs specific setup for structured output (which might have been partially handled in the agent adapter's `configure_structured_output` already).
+
+    ```python
+    def configure_structured_output(self, task: Task) -> None:
+        """Configure the expected structured output based on the task."""
+        if task.output_pydantic:
+            self._output_type = 'pydantic'
+            self._output_schema = task.output_pydantic
+            print(f"Converter: Configured for Pydantic output: {self._output_schema.__name__}")
+        elif task.output_json:
+            self._output_type = 'json'
+            self._output_schema = task.output_json
+            print(f"Converter: Configured for JSON output with schema: {self._output_schema}")
+        else:
+            self._output_type = 'text'
+            self._output_schema = None
+            print("Converter: Configured for standard text output.")
+
+        # Optionally, inform the agent adapter if needed
+        # self.agent_adapter.set_output_mode(self._output_type, self._output_schema)
+    ```
+
+4.  **Implement `enhance_system_prompt`**:
+    This method takes the agent's base system prompt string and should append instructions tailored to the currently configured `_output_type` and `_output_schema`. The goal is to guide the LLM powering the agent to produce output in the correct format.
+
+    ```python
+    def enhance_system_prompt(self, base_prompt: str) -> str:
+        """Enhance the system prompt with structured output instructions."""
+        if self._output_type == 'text':
+            return base_prompt # No enhancement needed for plain text
+
+        instructions = "\n\nYour final answer MUST be formatted as "
+        if self._output_type == 'json':
+            schema_str = json.dumps(self._output_schema, indent=2)
+            instructions += f"a JSON object conforming to the following schema:\n```json\n{schema_str}\n```"
+        elif self._output_type == 'pydantic':
+            schema_str = json.dumps(self._output_schema.model_json_schema(), indent=2)
+            instructions += f"a JSON object conforming to the Pydantic model '{self._output_schema.__name__}' with the following schema:\n```json\n{schema_str}\n```"
+
+        instructions += "\nEnsure your entire response is ONLY the valid JSON object, without any introductory text, explanations, or concluding remarks."
+        
+        print(f"Converter: Enhancing prompt for {self._output_type} output.")
+        return base_prompt + instructions
+    ```
+    *Note: The exact prompt engineering might need tuning based on the agent/LLM being used.* 
+
+5.  **Implement `post_process_result`**:
+    This method receives the raw string output from the agent. If structured output was requested (`json` or `pydantic`), you should attempt to parse the string into the expected format. Handle potential parsing errors (e.g., log them, attempt simple fixes, or raise an exception). Crucially, the method must **always return a string**, even if the intermediate format was a dictionary or Pydantic object (e.g., by serializing it back to a JSON string).
+
+    ```python
+    import json
+    from pydantic import ValidationError
+
+    def post_process_result(self, result: str) -> str:
+        """Post-process the agent's result to ensure it matches the expected format."""
+        print(f"Converter: Post-processing result for {self._output_type} output.")
+        if self._output_type == 'json':
+            try:
+                # Attempt to parse and re-serialize to ensure validity and consistent format
+                parsed_json = json.loads(result)
+                # Optional: Validate against self._output_schema if it's a JSON schema dictionary
+                # from jsonschema import validate
+                # validate(instance=parsed_json, schema=self._output_schema)
+                return json.dumps(parsed_json)
+            except json.JSONDecodeError as e:
+                print(f"Error: Failed to parse JSON output: {e}\nRaw output:\n{result}")
+                # Handle error: return raw, raise exception, or try to fix
+                return result # Example: return raw output on failure
+            # except Exception as e: # Catch validation errors if using jsonschema
+            #     print(f"Error: JSON output failed schema validation: {e}\nRaw output:\n{result}")
+            #     return result
+        elif self._output_type == 'pydantic':
+            try:
+                # Attempt to parse into the Pydantic model
+                model_instance = self._output_schema.model_validate_json(result)
+                # Return the model serialized back to JSON
+                return model_instance.model_dump_json()
+            except ValidationError as e:
+                print(f"Error: Failed to validate Pydantic output: {e}\nRaw output:\n{result}")
+                # Handle error
+                return result # Example: return raw output on failure
+            except json.JSONDecodeError as e:
+                 print(f"Error: Failed to parse JSON for Pydantic model: {e}\nRaw output:\n{result}")
+                 return result
+        else: # 'text'
+            return result # No processing needed for plain text
+    ```
+
+By implementing these methods, your `MyCustomConverterAdapter` ensures that structured output requests from CrewAI tasks are correctly handled by your integrated external agent, improving the reliability and usability of your custom agent within the CrewAI framework.
+
+## Out of the Box Adapters
+
+We provide out of the box adapters for the following frameworks:
+1. LangGraph
+2. OpenAI Agents
+
+## Kicking off a crew with adapted agents:
+
+```python
+import json
+import os
+from typing import List
+
+from crewai_tools import SerperDevTool
+from src.crewai import Agent, Crew, Task
+from langchain_openai import ChatOpenAI
+from pydantic import BaseModel
+
+from crewai.agents.agent_adapters.langgraph.langgraph_adapter import (
+    LangGraphAgentAdapter,
+)
+from crewai.agents.agent_adapters.openai_agents.openai_adapter import OpenAIAgentAdapter
+
+# CrewAI Agent
+code_helper_agent = Agent(
+    role="Code Helper",
+    goal="Help users solve coding problems effectively and provide clear explanations.",
+    backstory="You are an experienced programmer with deep knowledge across multiple programming languages and frameworks. You specialize in solving complex coding challenges and explaining solutions clearly.",
+    allow_delegation=False,
+    verbose=True,
+)
+# OpenAI Agent Adapter
+link_finder_agent = OpenAIAgentAdapter(
+    role="Link Finder",
+    goal="Find the most relevant and high-quality resources for coding tasks.",
+    backstory="You are a research specialist with a talent for finding the most helpful resources. You're skilled at using search tools to discover documentation, tutorials, and examples that directly address the user's coding needs.",
+    tools=[SerperDevTool()],
+    allow_delegation=False,
+    verbose=True,
+)
+
+# LangGraph Agent Adapter
+reporter_agent = LangGraphAgentAdapter(
+    role="Reporter",
+    goal="Report the results of the tasks.",
+    backstory="You are a reporter who reports the results of the other tasks",
+    llm=ChatOpenAI(model="gpt-4o"),
+    allow_delegation=True,
+    verbose=True,
+)
+
+
+class Code(BaseModel):
+    code: str
+
+
+task = Task(
+    description="Give an answer to the coding question: {task}",
+    expected_output="A thorough answer to the coding question: {task}",
+    agent=code_helper_agent,
+    output_json=Code,
+)
+task2 = Task(
+    description="Find links to resources that can help with coding tasks. Use the serper tool to find resources that can help.",
+    expected_output="A list of links to resources that can help with coding tasks",
+    agent=link_finder_agent,
+)
+
+
+class Report(BaseModel):
+    code: str
+    links: List[str]
+
+
+task3 = Task(
+    description="Report the results of the tasks.",
+    expected_output="A report of the results of the tasks. this is the code produced and then the links to the resources that can help with the coding task.",
+    agent=reporter_agent,
+    output_json=Report,
+)
+# Use in CrewAI
+crew = Crew(
+    agents=[code_helper_agent, link_finder_agent, reporter_agent],
+    tasks=[task, task2, task3],
+    verbose=True,
+)
+
+result = crew.kickoff(
+    inputs={"task": "How do you implement an abstract class in python?"}
+)
+
+# Print raw result first
+print("Raw result:", result)
+
+# Handle result based on its type
+if hasattr(result, "json_dict") and result.json_dict:
+    json_result = result.json_dict
+    print("\nStructured JSON result:")
+    print(f"{json.dumps(json_result, indent=2)}")
+
+    # Access fields safely
+    if isinstance(json_result, dict):
+        if "code" in json_result:
+            print("\nCode:")
+            print(
+                json_result["code"][:200] + "..."
+                if len(json_result["code"]) > 200
+                else json_result["code"]
+            )
+
+        if "links" in json_result:
+            print("\nLinks:")
+            for link in json_result["links"][:5]:  # Print first 5 links
+                print(f"- {link}")
+            if len(json_result["links"]) > 5:
+                print(f"...and {len(json_result['links']) - 5} more links")
+elif hasattr(result, "pydantic") and result.pydantic:
+    print("\nPydantic model result:")
+    print(result.pydantic.model_dump_json(indent=2))
+else:
+    # Fallback to raw output
+    print("\nNo structured result available, using raw output:")
+    print(result.raw[:500] + "..." if len(result.raw) > 500 else result.raw)
+
+```

docs/how-to/custom-llm.mdx

@@ -1,9 +1,13 @@
-# Custom LLM Implementations
+---
+title: Custom LLM Implementation
+description: Learn how to create custom LLM implementations in CrewAI.
+icon: code
+---
+
+## Custom LLM Implementations
 
 CrewAI now supports custom LLM implementations through the `BaseLLM` abstract base class. This allows you to create your own LLM implementations that don't rely on litellm's authentication mechanism.
 
-## Using Custom LLM Implementations
-
 To create a custom LLM implementation, you need to:
 
 1. Inherit from the `BaseLLM` abstract base class

docs/how-to/custom-manager-agent.mdx

@@ -1,5 +1,5 @@
 ---
-title: Create Your Own Manager Agent
+title: Custom Manager Agent
 description: Learn how to set a custom agent as the manager in CrewAI, providing more control over task management and coordination.
 icon: user-shield
 ---

docs/how-to/kickoff-async.mdx

@@ -92,12 +92,14 @@ coding_agent = Agent(
 # Create tasks that require code execution
 task_1 = Task(
     description="Analyze the first dataset and calculate the average age of participants. Ages: {ages}",
-    agent=coding_agent
+    agent=coding_agent,
+    expected_output="The average age of the participants."
 )
 
 task_2 = Task(
     description="Analyze the second dataset and calculate the average age of participants. Ages: {ages}",
-    agent=coding_agent
+    agent=coding_agent,
+    expected_output="The average age of the participants."
 )
 
 # Create two crews and add tasks

docs/how-to/replay-tasks-from-latest-crew-kickoff.mdx

@@ -20,10 +20,8 @@ Here's an example of how to replay from a task:
 To use the replay feature, follow these steps:
 
 <Steps>
-   <Step title="Open your terminal or command prompt.">
+   <Step title="Open your terminal or command prompt."></Step>
-   </Step>
+   <Step title="Navigate to the directory where your CrewAI project is located."></Step>
-   <Step title="Navigate to the directory where your CrewAI project is located.">
-   </Step>
    <Step title="Run the following commands:">
       To view the latest kickoff task_ids use:
 

docs/installation.mdx

@@ -4,6 +4,21 @@ description: Get started with CrewAI - Install, configure, and build your first
 icon: wrench
 ---
 
+## Video Tutorial
+Watch this video tutorial for a step-by-step demonstration of the installation process:
+
+<iframe
+  width="100%"
+  height="400"
+  src="https://www.youtube.com/embed/-kSOTtYzgEw"
+  title="CrewAI Installation Guide"
+  frameborder="0"
+  style={{ borderRadius: '10px' }}
+  allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture"
+  allowfullscreen
+></iframe>
+
+## Text Tutorial
 <Note>
   **Python Version Requirements**
 
@@ -140,6 +155,27 @@ We recommend using the `YAML` template scaffolding for a structured approach to
   </Step>
 </Steps>
 
+## Enterprise Installation Options
+
+<Note type="info">
+For teams and organizations, CrewAI offers enterprise deployment options that eliminate setup complexity:
+
+### CrewAI Enterprise (SaaS)
+- Zero installation required - just sign up for free at [app.crewai.com](https://app.crewai.com)
+- Automatic updates and maintenance
+- Managed infrastructure and scaling
+- Build Crews with no Code
+
+### CrewAI Factory (Self-hosted)
+- Containerized deployment for your infrastructure
+- Supports any hyperscaler including on prem depployments
+- Integration with your existing security systems
+
+<Card title="Explore Enterprise Options" icon="building" href="https://crewai.com/enterprise">
+  Learn about CrewAI's enterprise offerings and schedule a demo
+</Card>
+</Note>
+
 ## Next Steps
 
 <CardGroup cols={2}>

docs/introduction.mdx

@@ -15,6 +15,7 @@ CrewAI empowers developers with both high-level simplicity and precise low-level
 
 With over 100,000 developers certified through our community courses, CrewAI is rapidly becoming the standard for enterprise-ready AI automation.
 
+
 ## How Crews Work
 
 <Note>

docs/quickstart.mdx

@@ -87,15 +87,20 @@ Follow the steps below to get Crewing! 🚣‍♂️
     from crewai import Agent, Crew, Process, Task
     from crewai.project import CrewBase, agent, crew, task
     from crewai_tools import SerperDevTool
+    from crewai.agents.agent_builder.base_agent import BaseAgent
+    from typing import List
 
     @CrewBase
     class LatestAiDevelopmentCrew():
       """LatestAiDevelopment crew"""
 
+      agents: List[BaseAgent]
+      tasks: List[Task]
+
       @agent
       def researcher(self) -> Agent:
         return Agent(
-          config=self.agents_config['researcher'],
+          config=self.agents_config['researcher'], # type: ignore[index]
           verbose=True,
           tools=[SerperDevTool()]
         )
@@ -103,20 +108,20 @@ Follow the steps below to get Crewing! 🚣‍♂️
       @agent
       def reporting_analyst(self) -> Agent:
         return Agent(
-          config=self.agents_config['reporting_analyst'],
+          config=self.agents_config['reporting_analyst'], # type: ignore[index]
           verbose=True
         )
 
       @task
       def research_task(self) -> Task:
         return Task(
-          config=self.tasks_config['research_task'],
+          config=self.tasks_config['research_task'], # type: ignore[index]
         )
 
       @task
       def reporting_task(self) -> Task:
         return Task(
-          config=self.tasks_config['reporting_task'],
+          config=self.tasks_config['reporting_task'], # type: ignore[index]
           output_file='output/report.md' # This is the file that will be contain the final report.
         )
 
@@ -200,6 +205,22 @@ Follow the steps below to get Crewing! 🚣‍♂️
         ```
       </CodeGroup>
   </Step>
+
+  <Step title="Enterprise Alternative: Create in Crew Studio">
+  For CrewAI Enterprise users, you can create the same crew without writing code:
+
+  1. Log in to your CrewAI Enterprise account (create a free account at [app.crewai.com](https://app.crewai.com))
+  2. Open Crew Studio
+  3. Type what is the automation you're tryign to build
+  4. Create your tasks visually and connect them in sequence
+  5. Configure your inputs and click "Download Code" or "Deploy"
+
+  ![Crew Studio Quickstart](../images/enterprise/crew-studio-quickstart.png)
+
+  <Card title="Try CrewAI Enterprise" icon="rocket" href="https://app.crewai.com">
+    Start your free account at CrewAI Enterprise
+  </Card>
+  </Step>
   <Step title="View your final report">
   You should see the output in the console and the `report.md` file should be created in the root of your project with the final report.
 
@@ -315,9 +336,22 @@ email_summarizer_task:
       - research_task
 ```
 
-## Deploying Your Project
+## Deploying Your Crew
 
-The easiest way to deploy your crew is through [CrewAI Enterprise](http://app.crewai.com), where you can deploy your crew in a few clicks.
+The easiest way to deploy your crew to production is through [CrewAI Enterprise](http://app.crewai.com).
+
+Watch this video tutorial for a step-by-step demonstration of deploying your crew to [CrewAI Enterprise](http://app.crewai.com) using the CLI.
+
+<iframe
+  width="100%"
+  height="400"
+  src="https://www.youtube.com/embed/3EqSV-CYDZA"
+  title="CrewAI Deployment Guide"
+  frameborder="0"
+  style={{ borderRadius: '10px' }}
+  allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture"
+  allowfullscreen
+></iframe>
 
 <CardGroup cols={2}>
   <Card

docs/telemetry.mdx

@@ -22,7 +22,16 @@ usage of tools, API calls, responses, any data processed by the agents, or secre
 When the `share_crew` feature is enabled, detailed data including task descriptions, agents' backstories or goals, and other specific attributes are collected 
 to provide deeper insights. This expanded data collection may include personal information if users have incorporated it into their crews or tasks. 
 Users should carefully consider the content of their crews and tasks before enabling `share_crew`. 
-Users can disable telemetry by setting the environment variable `OTEL_SDK_DISABLED` to `true`.
+Users can disable telemetry by setting the environment variable `CREWAI_DISABLE_TELEMETRY` to `true` or by setting `OTEL_SDK_DISABLED` to `true` (note that the latter disables all OpenTelemetry instrumentation globally).
+
+### Examples:
+```python
+# Disable CrewAI telemetry only
+os.environ['CREWAI_DISABLE_TELEMETRY'] = 'true'
+
+# Disable all OpenTelemetry (including CrewAI)
+os.environ['OTEL_SDK_DISABLED'] = 'true'
+```
 
 ### Data Explanation:
 | Defaulted | Data                                      | Reason and Specifics                                                                                                       |

docs/tools/jsonsearchtool.mdx

@@ -30,7 +30,7 @@ pip install 'crewai[tools]'
 Here are updated examples on how to utilize the JSONSearchTool effectively for searching within JSON files. These examples take into account the current implementation and usage patterns identified in the codebase.
 
 ```python Code
-from crewai.json_tools import JSONSearchTool  # Updated import path
+from crewai_tools import JSONSearchTool
 
 # General JSON content search
 # This approach is suitable when the JSON path is either known beforehand or can be dynamically identified.

docs/tools/langchaintool.mdx

@@ -1,10 +1,10 @@
 ---
-title: Using LangChain Tools
+title: LangChain Tool
-description: Learn how to integrate LangChain tools with CrewAI agents to enhance search-based queries and more.
+description: The `LangChainTool` is a wrapper for LangChain tools and query engines.
 icon: link
 ---
 
-## Using LangChain Tools
+## `LangChainTool`
 
 <Info>
     CrewAI seamlessly integrates with LangChain's comprehensive [list of tools](https://python.langchain.com/docs/integrations/tools/), all of which can be used with CrewAI.

docs/tools/weaviatevectorsearchtool.mdx

@@ -25,7 +25,7 @@ uv add weaviate-client
 To effectively use the `WeaviateVectorSearchTool`, follow these steps:
 
 1. **Package Installation**: Confirm that the `crewai[tools]` and `weaviate-client` packages are installed in your Python environment.
-2. **Weaviate Setup**: Set up a Weaviate cluster. You can follow the [Weaviate documentation](https://weaviate.io/developers/wcs/connect) for instructions.
+2. **Weaviate Setup**: Set up a Weaviate cluster. You can follow the [Weaviate documentation](https://weaviate.io/developers/wcs/manage-clusters/connect) for instructions.
 3. **API Keys**: Obtain your Weaviate cluster URL and API key.
 4. **OpenAI API Key**: Ensure you have an OpenAI API key set in your environment variables as `OPENAI_API_KEY`.
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "crewai"
-version = "0.108.0"
+version = "0.114.0"
 description = "Cutting-edge framework for orchestrating role-playing, autonomous AI agents. By fostering collaborative intelligence, CrewAI empowers agents to work together seamlessly, tackling complex tasks."
 readme = "README.md"
 requires-python = ">=3.10,<3.13"
@@ -45,7 +45,7 @@ Documentation = "https://docs.crewai.com"
 Repository = "https://github.com/crewAIInc/crewAI"
 
 [project.optional-dependencies]
-tools = ["crewai-tools>=0.37.0"]
+tools = ["crewai-tools~=0.40.1"]
 embeddings = [
     "tiktoken~=0.7.0"
 ]
@@ -81,10 +81,10 @@ dev-dependencies = [
     "pillow>=10.2.0",
     "cairosvg>=2.7.1",
     "pytest>=8.0.0",
-    "pytest-vcr>=1.0.2",
     "python-dotenv>=1.0.0",
     "pytest-asyncio>=0.23.7",
     "pytest-subprocess>=1.5.2",
+    "pytest-recording>=0.13.2",
 ]
 
 [project.scripts]

src/crewai/__init__.py

@@ -2,12 +2,14 @@ import warnings
 
 from crewai.agent import Agent
 from crewai.crew import Crew
+from crewai.crews.crew_output import CrewOutput
 from crewai.flow.flow import Flow
 from crewai.knowledge.knowledge import Knowledge
 from crewai.llm import LLM
 from crewai.llms.base_llm import BaseLLM
 from crewai.process import Process
 from crewai.task import Task
+from crewai.tasks.task_output import TaskOutput
 
 warnings.filterwarnings(
     "ignore",
@@ -15,14 +17,16 @@ warnings.filterwarnings(
     category=UserWarning,
     module="pydantic.main",
 )
-__version__ = "0.108.0"
+__version__ = "0.114.0"
 __all__ = [
     "Agent",
     "Crew",
+    "CrewOutput",
     "Process",
     "Task",
     "LLM",
     "BaseLLM",
     "Flow",
     "Knowledge",
+    "TaskOutput",
 ]

src/crewai/agent.py

@@ -1,7 +1,6 @@
-import re
 import shutil
 import subprocess
-from typing import Any, Dict, List, Literal, Optional, Sequence, Union
+from typing import Any, Dict, List, Literal, Optional, Sequence, Type, Union
 
 from pydantic import Field, InstanceOf, PrivateAttr, model_validator
 
@@ -11,6 +10,7 @@ from crewai.agents.crew_agent_executor import CrewAgentExecutor
 from crewai.knowledge.knowledge import Knowledge
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
 from crewai.knowledge.utils.knowledge_utils import extract_knowledge_context
+from crewai.lite_agent import LiteAgent, LiteAgentOutput
 from crewai.llm import BaseLLM
 from crewai.memory.contextual.contextual_memory import ContextualMemory
 from crewai.security import Fingerprint
@@ -18,6 +18,11 @@ from crewai.task import Task
 from crewai.tools import BaseTool
 from crewai.tools.agent_tools.agent_tools import AgentTools
 from crewai.utilities import Converter, Prompts
+from crewai.utilities.agent_utils import (
+    get_tool_names,
+    parse_tools,
+    render_text_description_and_args,
+)
 from crewai.utilities.constants import TRAINED_AGENTS_DATA_FILE, TRAINING_DATA_FILE
 from crewai.utilities.converter import generate_model_description
 from crewai.utilities.events.agent_events import (
@@ -86,9 +91,6 @@ class Agent(BaseAgent):
     response_template: Optional[str] = Field(
         default=None, description="Response format for the agent."
     )
-    tools_results: Optional[List[Any]] = Field(
-        default=[], description="Results of the tools used by the agent."
-    )
     allow_code_execution: Optional[bool] = Field(
         default=False, description="Enable code execution for the agent."
     )
@@ -112,6 +114,14 @@ class Agent(BaseAgent):
         default=None,
         description="Embedder configuration for the agent.",
     )
+    agent_knowledge_context: Optional[str] = Field(
+        default=None,
+        description="Knowledge context for the agent.",
+    )
+    crew_knowledge_context: Optional[str] = Field(
+        default=None,
+        description="Knowledge context for the crew.",
+    )
 
     @model_validator(mode="after")
     def post_init_setup(self):
@@ -154,11 +164,28 @@ class Agent(BaseAgent):
         except (TypeError, ValueError) as e:
             raise ValueError(f"Invalid Knowledge Configuration: {str(e)}")
 
+    def _is_any_available_memory(self) -> bool:
+        """Check if any memory is available."""
+        if not self.crew:
+            return False
+
+        memory_attributes = [
+            "memory",
+            "memory_config",
+            "_short_term_memory",
+            "_long_term_memory",
+            "_entity_memory",
+            "_user_memory",
+            "_external_memory",
+        ]
+
+        return any(getattr(self.crew, attr) for attr in memory_attributes)
+
     def execute_task(
         self,
         task: Task,
         context: Optional[str] = None,
-        tools: Optional[List[BaseTool]] = None,
+        tools: Optional[List[BaseTool]] = None
     ) -> str:
         """Execute a task with the agent.
 
@@ -169,6 +196,11 @@ class Agent(BaseAgent):
 
         Returns:
             Output of the agent
+
+        Raises:
+            TimeoutError: If execution exceeds the maximum execution time.
+            ValueError: If the max execution time is not a positive integer.
+            RuntimeError: If the agent execution fails for other reasons.
         """
         if self.tools_handler:
             self.tools_handler.last_used_tool = {}  # type: ignore # Incompatible types in assignment (expression has type "dict[Never, Never]", variable has type "ToolCalling")
@@ -198,33 +230,42 @@ class Agent(BaseAgent):
                 task=task_prompt, context=context
             )
 
-        if self.crew and self.crew.memory:
+        if self._is_any_available_memory():
             contextual_memory = ContextualMemory(
                 self.crew.memory_config,
                 self.crew._short_term_memory,
                 self.crew._long_term_memory,
                 self.crew._entity_memory,
                 self.crew._user_memory,
+                self.crew._external_memory,
             )
             memory = contextual_memory.build_context_for_task(task, context)
             if memory.strip() != "":
                 task_prompt += self.i18n.slice("memory").format(memory=memory)
-
+        knowledge_config = (
+            self.knowledge_config.model_dump() if self.knowledge_config else {}
+        )
         if self.knowledge:
-            agent_knowledge_snippets = self.knowledge.query([task.prompt()])
+            agent_knowledge_snippets = self.knowledge.query(
+                [task.prompt()], **knowledge_config
+            )
             if agent_knowledge_snippets:
-                agent_knowledge_context = extract_knowledge_context(
+                self.agent_knowledge_context = extract_knowledge_context(
                     agent_knowledge_snippets
                 )
-                if agent_knowledge_context:
+                if self.agent_knowledge_context:
-                    task_prompt += agent_knowledge_context
+                    task_prompt += self.agent_knowledge_context
 
         if self.crew:
-            knowledge_snippets = self.crew.query_knowledge([task.prompt()])
+            knowledge_snippets = self.crew.query_knowledge(
+                [task.prompt()], **knowledge_config
+            )
             if knowledge_snippets:
-                crew_knowledge_context = extract_knowledge_context(knowledge_snippets)
+                self.crew_knowledge_context = extract_knowledge_context(
-                if crew_knowledge_context:
+                    knowledge_snippets
-                    task_prompt += crew_knowledge_context
+                )
+                if self.crew_knowledge_context:
+                    task_prompt += self.crew_knowledge_context
 
         tools = tools or self.tools or []
         self.create_agent_executor(tools=tools, task=task)
@@ -244,14 +285,26 @@ class Agent(BaseAgent):
                     task=task,
                 ),
             )
-            result = self.agent_executor.invoke(
+
-                {
+            # Determine execution method based on timeout setting
-                    "input": task_prompt,
+            if self.max_execution_time is not None:
-                    "tool_names": self.agent_executor.tools_names,
+                if not isinstance(self.max_execution_time, int) or self.max_execution_time <= 0:
-                    "tools": self.agent_executor.tools_description,
+                    raise ValueError("Max Execution time must be a positive integer greater than zero")
-                    "ask_for_human_input": task.human_input,
+                result = self._execute_with_timeout(task_prompt, task, self.max_execution_time)
-                }
+            else:
-            )["output"]
+                result = self._execute_without_timeout(task_prompt, task)
+                
+        except TimeoutError as e:
+            # Propagate TimeoutError without retry
+            crewai_event_bus.emit(
+                self,
+                event=AgentExecutionErrorEvent(
+                    agent=self,
+                    task=task,
+                    error=str(e),
+                ),
+            )
+            raise e
         except Exception as e:
             if e.__class__.__module__.startswith("litellm"):
                 # Do not retry on litellm errors
@@ -292,6 +345,66 @@ class Agent(BaseAgent):
         )
         return result
 
+    def _execute_with_timeout(
+        self,
+        task_prompt: str,
+        task: Task,
+        timeout: int
+    ) -> str:
+        """Execute a task with a timeout.
+        
+        Args:
+            task_prompt: The prompt to send to the agent.
+            task: The task being executed.
+            timeout: Maximum execution time in seconds.
+            
+        Returns:
+            The output of the agent.
+            
+        Raises:
+            TimeoutError: If execution exceeds the timeout.
+            RuntimeError: If execution fails for other reasons.
+        """
+        import concurrent.futures
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            future = executor.submit(
+                self._execute_without_timeout,
+                task_prompt=task_prompt,
+                task=task
+            )
+            
+            try:
+                return future.result(timeout=timeout)
+            except concurrent.futures.TimeoutError:
+                future.cancel()
+                raise TimeoutError(f"Task '{task.description}' execution timed out after {timeout} seconds. Consider increasing max_execution_time or optimizing the task.")
+            except Exception as e:
+                future.cancel()
+                raise RuntimeError(f"Task execution failed: {str(e)}")
+
+    def _execute_without_timeout(
+        self,
+        task_prompt: str,
+        task: Task
+    ) -> str:
+        """Execute a task without a timeout.
+        
+        Args:
+            task_prompt: The prompt to send to the agent.
+            task: The task being executed.
+            
+        Returns:
+            The output of the agent.
+        """
+        return self.agent_executor.invoke(
+            {
+                "input": task_prompt,
+                "tool_names": self.agent_executor.tools_names,
+                "tools": self.agent_executor.tools_description,
+                "ask_for_human_input": task.human_input,
+            }
+        )["output"]
+
     def create_agent_executor(
         self, tools: Optional[List[BaseTool]] = None, task=None
     ) -> None:
@@ -300,12 +413,12 @@ class Agent(BaseAgent):
         Returns:
             An instance of the CrewAgentExecutor class.
         """
-        tools = tools or self.tools or []
+        raw_tools: List[BaseTool] = tools or self.tools or []
-        parsed_tools = self._parse_tools(tools)
+        parsed_tools = parse_tools(raw_tools)
 
         prompt = Prompts(
             agent=self,
-            tools=tools,
+            has_tools=len(raw_tools) > 0,
             i18n=self.i18n,
             use_system_prompt=self.use_system_prompt,
             system_template=self.system_template,
@@ -327,12 +440,12 @@ class Agent(BaseAgent):
             crew=self.crew,
             tools=parsed_tools,
             prompt=prompt,
-            original_tools=tools,
+            original_tools=raw_tools,
             stop_words=stop_words,
             max_iter=self.max_iter,
             tools_handler=self.tools_handler,
-            tools_names=self.__tools_names(parsed_tools),
+            tools_names=get_tool_names(parsed_tools),
-            tools_description=self._render_text_description_and_args(parsed_tools),
+            tools_description=render_text_description_and_args(parsed_tools),
             step_callback=self.step_callback,
             function_calling_llm=self.function_calling_llm,
             respect_context_window=self.respect_context_window,
@@ -367,25 +480,6 @@ class Agent(BaseAgent):
     def get_output_converter(self, llm, text, model, instructions):
         return Converter(llm=llm, text=text, model=model, instructions=instructions)
 
-    def _parse_tools(self, tools: List[Any]) -> List[Any]:  # type: ignore
-        """Parse tools to be used for the task."""
-        tools_list = []
-        try:
-            # tentatively try to import from crewai_tools import BaseTool as CrewAITool
-            from crewai.tools import BaseTool as CrewAITool
-
-            for tool in tools:
-                if isinstance(tool, CrewAITool):
-                    tools_list.append(tool.to_structured_tool())
-                else:
-                    tools_list.append(tool)
-        except ModuleNotFoundError:
-            tools_list = []
-            for tool in tools:
-                tools_list.append(tool)
-
-        return tools_list
-
     def _training_handler(self, task_prompt: str) -> str:
         """Handle training data for the agent task prompt to improve output on Training."""
         if data := CrewTrainingHandler(TRAINING_DATA_FILE).load():
@@ -431,23 +525,6 @@ class Agent(BaseAgent):
 
         return description
 
-    def _render_text_description_and_args(self, tools: List[BaseTool]) -> str:
-        """Render the tool name, description, and args in plain text.
-
-            Output will be in the format of:
-
-            .. code-block:: markdown
-
-            search: This tool is used for search, args: {"query": {"type": "string"}}
-            calculator: This tool is used for math, \
-            args: {"expression": {"type": "string"}}
-        """
-        tool_strings = []
-        for tool in tools:
-            tool_strings.append(tool.description)
-
-        return "\n".join(tool_strings)
-
     def _validate_docker_installation(self) -> None:
         """Check if Docker is installed and running."""
         if not shutil.which("docker"):
@@ -467,10 +544,6 @@ class Agent(BaseAgent):
                 f"Docker is not running. Please start Docker to use code execution with agent: {self.role}"
             )
 
-    @staticmethod
-    def __tools_names(tools) -> str:
-        return ", ".join([t.name for t in tools])
-
     def __repr__(self):
         return f"Agent(role={self.role}, goal={self.goal}, backstory={self.backstory})"
 
@@ -483,3 +556,79 @@ class Agent(BaseAgent):
             Fingerprint: The agent's fingerprint
         """
         return self.security_config.fingerprint
+
+    def set_fingerprint(self, fingerprint: Fingerprint):
+        self.security_config.fingerprint = fingerprint
+
+    def kickoff(
+        self,
+        messages: Union[str, List[Dict[str, str]]],
+        response_format: Optional[Type[Any]] = None,
+    ) -> LiteAgentOutput:
+        """
+        Execute the agent with the given messages using a LiteAgent instance.
+
+        This method is useful when you want to use the Agent configuration but
+        with the simpler and more direct execution flow of LiteAgent.
+
+        Args:
+            messages: Either a string query or a list of message dictionaries.
+                     If a string is provided, it will be converted to a user message.
+                     If a list is provided, each dict should have 'role' and 'content' keys.
+            response_format: Optional Pydantic model for structured output.
+
+        Returns:
+            LiteAgentOutput: The result of the agent execution.
+        """
+        lite_agent = LiteAgent(
+            role=self.role,
+            goal=self.goal,
+            backstory=self.backstory,
+            llm=self.llm,
+            tools=self.tools or [],
+            max_iterations=self.max_iter,
+            max_execution_time=self.max_execution_time,
+            respect_context_window=self.respect_context_window,
+            verbose=self.verbose,
+            response_format=response_format,
+            i18n=self.i18n,
+            original_agent=self,
+        )
+
+        return lite_agent.kickoff(messages)
+
+    async def kickoff_async(
+        self,
+        messages: Union[str, List[Dict[str, str]]],
+        response_format: Optional[Type[Any]] = None,
+    ) -> LiteAgentOutput:
+        """
+        Execute the agent asynchronously with the given messages using a LiteAgent instance.
+
+        This is the async version of the kickoff method.
+
+        Args:
+            messages: Either a string query or a list of message dictionaries.
+                     If a string is provided, it will be converted to a user message.
+                     If a list is provided, each dict should have 'role' and 'content' keys.
+            response_format: Optional Pydantic model for structured output.
+
+        Returns:
+            LiteAgentOutput: The result of the agent execution.
+        """
+        lite_agent = LiteAgent(
+            role=self.role,
+            goal=self.goal,
+            backstory=self.backstory,
+            llm=self.llm,
+            tools=self.tools or [],
+            max_iterations=self.max_iter,
+            max_execution_time=self.max_execution_time,
+            respect_context_window=self.respect_context_window,
+            verbose=self.verbose,
+            response_format=response_format,
+            i18n=self.i18n,
+            original_agent=self,
+        )
+
+        return await lite_agent.kickoff_async(messages)

src/crewai/agents/agent_adapters/base_agent_adapter.py

@@ -0,0 +1,42 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List, Optional
+
+from pydantic import PrivateAttr
+
+from crewai.agent import BaseAgent
+from crewai.tools import BaseTool
+
+
+class BaseAgentAdapter(BaseAgent, ABC):
+    """Base class for all agent adapters in CrewAI.
+
+    This abstract class defines the common interface and functionality that all
+    agent adapters must implement. It extends BaseAgent to maintain compatibility
+    with the CrewAI framework while adding adapter-specific requirements.
+    """
+
+    adapted_structured_output: bool = False
+    _agent_config: Optional[Dict[str, Any]] = PrivateAttr(default=None)
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    def __init__(self, agent_config: Optional[Dict[str, Any]] = None, **kwargs: Any):
+        super().__init__(adapted_agent=True, **kwargs)
+        self._agent_config = agent_config
+
+    @abstractmethod
+    def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
+        """Configure and adapt tools for the specific agent implementation.
+
+        Args:
+            tools: Optional list of BaseTool instances to be configured
+        """
+        pass
+
+    def configure_structured_output(self, structured_output: Any) -> None:
+        """Configure the structured output for the specific agent implementation.
+
+        Args:
+            structured_output: The structured output to be configured
+        """
+        pass

src/crewai/agents/agent_adapters/base_converter_adapter.py

@@ -0,0 +1,29 @@
+from abc import ABC, abstractmethod
+
+
+class BaseConverterAdapter(ABC):
+    """Base class for all converter adapters in CrewAI.
+
+    This abstract class defines the common interface and functionality that all
+    converter adapters must implement for converting structured output.
+    """
+
+    def __init__(self, agent_adapter):
+        self.agent_adapter = agent_adapter
+
+    @abstractmethod
+    def configure_structured_output(self, task) -> None:
+        """Configure agents to return structured output.
+        Must support json and pydantic output.
+        """
+        pass
+
+    @abstractmethod
+    def enhance_system_prompt(self, base_prompt: str) -> str:
+        """Enhance the system prompt with structured output instructions."""
+        pass
+
+    @abstractmethod
+    def post_process_result(self, result: str) -> str:
+        """Post-process the result to ensure it matches the expected format: string."""
+        pass

src/crewai/agents/agent_adapters/base_tool_adapter.py

@@ -0,0 +1,37 @@
+from abc import ABC, abstractmethod
+from typing import Any, List, Optional
+
+from crewai.tools.base_tool import BaseTool
+
+
+class BaseToolAdapter(ABC):
+    """Base class for all tool adapters in CrewAI.
+
+    This abstract class defines the common interface that all tool adapters
+    must implement. It provides the structure for adapting CrewAI tools to
+    different frameworks and platforms.
+    """
+
+    original_tools: List[BaseTool]
+    converted_tools: List[Any]
+
+    def __init__(self, tools: Optional[List[BaseTool]] = None):
+        self.original_tools = tools or []
+        self.converted_tools = []
+
+    @abstractmethod
+    def configure_tools(self, tools: List[BaseTool]) -> None:
+        """Configure and convert tools for the specific implementation.
+
+        Args:
+            tools: List of BaseTool instances to be configured and converted
+        """
+        pass
+
+    def tools(self) -> List[Any]:
+        """Return all converted tools."""
+        return self.converted_tools
+
+    def sanitize_tool_name(self, tool_name: str) -> str:
+        """Sanitize tool name for API compatibility."""
+        return tool_name.replace(" ", "_")

src/crewai/agents/agent_adapters/langgraph/langgraph_adapter.py

@@ -0,0 +1,226 @@
+from typing import Any, AsyncIterable, Dict, List, Optional
+
+from pydantic import Field, PrivateAttr
+
+from crewai.agents.agent_adapters.base_agent_adapter import BaseAgentAdapter
+from crewai.agents.agent_adapters.langgraph.langgraph_tool_adapter import (
+    LangGraphToolAdapter,
+)
+from crewai.agents.agent_adapters.langgraph.structured_output_converter import (
+    LangGraphConverterAdapter,
+)
+from crewai.agents.agent_builder.base_agent import BaseAgent
+from crewai.tools.agent_tools.agent_tools import AgentTools
+from crewai.tools.base_tool import BaseTool
+from crewai.utilities import Logger
+from crewai.utilities.converter import Converter
+from crewai.utilities.events import crewai_event_bus
+from crewai.utilities.events.agent_events import (
+    AgentExecutionCompletedEvent,
+    AgentExecutionErrorEvent,
+    AgentExecutionStartedEvent,
+)
+
+try:
+    from langchain_core.messages import ToolMessage
+    from langgraph.checkpoint.memory import MemorySaver
+    from langgraph.prebuilt import create_react_agent
+
+    LANGGRAPH_AVAILABLE = True
+except ImportError:
+    LANGGRAPH_AVAILABLE = False
+
+
+class LangGraphAgentAdapter(BaseAgentAdapter):
+    """Adapter for LangGraph agents to work with CrewAI."""
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    _logger: Logger = PrivateAttr(default_factory=lambda: Logger())
+    _tool_adapter: LangGraphToolAdapter = PrivateAttr()
+    _graph: Any = PrivateAttr(default=None)
+    _memory: Any = PrivateAttr(default=None)
+    _max_iterations: int = PrivateAttr(default=10)
+    function_calling_llm: Any = Field(default=None)
+    step_callback: Any = Field(default=None)
+
+    model: str = Field(default="gpt-4o")
+    verbose: bool = Field(default=False)
+
+    def __init__(
+        self,
+        role: str,
+        goal: str,
+        backstory: str,
+        tools: Optional[List[BaseTool]] = None,
+        llm: Any = None,
+        max_iterations: int = 10,
+        agent_config: Optional[Dict[str, Any]] = None,
+        **kwargs,
+    ):
+        """Initialize the LangGraph agent adapter."""
+        if not LANGGRAPH_AVAILABLE:
+            raise ImportError(
+                "LangGraph Agent Dependencies are not installed. Please install it using `uv add langchain-core langgraph`"
+            )
+        super().__init__(
+            role=role,
+            goal=goal,
+            backstory=backstory,
+            tools=tools,
+            llm=llm or self.model,
+            agent_config=agent_config,
+            **kwargs,
+        )
+        self._tool_adapter = LangGraphToolAdapter(tools=tools)
+        self._converter_adapter = LangGraphConverterAdapter(self)
+        self._max_iterations = max_iterations
+        self._setup_graph()
+
+    def _setup_graph(self) -> None:
+        """Set up the LangGraph workflow graph."""
+        try:
+            self._memory = MemorySaver()
+
+            converted_tools: List[Any] = self._tool_adapter.tools()
+            if self._agent_config:
+                self._graph = create_react_agent(
+                    model=self.llm,
+                    tools=converted_tools,
+                    checkpointer=self._memory,
+                    debug=self.verbose,
+                    **self._agent_config,
+                )
+            else:
+                self._graph = create_react_agent(
+                    model=self.llm,
+                    tools=converted_tools or [],
+                    checkpointer=self._memory,
+                    debug=self.verbose,
+                )
+
+        except ImportError as e:
+            self._logger.log(
+                "error", f"Failed to import LangGraph dependencies: {str(e)}"
+            )
+            raise
+        except Exception as e:
+            self._logger.log("error", f"Error setting up LangGraph agent: {str(e)}")
+            raise
+
+    def _build_system_prompt(self) -> str:
+        """Build a system prompt for the LangGraph agent."""
+        base_prompt = f"""
+            You are {self.role}.
+        
+            Your goal is: {self.goal}
+
+            Your backstory: {self.backstory}
+
+            When working on tasks, think step-by-step and use the available tools when necessary.
+        """
+        return self._converter_adapter.enhance_system_prompt(base_prompt)
+
+    def execute_task(
+        self,
+        task: Any,
+        context: Optional[str] = None,
+        tools: Optional[List[BaseTool]] = None,
+    ) -> str:
+        """Execute a task using the LangGraph workflow."""
+        self.create_agent_executor(tools)
+
+        self.configure_structured_output(task)
+
+        try:
+            task_prompt = task.prompt() if hasattr(task, "prompt") else str(task)
+
+            if context:
+                task_prompt = self.i18n.slice("task_with_context").format(
+                    task=task_prompt, context=context
+                )
+
+            crewai_event_bus.emit(
+                self,
+                event=AgentExecutionStartedEvent(
+                    agent=self,
+                    tools=self.tools,
+                    task_prompt=task_prompt,
+                    task=task,
+                ),
+            )
+
+            session_id = f"task_{id(task)}"
+
+            config = {"configurable": {"thread_id": session_id}}
+
+            result = self._graph.invoke(
+                {
+                    "messages": [
+                        ("system", self._build_system_prompt()),
+                        ("user", task_prompt),
+                    ]
+                },
+                config,
+            )
+
+            messages = result.get("messages", [])
+            last_message = messages[-1] if messages else None
+
+            final_answer = ""
+            if isinstance(last_message, dict):
+                final_answer = last_message.get("content", "")
+            elif hasattr(last_message, "content"):
+                final_answer = getattr(last_message, "content", "")
+
+            final_answer = (
+                self._converter_adapter.post_process_result(final_answer)
+                or "Task execution completed but no clear answer was provided."
+            )
+            crewai_event_bus.emit(
+                self,
+                event=AgentExecutionCompletedEvent(
+                    agent=self, task=task, output=final_answer
+                ),
+            )
+
+            return final_answer
+
+        except Exception as e:
+            self._logger.log("error", f"Error executing LangGraph task: {str(e)}")
+            crewai_event_bus.emit(
+                self,
+                event=AgentExecutionErrorEvent(
+                    agent=self,
+                    task=task,
+                    error=str(e),
+                ),
+            )
+            raise
+
+    def create_agent_executor(self, tools: Optional[List[BaseTool]] = None) -> None:
+        """Configure the LangGraph agent for execution."""
+        self.configure_tools(tools)
+
+    def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
+        """Configure tools for the LangGraph agent."""
+        if tools:
+            all_tools = list(self.tools or []) + list(tools or [])
+            self._tool_adapter.configure_tools(all_tools)
+            available_tools = self._tool_adapter.tools()
+            self._graph.tools = available_tools
+
+    def get_delegation_tools(self, agents: List[BaseAgent]) -> List[BaseTool]:
+        """Implement delegation tools support for LangGraph."""
+        agent_tools = AgentTools(agents=agents)
+        return agent_tools.tools()
+
+    def get_output_converter(
+        self, llm: Any, text: str, model: Any, instructions: str
+    ) -> Any:
+        """Convert output format if needed."""
+        return Converter(llm=llm, text=text, model=model, instructions=instructions)
+
+    def configure_structured_output(self, task) -> None:
+        """Configure the structured output for LangGraph."""
+        self._converter_adapter.configure_structured_output(task)

src/crewai/agents/agent_adapters/langgraph/langgraph_tool_adapter.py

@@ -0,0 +1,61 @@
+import inspect
+from typing import Any, List, Optional
+
+from crewai.agents.agent_adapters.base_tool_adapter import BaseToolAdapter
+from crewai.tools.base_tool import BaseTool
+
+
+class LangGraphToolAdapter(BaseToolAdapter):
+    """Adapts CrewAI tools to LangGraph agent tool compatible format"""
+
+    def __init__(self, tools: Optional[List[BaseTool]] = None):
+        self.original_tools = tools or []
+        self.converted_tools = []
+
+    def configure_tools(self, tools: List[BaseTool]) -> None:
+        """
+        Configure and convert CrewAI tools to LangGraph-compatible format.
+        LangGraph expects tools in langchain_core.tools format.
+        """
+        from langchain_core.tools import BaseTool, StructuredTool
+
+        converted_tools = []
+        if self.original_tools:
+            all_tools = tools + self.original_tools
+        else:
+            all_tools = tools
+        for tool in all_tools:
+            if isinstance(tool, BaseTool):
+                converted_tools.append(tool)
+                continue
+
+            sanitized_name = self.sanitize_tool_name(tool.name)
+
+            async def tool_wrapper(*args, tool=tool, **kwargs):
+                output = None
+                if len(args) > 0 and isinstance(args[0], str):
+                    output = tool.run(args[0])
+                elif "input" in kwargs:
+                    output = tool.run(kwargs["input"])
+                else:
+                    output = tool.run(**kwargs)
+
+                if inspect.isawaitable(output):
+                    result = await output
+                else:
+                    result = output
+                return result
+
+            converted_tool = StructuredTool(
+                name=sanitized_name,
+                description=tool.description,
+                func=tool_wrapper,
+                args_schema=tool.args_schema,
+            )
+
+            converted_tools.append(converted_tool)
+
+        self.converted_tools = converted_tools
+
+    def tools(self) -> List[Any]:
+        return self.converted_tools or []

src/crewai/agents/agent_adapters/langgraph/structured_output_converter.py

@@ -0,0 +1,80 @@
+import json
+
+from crewai.agents.agent_adapters.base_converter_adapter import BaseConverterAdapter
+from crewai.utilities.converter import generate_model_description
+
+
+class LangGraphConverterAdapter(BaseConverterAdapter):
+    """Adapter for handling structured output conversion in LangGraph agents"""
+
+    def __init__(self, agent_adapter):
+        """Initialize the converter adapter with a reference to the agent adapter"""
+        self.agent_adapter = agent_adapter
+        self._output_format = None
+        self._schema = None
+        self._system_prompt_appendix = None
+
+    def configure_structured_output(self, task) -> None:
+        """Configure the structured output for LangGraph."""
+        if not (task.output_json or task.output_pydantic):
+            self._output_format = None
+            self._schema = None
+            self._system_prompt_appendix = None
+            return
+
+        if task.output_json:
+            self._output_format = "json"
+            self._schema = generate_model_description(task.output_json)
+        elif task.output_pydantic:
+            self._output_format = "pydantic"
+            self._schema = generate_model_description(task.output_pydantic)
+
+        self._system_prompt_appendix = self._generate_system_prompt_appendix()
+
+    def _generate_system_prompt_appendix(self) -> str:
+        """Generate an appendix for the system prompt to enforce structured output"""
+        if not self._output_format or not self._schema:
+            return ""
+
+        return f"""
+Important: Your final answer MUST be provided in the following structured format:
+
+{self._schema}
+
+DO NOT include any markdown code blocks, backticks, or other formatting around your response. 
+The output should be raw JSON that exactly matches the specified schema.
+"""
+
+    def enhance_system_prompt(self, original_prompt: str) -> str:
+        """Add structured output instructions to the system prompt if needed"""
+        if not self._system_prompt_appendix:
+            return original_prompt
+
+        return f"{original_prompt}\n{self._system_prompt_appendix}"
+
+    def post_process_result(self, result: str) -> str:
+        """Post-process the result to ensure it matches the expected format"""
+        if not self._output_format:
+            return result
+
+        # Try to extract valid JSON if it's wrapped in code blocks or other text
+        if self._output_format in ["json", "pydantic"]:
+            try:
+                # First, try to parse as is
+                json.loads(result)
+                return result
+            except json.JSONDecodeError:
+                # Try to extract JSON from the text
+                import re
+
+                json_match = re.search(r"(\{.*\})", result, re.DOTALL)
+                if json_match:
+                    try:
+                        extracted = json_match.group(1)
+                        # Validate it's proper JSON
+                        json.loads(extracted)
+                        return extracted
+                    except:
+                        pass
+
+        return result

src/crewai/agents/agent_adapters/openai_agents/openai_adapter.py

@@ -0,0 +1,178 @@
+from typing import Any, List, Optional
+
+from pydantic import Field, PrivateAttr
+
+from crewai.agents.agent_adapters.base_agent_adapter import BaseAgentAdapter
+from crewai.agents.agent_adapters.openai_agents.structured_output_converter import (
+    OpenAIConverterAdapter,
+)
+from crewai.agents.agent_builder.base_agent import BaseAgent
+from crewai.tools import BaseTool
+from crewai.tools.agent_tools.agent_tools import AgentTools
+from crewai.utilities import Logger
+from crewai.utilities.events import crewai_event_bus
+from crewai.utilities.events.agent_events import (
+    AgentExecutionCompletedEvent,
+    AgentExecutionErrorEvent,
+    AgentExecutionStartedEvent,
+)
+
+try:
+    from agents import Agent as OpenAIAgent  # type: ignore
+    from agents import Runner, enable_verbose_stdout_logging  # type: ignore
+
+    from .openai_agent_tool_adapter import OpenAIAgentToolAdapter
+
+    OPENAI_AVAILABLE = True
+except ImportError:
+    OPENAI_AVAILABLE = False
+
+
+class OpenAIAgentAdapter(BaseAgentAdapter):
+    """Adapter for OpenAI Assistants"""
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    _openai_agent: "OpenAIAgent" = PrivateAttr()
+    _logger: Logger = PrivateAttr(default_factory=lambda: Logger())
+    _active_thread: Optional[str] = PrivateAttr(default=None)
+    function_calling_llm: Any = Field(default=None)
+    step_callback: Any = Field(default=None)
+    _tool_adapter: "OpenAIAgentToolAdapter" = PrivateAttr()
+    _converter_adapter: OpenAIConverterAdapter = PrivateAttr()
+
+    def __init__(
+        self,
+        model: str = "gpt-4o-mini",
+        tools: Optional[List[BaseTool]] = None,
+        agent_config: Optional[dict] = None,
+        **kwargs,
+    ):
+        if not OPENAI_AVAILABLE:
+            raise ImportError(
+                "OpenAI Agent Dependencies are not installed. Please install it using `uv add openai-agents`"
+            )
+        else:
+            role = kwargs.pop("role", None)
+            goal = kwargs.pop("goal", None)
+            backstory = kwargs.pop("backstory", None)
+            super().__init__(
+                role=role,
+                goal=goal,
+                backstory=backstory,
+                tools=tools,
+                agent_config=agent_config,
+                **kwargs,
+            )
+            self._tool_adapter = OpenAIAgentToolAdapter(tools=tools)
+            self.llm = model
+            self._converter_adapter = OpenAIConverterAdapter(self)
+
+    def _build_system_prompt(self) -> str:
+        """Build a system prompt for the OpenAI agent."""
+        base_prompt = f"""
+            You are {self.role}.
+        
+            Your goal is: {self.goal}
+
+            Your backstory: {self.backstory}
+
+            When working on tasks, think step-by-step and use the available tools when necessary.
+        """
+        return self._converter_adapter.enhance_system_prompt(base_prompt)
+
+    def execute_task(
+        self,
+        task: Any,
+        context: Optional[str] = None,
+        tools: Optional[List[BaseTool]] = None,
+    ) -> str:
+        """Execute a task using the OpenAI Assistant"""
+        self._converter_adapter.configure_structured_output(task)
+        self.create_agent_executor(tools)
+
+        if self.verbose:
+            enable_verbose_stdout_logging()
+
+        try:
+            task_prompt = task.prompt()
+            if context:
+                task_prompt = self.i18n.slice("task_with_context").format(
+                    task=task_prompt, context=context
+                )
+            crewai_event_bus.emit(
+                self,
+                event=AgentExecutionStartedEvent(
+                    agent=self,
+                    tools=self.tools,
+                    task_prompt=task_prompt,
+                    task=task,
+                ),
+            )
+            result = self.agent_executor.run_sync(self._openai_agent, task_prompt)
+            final_answer = self.handle_execution_result(result)
+            crewai_event_bus.emit(
+                self,
+                event=AgentExecutionCompletedEvent(
+                    agent=self, task=task, output=final_answer
+                ),
+            )
+            return final_answer
+
+        except Exception as e:
+            self._logger.log("error", f"Error executing OpenAI task: {str(e)}")
+            crewai_event_bus.emit(
+                self,
+                event=AgentExecutionErrorEvent(
+                    agent=self,
+                    task=task,
+                    error=str(e),
+                ),
+            )
+            raise
+
+    def create_agent_executor(self, tools: Optional[List[BaseTool]] = None) -> None:
+        """
+        Configure the OpenAI agent for execution.
+        While OpenAI handles execution differently through Runner,
+        we can use this method to set up tools and configurations.
+        """
+        all_tools = list(self.tools or []) + list(tools or [])
+
+        instructions = self._build_system_prompt()
+        self._openai_agent = OpenAIAgent(
+            name=self.role,
+            instructions=instructions,
+            model=self.llm,
+            **self._agent_config or {},
+        )
+
+        if all_tools:
+            self.configure_tools(all_tools)
+
+        self.agent_executor = Runner
+
+    def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
+        """Configure tools for the OpenAI Assistant"""
+        if tools:
+            self._tool_adapter.configure_tools(tools)
+            if self._tool_adapter.converted_tools:
+                self._openai_agent.tools = self._tool_adapter.converted_tools
+
+    def handle_execution_result(self, result: Any) -> str:
+        """Process OpenAI Assistant execution result converting any structured output to a string"""
+        return self._converter_adapter.post_process_result(result.final_output)
+
+    def get_delegation_tools(self, agents: List[BaseAgent]) -> List[BaseTool]:
+        """Implement delegation tools support"""
+        agent_tools = AgentTools(agents=agents)
+        tools = agent_tools.tools()
+        return tools
+
+    def configure_structured_output(self, task) -> None:
+        """Configure the structured output for the specific agent implementation.
+
+        Args:
+            structured_output: The structured output to be configured
+        """
+        self._converter_adapter.configure_structured_output(task)

src/crewai/agents/agent_adapters/openai_agents/openai_agent_tool_adapter.py

@@ -0,0 +1,91 @@
+import inspect
+from typing import Any, List, Optional
+
+from agents import FunctionTool, Tool
+
+from crewai.agents.agent_adapters.base_tool_adapter import BaseToolAdapter
+from crewai.tools import BaseTool
+
+
+class OpenAIAgentToolAdapter(BaseToolAdapter):
+    """Adapter for OpenAI Assistant tools"""
+
+    def __init__(self, tools: Optional[List[BaseTool]] = None):
+        self.original_tools = tools or []
+
+    def configure_tools(self, tools: List[BaseTool]) -> None:
+        """Configure tools for the OpenAI Assistant"""
+        if self.original_tools:
+            all_tools = tools + self.original_tools
+        else:
+            all_tools = tools
+        if all_tools:
+            self.converted_tools = self._convert_tools_to_openai_format(all_tools)
+
+    def _convert_tools_to_openai_format(
+        self, tools: Optional[List[BaseTool]]
+    ) -> List[Tool]:
+        """Convert CrewAI tools to OpenAI Assistant tool format"""
+        if not tools:
+            return []
+
+        def sanitize_tool_name(name: str) -> str:
+            """Convert tool name to match OpenAI's required pattern"""
+            import re
+
+            sanitized = re.sub(r"[^a-zA-Z0-9_-]", "_", name).lower()
+            return sanitized
+
+        def create_tool_wrapper(tool: BaseTool):
+            """Create a wrapper function that handles the OpenAI function tool interface"""
+
+            async def wrapper(context_wrapper: Any, arguments: Any) -> Any:
+                # Get the parameter name from the schema
+                param_name = list(
+                    tool.args_schema.model_json_schema()["properties"].keys()
+                )[0]
+
+                # Handle different argument types
+                if isinstance(arguments, dict):
+                    args_dict = arguments
+                elif isinstance(arguments, str):
+                    try:
+                        import json
+
+                        args_dict = json.loads(arguments)
+                    except json.JSONDecodeError:
+                        args_dict = {param_name: arguments}
+                else:
+                    args_dict = {param_name: str(arguments)}
+
+                # Run the tool with the processed arguments
+                output = tool._run(**args_dict)
+
+                # Await if the tool returned a coroutine
+                if inspect.isawaitable(output):
+                    result = await output
+                else:
+                    result = output
+
+                # Ensure the result is JSON serializable
+                if isinstance(result, (dict, list, str, int, float, bool, type(None))):
+                    return result
+                return str(result)
+
+            return wrapper
+
+        openai_tools = []
+        for tool in tools:
+            schema = tool.args_schema.model_json_schema()
+
+            schema.update({"additionalProperties": False, "type": "object"})
+
+            openai_tool = FunctionTool(
+                name=sanitize_tool_name(tool.name),
+                description=tool.description,
+                params_json_schema=schema,
+                on_invoke_tool=create_tool_wrapper(tool),
+            )
+            openai_tools.append(openai_tool)
+
+        return openai_tools

src/crewai/agents/agent_adapters/openai_agents/structured_output_converter.py

@@ -0,0 +1,122 @@
+import json
+import re
+
+from crewai.agents.agent_adapters.base_converter_adapter import BaseConverterAdapter
+from crewai.utilities.converter import generate_model_description
+from crewai.utilities.i18n import I18N
+
+
+class OpenAIConverterAdapter(BaseConverterAdapter):
+    """
+    Adapter for handling structured output conversion in OpenAI agents.
+
+    This adapter enhances the OpenAI agent to handle structured output formats
+    and post-processes the results when needed.
+
+    Attributes:
+        _output_format: The expected output format (json, pydantic, or None)
+        _schema: The schema description for the expected output
+        _output_model: The Pydantic model for the output
+    """
+
+    def __init__(self, agent_adapter):
+        """Initialize the converter adapter with a reference to the agent adapter"""
+        self.agent_adapter = agent_adapter
+        self._output_format = None
+        self._schema = None
+        self._output_model = None
+
+    def configure_structured_output(self, task) -> None:
+        """
+        Configure the structured output for OpenAI agent based on task requirements.
+
+        Args:
+            task: The task containing output format requirements
+        """
+        # Reset configuration
+        self._output_format = None
+        self._schema = None
+        self._output_model = None
+
+        # If no structured output is required, return early
+        if not (task.output_json or task.output_pydantic):
+            return
+
+        # Configure based on task output format
+        if task.output_json:
+            self._output_format = "json"
+            self._schema = generate_model_description(task.output_json)
+            self.agent_adapter._openai_agent.output_type = task.output_json
+            self._output_model = task.output_json
+        elif task.output_pydantic:
+            self._output_format = "pydantic"
+            self._schema = generate_model_description(task.output_pydantic)
+            self.agent_adapter._openai_agent.output_type = task.output_pydantic
+            self._output_model = task.output_pydantic
+
+    def enhance_system_prompt(self, base_prompt: str) -> str:
+        """
+        Enhance the base system prompt with structured output requirements if needed.
+
+        Args:
+            base_prompt: The original system prompt
+
+        Returns:
+            Enhanced system prompt with output format instructions if needed
+        """
+        if not self._output_format:
+            return base_prompt
+
+        output_schema = (
+            I18N()
+            .slice("formatted_task_instructions")
+            .format(output_format=self._schema)
+        )
+
+        return f"{base_prompt}\n\n{output_schema}"
+
+    def post_process_result(self, result: str) -> str:
+        """
+        Post-process the result to ensure it matches the expected format.
+
+        This method attempts to extract valid JSON from the result if necessary.
+
+        Args:
+            result: The raw result from the agent
+
+        Returns:
+            Processed result conforming to the expected output format
+        """
+        if not self._output_format:
+            return result
+        # Try to extract valid JSON if it's wrapped in code blocks or other text
+        if isinstance(result, str) and self._output_format in ["json", "pydantic"]:
+            # First, try to parse as is
+            try:
+                json.loads(result)
+                return result
+            except json.JSONDecodeError:
+                # Try to extract JSON from markdown code blocks
+                code_block_pattern = r"```(?:json)?\s*([\s\S]*?)```"
+                code_blocks = re.findall(code_block_pattern, result)
+
+                for block in code_blocks:
+                    try:
+                        json.loads(block.strip())
+                        return block.strip()
+                    except json.JSONDecodeError:
+                        continue
+
+                # Try to extract any JSON-like structure
+                json_pattern = r"(\{[\s\S]*\})"
+                json_matches = re.findall(json_pattern, result, re.DOTALL)
+
+                for match in json_matches:
+                    try:
+                        json.loads(match)
+                        return match
+                    except json.JSONDecodeError:
+                        continue
+
+        # If all extraction attempts fail, return the original
+        return str(result)

src/crewai/agents/agent_builder/base_agent.py

@@ -2,7 +2,7 @@ import uuid
 from abc import ABC, abstractmethod
 from copy import copy as shallow_copy
 from hashlib import md5
-from typing import Any, Dict, List, Optional, TypeVar
+from typing import Any, Callable, Dict, List, Optional, TypeVar
 
 from pydantic import (
     UUID4,
@@ -19,6 +19,7 @@ from crewai.agents.agent_builder.utilities.base_token_process import TokenProces
 from crewai.agents.cache.cache_handler import CacheHandler
 from crewai.agents.tools_handler import ToolsHandler
 from crewai.knowledge.knowledge import Knowledge
+from crewai.knowledge.knowledge_config import KnowledgeConfig
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
 from crewai.security.security_config import SecurityConfig
 from crewai.tools.base_tool import BaseTool, Tool
@@ -62,8 +63,6 @@ class BaseAgent(ABC, BaseModel):
             Abstract method to execute a task.
         create_agent_executor(tools=None) -> None:
             Abstract method to create an agent executor.
-        _parse_tools(tools: List[BaseTool]) -> List[Any]:
-            Abstract method to parse tools.
         get_delegation_tools(agents: List["BaseAgent"]):
             Abstract method to set the agents task tools for handling delegation and question asking to other agents in crew.
         get_output_converter(llm, model, instructions):
@@ -72,8 +71,6 @@ class BaseAgent(ABC, BaseModel):
             Interpolate inputs into the agent description and backstory.
         set_cache_handler(cache_handler: CacheHandler) -> None:
             Set the cache handler for the agent.
-        increment_formatting_errors() -> None:
-            Increment formatting errors.
         copy() -> "BaseAgent":
             Create a copy of the agent.
         set_rpm_controller(rpm_controller: RPMController) -> None:
@@ -91,9 +88,6 @@ class BaseAgent(ABC, BaseModel):
     _original_backstory: Optional[str] = PrivateAttr(default=None)
     _token_process: TokenProcess = PrivateAttr(default_factory=TokenProcess)
     id: UUID4 = Field(default_factory=uuid.uuid4, frozen=True)
-    formatting_errors: int = Field(
-        default=0, description="Number of formatting errors."
-    )
     role: str = Field(description="Role of the agent")
     goal: str = Field(description="Objective of the agent")
     backstory: str = Field(description="Backstory of the agent")
@@ -135,6 +129,9 @@ class BaseAgent(ABC, BaseModel):
         default_factory=ToolsHandler,
         description="An instance of the ToolsHandler class.",
     )
+    tools_results: List[Dict[str, Any]] = Field(
+        default=[], description="Results of the tools used by the agent."
+    )
     max_tokens: Optional[int] = Field(
         default=None, description="Maximum number of tokens for the agent's execution."
     )
@@ -153,6 +150,16 @@ class BaseAgent(ABC, BaseModel):
         default_factory=SecurityConfig,
         description="Security configuration for the agent, including fingerprinting.",
     )
+    callbacks: List[Callable] = Field(
+        default=[], description="Callbacks to be used for the agent"
+    )
+    adapted_agent: bool = Field(
+        default=False, description="Whether the agent is adapted"
+    )
+    knowledge_config: Optional[KnowledgeConfig] = Field(
+        default=None,
+        description="Knowledge configuration for the agent such as limits and threshold",
+    )
 
     @model_validator(mode="before")
     @classmethod
@@ -169,15 +176,15 @@ class BaseAgent(ABC, BaseModel):
         tool meets these criteria, it is processed and added to the list of
         tools. Otherwise, a ValueError is raised.
         """
+        if not tools:
+            return []
+
         processed_tools = []
+        required_attrs = ["name", "func", "description"]
         for tool in tools:
             if isinstance(tool, BaseTool):
                 processed_tools.append(tool)
-            elif (
+            elif all(hasattr(tool, attr) for attr in required_attrs):
-                hasattr(tool, "name")
-                and hasattr(tool, "func")
-                and hasattr(tool, "description")
-            ):
                 # Tool has the required attributes, create a Tool instance
                 processed_tools.append(Tool.from_langchain(tool))
             else:
@@ -254,22 +261,11 @@ class BaseAgent(ABC, BaseModel):
     def create_agent_executor(self, tools=None) -> None:
         pass
 
-    @abstractmethod
-    def _parse_tools(self, tools: List[BaseTool]) -> List[BaseTool]:
-        pass
-
     @abstractmethod
     def get_delegation_tools(self, agents: List["BaseAgent"]) -> List[BaseTool]:
         """Set the task tools that init BaseAgenTools class."""
         pass
 
-    @abstractmethod
-    def get_output_converter(
-        self, llm: Any, text: str, model: type[BaseModel] | None, instructions: str
-    ) -> Converter:
-        """Get the converter class for the agent to create json/pydantic outputs."""
-        pass
-
     def copy(self: T) -> T:  # type: ignore # Signature of "copy" incompatible with supertype "BaseModel"
         """Create a deep copy of the Agent."""
         exclude = {
@@ -356,9 +352,6 @@ class BaseAgent(ABC, BaseModel):
             self.tools_handler.cache = cache_handler
         self.create_agent_executor()
 
-    def increment_formatting_errors(self) -> None:
-        self.formatting_errors += 1
-
     def set_rpm_controller(self, rpm_controller: RPMController) -> None:
         """Set the rpm controller for the agent.
 

src/crewai/agents/agent_builder/base_agent_executor_mixin.py

@@ -1,5 +1,5 @@
 import time
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from crewai.memory.entity.entity_memory_item import EntityMemoryItem
 from crewai.memory.long_term.long_term_memory_item import LongTermMemoryItem
@@ -15,9 +15,9 @@ if TYPE_CHECKING:
 
 
 class CrewAgentExecutorMixin:
-    crew: Optional["Crew"]
+    crew: "Crew"
-    agent: Optional["BaseAgent"]
+    agent: "BaseAgent"
-    task: Optional["Task"]
+    task: "Task"
     iterations: int
     max_iter: int
     _i18n: I18N
@@ -47,11 +47,31 @@ class CrewAgentExecutorMixin:
                 print(f"Failed to add to short term memory: {e}")
                 pass
 
+    def _create_external_memory(self, output) -> None:
+        """Create and save a external-term memory item if conditions are met."""
+        if (
+            self.crew
+            and self.agent
+            and self.task
+            and hasattr(self.crew, "_external_memory")
+            and self.crew._external_memory
+        ):
+            try:
+                self.crew._external_memory.save(
+                    value=output.text,
+                    metadata={
+                        "description": self.task.description,
+                    },
+                    agent=self.agent.role,
+                )
+            except Exception as e:
+                print(f"Failed to add to external memory: {e}")
+                pass
+
     def _create_long_term_memory(self, output) -> None:
         """Create and save long-term and entity memory items based on evaluation."""
         if (
             self.crew
-            and self.crew.memory
             and self.crew._long_term_memory
             and self.crew._entity_memory
             and self.task
@@ -93,6 +113,15 @@ class CrewAgentExecutorMixin:
             except Exception as e:
                 print(f"Failed to add to long term memory: {e}")
                 pass
+        elif (
+            self.crew
+            and self.crew._long_term_memory
+            and self.crew._entity_memory is None
+        ):
+            self._printer.print(
+                content="Long term memory is enabled, but entity memory is not enabled. Please configure entity memory or set memory=True to automatically enable it.",
+                color="bold_yellow",
+            )
 
     def _ask_human_input(self, final_answer: str) -> str:
         """Prompt human input with mode-appropriate messaging."""

src/crewai/agents/crew_agent_executor.py

@@ -1,42 +1,40 @@
 import json
 import re
-from dataclasses import dataclass
 from typing import Any, Callable, Dict, List, Optional, Union
 
 from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.agents.agent_builder.base_agent_executor_mixin import CrewAgentExecutorMixin
 from crewai.agents.parser import (
-    FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE,
     AgentAction,
     AgentFinish,
-    CrewAgentParser,
     OutputParserException,
 )
 from crewai.agents.tools_handler import ToolsHandler
 from crewai.llm import BaseLLM
 from crewai.tools.base_tool import BaseTool
-from crewai.tools.tool_usage import ToolUsage, ToolUsageErrorException
+from crewai.tools.structured_tool import CrewStructuredTool
+from crewai.tools.tool_types import ToolResult
 from crewai.utilities import I18N, Printer
+from crewai.utilities.agent_utils import (
+    enforce_rpm_limit,
+    format_message_for_llm,
+    get_llm_response,
+    handle_agent_action_core,
+    handle_context_length,
+    handle_max_iterations_exceeded,
+    handle_output_parser_exception,
+    handle_unknown_error,
+    has_reached_max_iterations,
+    is_context_length_exceeded,
+    process_llm_response,
+    show_agent_logs,
+)
 from crewai.utilities.constants import MAX_LLM_RETRY, TRAINING_DATA_FILE
-from crewai.utilities.events import (
-    ToolUsageErrorEvent,
-    ToolUsageStartedEvent,
-    crewai_event_bus,
-)
-from crewai.utilities.events.tool_usage_events import ToolUsageStartedEvent
-from crewai.utilities.exceptions.context_window_exceeding_exception import (
-    LLMContextLengthExceededException,
-)
 from crewai.utilities.logger import Logger
+from crewai.utilities.tool_utils import execute_tool_and_check_finality
 from crewai.utilities.training_handler import CrewTrainingHandler
 
 
-@dataclass
-class ToolResult:
-    result: Any
-    result_as_answer: bool
-
-
 class CrewAgentExecutor(CrewAgentExecutorMixin):
     _logger: Logger = Logger()
 
@@ -48,7 +46,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         agent: BaseAgent,
         prompt: dict[str, str],
         max_iter: int,
-        tools: List[BaseTool],
+        tools: List[CrewStructuredTool],
         tools_names: str,
         stop_words: List[str],
         tools_description: str,
@@ -84,7 +82,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         self.messages: List[Dict[str, str]] = []
         self.iterations = 0
         self.log_error_after = 3
-        self.tool_name_to_tool_map: Dict[str, BaseTool] = {
+        self.tool_name_to_tool_map: Dict[str, Union[CrewStructuredTool, BaseTool]] = {
             tool.name: tool for tool in self.tools
         }
         existing_stop = self.llm.stop or []
@@ -100,11 +98,11 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         if "system" in self.prompt:
             system_prompt = self._format_prompt(self.prompt.get("system", ""), inputs)
             user_prompt = self._format_prompt(self.prompt.get("user", ""), inputs)
-            self.messages.append(self._format_msg(system_prompt, role="system"))
+            self.messages.append(format_message_for_llm(system_prompt, role="system"))
-            self.messages.append(self._format_msg(user_prompt))
+            self.messages.append(format_message_for_llm(user_prompt))
         else:
             user_prompt = self._format_prompt(self.prompt.get("prompt", ""), inputs)
-            self.messages.append(self._format_msg(user_prompt))
+            self.messages.append(format_message_for_llm(user_prompt))
 
         self._show_start_logs()
 
@@ -119,7 +117,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
             )
             raise
         except Exception as e:
-            self._handle_unknown_error(e)
+            handle_unknown_error(self._printer, e)
             if e.__class__.__module__.startswith("litellm"):
                 # Do not retry on litellm errors
                 raise e
@@ -131,6 +129,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
 
         self._create_short_term_memory(formatted_answer)
         self._create_long_term_memory(formatted_answer)
+        self._create_external_memory(formatted_answer)
         return {"output": formatted_answer.output}
 
     def _invoke_loop(self) -> AgentFinish:
@@ -141,20 +140,51 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         formatted_answer = None
         while not isinstance(formatted_answer, AgentFinish):
             try:
-                if self._has_reached_max_iterations():
+                if has_reached_max_iterations(self.iterations, self.max_iter):
-                    formatted_answer = self._handle_max_iterations_exceeded(
+                    formatted_answer = handle_max_iterations_exceeded(
-                        formatted_answer
+                        formatted_answer,
+                        printer=self._printer,
+                        i18n=self._i18n,
+                        messages=self.messages,
+                        llm=self.llm,
+                        callbacks=self.callbacks,
                     )
-                    break
 
-                self._enforce_rpm_limit()
+                enforce_rpm_limit(self.request_within_rpm_limit)
 
-                answer = self._get_llm_response()
+                answer = get_llm_response(
-                formatted_answer = self._process_llm_response(answer)
+                    llm=self.llm,
+                    messages=self.messages,
+                    callbacks=self.callbacks,
+                    printer=self._printer,
+                )
+                formatted_answer = process_llm_response(answer, self.use_stop_words)
 
                 if isinstance(formatted_answer, AgentAction):
-                    tool_result = self._execute_tool_and_check_finality(
+                    # Extract agent fingerprint if available
-                        formatted_answer
+                    fingerprint_context = {}
+                    if (
+                        self.agent
+                        and hasattr(self.agent, "security_config")
+                        and hasattr(self.agent.security_config, "fingerprint")
+                    ):
+                        fingerprint_context = {
+                            "agent_fingerprint": str(
+                                self.agent.security_config.fingerprint
+                            )
+                        }
+
+                    tool_result = execute_tool_and_check_finality(
+                        agent_action=formatted_answer,
+                        fingerprint_context=fingerprint_context,
+                        tools=self.tools,
+                        i18n=self._i18n,
+                        agent_key=self.agent.key if self.agent else None,
+                        agent_role=self.agent.role if self.agent else None,
+                        tools_handler=self.tools_handler,
+                        task=self.task,
+                        agent=self.agent,
+                        function_calling_llm=self.function_calling_llm,
                     )
                     formatted_answer = self._handle_agent_action(
                         formatted_answer, tool_result
@@ -164,17 +194,30 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
                 self._append_message(formatted_answer.text, role="assistant")
 
             except OutputParserException as e:
-                formatted_answer = self._handle_output_parser_exception(e)
+                formatted_answer = handle_output_parser_exception(
+                    e=e,
+                    messages=self.messages,
+                    iterations=self.iterations,
+                    log_error_after=self.log_error_after,
+                    printer=self._printer,
+                )
 
             except Exception as e:
                 if e.__class__.__module__.startswith("litellm"):
                     # Do not retry on litellm errors
                     raise e
-                if self._is_context_length_exceeded(e):
+                if is_context_length_exceeded(e):
-                    self._handle_context_length()
+                    handle_context_length(
+                        respect_context_window=self.respect_context_window,
+                        printer=self._printer,
+                        messages=self.messages,
+                        llm=self.llm,
+                        callbacks=self.callbacks,
+                        i18n=self._i18n,
+                    )
                     continue
                 else:
-                    self._handle_unknown_error(e)
+                    handle_unknown_error(self._printer, e)
                     raise e
             finally:
                 self.iterations += 1
@@ -187,90 +230,28 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         self._show_logs(formatted_answer)
         return formatted_answer
 
-    def _handle_unknown_error(self, exception: Exception) -> None:
-        """Handle unknown errors by informing the user."""
-        self._printer.print(
-            content="An unknown error occurred. Please check the details below.",
-            color="red",
-        )
-        self._printer.print(
-            content=f"Error details: {exception}",
-            color="red",
-        )
-
-    def _has_reached_max_iterations(self) -> bool:
-        """Check if the maximum number of iterations has been reached."""
-        return self.iterations >= self.max_iter
-
-    def _enforce_rpm_limit(self) -> None:
-        """Enforce the requests per minute (RPM) limit if applicable."""
-        if self.request_within_rpm_limit:
-            self.request_within_rpm_limit()
-
-    def _get_llm_response(self) -> str:
-        """Call the LLM and return the response, handling any invalid responses."""
-        try:
-            answer = self.llm.call(
-                self.messages,
-                callbacks=self.callbacks,
-            )
-        except Exception as e:
-            self._printer.print(
-                content=f"Error during LLM call: {e}",
-                color="red",
-            )
-            raise e
-
-        if not answer:
-            self._printer.print(
-                content="Received None or empty response from LLM call.",
-                color="red",
-            )
-            raise ValueError("Invalid response from LLM call - None or empty.")
-
-        return answer
-
-    def _process_llm_response(self, answer: str) -> Union[AgentAction, AgentFinish]:
-        """Process the LLM response and format it into an AgentAction or AgentFinish."""
-        if not self.use_stop_words:
-            try:
-                # Preliminary parsing to check for errors.
-                self._format_answer(answer)
-            except OutputParserException as e:
-                if FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE in e.error:
-                    answer = answer.split("Observation:")[0].strip()
-
-        return self._format_answer(answer)
-
     def _handle_agent_action(
         self, formatted_answer: AgentAction, tool_result: ToolResult
     ) -> Union[AgentAction, AgentFinish]:
         """Handle the AgentAction, execute tools, and process the results."""
+        # Special case for add_image_tool
         add_image_tool = self._i18n.tools("add_image")
         if (
             isinstance(add_image_tool, dict)
             and formatted_answer.tool.casefold().strip()
             == add_image_tool.get("name", "").casefold().strip()
         ):
-            self.messages.append(tool_result.result)
+            self.messages.append({"role": "assistant", "content": tool_result.result})
-            return formatted_answer  # Continue the loop
-
-        if self.step_callback:
-            self.step_callback(tool_result)
-
-        formatted_answer.text += f"\nObservation: {tool_result.result}"
-        formatted_answer.result = tool_result.result
-
-        if tool_result.result_as_answer:
-            return AgentFinish(
-                thought="",
-                output=tool_result.result,
-                text=formatted_answer.text,
-            )
-
-        self._show_logs(formatted_answer)
             return formatted_answer
 
+        return handle_agent_action_core(
+            formatted_answer=formatted_answer,
+            tool_result=tool_result,
+            messages=self.messages,
+            step_callback=self.step_callback,
+            show_logs=self._show_logs,
+        )
+
     def _invoke_step_callback(self, formatted_answer) -> None:
         """Invoke the step callback if it exists."""
         if self.step_callback:
@@ -278,151 +259,33 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
 
     def _append_message(self, text: str, role: str = "assistant") -> None:
         """Append a message to the message list with the given role."""
-        self.messages.append(self._format_msg(text, role=role))
+        self.messages.append(format_message_for_llm(text, role=role))
-
-    def _handle_output_parser_exception(self, e: OutputParserException) -> AgentAction:
-        """Handle OutputParserException by updating messages and formatted_answer."""
-        self.messages.append({"role": "user", "content": e.error})
-
-        formatted_answer = AgentAction(
-            text=e.error,
-            tool="",
-            tool_input="",
-            thought="",
-        )
-
-        if self.iterations > self.log_error_after:
-            self._printer.print(
-                content=f"Error parsing LLM output, agent will retry: {e.error}",
-                color="red",
-            )
-
-        return formatted_answer
-
-    def _is_context_length_exceeded(self, exception: Exception) -> bool:
-        """Check if the exception is due to context length exceeding."""
-        return LLMContextLengthExceededException(
-            str(exception)
-        )._is_context_limit_error(str(exception))
 
     def _show_start_logs(self):
+        """Show logs for the start of agent execution."""
         if self.agent is None:
             raise ValueError("Agent cannot be None")
-        if self.agent.verbose or (
+        show_agent_logs(
-            hasattr(self, "crew") and getattr(self.crew, "verbose", False)
+            printer=self._printer,
-        ):
+            agent_role=self.agent.role,
-            agent_role = self.agent.role.split("\n")[0]
+            task_description=(
-            self._printer.print(
-                content=f"\033[1m\033[95m# Agent:\033[00m \033[1m\033[92m{agent_role}\033[00m"
-            )
-            description = (
                 getattr(self.task, "description") if self.task else "Not Found"
-            )
+            ),
-            self._printer.print(
+            verbose=self.agent.verbose
-                content=f"\033[95m## Task:\033[00m \033[92m{description}\033[00m"
+            or (hasattr(self, "crew") and getattr(self.crew, "verbose", False)),
         )
 
     def _show_logs(self, formatted_answer: Union[AgentAction, AgentFinish]):
+        """Show logs for the agent's execution."""
         if self.agent is None:
             raise ValueError("Agent cannot be None")
-        if self.agent.verbose or (
+        show_agent_logs(
-            hasattr(self, "crew") and getattr(self.crew, "verbose", False)
+            printer=self._printer,
-        ):
-            agent_role = self.agent.role.split("\n")[0]
-            if isinstance(formatted_answer, AgentAction):
-                thought = re.sub(r"\n+", "\n", formatted_answer.thought)
-                formatted_json = json.dumps(
-                    formatted_answer.tool_input,
-                    indent=2,
-                    ensure_ascii=False,
-                )
-                self._printer.print(
-                    content=f"\n\n\033[1m\033[95m# Agent:\033[00m \033[1m\033[92m{agent_role}\033[00m"
-                )
-                if thought and thought != "":
-                    self._printer.print(
-                        content=f"\033[95m## Thought:\033[00m \033[92m{thought}\033[00m"
-                    )
-                self._printer.print(
-                    content=f"\033[95m## Using tool:\033[00m \033[92m{formatted_answer.tool}\033[00m"
-                )
-                self._printer.print(
-                    content=f"\033[95m## Tool Input:\033[00m \033[92m\n{formatted_json}\033[00m"
-                )
-                self._printer.print(
-                    content=f"\033[95m## Tool Output:\033[00m \033[92m\n{formatted_answer.result}\033[00m"
-                )
-            elif isinstance(formatted_answer, AgentFinish):
-                self._printer.print(
-                    content=f"\n\n\033[1m\033[95m# Agent:\033[00m \033[1m\033[92m{agent_role}\033[00m"
-                )
-                self._printer.print(
-                    content=f"\033[95m## Final Answer:\033[00m \033[92m\n{formatted_answer.output}\033[00m\n\n"
-                )
-
-    def _execute_tool_and_check_finality(self, agent_action: AgentAction) -> ToolResult:
-        try:
-            if self.agent:
-                crewai_event_bus.emit(
-                    self,
-                    event=ToolUsageStartedEvent(
-                        agent_key=self.agent.key,
             agent_role=self.agent.role,
-                        tool_name=agent_action.tool,
+            formatted_answer=formatted_answer,
-                        tool_args=agent_action.tool_input,
+            verbose=self.agent.verbose
-                        tool_class=agent_action.tool,
+            or (hasattr(self, "crew") and getattr(self.crew, "verbose", False)),
-                    ),
         )
-            tool_usage = ToolUsage(
-                tools_handler=self.tools_handler,
-                tools=self.tools,
-                original_tools=self.original_tools,
-                tools_description=self.tools_description,
-                tools_names=self.tools_names,
-                function_calling_llm=self.function_calling_llm,
-                task=self.task,  # type: ignore[arg-type]
-                agent=self.agent,
-                action=agent_action,
-            )
-            tool_calling = tool_usage.parse_tool_calling(agent_action.text)
-
-            if isinstance(tool_calling, ToolUsageErrorException):
-                tool_result = tool_calling.message
-                return ToolResult(result=tool_result, result_as_answer=False)
-            else:
-                if tool_calling.tool_name.casefold().strip() in [
-                    name.casefold().strip() for name in self.tool_name_to_tool_map
-                ] or tool_calling.tool_name.casefold().replace("_", " ") in [
-                    name.casefold().strip() for name in self.tool_name_to_tool_map
-                ]:
-                    tool_result = tool_usage.use(tool_calling, agent_action.text)
-                    tool = self.tool_name_to_tool_map.get(tool_calling.tool_name)
-                    if tool:
-                        return ToolResult(
-                            result=tool_result, result_as_answer=tool.result_as_answer
-                        )
-                else:
-                    tool_result = self._i18n.errors("wrong_tool_name").format(
-                        tool=tool_calling.tool_name,
-                        tools=", ".join([tool.name.casefold() for tool in self.tools]),
-                    )
-                return ToolResult(result=tool_result, result_as_answer=False)
-
-        except Exception as e:
-            # TODO: drop
-            if self.agent:
-                crewai_event_bus.emit(
-                    self,
-                    event=ToolUsageErrorEvent(  # validation error
-                        agent_key=self.agent.key,
-                        agent_role=self.agent.role,
-                        tool_name=agent_action.tool,
-                        tool_args=agent_action.tool_input,
-                        tool_class=agent_action.tool,
-                        error=str(e),
-                    ),
-                )
-            raise e
 
     def _summarize_messages(self) -> None:
         messages_groups = []
@@ -430,47 +293,33 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
             content = message["content"]
             cut_size = self.llm.get_context_window_size()
             for i in range(0, len(content), cut_size):
-                messages_groups.append(content[i : i + cut_size])
+                messages_groups.append({"content": content[i : i + cut_size]})
 
         summarized_contents = []
         for group in messages_groups:
             summary = self.llm.call(
                 [
-                    self._format_msg(
+                    format_message_for_llm(
                         self._i18n.slice("summarizer_system_message"), role="system"
                     ),
-                    self._format_msg(
+                    format_message_for_llm(
-                        self._i18n.slice("summarize_instruction").format(group=group),
+                        self._i18n.slice("summarize_instruction").format(
+                            group=group["content"]
+                        ),
                     ),
                 ],
                 callbacks=self.callbacks,
             )
-            summarized_contents.append(summary)
+            summarized_contents.append({"content": str(summary)})
 
-        merged_summary = " ".join(str(content) for content in summarized_contents)
+        merged_summary = " ".join(content["content"] for content in summarized_contents)
 
         self.messages = [
-            self._format_msg(
+            format_message_for_llm(
                 self._i18n.slice("summary").format(merged_summary=merged_summary)
             )
         ]
 
-    def _handle_context_length(self) -> None:
-        if self.respect_context_window:
-            self._printer.print(
-                content="Context length exceeded. Summarizing content to fit the model context window.",
-                color="yellow",
-            )
-            self._summarize_messages()
-        else:
-            self._printer.print(
-                content="Context length exceeded. Consider using smaller text or RAG tools from crewai_tools.",
-                color="red",
-            )
-            raise SystemExit(
-                "Context length exceeded and user opted not to summarize. Consider using smaller text or RAG tools from crewai_tools."
-            )
-
     def _handle_crew_training_output(
         self, result: AgentFinish, human_feedback: Optional[str] = None
     ) -> None:
@@ -523,13 +372,6 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         prompt = prompt.replace("{tools}", inputs["tools"])
         return prompt
 
-    def _format_answer(self, answer: str) -> Union[AgentAction, AgentFinish]:
-        return CrewAgentParser(agent=self.agent).parse(answer)
-
-    def _format_msg(self, prompt: str, role: str = "user") -> Dict[str, str]:
-        prompt = prompt.rstrip()
-        return {"role": role, "content": prompt}
-
     def _handle_human_feedback(self, formatted_answer: AgentFinish) -> AgentFinish:
         """Handle human feedback with different flows for training vs regular use.
 
@@ -556,7 +398,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
         """Process feedback for training scenarios with single iteration."""
         self._handle_crew_training_output(initial_answer, feedback)
         self.messages.append(
-            self._format_msg(
+            format_message_for_llm(
                 self._i18n.slice("feedback_instructions").format(feedback=feedback)
             )
         )
@@ -585,7 +427,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
     def _process_feedback_iteration(self, feedback: str) -> AgentFinish:
         """Process a single feedback iteration."""
         self.messages.append(
-            self._format_msg(
+            format_message_for_llm(
                 self._i18n.slice("feedback_instructions").format(feedback=feedback)
             )
         )
@@ -610,45 +452,3 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
             ),
             color="red",
         )
-
-    def _handle_max_iterations_exceeded(self, formatted_answer):
-        """
-        Handles the case when the maximum number of iterations is exceeded.
-        Performs one more LLM call to get the final answer.
-
-        Parameters:
-            formatted_answer: The last formatted answer from the agent.
-
-        Returns:
-            The final formatted answer after exceeding max iterations.
-        """
-        self._printer.print(
-            content="Maximum iterations reached. Requesting final answer.",
-            color="yellow",
-        )
-
-        if formatted_answer and hasattr(formatted_answer, "text"):
-            assistant_message = (
-                formatted_answer.text + f'\n{self._i18n.errors("force_final_answer")}'
-            )
-        else:
-            assistant_message = self._i18n.errors("force_final_answer")
-
-        self.messages.append(self._format_msg(assistant_message, role="assistant"))
-
-        # Perform one more LLM call to get the final answer
-        answer = self.llm.call(
-            self.messages,
-            callbacks=self.callbacks,
-        )
-
-        if answer is None or answer == "":
-            self._printer.print(
-                content="Received None or empty response from LLM call.",
-                color="red",
-            )
-            raise ValueError("Invalid response from LLM call - None or empty.")
-
-        formatted_answer = self._format_answer(answer)
-        # Return the formatted answer, regardless of its type
-        return formatted_answer

src/crewai/agents/parser.py

@@ -1,5 +1,5 @@
 import re
-from typing import Any, Union
+from typing import Any, Optional, Union
 
 from json_repair import repair_json
 
@@ -67,9 +67,23 @@ class CrewAgentParser:
     _i18n: I18N = I18N()
     agent: Any = None
 
-    def __init__(self, agent: Any):
+    def __init__(self, agent: Optional[Any] = None):
         self.agent = agent
 
+    @staticmethod
+    def parse_text(text: str) -> Union[AgentAction, AgentFinish]:
+        """
+        Static method to parse text into an AgentAction or AgentFinish without needing to instantiate the class.
+
+        Args:
+            text: The text to parse.
+
+        Returns:
+            Either an AgentAction or AgentFinish based on the parsed content.
+        """
+        parser = CrewAgentParser()
+        return parser.parse(text)
+
     def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
         thought = self._extract_thought(text)
         includes_answer = FINAL_ANSWER_ACTION in text
@@ -77,22 +91,7 @@ class CrewAgentParser:
             r"Action\s*\d*\s*:[\s]*(.*?)[\s]*Action\s*\d*\s*Input\s*\d*\s*:[\s]*(.*)"
         )
         action_match = re.search(regex, text, re.DOTALL)
-        if action_match:
         if includes_answer:
-                raise OutputParserException(
-                    f"{FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE}"
-                )
-            action = action_match.group(1)
-            clean_action = self._clean_action(action)
-
-            action_input = action_match.group(2).strip()
-
-            tool_input = action_input.strip(" ").strip('"')
-            safe_tool_input = self._safe_repair_json(tool_input)
-
-            return AgentAction(thought, clean_action, safe_tool_input, text)
-
-        elif includes_answer:
             final_answer = text.split(FINAL_ANSWER_ACTION)[-1].strip()
             # Check whether the final answer ends with triple backticks.
             if final_answer.endswith("```"):
@@ -103,22 +102,30 @@ class CrewAgentParser:
                     final_answer = final_answer[:-3].rstrip()
             return AgentFinish(thought, final_answer, text)
 
+        elif action_match:
+            action = action_match.group(1)
+            clean_action = self._clean_action(action)
+
+            action_input = action_match.group(2).strip()
+
+            tool_input = action_input.strip(" ").strip('"')
+            safe_tool_input = self._safe_repair_json(tool_input)
+
+            return AgentAction(thought, clean_action, safe_tool_input, text)
+
         if not re.search(r"Action\s*\d*\s*:[\s]*(.*?)", text, re.DOTALL):
-            self.agent.increment_formatting_errors()
             raise OutputParserException(
                 f"{MISSING_ACTION_AFTER_THOUGHT_ERROR_MESSAGE}\n{self._i18n.slice('final_answer_format')}",
             )
         elif not re.search(
             r"[\s]*Action\s*\d*\s*Input\s*\d*\s*:[\s]*(.*)", text, re.DOTALL
         ):
-            self.agent.increment_formatting_errors()
             raise OutputParserException(
                 MISSING_ACTION_INPUT_AFTER_ACTION_ERROR_MESSAGE,
             )
         else:
             format = self._i18n.slice("format_without_tools")
             error = f"{format}"
-            self.agent.increment_formatting_errors()
             raise OutputParserException(
                 error,
             )

src/crewai/cli/constants.py

@@ -91,6 +91,12 @@ ENV_VARS = {
             "key_name": "CEREBRAS_API_KEY",
         },
     ],
+    "huggingface": [
+        {
+            "prompt": "Enter your Huggingface API key (HF_TOKEN) (press Enter to skip)",
+            "key_name": "HF_TOKEN",
+        },
+    ],
     "sambanova": [
         {
             "prompt": "Enter your SambaNovaCloud API key (press Enter to skip)",
@@ -106,6 +112,7 @@ PROVIDERS = [
     "gemini",
     "nvidia_nim",
     "groq",
+    "huggingface",
     "ollama",
     "watson",
     "bedrock",
@@ -270,6 +277,12 @@ MODELS = {
         "bedrock/mistral.mistral-7b-instruct-v0:2",
         "bedrock/mistral.mixtral-8x7b-instruct-v0:1",
     ],
+    "huggingface": [
+        "huggingface/meta-llama/Meta-Llama-3.1-8B-Instruct",
+        "huggingface/mistralai/Mixtral-8x7B-Instruct-v0.1",
+        "huggingface/tiiuae/falcon-180B-chat",
+        "huggingface/google/gemma-7b-it",
+    ],
     "sambanova": [
         "sambanova/Meta-Llama-3.3-70B-Instruct",
         "sambanova/QwQ-32B-Preview",

src/crewai/cli/install_crew.py

@@ -3,6 +3,10 @@ import subprocess
 import click
 
 
+# Be mindful about changing this.
+# on some enviorments we don't use this command but instead uv sync directly
+# so if you expect this to support more things you will need to replicate it there
+# ask @joaomdmoura if you are unsure
 def install_crew(proxy_options: list[str]) -> None:
     """
     Install the crew by running the UV command to lock and install.

src/crewai/cli/templates/crew/crew.py

@@ -1,6 +1,7 @@
 from crewai import Agent, Crew, Process, Task
 from crewai.project import CrewBase, agent, crew, task
-
+from crewai.agents.agent_builder.base_agent import BaseAgent
+from typing import List
 # If you want to run a snippet of code before or after the crew starts,
 # you can use the @before_kickoff and @after_kickoff decorators
 # https://docs.crewai.com/concepts/crews#example-crew-class-with-decorators
@@ -9,25 +10,26 @@ from crewai.project import CrewBase, agent, crew, task
 class {{crew_name}}():
     """{{crew_name}} crew"""
 
+    agents: List[BaseAgent]
+    tasks: List[Task]
+
     # Learn more about YAML configuration files here:
     # Agents: https://docs.crewai.com/concepts/agents#yaml-configuration-recommended
     # Tasks: https://docs.crewai.com/concepts/tasks#yaml-configuration-recommended
-    agents_config = 'config/agents.yaml'
-    tasks_config = 'config/tasks.yaml'
     
     # If you would like to add tools to your agents, you can learn more about it here:
     # https://docs.crewai.com/concepts/agents#agent-tools
     @agent
     def researcher(self) -> Agent:
         return Agent(
-            config=self.agents_config['researcher'],
+            config=self.agents_config['researcher'], # type: ignore[index]
             verbose=True
         )
 
     @agent
     def reporting_analyst(self) -> Agent:
         return Agent(
-            config=self.agents_config['reporting_analyst'],
+            config=self.agents_config['reporting_analyst'], # type: ignore[index]
             verbose=True
         )
 
@@ -37,13 +39,13 @@ class {{crew_name}}():
     @task
     def research_task(self) -> Task:
         return Task(
-            config=self.tasks_config['research_task'],
+            config=self.tasks_config['research_task'], # type: ignore[index]
         )
 
     @task
     def reporting_task(self) -> Task:
         return Task(
-            config=self.tasks_config['reporting_task'],
+            config=self.tasks_config['reporting_task'], # type: ignore[index]
             output_file='report.md'
         )
 

src/crewai/cli/templates/crew/main.py

@@ -33,7 +33,8 @@ def train():
     Train the crew for a given number of iterations.
     """
     inputs = {
-        "topic": "AI LLMs"
+        "topic": "AI LLMs",
+        'current_year': str(datetime.now().year)
     }
     try:
         {{crew_name}}().crew().train(n_iterations=int(sys.argv[1]), filename=sys.argv[2], inputs=inputs)
@@ -59,8 +60,9 @@ def test():
         "topic": "AI LLMs",
         "current_year": str(datetime.now().year)
     }
+    
     try:
-        {{crew_name}}().crew().test(n_iterations=int(sys.argv[1]), openai_model_name=sys.argv[2], inputs=inputs)
+        {{crew_name}}().crew().test(n_iterations=int(sys.argv[1]), eval_llm=sys.argv[2], inputs=inputs)
 
     except Exception as e:
         raise Exception(f"An error occurred while testing the crew: {e}")

src/crewai/cli/templates/crew/pyproject.toml

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
 authors = [{ name = "Your Name", email = "you@example.com" }]
 requires-python = ">=3.10,<3.13"
 dependencies = [
-    "crewai[tools]>=0.108.0,<1.0.0"
+    "crewai[tools]>=0.114.0,<1.0.0"
 ]
 
 [project.scripts]

src/crewai/cli/templates/flow/crews/poem_crew/poem_crew.py

@@ -1,5 +1,7 @@
 from crewai import Agent, Crew, Process, Task
 from crewai.project import CrewBase, agent, crew, task
+from crewai.agents.agent_builder.base_agent import BaseAgent
+from typing import List
 
 # If you want to run a snippet of code before or after the crew starts,
 # you can use the @before_kickoff and @after_kickoff decorators
@@ -10,6 +12,9 @@ from crewai.project import CrewBase, agent, crew, task
 class PoemCrew:
     """Poem Crew"""
 
+    agents: List[BaseAgent]
+    tasks: List[Task]
+
     # Learn more about YAML configuration files here:
     # Agents: https://docs.crewai.com/concepts/agents#yaml-configuration-recommended
     # Tasks: https://docs.crewai.com/concepts/tasks#yaml-configuration-recommended
@@ -21,7 +26,7 @@ class PoemCrew:
     @agent
     def poem_writer(self) -> Agent:
         return Agent(
-            config=self.agents_config["poem_writer"],
+            config=self.agents_config["poem_writer"],  # type: ignore[index]
         )
 
     # To learn more about structured task outputs,
@@ -30,7 +35,7 @@ class PoemCrew:
     @task
     def write_poem(self) -> Task:
         return Task(
-            config=self.tasks_config["write_poem"],
+            config=self.tasks_config["write_poem"],  # type: ignore[index]
         )
 
     @crew

src/crewai/cli/templates/flow/pyproject.toml

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
 authors = [{ name = "Your Name", email = "you@example.com" }]
 requires-python = ">=3.10,<3.13"
 dependencies = [
-    "crewai[tools]>=0.108.0,<1.0.0",
+    "crewai[tools]>=0.114.0,<1.0.0",
 ]
 
 [project.scripts]

src/crewai/cli/templates/tool/pyproject.toml

@@ -5,7 +5,7 @@ description = "Power up your crews with {{folder_name}}"
 readme = "README.md"
 requires-python = ">=3.10,<3.13"
 dependencies = [
-    "crewai[tools]>=0.108.0"
+    "crewai[tools]>=0.114.0"
 ]
 
 [tool.crewai]

src/crewai/cli/tools/main.py

@@ -117,7 +117,9 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
 
         published_handle = publish_response.json()["handle"]
         console.print(
-            f"Successfully published {published_handle} ({project_version}).\nInstall it in other projects with crewai tool install {published_handle}",
+            f"Successfully published `{published_handle}` ({project_version}).\n\n"
+            + "⚠️ Security checks are running in the background. Your tool will be available once these are complete.\n"
+            + f"You can monitor the status or access your tool here:\nhttps://app.crewai.com/crewai_plus/tools/{published_handle}",
             style="bold green",
         )
 
@@ -153,8 +155,12 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
         login_response_json = login_response.json()
 
         settings = Settings()
-        settings.tool_repository_username = login_response_json["credential"]["username"]
+        settings.tool_repository_username = login_response_json["credential"][
-        settings.tool_repository_password = login_response_json["credential"]["password"]
+            "username"
+        ]
+        settings.tool_repository_password = login_response_json["credential"][
+            "password"
+        ]
         settings.dump()
 
         console.print(
@@ -179,7 +185,7 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
             capture_output=False,
             env=self._build_env_with_credentials(repository_handle),
             text=True,
-            check=True
+            check=True,
         )
 
         if add_package_result.stderr:
@@ -204,7 +210,11 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
         settings = Settings()
 
         env = os.environ.copy()
-        env[f"UV_INDEX_{repository_handle}_USERNAME"] = str(settings.tool_repository_username or "")
+        env[f"UV_INDEX_{repository_handle}_USERNAME"] = str(
-        env[f"UV_INDEX_{repository_handle}_PASSWORD"] = str(settings.tool_repository_password or "")
+            settings.tool_repository_username or ""
+        )
+        env[f"UV_INDEX_{repository_handle}_PASSWORD"] = str(
+            settings.tool_repository_password or ""
+        )
 
         return env

src/crewai/cli/utils.py

@@ -273,11 +273,9 @@ def get_crew(crew_path: str = "crew.py", require: bool = False) -> Crew | None:
                         for attr_name in dir(module):
                             attr = getattr(module, attr_name)
                             try:
-                                if isinstance(attr, Crew) and hasattr(attr, "kickoff"):
+                                if callable(attr) and hasattr(attr, "crew"):
-                                    print(
+                                    crew_instance = attr().crew()
-                                        f"Found valid crew object in attribute '{attr_name}' at {crew_os_path}."
+                                    return crew_instance
-                                    )
-                                    return attr
 
                             except Exception as e:
                                 print(f"Error processing attribute {attr_name}: {e}")

src/crewai/crew.py

@@ -28,6 +28,7 @@ from crewai.knowledge.knowledge import Knowledge
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
 from crewai.llm import LLM, BaseLLM
 from crewai.memory.entity.entity_memory import EntityMemory
+from crewai.memory.external.external_memory import ExternalMemory
 from crewai.memory.long_term.long_term_memory import LongTermMemory
 from crewai.memory.short_term.short_term_memory import ShortTermMemory
 from crewai.memory.user.user_memory import UserMemory
@@ -105,6 +106,7 @@ class Crew(BaseModel):
     _long_term_memory: Optional[InstanceOf[LongTermMemory]] = PrivateAttr()
     _entity_memory: Optional[InstanceOf[EntityMemory]] = PrivateAttr()
     _user_memory: Optional[InstanceOf[UserMemory]] = PrivateAttr()
+    _external_memory: Optional[InstanceOf[ExternalMemory]] = PrivateAttr()
     _train: Optional[bool] = PrivateAttr(default=False)
     _train_iteration: Optional[int] = PrivateAttr()
     _inputs: Optional[Dict[str, Any]] = PrivateAttr(default=None)
@@ -145,6 +147,10 @@ class Crew(BaseModel):
         default=None,
         description="An instance of the UserMemory to be used by the Crew to store/fetch memories of a specific user.",
     )
+    external_memory: Optional[InstanceOf[ExternalMemory]] = Field(
+        default=None,
+        description="An Instance of the ExternalMemory to be used by the Crew",
+    )
     embedder: Optional[dict] = Field(
         default=None,
         description="Configuration for the embedder to be used for the crew.",
@@ -269,46 +275,49 @@ class Crew(BaseModel):
 
         return self
 
-    @model_validator(mode="after")
+    def _initialize_user_memory(self):
-    def create_crew_memory(self) -> "Crew":
-        """Set private attributes."""
-        if self.memory:
-            self._long_term_memory = (
-                self.long_term_memory if self.long_term_memory else LongTermMemory()
-            )
-            self._short_term_memory = (
-                self.short_term_memory
-                if self.short_term_memory
-                else ShortTermMemory(
-                    crew=self,
-                    embedder_config=self.embedder,
-                )
-            )
-            self._entity_memory = (
-                self.entity_memory
-                if self.entity_memory
-                else EntityMemory(crew=self, embedder_config=self.embedder)
-            )
         if (
-                self.memory_config and "user_memory" in self.memory_config
+            self.memory_config
+            and "user_memory" in self.memory_config
+            and self.memory_config.get("provider") == "mem0"
         ):  # Check for user_memory in config
             user_memory_config = self.memory_config["user_memory"]
             if isinstance(
-                    user_memory_config, UserMemory
-                ):  # Check if it is already an instance
-                    self._user_memory = user_memory_config
-                elif isinstance(
                 user_memory_config, dict
             ):  # Check if it's a configuration dict
-                    self._user_memory = UserMemory(
+                self._user_memory = UserMemory(crew=self)
-                        crew=self, **user_memory_config
-                    )  # Initialize with config
             else:
-                    raise TypeError(
+                raise TypeError("user_memory must be a configuration dictionary")
-                        "user_memory must be a UserMemory instance or a configuration dictionary"
+
+    def _initialize_default_memories(self):
+        self._long_term_memory = self._long_term_memory or LongTermMemory()
+        self._short_term_memory = self._short_term_memory or ShortTermMemory(
+            crew=self,
+            embedder_config=self.embedder,
         )
-            else:
+        self._entity_memory = self.entity_memory or EntityMemory(
-                self._user_memory = None  # No user memory if not in config
+            crew=self, embedder_config=self.embedder
+        )
+
+    @model_validator(mode="after")
+    def create_crew_memory(self) -> "Crew":
+        """Initialize private memory attributes."""
+        self._external_memory = (
+            # External memory doesn’t support a default value since it was designed to be managed entirely externally
+            self.external_memory.set_crew(self) if self.external_memory else None
+        )
+
+        self._long_term_memory = self.long_term_memory
+        self._short_term_memory = self.short_term_memory
+        self._entity_memory = self.entity_memory
+
+        # UserMemory is gonna to be deprecated in the future, but we have to initialize a default value for now
+        self._user_memory = None
+
+        if self.memory:
+            self._initialize_default_memories()
+            self._initialize_user_memory()
+
         return self
 
     @model_validator(mode="after")
@@ -1125,9 +1134,13 @@ class Crew(BaseModel):
         result = self._execute_tasks(self.tasks, start_index, True)
         return result
 
-    def query_knowledge(self, query: List[str]) -> Union[List[Dict[str, Any]], None]:
+    def query_knowledge(
+        self, query: List[str], results_limit: int = 3, score_threshold: float = 0.35
+    ) -> Union[List[Dict[str, Any]], None]:
         if self.knowledge:
-            return self.knowledge.query(query)
+            return self.knowledge.query(
+                query, results_limit=results_limit, score_threshold=score_threshold
+            )
         return None
 
     def fetch_inputs(self) -> Set[str]:
@@ -1156,7 +1169,12 @@ class Crew(BaseModel):
         return required_inputs
 
     def copy(self):
-        """Create a deep copy of the Crew."""
+        """
+        Creates a deep copy of the Crew instance.
+
+        Returns:
+            Crew: A new instance with copied components
+        """
 
         exclude = {
             "id",
@@ -1168,13 +1186,19 @@ class Crew(BaseModel):
             "_short_term_memory",
             "_long_term_memory",
             "_entity_memory",
+            "_external_memory",
+            "_telemetry",
             "agents",
             "tasks",
             "knowledge_sources",
             "knowledge",
+            "manager_agent",
+            "manager_llm",
         }
 
         cloned_agents = [agent.copy() for agent in self.agents]
+        manager_agent = self.manager_agent.copy() if self.manager_agent else None
+        manager_llm = shallow_copy(self.manager_llm) if self.manager_llm else None
 
         task_mapping = {}
 
@@ -1197,6 +1221,20 @@ class Crew(BaseModel):
 
         copied_data = self.model_dump(exclude=exclude)
         copied_data = {k: v for k, v in copied_data.items() if v is not None}
+        if self.short_term_memory:
+            copied_data["short_term_memory"] = self.short_term_memory.model_copy(
+                deep=True
+            )
+        if self.long_term_memory:
+            copied_data["long_term_memory"] = self.long_term_memory.model_copy(
+                deep=True
+            )
+        if self.entity_memory:
+            copied_data["entity_memory"] = self.entity_memory.model_copy(deep=True)
+        if self.external_memory:
+            copied_data["external_memory"] = self.external_memory.model_copy(deep=True)
+        if self.user_memory:
+            copied_data["user_memory"] = self.user_memory.model_copy(deep=True)
 
         copied_data.pop("agents", None)
         copied_data.pop("tasks", None)
@@ -1207,6 +1245,8 @@ class Crew(BaseModel):
             tasks=cloned_tasks,
             knowledge_sources=existing_knowledge_sources,
             knowledge=existing_knowledge,
+            manager_agent=manager_agent,
+            manager_llm=manager_llm,
         )
 
         return copied_crew
@@ -1307,7 +1347,15 @@ class Crew(BaseModel):
             RuntimeError: If memory reset operation fails.
         """
         VALID_TYPES = frozenset(
-            ["long", "short", "entity", "knowledge", "kickoff_outputs", "all"]
+            [
+                "long",
+                "short",
+                "entity",
+                "knowledge",
+                "kickoff_outputs",
+                "all",
+                "external",
+            ]
         )
 
         if command_type not in VALID_TYPES:
@@ -1333,6 +1381,7 @@ class Crew(BaseModel):
         memory_systems = [
             ("short term", getattr(self, "_short_term_memory", None)),
             ("entity", getattr(self, "_entity_memory", None)),
+            ("external", getattr(self, "_external_memory", None)),
             ("long term", getattr(self, "_long_term_memory", None)),
             ("task output", getattr(self, "_task_output_handler", None)),
             ("knowledge", getattr(self, "knowledge", None)),
@@ -1355,11 +1404,15 @@ class Crew(BaseModel):
             RuntimeError: If the specified memory system fails to reset
         """
         reset_functions = {
-            "long": (self._long_term_memory, "long term"),
+            "long": (getattr(self, "_long_term_memory", None), "long term"),
-            "short": (self._short_term_memory, "short term"),
+            "short": (getattr(self, "_short_term_memory", None), "short term"),
-            "entity": (self._entity_memory, "entity"),
+            "entity": (getattr(self, "_entity_memory", None), "entity"),
-            "knowledge": (self.knowledge, "knowledge"),
+            "knowledge": (getattr(self, "knowledge", None), "knowledge"),
-            "kickoff_outputs": (self._task_output_handler, "task output"),
+            "kickoff_outputs": (
+                getattr(self, "_task_output_handler", None),
+                "task output",
+            ),
+            "external": (getattr(self, "_external_memory", None), "external"),
         }
 
         memory_system, name = reset_functions[memory_type]

src/crewai/flow/flow.py

@@ -1043,6 +1043,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
             import traceback
 
             traceback.print_exc()
+            raise
 
     def _log_flow_event(
         self, message: str, color: str = "yellow", level: str = "info"

src/crewai/flow/persistence/sqlite.py

@@ -21,7 +21,7 @@ class SQLiteFlowPersistence(FlowPersistence):
     moderate performance requirements.
     """
 
-    db_path: str  # Type annotation for instance variable
+    db_path: str
 
     def __init__(self, db_path: Optional[str] = None):
         """Initialize SQLite persistence.

src/crewai/knowledge/knowledge.py

@@ -43,7 +43,9 @@ class Knowledge(BaseModel):
         self.storage.initialize_knowledge_storage()
         self._add_sources()
 
-    def query(self, query: List[str], limit: int = 3) -> List[Dict[str, Any]]:
+    def query(
+        self, query: List[str], results_limit: int = 3, score_threshold: float = 0.35
+    ) -> List[Dict[str, Any]]:
         """
         Query across all knowledge sources to find the most relevant information.
         Returns the top_k most relevant chunks.
@@ -56,7 +58,8 @@ class Knowledge(BaseModel):
 
         results = self.storage.search(
             query,
-            limit,
+            limit=results_limit,
+            score_threshold=score_threshold,
         )
         return results
 

src/crewai/knowledge/knowledge_config.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel, Field
+
+
+class KnowledgeConfig(BaseModel):
+    """Configuration for knowledge retrieval.
+
+    Args:
+        results_limit (int): The number of relevant documents to return.
+        score_threshold (float): The minimum score for a document to be considered relevant.
+    """
+
+    results_limit: int = Field(default=3, description="The number of results to return")
+    score_threshold: float = Field(
+        default=0.35,
+        description="The minimum score for a result to be considered relevant",
+    )

src/crewai/knowledge/storage/knowledge_storage.py

@@ -4,7 +4,7 @@ import io
 import logging
 import os
 import shutil
-from typing import Any, Dict, List, Optional, Union, cast
+from typing import Any, Dict, List, Optional, Union
 
 import chromadb
 import chromadb.errors

src/crewai/lite_agent.py

@@ -0,0 +1,478 @@
+import asyncio
+import uuid
+from datetime import datetime
+from typing import Any, Callable, Dict, List, Optional, Type, Union, cast
+
+from pydantic import BaseModel, Field, InstanceOf, PrivateAttr, model_validator
+
+from crewai.agents.agent_builder.base_agent import BaseAgent
+from crewai.agents.agent_builder.utilities.base_token_process import TokenProcess
+from crewai.agents.cache import CacheHandler
+from crewai.agents.parser import (
+    AgentAction,
+    AgentFinish,
+    OutputParserException,
+)
+from crewai.llm import LLM
+from crewai.tools.base_tool import BaseTool
+from crewai.tools.structured_tool import CrewStructuredTool
+from crewai.utilities import I18N
+from crewai.utilities.agent_utils import (
+    enforce_rpm_limit,
+    format_message_for_llm,
+    get_llm_response,
+    get_tool_names,
+    handle_agent_action_core,
+    handle_context_length,
+    handle_max_iterations_exceeded,
+    handle_output_parser_exception,
+    handle_unknown_error,
+    has_reached_max_iterations,
+    is_context_length_exceeded,
+    parse_tools,
+    process_llm_response,
+    render_text_description_and_args,
+    show_agent_logs,
+)
+from crewai.utilities.converter import convert_to_model, generate_model_description
+from crewai.utilities.events.agent_events import (
+    LiteAgentExecutionCompletedEvent,
+    LiteAgentExecutionErrorEvent,
+    LiteAgentExecutionStartedEvent,
+)
+from crewai.utilities.events.crewai_event_bus import crewai_event_bus
+from crewai.utilities.events.llm_events import (
+    LLMCallCompletedEvent,
+    LLMCallFailedEvent,
+    LLMCallStartedEvent,
+    LLMCallType,
+)
+from crewai.utilities.llm_utils import create_llm
+from crewai.utilities.printer import Printer
+from crewai.utilities.token_counter_callback import TokenCalcHandler
+from crewai.utilities.tool_utils import execute_tool_and_check_finality
+
+
+class LiteAgentOutput(BaseModel):
+    """Class that represents the result of a LiteAgent execution."""
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    raw: str = Field(description="Raw output of the agent", default="")
+    pydantic: Optional[BaseModel] = Field(
+        description="Pydantic output of the agent", default=None
+    )
+    agent_role: str = Field(description="Role of the agent that produced this output")
+    usage_metrics: Optional[Dict[str, Any]] = Field(
+        description="Token usage metrics for this execution", default=None
+    )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert pydantic_output to a dictionary."""
+        if self.pydantic:
+            return self.pydantic.model_dump()
+        return {}
+
+    def __str__(self) -> str:
+        """String representation of the output."""
+        if self.pydantic:
+            return str(self.pydantic)
+        return self.raw
+
+
+class LiteAgent(BaseModel):
+    """
+    A lightweight agent that can process messages and use tools.
+
+    This agent is simpler than the full Agent class, focusing on direct execution
+    rather than task delegation. It's designed to be used for simple interactions
+    where a full crew is not needed.
+
+    Attributes:
+        role: The role of the agent.
+        goal: The objective of the agent.
+        backstory: The backstory of the agent.
+        llm: The language model that will run the agent.
+        tools: Tools at the agent's disposal.
+        verbose: Whether the agent execution should be in verbose mode.
+        max_iterations: Maximum number of iterations for tool usage.
+        max_execution_time: Maximum execution time in seconds.
+        response_format: Optional Pydantic model for structured output.
+    """
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    # Core Agent Properties
+    role: str = Field(description="Role of the agent")
+    goal: str = Field(description="Goal of the agent")
+    backstory: str = Field(description="Backstory of the agent")
+    llm: Optional[Union[str, InstanceOf[LLM], Any]] = Field(
+        default=None, description="Language model that will run the agent"
+    )
+    tools: List[BaseTool] = Field(
+        default_factory=list, description="Tools at agent's disposal"
+    )
+
+    # Execution Control Properties
+    max_iterations: int = Field(
+        default=15, description="Maximum number of iterations for tool usage"
+    )
+    max_execution_time: Optional[int] = Field(
+        default=None, description="Maximum execution time in seconds"
+    )
+    respect_context_window: bool = Field(
+        default=True,
+        description="Whether to respect the context window of the LLM",
+    )
+    use_stop_words: bool = Field(
+        default=True,
+        description="Whether to use stop words to prevent the LLM from using tools",
+    )
+    request_within_rpm_limit: Optional[Callable[[], bool]] = Field(
+        default=None,
+        description="Callback to check if the request is within the RPM limit",
+    )
+    i18n: I18N = Field(default=I18N(), description="Internationalization settings.")
+
+    # Output and Formatting Properties
+    response_format: Optional[Type[BaseModel]] = Field(
+        default=None, description="Pydantic model for structured output"
+    )
+    verbose: bool = Field(
+        default=False, description="Whether to print execution details"
+    )
+    callbacks: List[Callable] = Field(
+        default=[], description="Callbacks to be used for the agent"
+    )
+
+    # State and Results
+    tools_results: List[Dict[str, Any]] = Field(
+        default=[], description="Results of the tools used by the agent."
+    )
+
+    # Reference of Agent
+    original_agent: Optional[BaseAgent] = Field(
+        default=None, description="Reference to the agent that created this LiteAgent"
+    )
+    # Private Attributes
+    _parsed_tools: List[CrewStructuredTool] = PrivateAttr(default_factory=list)
+    _token_process: TokenProcess = PrivateAttr(default_factory=TokenProcess)
+    _cache_handler: CacheHandler = PrivateAttr(default_factory=CacheHandler)
+    _key: str = PrivateAttr(default_factory=lambda: str(uuid.uuid4()))
+    _messages: List[Dict[str, str]] = PrivateAttr(default_factory=list)
+    _iterations: int = PrivateAttr(default=0)
+    _printer: Printer = PrivateAttr(default_factory=Printer)
+    
+    @model_validator(mode="after")
+    def setup_llm(self):
+        """Set up the LLM and other components after initialization."""
+        self.llm = create_llm(self.llm)
+        if not isinstance(self.llm, LLM):
+            raise ValueError("Unable to create LLM instance")
+
+        # Initialize callbacks
+        token_callback = TokenCalcHandler(token_cost_process=self._token_process)
+        self._callbacks = [token_callback]
+
+        return self
+
+    @model_validator(mode="after")
+    def parse_tools(self):
+        """Parse the tools and convert them to CrewStructuredTool instances."""
+        self._parsed_tools = parse_tools(self.tools)
+
+        return self
+
+    @property
+    def key(self) -> str:
+        """Get the unique key for this agent instance."""
+        return self._key
+
+    @property
+    def _original_role(self) -> str:
+        """Return the original role for compatibility with tool interfaces."""
+        return self.role
+
+    def kickoff(self, messages: Union[str, List[Dict[str, str]]]) -> LiteAgentOutput:
+        """
+        Execute the agent with the given messages.
+
+        Args:
+            messages: Either a string query or a list of message dictionaries.
+                     If a string is provided, it will be converted to a user message.
+                     If a list is provided, each dict should have 'role' and 'content' keys.
+
+        Returns:
+            LiteAgentOutput: The result of the agent execution.
+        """
+        # Create agent info for event emission
+        agent_info = {
+            "role": self.role,
+            "goal": self.goal,
+            "backstory": self.backstory,
+            "tools": self._parsed_tools,
+            "verbose": self.verbose,
+        }
+
+        try:
+            # Reset state for this run
+            self._iterations = 0
+            self.tools_results = []
+
+            # Format messages for the LLM
+            self._messages = self._format_messages(messages)
+
+            # Emit event for agent execution start
+            crewai_event_bus.emit(
+                self,
+                event=LiteAgentExecutionStartedEvent(
+                    agent_info=agent_info,
+                    tools=self._parsed_tools,
+                    messages=messages,
+                ),
+            )
+
+            # Execute the agent using invoke loop
+            agent_finish = self._invoke_loop()
+            formatted_result: Optional[BaseModel] = None
+            if self.response_format:
+                try:
+                    # Cast to BaseModel to ensure type safety
+                    result = self.response_format.model_validate_json(
+                        agent_finish.output
+                    )
+                    if isinstance(result, BaseModel):
+                        formatted_result = result
+                except Exception as e:
+                    self._printer.print(
+                        content=f"Failed to parse output into response format: {str(e)}",
+                        color="yellow",
+                    )
+
+            # Calculate token usage metrics
+            usage_metrics = self._token_process.get_summary()
+
+            # Create output
+            output = LiteAgentOutput(
+                raw=agent_finish.output,
+                pydantic=formatted_result,
+                agent_role=self.role,
+                usage_metrics=usage_metrics.model_dump() if usage_metrics else None,
+            )
+
+            # Emit completion event
+            crewai_event_bus.emit(
+                self,
+                event=LiteAgentExecutionCompletedEvent(
+                    agent_info=agent_info,
+                    output=agent_finish.output,
+                ),
+            )
+
+            return output
+
+        except Exception as e:
+            self._printer.print(
+                content="Agent failed to reach a final answer. This is likely a bug - please report it.",
+                color="red",
+            )
+            handle_unknown_error(self._printer, e)
+            # Emit error event
+            crewai_event_bus.emit(
+                self,
+                event=LiteAgentExecutionErrorEvent(
+                    agent_info=agent_info,
+                    error=str(e),
+                ),
+            )
+            raise e
+
+    async def kickoff_async(
+        self, messages: Union[str, List[Dict[str, str]]]
+    ) -> LiteAgentOutput:
+        """
+        Execute the agent asynchronously with the given messages.
+
+        Args:
+            messages: Either a string query or a list of message dictionaries.
+                     If a string is provided, it will be converted to a user message.
+                     If a list is provided, each dict should have 'role' and 'content' keys.
+
+        Returns:
+            LiteAgentOutput: The result of the agent execution.
+        """
+        return await asyncio.to_thread(self.kickoff, messages)
+
+    def _get_default_system_prompt(self) -> str:
+        """Get the default system prompt for the agent."""
+        base_prompt = ""
+        if self._parsed_tools:
+            # Use the prompt template for agents with tools
+            base_prompt = self.i18n.slice("lite_agent_system_prompt_with_tools").format(
+                role=self.role,
+                backstory=self.backstory,
+                goal=self.goal,
+                tools=render_text_description_and_args(self._parsed_tools),
+                tool_names=get_tool_names(self._parsed_tools),
+            )
+        else:
+            # Use the prompt template for agents without tools
+            base_prompt = self.i18n.slice(
+                "lite_agent_system_prompt_without_tools"
+            ).format(
+                role=self.role,
+                backstory=self.backstory,
+                goal=self.goal,
+            )
+
+        # Add response format instructions if specified
+        if self.response_format:
+            schema = generate_model_description(self.response_format)
+            base_prompt += self.i18n.slice("lite_agent_response_format").format(
+                response_format=schema
+            )
+
+        return base_prompt
+
+    def _format_messages(
+        self, messages: Union[str, List[Dict[str, str]]]
+    ) -> List[Dict[str, str]]:
+        """Format messages for the LLM."""
+        if isinstance(messages, str):
+            messages = [{"role": "user", "content": messages}]
+
+        system_prompt = self._get_default_system_prompt()
+
+        # Add system message at the beginning
+        formatted_messages = [{"role": "system", "content": system_prompt}]
+
+        # Add the rest of the messages
+        formatted_messages.extend(messages)
+
+        return formatted_messages
+
+    def _invoke_loop(self) -> AgentFinish:
+        """
+        Run the agent's thought process until it reaches a conclusion or max iterations.
+
+        Returns:
+            AgentFinish: The final result of the agent execution.
+        """
+        # Execute the agent loop
+        formatted_answer = None
+        while not isinstance(formatted_answer, AgentFinish):
+            try:
+                if has_reached_max_iterations(self._iterations, self.max_iterations):
+                    formatted_answer = handle_max_iterations_exceeded(
+                        formatted_answer,
+                        printer=self._printer,
+                        i18n=self.i18n,
+                        messages=self._messages,
+                        llm=cast(LLM, self.llm),
+                        callbacks=self._callbacks,
+                    )
+
+                enforce_rpm_limit(self.request_within_rpm_limit)
+
+                # Emit LLM call started event
+                crewai_event_bus.emit(
+                    self,
+                    event=LLMCallStartedEvent(
+                        messages=self._messages,
+                        tools=None,
+                        callbacks=self._callbacks,
+                    ),
+                )
+
+                try:
+                    answer = get_llm_response(
+                        llm=cast(LLM, self.llm),
+                        messages=self._messages,
+                        callbacks=self._callbacks,
+                        printer=self._printer,
+                    )
+
+                    # Emit LLM call completed event
+                    crewai_event_bus.emit(
+                        self,
+                        event=LLMCallCompletedEvent(
+                            response=answer,
+                            call_type=LLMCallType.LLM_CALL,
+                        ),
+                    )
+                except Exception as e:
+                    # Emit LLM call failed event
+                    crewai_event_bus.emit(
+                        self,
+                        event=LLMCallFailedEvent(error=str(e)),
+                    )
+                    raise e
+
+                formatted_answer = process_llm_response(answer, self.use_stop_words)
+
+                if isinstance(formatted_answer, AgentAction):
+                    try:
+                        tool_result = execute_tool_and_check_finality(
+                            agent_action=formatted_answer,
+                            tools=self._parsed_tools,
+                            i18n=self.i18n,
+                            agent_key=self.key,
+                            agent_role=self.role,
+                            agent=self.original_agent,
+                        )
+                    except Exception as e:
+                        raise e
+
+                    formatted_answer = handle_agent_action_core(
+                        formatted_answer=formatted_answer,
+                        tool_result=tool_result,
+                        show_logs=self._show_logs,
+                    )
+
+                self._append_message(formatted_answer.text, role="assistant")
+            except OutputParserException as e:
+                formatted_answer = handle_output_parser_exception(
+                    e=e,
+                    messages=self._messages,
+                    iterations=self._iterations,
+                    log_error_after=3,
+                    printer=self._printer,
+                )
+
+            except Exception as e:
+                if e.__class__.__module__.startswith("litellm"):
+                    # Do not retry on litellm errors
+                    raise e
+                if is_context_length_exceeded(e):
+                    handle_context_length(
+                        respect_context_window=self.respect_context_window,
+                        printer=self._printer,
+                        messages=self._messages,
+                        llm=cast(LLM, self.llm),
+                        callbacks=self._callbacks,
+                        i18n=self.i18n,
+                    )
+                    continue
+                else:
+                    handle_unknown_error(self._printer, e)
+                    raise e
+
+            finally:
+                self._iterations += 1
+
+        assert isinstance(formatted_answer, AgentFinish)
+        self._show_logs(formatted_answer)
+        return formatted_answer
+
+    def _show_logs(self, formatted_answer: Union[AgentAction, AgentFinish]):
+        """Show logs for the agent's execution."""
+        show_agent_logs(
+            printer=self._printer,
+            agent_role=self.role,
+            formatted_answer=formatted_answer,
+            verbose=self.verbose,
+        )
+
+    def _append_message(self, text: str, role: str = "assistant") -> None:
+        """Append a message to the message list with the given role."""
+        self._messages.append(format_message_for_llm(text, role=role))

src/crewai/llm.py

@@ -4,9 +4,12 @@ import os
 import sys
 import threading
 import warnings
+from collections import defaultdict
 from contextlib import contextmanager
+from types import SimpleNamespace
 from typing import (
     Any,
+    DefaultDict,
     Dict,
     List,
     Literal,
@@ -18,7 +21,8 @@ from typing import (
 )
 
 from dotenv import load_dotenv
-from pydantic import BaseModel
+from litellm.types.utils import ChatCompletionDeltaToolCall
+from pydantic import BaseModel, Field
 
 from crewai.utilities.events.llm_events import (
     LLMCallCompletedEvent,
@@ -219,6 +223,15 @@ class StreamingChoices(TypedDict):
     finish_reason: Optional[str]
 
 
+class FunctionArgs(BaseModel):
+    name: str = ""
+    arguments: str = ""
+
+
+class AccumulatedToolArgs(BaseModel):
+    function: FunctionArgs = Field(default_factory=FunctionArgs)
+
+
 class LLM(BaseLLM):
     def __init__(
         self,
@@ -371,6 +384,11 @@ class LLM(BaseLLM):
         last_chunk = None
         chunk_count = 0
         usage_info = None
+        tool_calls = None
+
+        accumulated_tool_args: DefaultDict[int, AccumulatedToolArgs] = defaultdict(
+            AccumulatedToolArgs
+        )
 
         # --- 2) Make sure stream is set to True and include usage metrics
         params["stream"] = True
@@ -428,6 +446,20 @@ class LLM(BaseLLM):
                             if chunk_content is None and isinstance(delta, dict):
                                 # Some models might send empty content chunks
                                 chunk_content = ""
+
+                            # Enable tool calls using streaming
+                            if "tool_calls" in delta:
+                                tool_calls = delta["tool_calls"]
+
+                                if tool_calls:
+                                    result = self._handle_streaming_tool_calls(
+                                        tool_calls=tool_calls,
+                                        accumulated_tool_args=accumulated_tool_args,
+                                        available_functions=available_functions,
+                                    )
+                                    if result is not None:
+                                        chunk_content = result
+
                 except Exception as e:
                     logging.debug(f"Error extracting content from chunk: {e}")
                     logging.debug(f"Chunk format: {type(chunk)}, content: {chunk}")
@@ -442,7 +474,6 @@ class LLM(BaseLLM):
                         self,
                         event=LLMStreamChunkEvent(chunk=chunk_content),
                     )
-
             # --- 4) Fallback to non-streaming if no content received
             if not full_response.strip() and chunk_count == 0:
                 logging.warning(
@@ -501,7 +532,7 @@ class LLM(BaseLLM):
                         )
 
             # --- 6) If still empty, raise an error instead of using a default response
-            if not full_response.strip():
+            if not full_response.strip() and len(accumulated_tool_args) == 0:
                 raise Exception(
                     "No content received from streaming response. Received empty chunks or failed to extract content."
                 )
@@ -533,8 +564,8 @@ class LLM(BaseLLM):
                                 tool_calls = getattr(message, "tool_calls")
             except Exception as e:
                 logging.debug(f"Error checking for tool calls: {e}")
-
             # --- 8) If no tool calls or no available functions, return the text response directly
+
             if not tool_calls or not available_functions:
                 # Log token usage if available in streaming mode
                 self._handle_streaming_callbacks(callbacks, usage_info, last_chunk)
@@ -568,6 +599,47 @@ class LLM(BaseLLM):
             )
             raise Exception(f"Failed to get streaming response: {str(e)}")
 
+    def _handle_streaming_tool_calls(
+        self,
+        tool_calls: List[ChatCompletionDeltaToolCall],
+        accumulated_tool_args: DefaultDict[int, AccumulatedToolArgs],
+        available_functions: Optional[Dict[str, Any]] = None,
+    ) -> None | str:
+        for tool_call in tool_calls:
+            current_tool_accumulator = accumulated_tool_args[tool_call.index]
+
+            if tool_call.function.name:
+                current_tool_accumulator.function.name = tool_call.function.name
+
+            if tool_call.function.arguments:
+                current_tool_accumulator.function.arguments += (
+                    tool_call.function.arguments
+                )
+
+            crewai_event_bus.emit(
+                self,
+                event=LLMStreamChunkEvent(
+                    tool_call=tool_call.to_dict(),
+                    chunk=tool_call.function.arguments,
+                ),
+            )
+
+            if (
+                current_tool_accumulator.function.name
+                and current_tool_accumulator.function.arguments
+                and available_functions
+            ):
+                try:
+                    json.loads(current_tool_accumulator.function.arguments)
+
+                    return self._handle_tool_call(
+                        [current_tool_accumulator],
+                        available_functions,
+                    )
+                except json.JSONDecodeError:
+                    continue
+        return None
+
     def _handle_streaming_callbacks(
         self,
         callbacks: Optional[List[Any]],
@@ -707,15 +779,6 @@ class LLM(BaseLLM):
                     function_name, lambda: None
                 )  # Ensure fn is always a callable
                 logging.error(f"Error executing function '{function_name}': {e}")
-                crewai_event_bus.emit(
-                    self,
-                    event=ToolExecutionErrorEvent(
-                        tool_name=function_name,
-                        tool_args=function_args,
-                        tool_class=fn,
-                        error=str(e),
-                    ),
-                )
                 crewai_event_bus.emit(
                     self,
                     event=LLMCallFailedEvent(error=f"Tool execution error: {str(e)}"),

src/crewai/memory/__init__.py

@@ -2,5 +2,12 @@ from .entity.entity_memory import EntityMemory
 from .long_term.long_term_memory import LongTermMemory
 from .short_term.short_term_memory import ShortTermMemory
 from .user.user_memory import UserMemory
+from .external.external_memory import ExternalMemory
 
-__all__ = ["UserMemory", "EntityMemory", "LongTermMemory", "ShortTermMemory"]
+__all__ = [
+    "UserMemory",
+    "EntityMemory",
+    "LongTermMemory",
+    "ShortTermMemory",
+    "ExternalMemory",
+]

src/crewai/memory/contextual/contextual_memory.py

@@ -1,6 +1,12 @@
 from typing import Any, Dict, Optional
 
-from crewai.memory import EntityMemory, LongTermMemory, ShortTermMemory, UserMemory
+from crewai.memory import (
+    EntityMemory,
+    ExternalMemory,
+    LongTermMemory,
+    ShortTermMemory,
+    UserMemory,
+)
 
 
 class ContextualMemory:
@@ -11,6 +17,7 @@ class ContextualMemory:
         ltm: LongTermMemory,
         em: EntityMemory,
         um: UserMemory,
+        exm: ExternalMemory,
     ):
         if memory_config is not None:
             self.memory_provider = memory_config.get("provider")
@@ -20,6 +27,7 @@ class ContextualMemory:
         self.ltm = ltm
         self.em = em
         self.um = um
+        self.exm = exm
 
     def build_context_for_task(self, task, context) -> str:
         """
@@ -35,6 +43,7 @@ class ContextualMemory:
         context.append(self._fetch_ltm_context(task.description))
         context.append(self._fetch_stm_context(query))
         context.append(self._fetch_entity_context(query))
+        context.append(self._fetch_external_context(query))
         if self.memory_provider == "mem0":
             context.append(self._fetch_user_context(query))
         return "\n".join(filter(None, context))
@@ -44,6 +53,10 @@ class ContextualMemory:
         Fetches recent relevant insights from STM related to the task's description and expected_output,
         formatted as bullet points.
         """
+
+        if self.stm is None:
+            return ""
+
         stm_results = self.stm.search(query)
         formatted_results = "\n".join(
             [
@@ -58,6 +71,10 @@ class ContextualMemory:
         Fetches historical data or insights from LTM that are relevant to the task's description and expected_output,
         formatted as bullet points.
         """
+
+        if self.ltm is None:
+            return ""
+
         ltm_results = self.ltm.search(task, latest_n=2)
         if not ltm_results:
             return None
@@ -77,6 +94,9 @@ class ContextualMemory:
         Fetches relevant entity information from Entity Memory related to the task's description and expected_output,
         formatted as bullet points.
         """
+        if self.em is None:
+            return ""
+
         em_results = self.em.search(query)
         formatted_results = "\n".join(
             [
@@ -94,6 +114,10 @@ class ContextualMemory:
         Returns:
             str: Formatted user memories as bullet points, or an empty string if none found.
         """
+
+        if self.um is None:
+            return ""
+
         user_memories = self.um.search(query)
         if not user_memories:
             return ""
@@ -102,3 +126,24 @@ class ContextualMemory:
             f"- {result['memory']}" for result in user_memories
         )
         return f"User memories/preferences:\n{formatted_memories}"
+
+    def _fetch_external_context(self, query: str) -> str:
+        """
+        Fetches and formats relevant information from External Memory.
+        Args:
+            query (str): The search query to find relevant information.
+        Returns:
+            str: Formatted information as bullet points, or an empty string if none found.
+        """
+        if self.exm is None:
+            return ""
+
+        external_memories = self.exm.search(query)
+
+        if not external_memories:
+            return ""
+
+        formatted_memories = "\n".join(
+            f"- {result['memory']}" for result in external_memories
+        )
+        return f"External memories:\n{formatted_memories}"

src/crewai/memory/external/external_memory.py

@@ -0,0 +1,61 @@
+from typing import TYPE_CHECKING, Any, Dict, Optional
+
+from crewai.memory.external.external_memory_item import ExternalMemoryItem
+from crewai.memory.memory import Memory
+from crewai.memory.storage.interface import Storage
+
+if TYPE_CHECKING:
+    from crewai.memory.storage.mem0_storage import Mem0Storage
+
+
+class ExternalMemory(Memory):
+    def __init__(self, storage: Optional[Storage] = None, **data: Any):
+        super().__init__(storage=storage, **data)
+
+    @staticmethod
+    def _configure_mem0(crew: Any, config: Dict[str, Any]) -> "Mem0Storage":
+        from crewai.memory.storage.mem0_storage import Mem0Storage
+
+        return Mem0Storage(type="external", crew=crew, config=config)
+
+    @staticmethod
+    def external_supported_storages() -> Dict[str, Any]:
+        return {
+            "mem0": ExternalMemory._configure_mem0,
+        }
+
+    @staticmethod
+    def create_storage(crew: Any, embedder_config: Optional[Dict[str, Any]]) -> Storage:
+        if not embedder_config:
+            raise ValueError("embedder_config is required")
+
+        if "provider" not in embedder_config:
+            raise ValueError("embedder_config must include a 'provider' key")
+
+        provider = embedder_config["provider"]
+        supported_storages = ExternalMemory.external_supported_storages()
+        if provider not in supported_storages:
+            raise ValueError(f"Provider {provider} not supported")
+
+        return supported_storages[provider](crew, embedder_config.get("config", {}))
+
+    def save(
+        self,
+        value: Any,
+        metadata: Optional[Dict[str, Any]] = None,
+        agent: Optional[str] = None,
+    ) -> None:
+        """Saves a value into the external storage."""
+        item = ExternalMemoryItem(value=value, metadata=metadata, agent=agent)
+        super().save(value=item.value, metadata=item.metadata, agent=item.agent)
+
+    def reset(self) -> None:
+        self.storage.reset()
+
+    def set_crew(self, crew: Any) -> "ExternalMemory":
+        super().set_crew(crew)
+
+        if not self.storage:
+            self.storage = self.create_storage(crew, self.embedder_config)
+
+        return self

src/crewai/memory/external/external_memory_item.py

@@ -0,0 +1,13 @@
+from typing import Any, Dict, Optional
+
+
+class ExternalMemoryItem:
+    def __init__(
+        self,
+        value: Any,
+        metadata: Optional[Dict[str, Any]] = None,
+        agent: Optional[str] = None,
+    ):
+        self.value = value
+        self.metadata = metadata
+        self.agent = agent

src/crewai/memory/memory.py

@@ -9,6 +9,7 @@ class Memory(BaseModel):
     """
 
     embedder_config: Optional[Dict[str, Any]] = None
+    crew: Optional[Any] = None
 
     storage: Any
 
@@ -36,3 +37,7 @@ class Memory(BaseModel):
         return self.storage.search(
             query=query, limit=limit, score_threshold=score_threshold
         )
+
+    def set_crew(self, crew: Any) -> "Memory":
+        self.crew = crew
+        return self

src/crewai/memory/storage/mem0_storage.py

@@ -11,15 +11,20 @@ class Mem0Storage(Storage):
     Extends Storage to handle embedding and searching across entities using Mem0.
     """
 
-    def __init__(self, type, crew=None):
+    def __init__(self, type, crew=None, config=None):
         super().__init__()
-
+        supported_types = ["user", "short_term", "long_term", "entities", "external"]
-        if type not in ["user", "short_term", "long_term", "entities"]:
+        if type not in supported_types:
-            raise ValueError("Invalid type for Mem0Storage. Must be 'user' or 'agent'.")
+            raise ValueError(
+                f"Invalid type '{type}' for Mem0Storage. Must be one of: "
+                + ", ".join(supported_types)
+            )
 
         self.memory_type = type
         self.crew = crew
-        self.memory_config = crew.memory_config
+        self.config = config or {}
+        # TODO: Memory config will be removed in the future the config will be passed as a parameter
+        self.memory_config = self.config or getattr(crew, "memory_config", {}) or {}
 
         # User ID is required for user memory type "user" since it's used as a unique identifier for the user.
         user_id = self._get_user_id()
@@ -27,10 +32,11 @@ class Mem0Storage(Storage):
             raise ValueError("User ID is required for user memory type")
 
         # API key in memory config overrides the environment variable
-        config = self.memory_config.get("config", {})
+        config = self._get_config()
         mem0_api_key = config.get("api_key") or os.getenv("MEM0_API_KEY")
         mem0_org_id = config.get("org_id")
         mem0_project_id = config.get("project_id")
+        mem0_local_config = config.get("local_mem0_config")
 
         # Initialize MemoryClient or Memory based on the presence of the mem0_api_key
         if mem0_api_key:
@@ -41,7 +47,10 @@ class Mem0Storage(Storage):
             else:
                 self.memory = MemoryClient(api_key=mem0_api_key)
         else:
-            self.memory = Memory()  # Fallback to Memory if no Mem0 API key is provided
+            if mem0_local_config and len(mem0_local_config):
+                self.memory = Memory.from_config(mem0_local_config)
+            else:
+                self.memory = Memory()
 
     def _sanitize_role(self, role: str) -> str:
         """
@@ -52,26 +61,34 @@ class Mem0Storage(Storage):
     def save(self, value: Any, metadata: Dict[str, Any]) -> None:
         user_id = self._get_user_id()
         agent_name = self._get_agent_name()
-        if self.memory_type == "user":
+        params = None
-            self.memory.add(value, user_id=user_id, metadata={**metadata})
+        if self.memory_type == "short_term":
-        elif self.memory_type == "short_term":
+            params = {
-            agent_name = self._get_agent_name()
+                "agent_id": agent_name,
-            self.memory.add(
+                "infer": False,
-                value, agent_id=agent_name, metadata={"type": "short_term", **metadata}
+                "metadata": {"type": "short_term", **metadata},
-            )
+            }
         elif self.memory_type == "long_term":
-            agent_name = self._get_agent_name()
+            params = {
-            self.memory.add(
+                "agent_id": agent_name,
-                value,
+                "infer": False,
-                agent_id=agent_name,
+                "metadata": {"type": "long_term", **metadata},
-                infer=False,
+            }
-                metadata={"type": "long_term", **metadata},
-            )
         elif self.memory_type == "entities":
-            entity_name = self._get_agent_name()
+            params = {
-            self.memory.add(
+                "agent_id": agent_name,
-                value, user_id=entity_name, metadata={"type": "entity", **metadata}
+                "infer": False,
-            )
+                "metadata": {"type": "entity", **metadata},
+            }
+        elif self.memory_type == "external":
+            params = {
+                "user_id": user_id,
+                "agent_id": agent_name,
+                "metadata": {"type": "external", **metadata},
+            }
+
+        if params:
+            self.memory.add(value, **params | {"output_format": "v1.1"})
 
     def search(
         self,
@@ -80,37 +97,43 @@ class Mem0Storage(Storage):
         score_threshold: float = 0.35,
     ) -> List[Any]:
         params = {"query": query, "limit": limit}
-        if self.memory_type == "user":
+        if user_id := self._get_user_id():
-            user_id = self._get_user_id()
             params["user_id"] = user_id
-        elif self.memory_type == "short_term":
+
         agent_name = self._get_agent_name()
+        if self.memory_type == "short_term":
             params["agent_id"] = agent_name
             params["metadata"] = {"type": "short_term"}
         elif self.memory_type == "long_term":
-            agent_name = self._get_agent_name()
             params["agent_id"] = agent_name
             params["metadata"] = {"type": "long_term"}
         elif self.memory_type == "entities":
-            agent_name = self._get_agent_name()
             params["agent_id"] = agent_name
             params["metadata"] = {"type": "entity"}
+        elif self.memory_type == "external":
+            params["agent_id"] = agent_name
+            params["metadata"] = {"type": "external"}
 
         # Discard the filters for now since we create the filters
         # automatically when the crew is created.
         results = self.memory.search(**params)
         return [r for r in results if r["score"] >= score_threshold]
 
-    def _get_user_id(self):
+    def _get_user_id(self) -> str:
-        if self.memory_type == "user":
+        return self._get_config().get("user_id", "")
-            if hasattr(self, "memory_config") and self.memory_config is not None:
-                return self.memory_config.get("config", {}).get("user_id")
-            else:
-                return None
-        return None
 
-    def _get_agent_name(self):
+    def _get_agent_name(self) -> str:
-        agents = self.crew.agents if self.crew else []
+        if not self.crew:
+            return ""
+
+        agents = self.crew.agents
         agents = [self._sanitize_role(agent.role) for agent in agents]
         agents = "_".join(agents)
         return agents
+
+    def _get_config(self) -> Dict[str, Any]:
+        return self.config or getattr(self, "memory_config", {}).get("config", {}) or {}
+
+    def reset(self):
+        if self.memory:
+            self.memory.reset()

src/crewai/memory/user/user_memory.py

@@ -1,3 +1,4 @@
+import warnings
 from typing import Any, Dict, Optional
 
 from crewai.memory.memory import Memory
@@ -12,6 +13,12 @@ class UserMemory(Memory):
     """
 
     def __init__(self, crew=None):
+        warnings.warn(
+            "UserMemory is deprecated and will be removed in a future version. "
+            "Please use ExternalMemory instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         try:
             from crewai.memory.storage.mem0_storage import Mem0Storage
         except ImportError:
@@ -43,3 +50,9 @@ class UserMemory(Memory):
             score_threshold=score_threshold,
         )
         return results
+
+    def reset(self) -> None:
+        try:
+            self.storage.reset()
+        except Exception as e:
+            raise Exception(f"An error occurred while resetting the user memory: {e}")

src/crewai/project/crew_base.py

@@ -137,13 +137,11 @@ def CrewBase(cls: T) -> T:
                 all_functions, "is_cache_handler"
             )
             callbacks = self._filter_functions(all_functions, "is_callback")
-            agents = self._filter_functions(all_functions, "is_agent")
 
             for agent_name, agent_info in self.agents_config.items():
                 self._map_agent_variables(
                     agent_name,
                     agent_info,
-                    agents,
                     llms,
                     tool_functions,
                     cache_handler_functions,
@@ -154,7 +152,6 @@ def CrewBase(cls: T) -> T:
             self,
             agent_name: str,
             agent_info: Dict[str, Any],
-            agents: Dict[str, Callable],
             llms: Dict[str, Callable],
             tool_functions: Dict[str, Callable],
             cache_handler_functions: Dict[str, Callable],
@@ -172,9 +169,10 @@ def CrewBase(cls: T) -> T:
                 ]
 
             if function_calling_llm := agent_info.get("function_calling_llm"):
-                self.agents_config[agent_name]["function_calling_llm"] = agents[
+                try:
-                    function_calling_llm
+                    self.agents_config[agent_name]["function_calling_llm"] = llms[function_calling_llm]()
-                ]()
+                except KeyError:
+                    self.agents_config[agent_name]["function_calling_llm"] = function_calling_llm
 
             if step_callback := agent_info.get("step_callback"):
                 self.agents_config[agent_name]["step_callback"] = callbacks[

src/crewai/task.py

@@ -388,7 +388,7 @@ class Task(BaseModel):
             tools = tools or self.tools or []
 
             self.processed_by_agents.add(agent.role)
-            crewai_event_bus.emit(self, TaskStartedEvent(context=context))
+            crewai_event_bus.emit(self, TaskStartedEvent(context=context, task=self))
             result = agent.execute_task(
                 task=self,
                 context=context,
@@ -464,11 +464,11 @@ class Task(BaseModel):
                     )
                 )
                 self._save_file(content)
-            crewai_event_bus.emit(self, TaskCompletedEvent(output=task_output))
+            crewai_event_bus.emit(self, TaskCompletedEvent(output=task_output, task=self))
             return task_output
         except Exception as e:
             self.end_time = datetime.datetime.now()
-            crewai_event_bus.emit(self, TaskFailedEvent(error=str(e)))
+            crewai_event_bus.emit(self, TaskFailedEvent(error=str(e), task=self))
             raise e  # Re-raise the exception after emitting the event
 
     def prompt(self) -> str:

src/crewai/telemetry/constants.py

@@ -0,0 +1,2 @@
+CREWAI_TELEMETRY_BASE_URL: str = "https://telemetry.crewai.com:4319"
+CREWAI_TELEMETRY_SERVICE_NAME: str = "crewAI-telemetry"

src/crewai/telemetry/telemetry.py

@@ -9,6 +9,11 @@ from contextlib import contextmanager
 from importlib.metadata import version
 from typing import TYPE_CHECKING, Any, Optional
 
+from crewai.telemetry.constants import (
+    CREWAI_TELEMETRY_BASE_URL,
+    CREWAI_TELEMETRY_SERVICE_NAME,
+)
+
 
 @contextmanager
 def suppress_warnings():
@@ -45,23 +50,22 @@ class Telemetry:
     """
 
     def __init__(self):
-        self.ready = False
+        self.ready: bool = False
-        self.trace_set = False
+        self.trace_set: bool = False
 
-        if os.getenv("OTEL_SDK_DISABLED", "false").lower() == "true":
+        if self._is_telemetry_disabled():
             return
 
         try:
-            telemetry_endpoint = "https://telemetry.crewai.com:4319"
             self.resource = Resource(
-                attributes={SERVICE_NAME: "crewAI-telemetry"},
+                attributes={SERVICE_NAME: CREWAI_TELEMETRY_SERVICE_NAME},
             )
             with suppress_warnings():
                 self.provider = TracerProvider(resource=self.resource)
 
             processor = BatchSpanProcessor(
                 OTLPSpanExporter(
-                    endpoint=f"{telemetry_endpoint}/v1/traces",
+                    endpoint=f"{CREWAI_TELEMETRY_BASE_URL}/v1/traces",
                     timeout=30,
                 )
             )
@@ -76,6 +80,13 @@ class Telemetry:
                 raise  # Re-raise the exception to not interfere with system signals
             self.ready = False
 
+    def _is_telemetry_disabled(self) -> bool:
+        """Check if telemetry should be disabled based on environment variables."""
+        return (
+            os.getenv("OTEL_SDK_DISABLED", "false").lower() == "true"
+            or os.getenv("CREWAI_DISABLE_TELEMETRY", "false").lower() == "true"
+        )
+
     def set_tracer(self):
         if self.ready and not self.trace_set:
             try:
@@ -112,6 +123,23 @@ class Telemetry:
             self._add_attribute(span, "crew_memory", crew.memory)
             self._add_attribute(span, "crew_number_of_tasks", len(crew.tasks))
             self._add_attribute(span, "crew_number_of_agents", len(crew.agents))
+
+            # Add fingerprint data
+            if hasattr(crew, "fingerprint") and crew.fingerprint:
+                self._add_attribute(span, "crew_fingerprint", crew.fingerprint.uuid_str)
+                self._add_attribute(
+                    span,
+                    "crew_fingerprint_created_at",
+                    crew.fingerprint.created_at.isoformat(),
+                )
+                # Add fingerprint metadata if it exists
+                if hasattr(crew.fingerprint, "metadata") and crew.fingerprint.metadata:
+                    self._add_attribute(
+                        span,
+                        "crew_fingerprint_metadata",
+                        json.dumps(crew.fingerprint.metadata),
+                    )
+
             if crew.share_crew:
                 self._add_attribute(
                     span,
@@ -129,17 +157,43 @@ class Telemetry:
                                 "max_rpm": agent.max_rpm,
                                 "i18n": agent.i18n.prompt_file,
                                 "function_calling_llm": (
-                                    agent.function_calling_llm.model
+                                    getattr(
-                                    if agent.function_calling_llm
+                                        getattr(agent, "function_calling_llm", None),
+                                        "model",
+                                        "",
+                                    )
+                                    if getattr(agent, "function_calling_llm", None)
                                     else ""
                                 ),
                                 "llm": agent.llm.model,
                                 "delegation_enabled?": agent.allow_delegation,
-                                "allow_code_execution?": agent.allow_code_execution,
+                                "allow_code_execution?": getattr(
-                                "max_retry_limit": agent.max_retry_limit,
+                                    agent, "allow_code_execution", False
+                                ),
+                                "max_retry_limit": getattr(agent, "max_retry_limit", 3),
                                 "tools_names": [
                                     tool.name.casefold() for tool in agent.tools or []
                                 ],
+                                # Add agent fingerprint data if sharing crew details
+                                "fingerprint": (
+                                    getattr(
+                                        getattr(agent, "fingerprint", None),
+                                        "uuid_str",
+                                        None,
+                                    )
+                                ),
+                                "fingerprint_created_at": (
+                                    created_at.isoformat()
+                                    if (
+                                        created_at := getattr(
+                                            getattr(agent, "fingerprint", None),
+                                            "created_at",
+                                            None,
+                                        )
+                                    )
+                                    is not None
+                                    else None
+                                ),
                             }
                             for agent in crew.agents
                         ]
@@ -169,6 +223,17 @@ class Telemetry:
                                 "tools_names": [
                                     tool.name.casefold() for tool in task.tools or []
                                 ],
+                                # Add task fingerprint data if sharing crew details
+                                "fingerprint": (
+                                    task.fingerprint.uuid_str
+                                    if hasattr(task, "fingerprint") and task.fingerprint
+                                    else None
+                                ),
+                                "fingerprint_created_at": (
+                                    task.fingerprint.created_at.isoformat()
+                                    if hasattr(task, "fingerprint") and task.fingerprint
+                                    else None
+                                ),
                             }
                             for task in crew.tasks
                         ]
@@ -196,14 +261,20 @@ class Telemetry:
                                 "max_iter": agent.max_iter,
                                 "max_rpm": agent.max_rpm,
                                 "function_calling_llm": (
-                                    agent.function_calling_llm.model
+                                    getattr(
-                                    if agent.function_calling_llm
+                                        getattr(agent, "function_calling_llm", None),
+                                        "model",
+                                        "",
+                                    )
+                                    if getattr(agent, "function_calling_llm", None)
                                     else ""
                                 ),
                                 "llm": agent.llm.model,
                                 "delegation_enabled?": agent.allow_delegation,
-                                "allow_code_execution?": agent.allow_code_execution,
+                                "allow_code_execution?": getattr(
-                                "max_retry_limit": agent.max_retry_limit,
+                                    agent, "allow_code_execution", False
+                                ),
+                                "max_retry_limit": getattr(agent, "max_retry_limit", 3),
                                 "tools_names": [
                                     tool.name.casefold() for tool in agent.tools or []
                                 ],
@@ -252,6 +323,39 @@ class Telemetry:
             self._add_attribute(created_span, "task_key", task.key)
             self._add_attribute(created_span, "task_id", str(task.id))
 
+            # Add fingerprint data
+            if hasattr(crew, "fingerprint") and crew.fingerprint:
+                self._add_attribute(
+                    created_span, "crew_fingerprint", crew.fingerprint.uuid_str
+                )
+
+            if hasattr(task, "fingerprint") and task.fingerprint:
+                self._add_attribute(
+                    created_span, "task_fingerprint", task.fingerprint.uuid_str
+                )
+                self._add_attribute(
+                    created_span,
+                    "task_fingerprint_created_at",
+                    task.fingerprint.created_at.isoformat(),
+                )
+                # Add fingerprint metadata if it exists
+                if hasattr(task.fingerprint, "metadata") and task.fingerprint.metadata:
+                    self._add_attribute(
+                        created_span,
+                        "task_fingerprint_metadata",
+                        json.dumps(task.fingerprint.metadata),
+                    )
+
+            # Add agent fingerprint if task has an assigned agent
+            if hasattr(task, "agent") and task.agent:
+                agent_fingerprint = getattr(
+                    getattr(task.agent, "fingerprint", None), "uuid_str", None
+                )
+                if agent_fingerprint:
+                    self._add_attribute(
+                        created_span, "agent_fingerprint", agent_fingerprint
+                    )
+
             if crew.share_crew:
                 self._add_attribute(
                     created_span, "formatted_description", task.description
@@ -270,6 +374,21 @@ class Telemetry:
             self._add_attribute(span, "task_key", task.key)
             self._add_attribute(span, "task_id", str(task.id))
 
+            # Add fingerprint data to execution span
+            if hasattr(crew, "fingerprint") and crew.fingerprint:
+                self._add_attribute(span, "crew_fingerprint", crew.fingerprint.uuid_str)
+
+            if hasattr(task, "fingerprint") and task.fingerprint:
+                self._add_attribute(span, "task_fingerprint", task.fingerprint.uuid_str)
+
+            # Add agent fingerprint if task has an assigned agent
+            if hasattr(task, "agent") and task.agent:
+                agent_fingerprint = getattr(
+                    getattr(task.agent, "fingerprint", None), "uuid_str", None
+                )
+                if agent_fingerprint:
+                    self._add_attribute(span, "agent_fingerprint", agent_fingerprint)
+
             if crew.share_crew:
                 self._add_attribute(span, "formatted_description", task.description)
                 self._add_attribute(
@@ -291,7 +410,12 @@ class Telemetry:
         Note:
             If share_crew is enabled, this will also record the task output
         """
+
         def operation():
+            # Ensure fingerprint data is present on completion span
+            if hasattr(task, "fingerprint") and task.fingerprint:
+                self._add_attribute(span, "task_fingerprint", task.fingerprint.uuid_str)
+
             if crew.share_crew:
                 self._add_attribute(
                     span,
@@ -312,6 +436,7 @@ class Telemetry:
             tool_name (str): Name of the tool being repeatedly used
             attempts (int): Number of attempts made with this tool
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Tool Repeated Usage")
@@ -329,14 +454,16 @@ class Telemetry:
 
         self._safe_telemetry_operation(operation)
 
-    def tool_usage(self, llm: Any, tool_name: str, attempts: int):
+    def tool_usage(self, llm: Any, tool_name: str, attempts: int, agent: Any = None):
         """Records the usage of a tool by an agent.
 
         Args:
             llm (Any): The language model being used
             tool_name (str): Name of the tool being used
             attempts (int): Number of attempts made with this tool
+            agent (Any, optional): The agent using the tool
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Tool Usage")
@@ -349,17 +476,31 @@ class Telemetry:
             self._add_attribute(span, "attempts", attempts)
             if llm:
                 self._add_attribute(span, "llm", llm.model)
+
+            # Add agent fingerprint data if available
+            if agent and hasattr(agent, "fingerprint") and agent.fingerprint:
+                self._add_attribute(
+                    span, "agent_fingerprint", agent.fingerprint.uuid_str
+                )
+                if hasattr(agent, "role"):
+                    self._add_attribute(span, "agent_role", agent.role)
+
             span.set_status(Status(StatusCode.OK))
             span.end()
 
         self._safe_telemetry_operation(operation)
 
-    def tool_usage_error(self, llm: Any):
+    def tool_usage_error(
+        self, llm: Any, agent: Any = None, tool_name: Optional[str] = None
+    ):
         """Records when a tool usage results in an error.
 
         Args:
             llm (Any): The language model being used when the error occurred
+            agent (Any, optional): The agent using the tool
+            tool_name (str, optional): Name of the tool that caused the error
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Tool Usage Error")
@@ -370,6 +511,18 @@ class Telemetry:
             )
             if llm:
                 self._add_attribute(span, "llm", llm.model)
+
+            if tool_name:
+                self._add_attribute(span, "tool_name", tool_name)
+
+            # Add agent fingerprint data if available
+            if agent and hasattr(agent, "fingerprint") and agent.fingerprint:
+                self._add_attribute(
+                    span, "agent_fingerprint", agent.fingerprint.uuid_str
+                )
+                if hasattr(agent, "role"):
+                    self._add_attribute(span, "agent_role", agent.role)
+
             span.set_status(Status(StatusCode.OK))
             span.end()
 
@@ -386,6 +539,7 @@ class Telemetry:
             exec_time (int): Execution time in seconds
             model_name (str): Name of the model used
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Crew Individual Test Result")
@@ -420,6 +574,7 @@ class Telemetry:
             inputs (dict[str, Any] | None): Input parameters for the test
             model_name (str): Name of the model used in testing
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Crew Test Execution")
@@ -446,6 +601,7 @@ class Telemetry:
 
     def deploy_signup_error_span(self):
         """Records when an error occurs during the deployment signup process."""
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Deploy Signup Error")
@@ -460,6 +616,7 @@ class Telemetry:
         Args:
             uuid (Optional[str]): Unique identifier for the deployment
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Start Deployment")
@@ -472,6 +629,7 @@ class Telemetry:
 
     def create_crew_deployment_span(self):
         """Records the creation of a new crew deployment."""
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Create Crew Deployment")
@@ -487,6 +645,7 @@ class Telemetry:
             uuid (Optional[str]): Unique identifier for the crew
             log_type (str, optional): Type of logs being retrieved. Defaults to "deployment".
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Get Crew Logs")
@@ -504,6 +663,7 @@ class Telemetry:
         Args:
             uuid (Optional[str]): Unique identifier for the crew being removed
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Remove Crew")
@@ -634,6 +794,7 @@ class Telemetry:
         Args:
             flow_name (str): Name of the flow being created
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Flow Creation")
@@ -650,6 +811,7 @@ class Telemetry:
             flow_name (str): Name of the flow being plotted
             node_names (list[str]): List of node names in the flow
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Flow Plotting")
@@ -667,6 +829,7 @@ class Telemetry:
             flow_name (str): Name of the flow being executed
             node_names (list[str]): List of nodes being executed in the flow
         """
+
         def operation():
             tracer = trace.get_tracer("crewai.telemetry")
             span = tracer.start_span("Flow Execution")

src/crewai/tools/base_tool.py

@@ -1,3 +1,4 @@
+import asyncio
 import warnings
 from abc import ABC, abstractmethod
 from inspect import signature
@@ -7,29 +8,27 @@ from pydantic import (
     BaseModel,
     ConfigDict,
     Field,
-    PydanticDeprecatedSince20,
     create_model,
-    validator,
+    field_validator,
 )
 from pydantic import BaseModel as PydanticBaseModel
 
 from crewai.tools.structured_tool import CrewStructuredTool
 
-# Ignore all "PydanticDeprecatedSince20" warnings globally
-warnings.filterwarnings("ignore", category=PydanticDeprecatedSince20)
-
 
 class BaseTool(BaseModel, ABC):
     class _ArgsSchemaPlaceholder(PydanticBaseModel):
         pass
 
-    model_config = ConfigDict()
+    model_config = ConfigDict(arbitrary_types_allowed=True)
 
     name: str
     """The unique name of the tool that clearly communicates its purpose."""
     description: str
     """Used to tell the model how/when/why to use the tool."""
-    args_schema: Type[PydanticBaseModel] = Field(default_factory=_ArgsSchemaPlaceholder)
+    args_schema: Type[PydanticBaseModel] = Field(
+        default_factory=_ArgsSchemaPlaceholder, validate_default=True
+    )
     """The schema for the arguments that the tool accepts."""
     description_updated: bool = False
     """Flag to check if the description has been updated."""
@@ -38,7 +37,8 @@ class BaseTool(BaseModel, ABC):
     result_as_answer: bool = False
     """Flag to check if the tool should be the final agent answer."""
 
-    @validator("args_schema", always=True, pre=True)
+    @field_validator("args_schema", mode="before")
+    @classmethod
     def _default_args_schema(
         cls, v: Type[PydanticBaseModel]
     ) -> Type[PydanticBaseModel]:
@@ -66,7 +66,13 @@ class BaseTool(BaseModel, ABC):
         **kwargs: Any,
     ) -> Any:
         print(f"Using Tool: {self.name}")
-        return self._run(*args, **kwargs)
+        result = self._run(*args, **kwargs)
+
+        # If _run is async, we safely run it
+        if asyncio.iscoroutine(result):
+            return asyncio.run(result)
+
+        return result
 
     @abstractmethod
     def _run(
@@ -245,9 +251,13 @@ def to_langchain(
     return [t.to_structured_tool() if isinstance(t, BaseTool) else t for t in tools]
 
 
-def tool(*args):
+def tool(*args, result_as_answer=False):
     """
     Decorator to create a tool from a function.
+    
+    Args:
+        *args: Positional arguments, either the function to decorate or the tool name.
+        result_as_answer: Flag to indicate if the tool result should be used as the final agent answer.
     """
 
     def _make_with_name(tool_name: str) -> Callable:
@@ -273,6 +283,7 @@ def tool(*args):
                 description=f.__doc__,
                 func=f,
                 args_schema=args_schema,
+                result_as_answer=result_as_answer,
             )
 
         return _make_tool

src/crewai/tools/tool_types.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class ToolResult:
+    """Result of tool execution."""
+
+    result: str
+    result_as_answer: bool = False

src/crewai/tools/tool_usage.py

@@ -5,7 +5,7 @@ import time
 from difflib import SequenceMatcher
 from json import JSONDecodeError
 from textwrap import dedent
-from typing import Any, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 
 import json5
 from json_repair import repair_json
@@ -13,18 +13,26 @@ from json_repair import repair_json
 from crewai.agents.tools_handler import ToolsHandler
 from crewai.task import Task
 from crewai.telemetry import Telemetry
-from crewai.tools import BaseTool
 from crewai.tools.structured_tool import CrewStructuredTool
 from crewai.tools.tool_calling import InstructorToolCalling, ToolCalling
-from crewai.utilities import I18N, Converter, ConverterError, Printer
+from crewai.utilities import I18N, Converter, Printer
+from crewai.utilities.agent_utils import (
+    get_tool_names,
+    render_text_description_and_args,
+)
 from crewai.utilities.events.crewai_event_bus import crewai_event_bus
 from crewai.utilities.events.tool_usage_events import (
     ToolSelectionErrorEvent,
     ToolUsageErrorEvent,
     ToolUsageFinishedEvent,
+    ToolUsageStartedEvent,
     ToolValidateInputErrorEvent,
 )
 
+if TYPE_CHECKING:
+    from crewai.agents.agent_builder.base_agent import BaseAgent
+    from crewai.lite_agent import LiteAgent
+
 OPENAI_BIGGER_MODELS = [
     "gpt-4",
     "gpt-4o",
@@ -60,31 +68,29 @@ class ToolUsage:
 
     def __init__(
         self,
-        tools_handler: ToolsHandler,
+        tools_handler: Optional[ToolsHandler],
-        tools: List[BaseTool],
+        tools: List[CrewStructuredTool],
-        original_tools: List[Any],
+        task: Optional[Task],
-        tools_description: str,
-        tools_names: str,
-        task: Task,
         function_calling_llm: Any,
-        agent: Any,
+        agent: Optional[Union["BaseAgent", "LiteAgent"]] = None,
-        action: Any,
+        action: Any = None,
+        fingerprint_context: Optional[Dict[str, str]] = None,
     ) -> None:
-        self._i18n: I18N = agent.i18n
+        self._i18n: I18N = agent.i18n if agent else I18N()
         self._printer: Printer = Printer()
         self._telemetry: Telemetry = Telemetry()
         self._run_attempts: int = 1
         self._max_parsing_attempts: int = 3
         self._remember_format_after_usages: int = 3
         self.agent = agent
-        self.tools_description = tools_description
+        self.tools_description = render_text_description_and_args(tools)
-        self.tools_names = tools_names
+        self.tools_names = get_tool_names(tools)
         self.tools_handler = tools_handler
-        self.original_tools = original_tools
         self.tools = tools
         self.task = task
         self.action = action
         self.function_calling_llm = function_calling_llm
+        self.fingerprint_context = fingerprint_context or {}
 
         # Set the maximum parsing attempts for bigger models
         if (
@@ -103,8 +109,9 @@ class ToolUsage:
     ) -> str:
         if isinstance(calling, ToolUsageErrorException):
             error = calling.message
-            if self.agent.verbose:
+            if self.agent and self.agent.verbose:
                 self._printer.print(content=f"\n\n{error}\n", color="red")
+            if self.task:
                 self.task.increment_tools_errors()
             return error
 
@@ -112,8 +119,9 @@ class ToolUsage:
             tool = self._select_tool(calling.tool_name)
         except Exception as e:
             error = getattr(e, "message", str(e))
+            if self.task:
                 self.task.increment_tools_errors()
-            if self.agent.verbose:
+            if self.agent and self.agent.verbose:
                 self._printer.print(content=f"\n\n{error}\n", color="red")
             return error
 
@@ -127,8 +135,9 @@ class ToolUsage:
 
             except Exception as e:
                 error = getattr(e, "message", str(e))
+                if self.task:
                     self.task.increment_tools_errors()
-                if self.agent.verbose:
+                if self.agent and self.agent.verbose:
                     self._printer.print(content=f"\n\n{error}\n", color="red")
                 return error
 
@@ -137,9 +146,9 @@ class ToolUsage:
     def _use(
         self,
         tool_string: str,
-        tool: Any,
+        tool: CrewStructuredTool,
         calling: Union[ToolCalling, InstructorToolCalling],
-    ) -> str:  # TODO: Fix this return type
+    ) -> str:
         if self._check_tool_repeated_usage(calling=calling):  # type: ignore # _check_tool_repeated_usage of "ToolUsage" does not return a value (it only ever returns None)
             try:
                 result = self._i18n.errors("task_repeated_usage").format(
@@ -154,24 +163,44 @@ class ToolUsage:
                 return result  # type: ignore # Fix the return type of this function
 
             except Exception:
+                if self.task:
                     self.task.increment_tools_errors()
 
+        if self.agent:
+            event_data = {
+                "agent_key": self.agent.key,
+                "agent_role": self.agent.role,
+                "tool_name": self.action.tool,
+                "tool_args": self.action.tool_input,
+                "tool_class": self.action.tool,
+                "agent": self.agent,
+            }
+
+            if self.agent.fingerprint:
+                event_data.update(self.agent.fingerprint)
+
+            crewai_event_bus.emit(self,ToolUsageStartedEvent(**event_data))
+            
         started_at = time.time()
         from_cache = False
+        result = None  # type: ignore
 
-        result = None  # type: ignore # Incompatible types in assignment (expression has type "None", variable has type "str")
+        if self.tools_handler and self.tools_handler.cache:
-        # check if cache is available
+            result = self.tools_handler.cache.read(
-        if self.tools_handler.cache:
-            result = self.tools_handler.cache.read(  # type: ignore # Incompatible types in assignment (expression has type "str | None", variable has type "str")
                 tool=calling.tool_name, input=calling.arguments
-            )
+            )  # type: ignore
             from_cache = result is not None
 
-        original_tool = next(
+        available_tool = next(
-            (ot for ot in self.original_tools if ot.name == tool.name), None
+            (
+                available_tool
+                for available_tool in self.tools
+                if available_tool.name == tool.name
+            ),
+            None,
         )
 
-        if result is None:  #! finecwg: if not result --> if result is None
+        if result is None:
             try:
                 if calling.tool_name in [
                     "Delegate work to coworker",
@@ -180,6 +209,7 @@ class ToolUsage:
                     coworker = (
                         calling.arguments.get("coworker") if calling.arguments else None
                     )
+                    if self.task:
                         self.task.increment_delegations(coworker)
 
                 if calling.arguments:
@@ -192,12 +222,18 @@ class ToolUsage:
                             for k, v in calling.arguments.items()
                             if k in acceptable_args
                         }
+                        # Add fingerprint metadata if available
+                        arguments = self._add_fingerprint_metadata(arguments)
                         result = tool.invoke(input=arguments)
                     except Exception:
                         arguments = calling.arguments
+                        # Add fingerprint metadata if available
+                        arguments = self._add_fingerprint_metadata(arguments)
                         result = tool.invoke(input=arguments)
                 else:
-                    result = tool.invoke(input={})
+                    # Add fingerprint metadata even to empty arguments
+                    arguments = self._add_fingerprint_metadata({})
+                    result = tool.invoke(input=arguments)
             except Exception as e:
                 self.on_tool_error(tool=tool, tool_calling=calling, e=e)
                 self._run_attempts += 1
@@ -209,23 +245,25 @@ class ToolUsage:
                     error = ToolUsageErrorException(
                         f"\n{error_message}.\nMoving on then. {self._i18n.slice('format').format(tool_names=self.tools_names)}"
                     ).message
+                    if self.task:
                         self.task.increment_tools_errors()
-                    if self.agent.verbose:
+                    if self.agent and self.agent.verbose:
                         self._printer.print(
                             content=f"\n\n{error_message}\n", color="red"
                         )
                     return error  # type: ignore # No return value expected
 
+                if self.task:
                     self.task.increment_tools_errors()
                 return self.use(calling=calling, tool_string=tool_string)  # type: ignore # No return value expected
 
             if self.tools_handler:
                 should_cache = True
                 if (
-                    hasattr(original_tool, "cache_function")
+                    hasattr(available_tool, "cache_function")
-                    and original_tool.cache_function  # type: ignore # Item "None" of "Any | None" has no attribute "cache_function"
+                    and available_tool.cache_function  # type: ignore # Item "None" of "Any | None" has no attribute "cache_function"
                 ):
-                    should_cache = original_tool.cache_function(  # type: ignore # Item "None" of "Any | None" has no attribute "cache_function"
+                    should_cache = available_tool.cache_function(  # type: ignore # Item "None" of "Any | None" has no attribute "cache_function"
                         calling.arguments, result
                     )
 
@@ -253,41 +291,46 @@ class ToolUsage:
         )
 
         if (
-            hasattr(original_tool, "result_as_answer")
+            hasattr(available_tool, "result_as_answer")
-            and original_tool.result_as_answer  # type: ignore # Item "None" of "Any | None" has no attribute "cache_function"
+            and available_tool.result_as_answer  # type: ignore # Item "None" of "Any | None" has no attribute "cache_function"
         ):
-            result_as_answer = original_tool.result_as_answer  # type: ignore # Item "None" of "Any | None" has no attribute "result_as_answer"
+            result_as_answer = available_tool.result_as_answer  # type: ignore # Item "None" of "Any | None" has no attribute "result_as_answer"
-            data["result_as_answer"] = result_as_answer
+            data["result_as_answer"] = result_as_answer  # type: ignore
 
+        if self.agent and hasattr(self.agent, "tools_results"):
             self.agent.tools_results.append(data)
 
-        return result  # type: ignore # No return value expected
-
-    def _format_result(self, result: Any) -> None:
-        self.task.used_tools += 1
-        if self._should_remember_format():  # type: ignore # "_should_remember_format" of "ToolUsage" does not return a value (it only ever returns None)
-            result = self._remember_format(result=result)  # type: ignore # "_remember_format" of "ToolUsage" does not return a value (it only ever returns None)
         return result
 
-    def _should_remember_format(self) -> bool:
+    def _format_result(self, result: Any) -> str:
-        return self.task.used_tools % self._remember_format_after_usages == 0
+        if self.task:
+            self.task.used_tools += 1
+        if self._should_remember_format():
+            result = self._remember_format(result=result)
+        return str(result)
 
-    def _remember_format(self, result: str) -> None:
+    def _should_remember_format(self) -> bool:
+        if self.task:
+            return self.task.used_tools % self._remember_format_after_usages == 0
+        return False
+
+    def _remember_format(self, result: str) -> str:
         result = str(result)
         result += "\n\n" + self._i18n.slice("tools").format(
             tools=self.tools_description, tool_names=self.tools_names
         )
-        return result  # type: ignore # No return value expected
+        return result
 
     def _check_tool_repeated_usage(
         self, calling: Union[ToolCalling, InstructorToolCalling]
-    ) -> None:
+    ) -> bool:
         if not self.tools_handler:
-            return False  # type: ignore # No return value expected
+            return False
         if last_tool_usage := self.tools_handler.last_used_tool:
-            return (calling.tool_name == last_tool_usage.tool_name) and (  # type: ignore # No return value expected
+            return (calling.tool_name == last_tool_usage.tool_name) and (
                 calling.arguments == last_tool_usage.arguments
             )
+        return False
 
     def _select_tool(self, tool_name: str) -> Any:
         order_tools = sorted(
@@ -306,10 +349,11 @@ class ToolUsage:
                 > 0.85
             ):
                 return tool
+        if self.task:
             self.task.increment_tools_errors()
-        tool_selection_data = {
+        tool_selection_data: Dict[str, Any] = {
-            "agent_key": self.agent.key,
+            "agent_key": getattr(self.agent, "key", None) if self.agent else None,
-            "agent_role": self.agent.role,
+            "agent_role": getattr(self.agent, "role", None) if self.agent else None,
             "tool_name": tool_name,
             "tool_args": {},
             "tool_class": self.tools_description,
@@ -342,7 +386,9 @@ class ToolUsage:
             descriptions.append(tool.description)
         return "\n--\n".join(descriptions)
 
-    def _function_calling(self, tool_string: str):
+    def _function_calling(
+        self, tool_string: str
+    ) -> Union[ToolCalling, InstructorToolCalling]:
         model = (
             InstructorToolCalling
             if self.function_calling_llm.supports_function_calling()
@@ -364,18 +410,14 @@ class ToolUsage:
             max_attempts=1,
         )
         tool_object = converter.to_pydantic()
-        calling = ToolCalling(
+        if not isinstance(tool_object, (ToolCalling, InstructorToolCalling)):
-            tool_name=tool_object["tool_name"],
+            raise ToolUsageErrorException("Failed to parse tool calling")
-            arguments=tool_object["arguments"],
-            log=tool_string,  # type: ignore
-        )
 
-        if isinstance(calling, ConverterError):
+        return tool_object
-            raise calling
 
-        return calling
+    def _original_tool_calling(
-
+        self, tool_string: str, raise_error: bool = False
-    def _original_tool_calling(self, tool_string: str, raise_error: bool = False):
+    ) -> Union[ToolCalling, InstructorToolCalling, ToolUsageErrorException]:
         tool_name = self.action.tool
         tool = self._select_tool(tool_name)
         try:
@@ -400,12 +442,11 @@ class ToolUsage:
         return ToolCalling(
             tool_name=tool.name,
             arguments=arguments,
-            log=tool_string,
         )
 
     def _tool_calling(
         self, tool_string: str
-    ) -> Union[ToolCalling, InstructorToolCalling]:
+    ) -> Union[ToolCalling, InstructorToolCalling, ToolUsageErrorException]:
         try:
             try:
                 return self._original_tool_calling(tool_string, raise_error=True)
@@ -418,8 +459,9 @@ class ToolUsage:
             self._run_attempts += 1
             if self._run_attempts > self._max_parsing_attempts:
                 self._telemetry.tool_usage_error(llm=self.function_calling_llm)
+                if self.task:
                     self.task.increment_tools_errors()
-                if self.agent.verbose:
+                if self.agent and self.agent.verbose:
                     self._printer.print(content=f"\n\n{e}\n", color="red")
                 return ToolUsageErrorException(  # type: ignore # Incompatible return value type (got "ToolUsageErrorException", expected "ToolCalling | InstructorToolCalling")
                     f"{self._i18n.errors('tool_usage_error').format(error=e)}\nMoving on then. {self._i18n.slice('format').format(tool_names=self.tools_names)}"
@@ -449,6 +491,7 @@ class ToolUsage:
             if isinstance(arguments, dict):
                 return arguments
         except (ValueError, SyntaxError):
+            repaired_input = repair_json(tool_input)
             pass  # Continue to the next parsing attempt
 
         # Attempt 3: Parse as JSON5
@@ -461,7 +504,7 @@ class ToolUsage:
 
         # Attempt 4: Repair JSON
         try:
-            repaired_input = repair_json(tool_input, skip_json_loads=True)
+            repaired_input = str(repair_json(tool_input, skip_json_loads=True))
             self._printer.print(
                 content=f"Repaired JSON: {repaired_input}", color="blue"
             )
@@ -481,26 +524,36 @@ class ToolUsage:
 
     def _emit_validate_input_error(self, final_error: str):
         tool_selection_data = {
-            "agent_key": self.agent.key,
+            "agent_key": getattr(self.agent, "key", None) if self.agent else None,
-            "agent_role": self.agent.role,
+            "agent_role": getattr(self.agent, "role", None) if self.agent else None,
             "tool_name": self.action.tool,
             "tool_args": str(self.action.tool_input),
             "tool_class": self.__class__.__name__,
+            "agent": self.agent,  # Adding agent for fingerprint extraction
         }
 
+        # Include fingerprint context if available
+        if self.fingerprint_context:
+            tool_selection_data.update(self.fingerprint_context)
+
         crewai_event_bus.emit(
             self,
             ToolValidateInputErrorEvent(**tool_selection_data, error=final_error),
         )
 
-    def on_tool_error(self, tool: Any, tool_calling: ToolCalling, e: Exception) -> None:
+    def on_tool_error(
+        self,
+        tool: Any,
+        tool_calling: Union[ToolCalling, InstructorToolCalling],
+        e: Exception,
+    ) -> None:
         event_data = self._prepare_event_data(tool, tool_calling)
         crewai_event_bus.emit(self, ToolUsageErrorEvent(**{**event_data, "error": e}))
 
     def on_tool_use_finished(
         self,
         tool: Any,
-        tool_calling: ToolCalling,
+        tool_calling: Union[ToolCalling, InstructorToolCalling],
         from_cache: bool,
         started_at: float,
         result: Any,
@@ -517,13 +570,70 @@ class ToolUsage:
         )
         crewai_event_bus.emit(self, ToolUsageFinishedEvent(**event_data))
 
-    def _prepare_event_data(self, tool: Any, tool_calling: ToolCalling) -> dict:
+    def _prepare_event_data(
-        return {
+        self, tool: Any, tool_calling: Union[ToolCalling, InstructorToolCalling]
-            "agent_key": self.agent.key,
+    ) -> dict:
-            "agent_role": (self.agent._original_role or self.agent.role),
+        event_data = {
             "run_attempts": self._run_attempts,
-            "delegations": self.task.delegations,
+            "delegations": self.task.delegations if self.task else 0,
             "tool_name": tool.name,
             "tool_args": tool_calling.arguments,
             "tool_class": tool.__class__.__name__,
+            "agent_key": (
+                getattr(self.agent, "key", "unknown") if self.agent else "unknown"
+            ),
+            "agent_role": (
+                getattr(self.agent, "_original_role", None)
+                or getattr(self.agent, "role", "unknown")
+                if self.agent
+                else "unknown"
+            ),
         }
+
+        # Include fingerprint context if available
+        if self.fingerprint_context:
+            event_data.update(self.fingerprint_context)
+
+        return event_data
+
+    def _add_fingerprint_metadata(self, arguments: dict) -> dict:
+        """Add fingerprint metadata to tool arguments if available.
+
+        Args:
+            arguments: The original tool arguments
+
+        Returns:
+            Updated arguments dictionary with fingerprint metadata
+        """
+        # Create a shallow copy to avoid modifying the original
+        arguments = arguments.copy()
+
+        # Add security metadata under a designated key
+        if "security_context" not in arguments:
+            arguments["security_context"] = {}
+
+        security_context = arguments["security_context"]
+
+        # Add agent fingerprint if available
+        if self.agent and hasattr(self.agent, "security_config"):
+            security_config = getattr(self.agent, "security_config", None)
+            if security_config and hasattr(security_config, "fingerprint"):
+                try:
+                    security_context["agent_fingerprint"] = (
+                        security_config.fingerprint.to_dict()
+                    )
+                except AttributeError:
+                    pass
+
+        # Add task fingerprint if available
+        if self.task and hasattr(self.task, "security_config"):
+            security_config = getattr(self.task, "security_config", None)
+            if security_config and hasattr(security_config, "fingerprint"):
+                try:
+                    security_context["task_fingerprint"] = (
+                        security_config.fingerprint.to_dict()
+                    )
+                except AttributeError:
+                    pass
+
+        return arguments

src/crewai/translations/en.json

@@ -24,7 +24,10 @@
     "manager_request": "Your best answer to your coworker asking you this, accounting for the context shared.",
     "formatted_task_instructions": "Ensure your final answer contains only the content in the following format: {output_format}\n\nEnsure the final output does not include any code block markers like ```json or ```python.",
     "conversation_history_instruction": "You are a member of a crew collaborating to achieve a common goal. Your task is a specific action that contributes to this larger objective. For additional context, please review the conversation history between you and the user that led to the initiation of this crew. Use any relevant information or feedback from the conversation to inform your task execution and ensure your response aligns with both the immediate task and the crew's overall goals.",
-    "feedback_instructions": "User feedback: {feedback}\nInstructions: Use this feedback to enhance the next output iteration.\nNote: Do not respond or add commentary."
+    "feedback_instructions": "User feedback: {feedback}\nInstructions: Use this feedback to enhance the next output iteration.\nNote: Do not respond or add commentary.",
+    "lite_agent_system_prompt_with_tools": "You are {role}. {backstory}\nYour personal goal is: {goal}\n\nYou ONLY have access to the following tools, and should NEVER make up tools that are not listed here:\n\n{tools}\n\nIMPORTANT: Use the following format in your response:\n\n```\nThought: you should always think about what to do\nAction: the action to take, only one name of [{tool_names}], just the name, exactly as it's written.\nAction Input: the input to the action, just a simple JSON object, enclosed in curly braces, using \" to wrap keys and values.\nObservation: the result of the action\n```\n\nOnce all necessary information is gathered, return the following format:\n\n```\nThought: I now know the final answer\nFinal Answer: the final answer to the original input question\n```",
+    "lite_agent_system_prompt_without_tools": "You are {role}. {backstory}\nYour personal goal is: {goal}\n\nTo give my best complete final answer to the task respond using the exact following format:\n\nThought: I now can give a great answer\nFinal Answer: Your final answer must be the great and the most complete as possible, it must be outcome described.\n\nI MUST use these formats, my job depends on it!",
+    "lite_agent_response_format": "\nIMPORTANT: Your final answer MUST contain all the information requested in the following format: {response_format}\n\nIMPORTANT: Ensure the final output does not include any code block markers like ```json or ```python."
   },
   "errors": {
     "force_final_answer_error": "You can't keep going, here is the best final answer you generated:\n\n {formatted_answer}",

src/crewai/utilities/agent_utils.py

@@ -0,0 +1,430 @@
+import json
+import re
+from typing import Any, Callable, Dict, List, Optional, Sequence, Union
+
+from crewai.agents.parser import (
+    FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE,
+    AgentAction,
+    AgentFinish,
+    CrewAgentParser,
+    OutputParserException,
+)
+from crewai.llm import LLM
+from crewai.llms.base_llm import BaseLLM
+from crewai.tools import BaseTool as CrewAITool
+from crewai.tools.base_tool import BaseTool
+from crewai.tools.structured_tool import CrewStructuredTool
+from crewai.tools.tool_types import ToolResult
+from crewai.utilities import I18N, Printer
+from crewai.utilities.exceptions.context_window_exceeding_exception import (
+    LLMContextLengthExceededException,
+)
+
+
+def parse_tools(tools: List[BaseTool]) -> List[CrewStructuredTool]:
+    """Parse tools to be used for the task."""
+    tools_list = []
+
+    for tool in tools:
+        if isinstance(tool, CrewAITool):
+            tools_list.append(tool.to_structured_tool())
+        else:
+            raise ValueError("Tool is not a CrewStructuredTool or BaseTool")
+
+    return tools_list
+
+
+def get_tool_names(tools: Sequence[Union[CrewStructuredTool, BaseTool]]) -> str:
+    """Get the names of the tools."""
+    return ", ".join([t.name for t in tools])
+
+
+def render_text_description_and_args(
+    tools: Sequence[Union[CrewStructuredTool, BaseTool]],
+) -> str:
+    """Render the tool name, description, and args in plain text.
+    
+        search: This tool is used for search, args: {"query": {"type": "string"}}
+        calculator: This tool is used for math, \
+        args: {"expression": {"type": "string"}}
+    """
+    tool_strings = []
+    for tool in tools:
+        tool_strings.append(tool.description)
+
+    return "\n".join(tool_strings)
+
+
+def has_reached_max_iterations(iterations: int, max_iterations: int) -> bool:
+    """Check if the maximum number of iterations has been reached."""
+    return iterations >= max_iterations
+
+
+def handle_max_iterations_exceeded(
+    formatted_answer: Union[AgentAction, AgentFinish, None],
+    printer: Printer,
+    i18n: I18N,
+    messages: List[Dict[str, str]],
+    llm: Union[LLM, BaseLLM],
+    callbacks: List[Any],
+) -> Union[AgentAction, AgentFinish]:
+    """
+    Handles the case when the maximum number of iterations is exceeded.
+    Performs one more LLM call to get the final answer.
+
+    Parameters:
+        formatted_answer: The last formatted answer from the agent.
+
+    Returns:
+        The final formatted answer after exceeding max iterations.
+    """
+    printer.print(
+        content="Maximum iterations reached. Requesting final answer.",
+        color="yellow",
+    )
+
+    if formatted_answer and hasattr(formatted_answer, "text"):
+        assistant_message = (
+            formatted_answer.text + f'\n{i18n.errors("force_final_answer")}'
+        )
+    else:
+        assistant_message = i18n.errors("force_final_answer")
+
+    messages.append(format_message_for_llm(assistant_message, role="assistant"))
+
+    # Perform one more LLM call to get the final answer
+    answer = llm.call(
+        messages,
+        callbacks=callbacks,
+    )
+
+    if answer is None or answer == "":
+        printer.print(
+            content="Received None or empty response from LLM call.",
+            color="red",
+        )
+        raise ValueError("Invalid response from LLM call - None or empty.")
+
+    formatted_answer = format_answer(answer)
+    # Return the formatted answer, regardless of its type
+    return formatted_answer
+
+
+def format_message_for_llm(prompt: str, role: str = "user") -> Dict[str, str]:
+    prompt = prompt.rstrip()
+    return {"role": role, "content": prompt}
+
+
+def format_answer(answer: str) -> Union[AgentAction, AgentFinish]:
+    """Format a response from the LLM into an AgentAction or AgentFinish."""
+    try:
+        return CrewAgentParser.parse_text(answer)
+    except Exception:
+        # If parsing fails, return a default AgentFinish
+        return AgentFinish(
+            thought="Failed to parse LLM response",
+            output=answer,
+            text=answer,
+        )
+
+
+def enforce_rpm_limit(
+    request_within_rpm_limit: Optional[Callable[[], bool]] = None,
+) -> None:
+    """Enforce the requests per minute (RPM) limit if applicable."""
+    if request_within_rpm_limit:
+        request_within_rpm_limit()
+
+
+def get_llm_response(
+    llm: Union[LLM, BaseLLM],
+    messages: List[Dict[str, str]],
+    callbacks: List[Any],
+    printer: Printer,
+) -> str:
+    """Call the LLM and return the response, handling any invalid responses."""
+    try:
+        answer = llm.call(
+            messages,
+            callbacks=callbacks,
+        )
+    except Exception as e:
+        printer.print(
+            content=f"Error during LLM call: {e}",
+            color="red",
+        )
+        raise e
+    if not answer:
+        printer.print(
+            content="Received None or empty response from LLM call.",
+            color="red",
+        )
+        raise ValueError("Invalid response from LLM call - None or empty.")
+
+    return answer
+
+
+def process_llm_response(
+    answer: str, use_stop_words: bool
+) -> Union[AgentAction, AgentFinish]:
+    """Process the LLM response and format it into an AgentAction or AgentFinish."""
+    if not use_stop_words:
+        try:
+            # Preliminary parsing to check for errors.
+            format_answer(answer)
+        except OutputParserException as e:
+            if FINAL_ANSWER_AND_PARSABLE_ACTION_ERROR_MESSAGE in e.error:
+                answer = answer.split("Observation:")[0].strip()
+
+    return format_answer(answer)
+
+
+def handle_agent_action_core(
+    formatted_answer: AgentAction,
+    tool_result: ToolResult,
+    messages: Optional[List[Dict[str, str]]] = None,
+    step_callback: Optional[Callable] = None,
+    show_logs: Optional[Callable] = None,
+) -> Union[AgentAction, AgentFinish]:
+    """Core logic for handling agent actions and tool results.
+
+    Args:
+        formatted_answer: The agent's action
+        tool_result: The result of executing the tool
+        messages: Optional list of messages to append results to
+        step_callback: Optional callback to execute after processing
+        show_logs: Optional function to show logs
+
+    Returns:
+        Either an AgentAction or AgentFinish
+    """
+    if step_callback:
+        step_callback(tool_result)
+
+    formatted_answer.text += f"\nObservation: {tool_result.result}"
+    formatted_answer.result = tool_result.result
+
+    if tool_result.result_as_answer:
+        return AgentFinish(
+            thought="",
+            output=tool_result.result,
+            text=formatted_answer.text,
+        )
+
+    if show_logs:
+        show_logs(formatted_answer)
+
+    if messages is not None:
+        messages.append({"role": "assistant", "content": tool_result.result})
+
+    return formatted_answer
+
+
+def handle_unknown_error(printer: Any, exception: Exception) -> None:
+    """Handle unknown errors by informing the user.
+
+    Args:
+        printer: Printer instance for output
+        exception: The exception that occurred
+    """
+    printer.print(
+        content="An unknown error occurred. Please check the details below.",
+        color="red",
+    )
+    printer.print(
+        content=f"Error details: {exception}",
+        color="red",
+    )
+
+
+def handle_output_parser_exception(
+    e: OutputParserException,
+    messages: List[Dict[str, str]],
+    iterations: int,
+    log_error_after: int = 3,
+    printer: Optional[Any] = None,
+) -> AgentAction:
+    """Handle OutputParserException by updating messages and formatted_answer.
+
+    Args:
+        e: The OutputParserException that occurred
+        messages: List of messages to append to
+        iterations: Current iteration count
+        log_error_after: Number of iterations after which to log errors
+        printer: Optional printer instance for logging
+
+    Returns:
+        AgentAction: A formatted answer with the error
+    """
+    messages.append({"role": "user", "content": e.error})
+
+    formatted_answer = AgentAction(
+        text=e.error,
+        tool="",
+        tool_input="",
+        thought="",
+    )
+
+    if iterations > log_error_after and printer:
+        printer.print(
+            content=f"Error parsing LLM output, agent will retry: {e.error}",
+            color="red",
+        )
+
+    return formatted_answer
+
+
+def is_context_length_exceeded(exception: Exception) -> bool:
+    """Check if the exception is due to context length exceeding.
+
+    Args:
+        exception: The exception to check
+
+    Returns:
+        bool: True if the exception is due to context length exceeding
+    """
+    return LLMContextLengthExceededException(str(exception))._is_context_limit_error(
+        str(exception)
+    )
+
+
+def handle_context_length(
+    respect_context_window: bool,
+    printer: Any,
+    messages: List[Dict[str, str]],
+    llm: Any,
+    callbacks: List[Any],
+    i18n: Any,
+) -> None:
+    """Handle context length exceeded by either summarizing or raising an error.
+
+    Args:
+        respect_context_window: Whether to respect context window
+        printer: Printer instance for output
+        messages: List of messages to summarize
+        llm: LLM instance for summarization
+        callbacks: List of callbacks for LLM
+        i18n: I18N instance for messages
+    """
+    if respect_context_window:
+        printer.print(
+            content="Context length exceeded. Summarizing content to fit the model context window.",
+            color="yellow",
+        )
+        summarize_messages(messages, llm, callbacks, i18n)
+    else:
+        printer.print(
+            content="Context length exceeded. Consider using smaller text or RAG tools from crewai_tools.",
+            color="red",
+        )
+        raise SystemExit(
+            "Context length exceeded and user opted not to summarize. Consider using smaller text or RAG tools from crewai_tools."
+        )
+
+
+def summarize_messages(
+    messages: List[Dict[str, str]],
+    llm: Any,
+    callbacks: List[Any],
+    i18n: Any,
+) -> None:
+    """Summarize messages to fit within context window.
+
+    Args:
+        messages: List of messages to summarize
+        llm: LLM instance for summarization
+        callbacks: List of callbacks for LLM
+        i18n: I18N instance for messages
+    """
+    messages_groups = []
+    for message in messages:
+        content = message["content"]
+        cut_size = llm.get_context_window_size()
+        for i in range(0, len(content), cut_size):
+            messages_groups.append({"content": content[i : i + cut_size]})
+
+    summarized_contents = []
+    for group in messages_groups:
+        summary = llm.call(
+            [
+                format_message_for_llm(
+                    i18n.slice("summarizer_system_message"), role="system"
+                ),
+                format_message_for_llm(
+                    i18n.slice("summarize_instruction").format(group=group["content"]),
+                ),
+            ],
+            callbacks=callbacks,
+        )
+        summarized_contents.append({"content": str(summary)})
+
+    merged_summary = " ".join(content["content"] for content in summarized_contents)
+
+    messages.clear()
+    messages.append(
+        format_message_for_llm(
+            i18n.slice("summary").format(merged_summary=merged_summary)
+        )
+    )
+
+
+def show_agent_logs(
+    printer: Printer,
+    agent_role: str,
+    formatted_answer: Optional[Union[AgentAction, AgentFinish]] = None,
+    task_description: Optional[str] = None,
+    verbose: bool = False,
+) -> None:
+    """Show agent logs for both start and execution states.
+
+    Args:
+        printer: Printer instance for output
+        agent_role: Role of the agent
+        formatted_answer: Optional AgentAction or AgentFinish for execution logs
+        task_description: Optional task description for start logs
+        verbose: Whether to show verbose output
+    """
+    if not verbose:
+        return
+
+    agent_role = agent_role.split("\n")[0]
+
+    if formatted_answer is None:
+        # Start logs
+        printer.print(
+            content=f"\033[1m\033[95m# Agent:\033[00m \033[1m\033[92m{agent_role}\033[00m"
+        )
+        if task_description:
+            printer.print(
+                content=f"\033[95m## Task:\033[00m \033[92m{task_description}\033[00m"
+            )
+    else:
+        # Execution logs
+        printer.print(
+            content=f"\n\n\033[1m\033[95m# Agent:\033[00m \033[1m\033[92m{agent_role}\033[00m"
+        )
+
+        if isinstance(formatted_answer, AgentAction):
+            thought = re.sub(r"\n+", "\n", formatted_answer.thought)
+            formatted_json = json.dumps(
+                formatted_answer.tool_input,
+                indent=2,
+                ensure_ascii=False,
+            )
+            if thought and thought != "":
+                printer.print(
+                    content=f"\033[95m## Thought:\033[00m \033[92m{thought}\033[00m"
+                )
+            printer.print(
+                content=f"\033[95m## Using tool:\033[00m \033[92m{formatted_answer.tool}\033[00m"
+            )
+            printer.print(
+                content=f"\033[95m## Tool Input:\033[00m \033[92m\n{formatted_json}\033[00m"
+            )
+            printer.print(
+                content=f"\033[95m## Tool Output:\033[00m \033[92m\n{formatted_answer.result}\033[00m"
+            )
+        elif isinstance(formatted_answer, AgentFinish):
+            printer.print(
+                content=f"\033[95m## Final Answer:\033[00m \033[92m\n{formatted_answer.output}\033[00m\n\n"
+            )

src/crewai/utilities/converter.py

@@ -216,7 +216,7 @@ def convert_with_instructions(
 
 def get_conversion_instructions(model: Type[BaseModel], llm: Any) -> str:
     instructions = "Please convert the following text into valid JSON."
-    if llm.supports_function_calling():
+    if llm and not isinstance(llm, str) and llm.supports_function_calling():
         model_schema = PydanticSchemaParser(model=model).get_schema()
         instructions += (
             f"\n\nOutput ONLY the valid JSON and nothing else.\n\n"
@@ -287,8 +287,9 @@ def generate_model_description(model: Type[BaseModel]) -> str:
         else:
             return str(field_type)
 
-    fields = model.__annotations__
+    fields = model.model_fields
     field_descriptions = [
-        f'"{name}": {describe_field(type_)}' for name, type_ in fields.items()
+        f'"{name}": {describe_field(field.annotation)}'
+        for name, field in fields.items()
     ]
     return "{\n  " + ",\n  ".join(field_descriptions) + "\n}"

src/crewai/utilities/evaluators/task_evaluator.py

@@ -45,7 +45,7 @@ class TaskEvaluator:
 
     def evaluate(self, task, output) -> TaskEvaluation:
         crewai_event_bus.emit(
-            self, TaskEvaluationEvent(evaluation_type="task_evaluation")
+            self, TaskEvaluationEvent(evaluation_type="task_evaluation", task=task)
         )
         evaluation_query = (
             f"Assess the quality of the task completed based on the description, expected output, and actual results.\n\n"

src/crewai/utilities/events/agent_events.py

@@ -1,16 +1,16 @@
-from typing import TYPE_CHECKING, Any, Dict, Optional, Sequence, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union
 
 from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.tools.base_tool import BaseTool
 from crewai.tools.structured_tool import CrewStructuredTool
 
-from .base_events import CrewEvent
+from .base_events import BaseEvent
 
 if TYPE_CHECKING:
     from crewai.agents.agent_builder.base_agent import BaseAgent
 
 
-class AgentExecutionStartedEvent(CrewEvent):
+class AgentExecutionStartedEvent(BaseEvent):
     """Event emitted when an agent starts executing a task"""
 
     agent: BaseAgent
@@ -21,8 +21,20 @@ class AgentExecutionStartedEvent(CrewEvent):
 
     model_config = {"arbitrary_types_allowed": True}
 
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Set fingerprint data from the agent
+        if hasattr(self.agent, "fingerprint") and self.agent.fingerprint:
+            self.source_fingerprint = self.agent.fingerprint.uuid_str
+            self.source_type = "agent"
+            if (
+                hasattr(self.agent.fingerprint, "metadata")
+                and self.agent.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.agent.fingerprint.metadata
 
-class AgentExecutionCompletedEvent(CrewEvent):
+
+class AgentExecutionCompletedEvent(BaseEvent):
     """Event emitted when an agent completes executing a task"""
 
     agent: BaseAgent
@@ -30,11 +42,63 @@ class AgentExecutionCompletedEvent(CrewEvent):
     output: str
     type: str = "agent_execution_completed"
 
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Set fingerprint data from the agent
+        if hasattr(self.agent, "fingerprint") and self.agent.fingerprint:
+            self.source_fingerprint = self.agent.fingerprint.uuid_str
+            self.source_type = "agent"
+            if (
+                hasattr(self.agent.fingerprint, "metadata")
+                and self.agent.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.agent.fingerprint.metadata
 
-class AgentExecutionErrorEvent(CrewEvent):
+
+class AgentExecutionErrorEvent(BaseEvent):
     """Event emitted when an agent encounters an error during execution"""
 
     agent: BaseAgent
     task: Any
     error: str
     type: str = "agent_execution_error"
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Set fingerprint data from the agent
+        if hasattr(self.agent, "fingerprint") and self.agent.fingerprint:
+            self.source_fingerprint = self.agent.fingerprint.uuid_str
+            self.source_type = "agent"
+            if (
+                hasattr(self.agent.fingerprint, "metadata")
+                and self.agent.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.agent.fingerprint.metadata
+
+
+# New event classes for LiteAgent
+class LiteAgentExecutionStartedEvent(BaseEvent):
+    """Event emitted when a LiteAgent starts executing"""
+
+    agent_info: Dict[str, Any]
+    tools: Optional[Sequence[Union[BaseTool, CrewStructuredTool]]]
+    messages: Union[str, List[Dict[str, str]]]
+    type: str = "lite_agent_execution_started"
+
+    model_config = {"arbitrary_types_allowed": True}
+
+
+class LiteAgentExecutionCompletedEvent(BaseEvent):
+    """Event emitted when a LiteAgent completes execution"""
+
+    agent_info: Dict[str, Any]
+    output: str
+    type: str = "lite_agent_execution_completed"
+
+
+class LiteAgentExecutionErrorEvent(BaseEvent):
+    """Event emitted when a LiteAgent encounters an error during execution"""
+
+    agent_info: Dict[str, Any]
+    error: str
+    type: str = "lite_agent_execution_error"

src/crewai/utilities/events/base_events.py

@@ -1,10 +1,28 @@
 from datetime import datetime
+from typing import Any, Dict, Optional
 
 from pydantic import BaseModel, Field
 
+from crewai.utilities.serialization import to_serializable
 
-class CrewEvent(BaseModel):
+
-    """Base class for all crew events"""
+class BaseEvent(BaseModel):
+    """Base class for all events"""
 
     timestamp: datetime = Field(default_factory=datetime.now)
     type: str
+    source_fingerprint: Optional[str] = None  # UUID string of the source entity
+    source_type: Optional[str] = None  # "agent", "task", "crew"
+    fingerprint_metadata: Optional[Dict[str, Any]] = None  # Any relevant metadata
+
+    def to_json(self, exclude: set[str] | None = None):
+        """
+        Converts the event to a JSON-serializable dictionary.
+
+        Args:
+            exclude (set[str], optional): Set of keys to exclude from the result. Defaults to None.
+
+        Returns:
+            dict: A JSON-serializable dictionary.
+        """
+        return to_serializable(self, exclude=exclude)

src/crewai/utilities/events/crew_events.py

@@ -1,81 +1,102 @@
-from typing import Any, Dict, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, Optional, Union
 
-from pydantic import InstanceOf
+from crewai.utilities.events.base_events import BaseEvent
 
-from crewai.utilities.events.base_events import CrewEvent
+if TYPE_CHECKING:
+    from crewai.crew import Crew
+else:
+    Crew = Any
 
 
-class CrewKickoffStartedEvent(CrewEvent):
+class CrewBaseEvent(BaseEvent):
-    """Event emitted when a crew starts execution"""
+    """Base class for crew events with fingerprint handling"""
 
     crew_name: Optional[str]
+    crew: Optional[Crew] = None
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        self.set_crew_fingerprint()
+
+    def set_crew_fingerprint(self) -> None:
+        if self.crew and hasattr(self.crew, "fingerprint") and self.crew.fingerprint:
+            self.source_fingerprint = self.crew.fingerprint.uuid_str
+            self.source_type = "crew"
+            if (
+                hasattr(self.crew.fingerprint, "metadata")
+                and self.crew.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.crew.fingerprint.metadata
+
+    def to_json(self, exclude: set[str] | None = None):
+        if exclude is None:
+            exclude = set()
+        exclude.add("crew")
+        return super().to_json(exclude=exclude)
+
+
+class CrewKickoffStartedEvent(CrewBaseEvent):
+    """Event emitted when a crew starts execution"""
+
     inputs: Optional[Dict[str, Any]]
     type: str = "crew_kickoff_started"
 
 
-class CrewKickoffCompletedEvent(CrewEvent):
+class CrewKickoffCompletedEvent(CrewBaseEvent):
     """Event emitted when a crew completes execution"""
 
-    crew_name: Optional[str]
     output: Any
     type: str = "crew_kickoff_completed"
 
 
-class CrewKickoffFailedEvent(CrewEvent):
+class CrewKickoffFailedEvent(CrewBaseEvent):
     """Event emitted when a crew fails to complete execution"""
 
     error: str
-    crew_name: Optional[str]
     type: str = "crew_kickoff_failed"
 
 
-class CrewTrainStartedEvent(CrewEvent):
+class CrewTrainStartedEvent(CrewBaseEvent):
     """Event emitted when a crew starts training"""
 
-    crew_name: Optional[str]
     n_iterations: int
     filename: str
     inputs: Optional[Dict[str, Any]]
     type: str = "crew_train_started"
 
 
-class CrewTrainCompletedEvent(CrewEvent):
+class CrewTrainCompletedEvent(CrewBaseEvent):
     """Event emitted when a crew completes training"""
 
-    crew_name: Optional[str]
     n_iterations: int
     filename: str
     type: str = "crew_train_completed"
 
 
-class CrewTrainFailedEvent(CrewEvent):
+class CrewTrainFailedEvent(CrewBaseEvent):
     """Event emitted when a crew fails to complete training"""
 
     error: str
-    crew_name: Optional[str]
     type: str = "crew_train_failed"
 
 
-class CrewTestStartedEvent(CrewEvent):
+class CrewTestStartedEvent(CrewBaseEvent):
     """Event emitted when a crew starts testing"""
 
-    crew_name: Optional[str]
     n_iterations: int
     eval_llm: Optional[Union[str, Any]]
     inputs: Optional[Dict[str, Any]]
     type: str = "crew_test_started"
 
 
-class CrewTestCompletedEvent(CrewEvent):
+class CrewTestCompletedEvent(CrewBaseEvent):
     """Event emitted when a crew completes testing"""
 
-    crew_name: Optional[str]
     type: str = "crew_test_completed"
 
 
-class CrewTestFailedEvent(CrewEvent):
+class CrewTestFailedEvent(CrewBaseEvent):
     """Event emitted when a crew fails to complete testing"""
 
     error: str
-    crew_name: Optional[str]
     type: str = "crew_test_failed"

src/crewai/utilities/events/crewai_event_bus.py

@@ -4,10 +4,10 @@ from typing import Any, Callable, Dict, List, Type, TypeVar, cast
 
 from blinker import Signal
 
-from crewai.utilities.events.base_events import CrewEvent
+from crewai.utilities.events.base_events import BaseEvent
 from crewai.utilities.events.event_types import EventTypes
 
-EventT = TypeVar("EventT", bound=CrewEvent)
+EventT = TypeVar("EventT", bound=BaseEvent)
 
 
 class CrewAIEventsBus:
@@ -30,7 +30,7 @@ class CrewAIEventsBus:
     def _initialize(self) -> None:
         """Initialize the event bus internal state"""
         self._signal = Signal("crewai_event_bus")
-        self._handlers: Dict[Type[CrewEvent], List[Callable]] = {}
+        self._handlers: Dict[Type[BaseEvent], List[Callable]] = {}
 
     def on(
         self, event_type: Type[EventT]
@@ -59,7 +59,7 @@ class CrewAIEventsBus:
 
         return decorator
 
-    def emit(self, source: Any, event: CrewEvent) -> None:
+    def emit(self, source: Any, event: BaseEvent) -> None:
         """
         Emit an event to all registered handlers
 

src/crewai/utilities/events/event_listener.py

@@ -16,7 +16,13 @@ from crewai.utilities.events.llm_events import (
 )
 from crewai.utilities.events.utils.console_formatter import ConsoleFormatter
 
-from .agent_events import AgentExecutionCompletedEvent, AgentExecutionStartedEvent
+from .agent_events import (
+    AgentExecutionCompletedEvent,
+    AgentExecutionStartedEvent,
+    LiteAgentExecutionCompletedEvent,
+    LiteAgentExecutionErrorEvent,
+    LiteAgentExecutionStartedEvent,
+)
 from .crew_events import (
     CrewKickoffCompletedEvent,
     CrewKickoffFailedEvent,
@@ -65,7 +71,7 @@ class EventListener(BaseEventListener):
             self._telemetry.set_tracer()
             self.execution_spans = {}
             self._initialized = True
-            self.formatter = ConsoleFormatter()
+            self.formatter = ConsoleFormatter(verbose=True)
 
     # ----------- CREW EVENTS -----------
 
@@ -171,6 +177,36 @@ class EventListener(BaseEventListener):
                 self.formatter.current_crew_tree,
             )
 
+        # ----------- LITE AGENT EVENTS -----------
+
+        @crewai_event_bus.on(LiteAgentExecutionStartedEvent)
+        def on_lite_agent_execution_started(
+            source, event: LiteAgentExecutionStartedEvent
+        ):
+            """Handle LiteAgent execution started event."""
+            self.formatter.handle_lite_agent_execution(
+                event.agent_info["role"], status="started", **event.agent_info
+            )
+
+        @crewai_event_bus.on(LiteAgentExecutionCompletedEvent)
+        def on_lite_agent_execution_completed(
+            source, event: LiteAgentExecutionCompletedEvent
+        ):
+            """Handle LiteAgent execution completed event."""
+            self.formatter.handle_lite_agent_execution(
+                event.agent_info["role"], status="completed", **event.agent_info
+            )
+
+        @crewai_event_bus.on(LiteAgentExecutionErrorEvent)
+        def on_lite_agent_execution_error(source, event: LiteAgentExecutionErrorEvent):
+            """Handle LiteAgent execution error event."""
+            self.formatter.handle_lite_agent_execution(
+                event.agent_info["role"],
+                status="failed",
+                error=event.error,
+                **event.agent_info,
+            )
+
         # ----------- FLOW EVENTS -----------
 
         @crewai_event_bus.on(FlowCreatedEvent)

src/crewai/utilities/events/flow_events.py

@@ -2,10 +2,10 @@ from typing import Any, Dict, Optional, Union
 
 from pydantic import BaseModel, ConfigDict
 
-from .base_events import CrewEvent
+from .base_events import BaseEvent
 
 
-class FlowEvent(CrewEvent):
+class FlowEvent(BaseEvent):
     """Base class for all flow events"""
 
     type: str

src/crewai/utilities/events/llm_events.py

@@ -1,7 +1,9 @@
 from enum import Enum
 from typing import Any, Dict, List, Optional, Union
 
-from crewai.utilities.events.base_events import CrewEvent
+from pydantic import BaseModel
+
+from crewai.utilities.events.base_events import BaseEvent
 
 
 class LLMCallType(Enum):
@@ -11,7 +13,7 @@ class LLMCallType(Enum):
     LLM_CALL = "llm_call"
 
 
-class LLMCallStartedEvent(CrewEvent):
+class LLMCallStartedEvent(BaseEvent):
     """Event emitted when a LLM call starts
 
     Attributes:
@@ -26,7 +28,7 @@ class LLMCallStartedEvent(CrewEvent):
     available_functions: Optional[Dict[str, Any]] = None
 
 
-class LLMCallCompletedEvent(CrewEvent):
+class LLMCallCompletedEvent(BaseEvent):
     """Event emitted when a LLM call completes"""
 
     type: str = "llm_call_completed"
@@ -34,15 +36,28 @@ class LLMCallCompletedEvent(CrewEvent):
     call_type: LLMCallType
 
 
-class LLMCallFailedEvent(CrewEvent):
+class LLMCallFailedEvent(BaseEvent):
     """Event emitted when a LLM call fails"""
 
     error: str
     type: str = "llm_call_failed"
 
 
-class LLMStreamChunkEvent(CrewEvent):
+class FunctionCall(BaseModel):
+    arguments: str
+    name: Optional[str] = None
+
+
+class ToolCall(BaseModel):
+    id: Optional[str] = None
+    function: FunctionCall
+    type: Optional[str] = None
+    index: int
+
+
+class LLMStreamChunkEvent(BaseEvent):
     """Event emitted when a streaming chunk is received"""
 
     type: str = "llm_stream_chunk"
     chunk: str
+    tool_call: Optional[ToolCall] = None

src/crewai/utilities/events/task_events.py

@@ -1,32 +1,84 @@
-from typing import Optional
+from typing import Any, Optional
 
 from crewai.tasks.task_output import TaskOutput
-from crewai.utilities.events.base_events import CrewEvent
+from crewai.utilities.events.base_events import BaseEvent
 
 
-class TaskStartedEvent(CrewEvent):
+class TaskStartedEvent(BaseEvent):
     """Event emitted when a task starts"""
 
     type: str = "task_started"
     context: Optional[str]
+    task: Optional[Any] = None
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Set fingerprint data from the task
+        if hasattr(self.task, "fingerprint") and self.task.fingerprint:
+            self.source_fingerprint = self.task.fingerprint.uuid_str
+            self.source_type = "task"
+            if (
+                hasattr(self.task.fingerprint, "metadata")
+                and self.task.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.task.fingerprint.metadata
 
 
-class TaskCompletedEvent(CrewEvent):
+class TaskCompletedEvent(BaseEvent):
     """Event emitted when a task completes"""
 
     output: TaskOutput
     type: str = "task_completed"
+    task: Optional[Any] = None
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Set fingerprint data from the task
+        if hasattr(self.task, "fingerprint") and self.task.fingerprint:
+            self.source_fingerprint = self.task.fingerprint.uuid_str
+            self.source_type = "task"
+            if (
+                hasattr(self.task.fingerprint, "metadata")
+                and self.task.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.task.fingerprint.metadata
 
 
-class TaskFailedEvent(CrewEvent):
+class TaskFailedEvent(BaseEvent):
     """Event emitted when a task fails"""
 
     error: str
     type: str = "task_failed"
+    task: Optional[Any] = None
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Set fingerprint data from the task
+        if hasattr(self.task, "fingerprint") and self.task.fingerprint:
+            self.source_fingerprint = self.task.fingerprint.uuid_str
+            self.source_type = "task"
+            if (
+                hasattr(self.task.fingerprint, "metadata")
+                and self.task.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.task.fingerprint.metadata
 
 
-class TaskEvaluationEvent(CrewEvent):
+class TaskEvaluationEvent(BaseEvent):
     """Event emitted when a task evaluation is completed"""
 
     type: str = "task_evaluation"
     evaluation_type: str
+    task: Optional[Any] = None
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Set fingerprint data from the task
+        if hasattr(self.task, "fingerprint") and self.task.fingerprint:
+            self.source_fingerprint = self.task.fingerprint.uuid_str
+            self.source_type = "task"
+            if (
+                hasattr(self.task.fingerprint, "metadata")
+                and self.task.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.task.fingerprint.metadata

src/crewai/utilities/events/tool_usage_events.py

@@ -1,10 +1,10 @@
 from datetime import datetime
-from typing import Any, Callable, Dict
+from typing import Any, Callable, Dict, Optional
 
-from .base_events import CrewEvent
+from .base_events import BaseEvent
 
 
-class ToolUsageEvent(CrewEvent):
+class ToolUsageEvent(BaseEvent):
     """Base event for tool usage tracking"""
 
     agent_key: str
@@ -14,9 +14,22 @@ class ToolUsageEvent(CrewEvent):
     tool_class: str
     run_attempts: int | None = None
     delegations: int | None = None
+    agent: Optional[Any] = None
 
     model_config = {"arbitrary_types_allowed": True}
 
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Set fingerprint data from the agent
+        if self.agent and hasattr(self.agent, "fingerprint") and self.agent.fingerprint:
+            self.source_fingerprint = self.agent.fingerprint.uuid_str
+            self.source_type = "agent"
+            if (
+                hasattr(self.agent.fingerprint, "metadata")
+                and self.agent.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.agent.fingerprint.metadata
+
 
 class ToolUsageStartedEvent(ToolUsageEvent):
     """Event emitted when a tool execution is started"""
@@ -55,7 +68,7 @@ class ToolSelectionErrorEvent(ToolUsageEvent):
     type: str = "tool_selection_error"
 
 
-class ToolExecutionErrorEvent(CrewEvent):
+class ToolExecutionErrorEvent(BaseEvent):
     """Event emitted when a tool execution encounters an error"""
 
     error: Any
@@ -63,3 +76,16 @@ class ToolExecutionErrorEvent(CrewEvent):
     tool_name: str
     tool_args: Dict[str, Any]
     tool_class: Callable
+    agent: Optional[Any] = None
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        # Set fingerprint data from the agent
+        if self.agent and hasattr(self.agent, "fingerprint") and self.agent.fingerprint:
+            self.source_fingerprint = self.agent.fingerprint.uuid_str
+            self.source_type = "agent"
+            if (
+                hasattr(self.agent.fingerprint, "metadata")
+                and self.agent.fingerprint.metadata
+            ):
+                self.fingerprint_metadata = self.agent.fingerprint.metadata

src/crewai/utilities/events/utils/console_formatter.py

@@ -1,4 +1,4 @@
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
 
 from rich.console import Console
 from rich.panel import Panel
@@ -13,6 +13,7 @@ class ConsoleFormatter:
     current_tool_branch: Optional[Tree] = None
     current_flow_tree: Optional[Tree] = None
     current_method_branch: Optional[Tree] = None
+    current_lite_agent_branch: Optional[Tree] = None
     tool_usage_counts: Dict[str, int] = {}
 
     def __init__(self, verbose: bool = False):
@@ -390,21 +391,24 @@ class ConsoleFormatter:
         crew_tree: Optional[Tree],
     ) -> Optional[Tree]:
         """Handle tool usage started event."""
-        if not self.verbose or agent_branch is None or crew_tree is None:
+        if not self.verbose:
+            return None
+
+        # Use LiteAgent branch if available, otherwise use regular agent branch
+        branch_to_use = self.current_lite_agent_branch or agent_branch
+        tree_to_use = branch_to_use or crew_tree
+
+        if branch_to_use is None or tree_to_use is None:
             return None
 
         # Update tool usage count
         self.tool_usage_counts[tool_name] = self.tool_usage_counts.get(tool_name, 0) + 1
 
-        # Find existing tool node or create new one
+        # Find or create tool node
-        tool_branch = None
+        tool_branch = self.current_tool_branch
-        for child in agent_branch.children:
+        if tool_branch is None:
-            if tool_name in str(child.label):
+            tool_branch = branch_to_use.add("")
-                tool_branch = child
+            self.current_tool_branch = tool_branch
-                break
-
-        if not tool_branch:
-            tool_branch = agent_branch.add("")
 
         # Update label with current count
         self.update_tree_label(
@@ -414,12 +418,11 @@ class ConsoleFormatter:
             "yellow",
         )
 
-        self.print(crew_tree)
+        # Only print if this is a new tool usage
+        if tool_branch not in branch_to_use.children:
+            self.print(tree_to_use)
             self.print()
 
-        # Set the current_tool_branch attribute directly
-        self.current_tool_branch = tool_branch
-
         return tool_branch
 
     def handle_tool_usage_finished(
@@ -429,16 +432,28 @@ class ConsoleFormatter:
         crew_tree: Optional[Tree],
     ) -> None:
         """Handle tool usage finished event."""
-        if not self.verbose or tool_branch is None or crew_tree is None:
+        if not self.verbose or tool_branch is None:
             return
 
+        # Use LiteAgent branch if available, otherwise use crew tree
+        tree_to_use = self.current_lite_agent_branch or crew_tree
+        if tree_to_use is None:
+            return
+
+        # Update the existing tool node's label
         self.update_tree_label(
             tool_branch,
             "🔧",
             f"Used {tool_name} ({self.tool_usage_counts[tool_name]})",
             "green",
         )
-        self.print(crew_tree)
+
+        # Clear the current tool branch as we're done with it
+        self.current_tool_branch = None
+
+        # Only print if we have a valid tree and the tool node is still in it
+        if isinstance(tree_to_use, Tree) and tool_branch in tree_to_use.children:
+            self.print(tree_to_use)
             self.print()
 
     def handle_tool_usage_error(
@@ -452,6 +467,9 @@ class ConsoleFormatter:
         if not self.verbose:
             return
 
+        # Use LiteAgent branch if available, otherwise use crew tree
+        tree_to_use = self.current_lite_agent_branch or crew_tree
+
         if tool_branch:
             self.update_tree_label(
                 tool_branch,
@@ -459,7 +477,8 @@ class ConsoleFormatter:
                 f"{tool_name} ({self.tool_usage_counts[tool_name]})",
                 "red",
             )
-            self.print(crew_tree)
+            if tree_to_use:
+                self.print(tree_to_use)
                 self.print()
 
         # Show error panel
@@ -474,19 +493,23 @@ class ConsoleFormatter:
         crew_tree: Optional[Tree],
     ) -> Optional[Tree]:
         """Handle LLM call started event."""
-        if not self.verbose or agent_branch is None or crew_tree is None:
+        if not self.verbose:
             return None
 
-        # Only add thinking status if it doesn't exist
+        # Use LiteAgent branch if available, otherwise use regular agent branch
-        if not any("Thinking" in str(child.label) for child in agent_branch.children):
+        branch_to_use = self.current_lite_agent_branch or agent_branch
-            tool_branch = agent_branch.add("")
+        tree_to_use = branch_to_use or crew_tree
+
+        if branch_to_use is None or tree_to_use is None:
+            return None
+
+        # Only add thinking status if we don't have a current tool branch
+        if self.current_tool_branch is None:
+            tool_branch = branch_to_use.add("")
             self.update_tree_label(tool_branch, "🧠", "Thinking...", "blue")
-            self.print(crew_tree)
-            self.print()
-
-            # Set the current_tool_branch attribute directly
             self.current_tool_branch = tool_branch
-
+            self.print(tree_to_use)
+            self.print()
             return tool_branch
         return None
 
@@ -497,20 +520,27 @@ class ConsoleFormatter:
         crew_tree: Optional[Tree],
     ) -> None:
         """Handle LLM call completed event."""
-        if (
+        if not self.verbose or tool_branch is None:
-            not self.verbose
-            or tool_branch is None
-            or agent_branch is None
-            or crew_tree is None
-        ):
             return
 
-        # Remove the thinking status node when complete
+        # Use LiteAgent branch if available, otherwise use regular agent branch
+        branch_to_use = self.current_lite_agent_branch or agent_branch
+        tree_to_use = branch_to_use or crew_tree
+
+        if branch_to_use is None or tree_to_use is None:
+            return
+
+        # Remove the thinking status node when complete, but only if it exists
         if "Thinking" in str(tool_branch.label):
-            if tool_branch in agent_branch.children:
+            try:
-                agent_branch.children.remove(tool_branch)
+                # Check if the node is actually in the children list
-                self.print(crew_tree)
+                if tool_branch in branch_to_use.children:
+                    branch_to_use.children.remove(tool_branch)
+                    self.print(tree_to_use)
                     self.print()
+            except Exception:
+                # If any error occurs during removal, just continue without removing
+                pass
 
     def handle_llm_call_failed(
         self, tool_branch: Optional[Tree], error: str, crew_tree: Optional[Tree]
@@ -519,10 +549,14 @@ class ConsoleFormatter:
         if not self.verbose:
             return
 
+        # Use LiteAgent branch if available, otherwise use crew tree
+        tree_to_use = self.current_lite_agent_branch or crew_tree
+
         # Update tool branch if it exists
         if tool_branch:
             tool_branch.label = Text("❌ LLM Failed", style="red bold")
-            self.print(crew_tree)
+            if tree_to_use:
+                self.print(tree_to_use)
                 self.print()
 
         # Show error panel
@@ -658,3 +692,94 @@ class ConsoleFormatter:
 
         self.print_panel(failure_content, "Test Failure", "red")
         self.print()
+
+    def create_lite_agent_branch(self, lite_agent_role: str) -> Optional[Tree]:
+        """Create and initialize a lite agent branch."""
+        if not self.verbose:
+            return None
+
+        # Create initial tree for LiteAgent if it doesn't exist
+        if not self.current_lite_agent_branch:
+            lite_agent_label = Text()
+            lite_agent_label.append("🤖 LiteAgent: ", style="cyan bold")
+            lite_agent_label.append(lite_agent_role, style="cyan")
+            lite_agent_label.append("\n    Status: ", style="white")
+            lite_agent_label.append("In Progress", style="yellow")
+
+            lite_agent_tree = Tree(lite_agent_label)
+            self.current_lite_agent_branch = lite_agent_tree
+            self.print(lite_agent_tree)
+            self.print()
+
+        return self.current_lite_agent_branch
+
+    def update_lite_agent_status(
+        self,
+        lite_agent_branch: Optional[Tree],
+        lite_agent_role: str,
+        status: str = "completed",
+        **fields: Dict[str, Any],
+    ) -> None:
+        """Update lite agent status in the tree."""
+        if not self.verbose or lite_agent_branch is None:
+            return
+
+        # Determine style based on status
+        if status == "completed":
+            prefix, style = "✅ LiteAgent:", "green"
+            status_text = "Completed"
+            title = "LiteAgent Completion"
+        elif status == "failed":
+            prefix, style = "❌ LiteAgent:", "red"
+            status_text = "Failed"
+            title = "LiteAgent Error"
+        else:
+            prefix, style = "🤖 LiteAgent:", "yellow"
+            status_text = "In Progress"
+            title = "LiteAgent Status"
+
+        # Update the tree label
+        lite_agent_label = Text()
+        lite_agent_label.append(f"{prefix} ", style=f"{style} bold")
+        lite_agent_label.append(lite_agent_role, style=style)
+        lite_agent_label.append("\n    Status: ", style="white")
+        lite_agent_label.append(status_text, style=f"{style} bold")
+        lite_agent_branch.label = lite_agent_label
+
+        self.print(lite_agent_branch)
+        self.print()
+
+        # Show status panel if additional fields are provided
+        if fields:
+            content = self.create_status_content(
+                f"LiteAgent {status.title()}", lite_agent_role, style, **fields
+            )
+            self.print_panel(content, title, style)
+
+    def handle_lite_agent_execution(
+        self,
+        lite_agent_role: str,
+        status: str = "started",
+        error: Any = None,
+        **fields: Dict[str, Any],
+    ) -> None:
+        """Handle lite agent execution events with consistent formatting."""
+        if not self.verbose:
+            return
+
+        if status == "started":
+            # Create or get the LiteAgent branch
+            lite_agent_branch = self.create_lite_agent_branch(lite_agent_role)
+            if lite_agent_branch and fields:
+                # Show initial status panel
+                content = self.create_status_content(
+                    "LiteAgent Session Started", lite_agent_role, "cyan", **fields
+                )
+                self.print_panel(content, "LiteAgent Started", "cyan")
+        else:
+            # Update existing LiteAgent branch
+            if error:
+                fields["Error"] = error
+            self.update_lite_agent_status(
+                self.current_lite_agent_branch, lite_agent_role, status, **fields
+            )

src/crewai/utilities/prompts.py

@@ -9,7 +9,7 @@ class Prompts(BaseModel):
     """Manages and generates prompts for a generic agent."""
 
     i18n: I18N = Field(default=I18N())
-    tools: list[Any] = Field(default=[])
+    has_tools: bool = False
     system_template: Optional[str] = None
     prompt_template: Optional[str] = None
     response_template: Optional[str] = None
@@ -19,7 +19,7 @@ class Prompts(BaseModel):
     def task_execution(self) -> dict[str, str]:
         """Generate a standard prompt for task execution."""
         slices = ["role_playing"]
-        if len(self.tools) > 0:
+        if self.has_tools:
             slices.append("tools")
         else:
             slices.append("no_tools")

src/crewai/utilities/serialization.py

@@ -5,35 +5,17 @@ from typing import Any, Dict, List, Union
 
 from pydantic import BaseModel
 
-from crewai.flow import Flow
-
 SerializablePrimitive = Union[str, int, float, bool, None]
 Serializable = Union[
     SerializablePrimitive, List["Serializable"], Dict[str, "Serializable"]
 ]
 
 
-def export_state(flow: Flow) -> dict[str, Serializable]:
-    """Exports the Flow's internal state as JSON-compatible data structures.
-
-    Performs a one-way transformation of a Flow's state into basic Python types
-    that can be safely serialized to JSON. To prevent infinite recursion with
-    circular references, the conversion is limited to a depth of 5 levels.
-
-    Args:
-        flow: The Flow object whose state needs to be exported
-
-    Returns:
-        dict[str, Any]: The transformed state using JSON-compatible Python
-            types.
-    """
-    result = to_serializable(flow._state)
-    assert isinstance(result, dict)
-    return result
-
-
 def to_serializable(
-    obj: Any, exclude: set[str] | None = None, max_depth: int = 5, _current_depth: int = 0
+    obj: Any,
+    exclude: set[str] | None = None,
+    max_depth: int = 5,
+    _current_depth: int = 0,
 ) -> Serializable:
     """Converts a Python object into a JSON-compatible representation.