fix: ignore expected security warnings in subprocess_utils.py

- Add per-file-ignores for S602/S603 in subprocess_utils.py
- These warnings are expected for Windows shell=True compatibility fix
- Resolves remaining CI lint failures for Windows subprocess fix

Co-Authored-By: João <joao@crewai.com>

.github/workflows/tests.yml

@@ -22,6 +22,8 @@ jobs:
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
+        with:
+          fetch-depth: 0  # Fetch all history for proper diff
 
       - name: Restore global uv cache
         id: cache-restore
@@ -45,14 +47,41 @@ jobs:
       - name: Install the project
         run: uv sync --all-groups --all-extras
 
+      - name: Restore test durations
+        uses: actions/cache/restore@v4
+        with:
+          path: .test_durations_py*
+          key: test-durations-py${{ matrix.python-version }}
+
       - name: Run tests (group ${{ matrix.group }} of 8)
         run: |
+          PYTHON_VERSION_SAFE=$(echo "${{ matrix.python-version }}" | tr '.' '_')
+          DURATION_FILE=".test_durations_py${PYTHON_VERSION_SAFE}"
+          
+          # Temporarily always skip cached durations to fix test splitting
+          # When durations don't match, pytest-split runs duplicate tests instead of splitting
+          echo "Using even test splitting (duration cache disabled until fix merged)"
+          DURATIONS_ARG=""
+          
+          # Original logic (disabled temporarily):
+          # if [ ! -f "$DURATION_FILE" ]; then
+          #   echo "No cached durations found, tests will be split evenly"
+          #   DURATIONS_ARG=""
+          # elif git diff origin/${{ github.base_ref }}...HEAD --name-only 2>/dev/null | grep -q "^tests/.*\.py$"; then
+          #   echo "Test files have changed, skipping cached durations to avoid mismatches"
+          #   DURATIONS_ARG=""
+          # else
+          #   echo "No test changes detected, using cached test durations for optimal splitting"
+          #   DURATIONS_ARG="--durations-path=${DURATION_FILE}"
+          # fi
+          
           uv run pytest \
             --block-network \
             --timeout=30 \
             -vv \
             --splits 8 \
             --group ${{ matrix.group }} \
+            $DURATIONS_ARG \
             --durations=10 \
             -n auto \
             --maxfail=3

.github/workflows/update-test-durations.yml

@@ -0,0 +1,71 @@
+name: Update Test Durations
+
+on:
+  push:
+    branches:
+      - main
+    paths:
+      - 'tests/**/*.py'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  update-durations:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ['3.10', '3.11', '3.12', '3.13']
+    env:
+      OPENAI_API_KEY: fake-api-key
+      PYTHONUNBUFFERED: 1
+    
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Restore global uv cache
+        id: cache-restore
+        uses: actions/cache/restore@v4
+        with:
+          path: |
+            ~/.cache/uv
+            ~/.local/share/uv
+            .venv
+          key: uv-main-py${{ matrix.python-version }}-${{ hashFiles('uv.lock') }}
+          restore-keys: |
+            uv-main-py${{ matrix.python-version }}-
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v6
+        with:
+          version: "0.8.4"
+          python-version: ${{ matrix.python-version }}
+          enable-cache: false
+
+      - name: Install the project
+        run: uv sync --all-groups --all-extras
+
+      - name: Run all tests and store durations
+        run: |
+          PYTHON_VERSION_SAFE=$(echo "${{ matrix.python-version }}" | tr '.' '_')
+          uv run pytest --store-durations --durations-path=.test_durations_py${PYTHON_VERSION_SAFE} -n auto
+        continue-on-error: true
+
+      - name: Save durations to cache
+        if: always()
+        uses: actions/cache/save@v4
+        with:
+          path: .test_durations_py*
+          key: test-durations-py${{ matrix.python-version }}
+
+      - name: Save uv caches
+        if: steps.cache-restore.outputs.cache-hit != 'true'
+        uses: actions/cache/save@v4
+        with:
+          path: |
+            ~/.cache/uv
+            ~/.local/share/uv
+            .venv
+          key: uv-main-py${{ matrix.python-version }}-${{ hashFiles('uv.lock') }}

pyproject.toml

@@ -135,6 +135,7 @@ ignore = ["E501"]  # ignore line too long
 
 [tool.ruff.lint.per-file-ignores]
 "tests/**/*.py" = ["S101"]  # Allow assert statements in tests
+"src/crewai/cli/subprocess_utils.py" = ["S602", "S603"]  # Allow shell=True for Windows compatibility
 
 [tool.mypy]
 exclude = ["src/crewai/cli/templates", "tests"]

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

@@ -1,47 +1,56 @@
-from typing import Any, Dict, List, Optional
+"""LangGraph agent adapter for CrewAI integration.
 
-from pydantic import Field, PrivateAttr
+This module contains the LangGraphAgentAdapter class that integrates LangGraph ReAct agents
+with CrewAI's agent system. Provides memory persistence, tool integration, and structured
+output functionality.
+"""
+
+from collections.abc import Callable
+from typing import Any, cast
+
+from pydantic import ConfigDict, 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.protocols import (
+    LangGraphCheckPointMemoryModule,
+    LangGraphPrebuiltModule,
+)
 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.events.event_bus import crewai_event_bus
 from crewai.events.types.agent_events import (
     AgentExecutionCompletedEvent,
     AgentExecutionErrorEvent,
     AgentExecutionStartedEvent,
 )
-
-try:
-    from langgraph.checkpoint.memory import MemorySaver
-    from langgraph.prebuilt import create_react_agent
-
-    LANGGRAPH_AVAILABLE = True
-except ImportError:
-    LANGGRAPH_AVAILABLE = False
+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.import_utils import require
 
 
 class LangGraphAgentAdapter(BaseAgentAdapter):
-    """Adapter for LangGraph agents to work with CrewAI."""
+    """Adapter for LangGraph agents to work with CrewAI.
 
-    model_config = {"arbitrary_types_allowed": True}
+    This adapter integrates LangGraph's ReAct agents with CrewAI's agent system,
+    providing memory persistence, tool integration, and structured output support.
+    """
 
-    _logger: Logger = PrivateAttr(default_factory=lambda: Logger())
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    _logger: Logger = PrivateAttr(default_factory=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)
+    step_callback: Callable[..., Any] | None = Field(default=None)
 
     model: str = Field(default="gpt-4o")
     verbose: bool = Field(default=False)
@@ -51,17 +60,24 @@ class LangGraphAgentAdapter(BaseAgentAdapter):
         role: str,
         goal: str,
         backstory: str,
-        tools: Optional[List[BaseTool]] = None,
+        tools: list[BaseTool] | None = None,
         llm: Any = None,
         max_iterations: int = 10,
-        agent_config: Optional[Dict[str, Any]] = None,
+        agent_config: dict[str, Any] | None = 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`"
-            )
+    ) -> None:
+        """Initialize the LangGraph agent adapter.
+
+        Args:
+            role: The role description for the agent.
+            goal: The primary goal the agent should achieve.
+            backstory: Background information about the agent.
+            tools: Optional list of tools available to the agent.
+            llm: Language model to use, defaults to gpt-4o.
+            max_iterations: Maximum number of iterations for task execution.
+            agent_config: Additional configuration for the LangGraph agent.
+            **kwargs: Additional arguments passed to the base adapter.
+        """
         super().__init__(
             role=role,
             goal=goal,
@@ -72,46 +88,65 @@ class LangGraphAgentAdapter(BaseAgentAdapter):
             **kwargs,
         )
         self._tool_adapter = LangGraphToolAdapter(tools=tools)
-        self._converter_adapter = LangGraphConverterAdapter(self)
+        self._converter_adapter: LangGraphConverterAdapter = 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()
+        """Set up the LangGraph workflow graph.
 
-            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,
-                )
+        Initializes the memory saver and creates a ReAct agent with the configured
+        tools, memory checkpointer, and debug settings.
+        """
 
-        except ImportError as e:
-            self._logger.log(
-                "error", f"Failed to import LangGraph dependencies: {str(e)}"
+        memory_saver: type[Any] = cast(
+            LangGraphCheckPointMemoryModule,
+            require(
+                "langgraph.checkpoint.memory",
+                purpose="LangGraph core functionality",
+            ),
+        ).MemorySaver
+        create_react_agent: Callable[..., Any] = cast(
+            LangGraphPrebuiltModule,
+            require(
+                "langgraph.prebuilt",
+                purpose="LangGraph core functionality",
+            ),
+        ).create_react_agent
+
+        self._memory = memory_saver()
+
+        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,
             )
-            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."""
+        """Build a system prompt for the LangGraph agent.
+
+        Creates a prompt that includes the agent's role, goal, and backstory,
+        then enhances it through the converter adapter for structured output.
+
+        Returns:
+            The complete system prompt string.
+        """
         base_prompt = f"""
             You are {self.role}.
-        
+
             Your goal is: {self.goal}
 
             Your backstory: {self.backstory}
@@ -123,10 +158,25 @@ class LangGraphAgentAdapter(BaseAgentAdapter):
     def execute_task(
         self,
         task: Any,
-        context: Optional[str] = None,
-        tools: Optional[List[BaseTool]] = None,
+        context: str | None = None,
+        tools: list[BaseTool] | None = None,
     ) -> str:
-        """Execute a task using the LangGraph workflow."""
+        """Execute a task using the LangGraph workflow.
+
+        Configures the agent, processes the task through the LangGraph workflow,
+        and handles event emission for execution tracking.
+
+        Args:
+            task: The task object to execute.
+            context: Optional context information for the task.
+            tools: Optional additional tools for this specific execution.
+
+        Returns:
+            The final answer from the task execution.
+
+        Raises:
+            Exception: If task execution fails.
+        """
         self.create_agent_executor(tools)
 
         self.configure_structured_output(task)
@@ -151,9 +201,11 @@ class LangGraphAgentAdapter(BaseAgentAdapter):
 
             session_id = f"task_{id(task)}"
 
-            config = {"configurable": {"thread_id": session_id}}
+            config: dict[str, dict[str, str]] = {
+                "configurable": {"thread_id": session_id}
+            }
 
-            result = self._graph.invoke(
+            result: dict[str, Any] = self._graph.invoke(
                 {
                     "messages": [
                         ("system", self._build_system_prompt()),
@@ -163,10 +215,10 @@ class LangGraphAgentAdapter(BaseAgentAdapter):
                 config,
             )
 
-            messages = result.get("messages", [])
-            last_message = messages[-1] if messages else None
+            messages: list[Any] = result.get("messages", [])
+            last_message: Any = messages[-1] if messages else None
 
-            final_answer = ""
+            final_answer: str = ""
             if isinstance(last_message, dict):
                 final_answer = last_message.get("content", "")
             elif hasattr(last_message, "content"):
@@ -186,7 +238,7 @@ class LangGraphAgentAdapter(BaseAgentAdapter):
             return final_answer
 
         except Exception as e:
-            self._logger.log("error", f"Error executing LangGraph task: {str(e)}")
+            self._logger.log("error", f"Error executing LangGraph task: {e!s}")
             crewai_event_bus.emit(
                 self,
                 event=AgentExecutionErrorEvent(
@@ -197,29 +249,67 @@ class LangGraphAgentAdapter(BaseAgentAdapter):
             )
             raise
 
-    def create_agent_executor(self, tools: Optional[List[BaseTool]] = None) -> None:
-        """Configure the LangGraph agent for execution."""
+    def create_agent_executor(self, tools: list[BaseTool] | None = None) -> None:
+        """Configure the LangGraph agent for execution.
+
+        Args:
+            tools: Optional tools to configure for the agent.
+        """
         self.configure_tools(tools)
 
-    def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
-        """Configure tools for the LangGraph agent."""
+    def configure_tools(self, tools: list[BaseTool] | None = None) -> None:
+        """Configure tools for the LangGraph agent.
+
+        Merges additional tools with existing ones and updates the graph's
+        available tools through the tool adapter.
+
+        Args:
+            tools: Optional additional tools to configure.
+        """
         if tools:
-            all_tools = list(self.tools or []) + list(tools or [])
+            all_tools: list[BaseTool] = list(self.tools or []) + list(tools or [])
             self._tool_adapter.configure_tools(all_tools)
-            available_tools = self._tool_adapter.tools()
+            available_tools: list[Any] = 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)
+    def get_delegation_tools(self, agents: list[BaseAgent]) -> list[BaseTool]:
+        """Implement delegation tools support for LangGraph.
+
+        Creates delegation tools that allow this agent to delegate tasks to other agents.
+
+        Args:
+            agents: List of agents available for delegation.
+
+        Returns:
+            List of delegation tools.
+        """
+        agent_tools: AgentTools = AgentTools(agents=agents)
         return agent_tools.tools()
 
+    @staticmethod
     def get_output_converter(
-        self, llm: Any, text: str, model: Any, instructions: str
-    ) -> Any:
-        """Convert output format if needed."""
+        llm: Any, text: str, model: Any, instructions: str
+    ) -> Converter:
+        """Convert output format if needed.
+
+        Args:
+            llm: Language model instance.
+            text: Text to convert.
+            model: Model configuration.
+            instructions: Conversion instructions.
+
+        Returns:
+            Converter instance for output transformation.
+        """
         return Converter(llm=llm, text=text, model=model, instructions=instructions)
 
-    def configure_structured_output(self, task) -> None:
-        """Configure the structured output for LangGraph."""
+    def configure_structured_output(self, task: Any) -> None:
+        """Configure the structured output for LangGraph.
+
+        Uses the converter adapter to set up structured output formatting
+        based on the task requirements.
+
+        Args:
+            task: Task object containing output requirements.
+        """
         self._converter_adapter.configure_structured_output(task)

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

@@ -1,38 +1,72 @@
+"""LangGraph tool adapter for CrewAI tool integration.
+
+This module contains the LangGraphToolAdapter class that converts CrewAI tools
+to LangGraph-compatible format using langchain_core.tools.
+"""
+
 import inspect
-from typing import Any, List, Optional
+from collections.abc import Awaitable
+from typing import Any
 
 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"""
+    """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 = []
+    Converts CrewAI BaseTool instances to langchain_core.tools format
+    that can be used by LangGraph agents.
+    """
 
-    def configure_tools(self, tools: List[BaseTool]) -> None:
+    def __init__(self, tools: list[BaseTool] | None = None) -> None:
+        """Initialize the tool adapter.
+
+        Args:
+            tools: Optional list of CrewAI tools to adapt.
         """
-        Configure and convert CrewAI tools to LangGraph-compatible format.
-        LangGraph expects tools in langchain_core.tools format.
-        """
-        from langchain_core.tools import BaseTool, StructuredTool
+        super().__init__()
+        self.original_tools: list[BaseTool] = tools or []
+        self.converted_tools: list[Any] = []
 
-        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. This method
+        converts CrewAI BaseTool instances to StructuredTool instances.
+
+        Args:
+            tools: List of CrewAI tools to convert.
+        """
+        from langchain_core.tools import BaseTool as LangChainBaseTool
+        from langchain_core.tools import StructuredTool
+
+        converted_tools: list[Any] = []
         if self.original_tools:
-            all_tools = tools + self.original_tools
+            all_tools: list[BaseTool] = tools + self.original_tools
         else:
             all_tools = tools
         for tool in all_tools:
-            if isinstance(tool, BaseTool):
+            if isinstance(tool, LangChainBaseTool):
                 converted_tools.append(tool)
                 continue
 
-            sanitized_name = self.sanitize_tool_name(tool.name)
+            sanitized_name: str = self.sanitize_tool_name(tool.name)
 
-            async def tool_wrapper(*args, tool=tool, **kwargs):
-                output = None
+            async def tool_wrapper(
+                *args: Any, tool: BaseTool = tool, **kwargs: Any
+            ) -> Any:
+                """Wrapper function to adapt CrewAI tool calls to LangGraph format.
+
+                Args:
+                    *args: Positional arguments for the tool.
+                    tool: The CrewAI tool to wrap.
+                    **kwargs: Keyword arguments for the tool.
+
+                Returns:
+                    The result from the tool execution.
+                """
+                output: Any | Awaitable[Any]
                 if len(args) > 0 and isinstance(args[0], str):
                     output = tool.run(args[0])
                 elif "input" in kwargs:
@@ -41,12 +75,12 @@ class LangGraphToolAdapter(BaseToolAdapter):
                     output = tool.run(**kwargs)
 
                 if inspect.isawaitable(output):
-                    result = await output
+                    result: Any = await output
                 else:
                     result = output
                 return result
 
-            converted_tool = StructuredTool(
+            converted_tool: StructuredTool = StructuredTool(
                 name=sanitized_name,
                 description=tool.description,
                 func=tool_wrapper,
@@ -57,5 +91,10 @@ class LangGraphToolAdapter(BaseToolAdapter):
 
         self.converted_tools = converted_tools
 
-    def tools(self) -> List[Any]:
+    def tools(self) -> list[Any]:
+        """Get the list of converted tools.
+
+        Returns:
+            List of LangGraph-compatible tools.
+        """
         return self.converted_tools or []

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

@@ -0,0 +1,55 @@
+"""Type protocols for LangGraph modules."""
+
+from typing import Any, Protocol, runtime_checkable
+
+
+@runtime_checkable
+class LangGraphMemorySaver(Protocol):
+    """Protocol for LangGraph MemorySaver.
+
+    Defines the interface for LangGraph's memory persistence mechanism.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the memory saver."""
+        ...
+
+
+@runtime_checkable
+class LangGraphCheckPointMemoryModule(Protocol):
+    """Protocol for LangGraph checkpoint memory module.
+
+    Defines the interface for modules containing memory checkpoint functionality.
+    """
+
+    MemorySaver: type[LangGraphMemorySaver]
+
+
+@runtime_checkable
+class LangGraphPrebuiltModule(Protocol):
+    """Protocol for LangGraph prebuilt module.
+
+    Defines the interface for modules containing prebuilt agent factories.
+    """
+
+    def create_react_agent(
+        self,
+        model: Any,
+        tools: list[Any],
+        checkpointer: Any,
+        debug: bool = False,
+        **kwargs: Any,
+    ) -> Any:
+        """Create a ReAct agent with the given configuration.
+
+        Args:
+            model: The language model to use for the agent.
+            tools: List of tools available to the agent.
+            checkpointer: Memory checkpointer for state persistence.
+            debug: Whether to enable debug mode.
+            **kwargs: Additional configuration options.
+
+        Returns:
+            The configured ReAct agent instance.
+        """
+        ...

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

@@ -1,21 +1,45 @@
+"""LangGraph structured output converter for CrewAI task integration.
+
+This module contains the LangGraphConverterAdapter class that handles structured
+output conversion for LangGraph agents, supporting JSON and Pydantic model formats.
+"""
+
 import json
+import re
+from typing import Any, Literal
 
 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"""
+    """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
+    Converts task output requirements into system prompt modifications and
+    post-processing logic to ensure agents return properly structured outputs.
+    """
 
-    def configure_structured_output(self, task) -> None:
-        """Configure the structured output for LangGraph."""
+    def __init__(self, agent_adapter: Any) -> None:
+        """Initialize the converter adapter with a reference to the agent adapter.
+
+        Args:
+            agent_adapter: The LangGraph agent adapter instance.
+        """
+        super().__init__(agent_adapter=agent_adapter)
+        self.agent_adapter: Any = agent_adapter
+        self._output_format: Literal["json", "pydantic"] | None = None
+        self._schema: str | None = None
+        self._system_prompt_appendix: str | None = None
+
+    def configure_structured_output(self, task: Any) -> None:
+        """Configure the structured output for LangGraph.
+
+        Analyzes the task's output requirements and sets up the necessary
+        formatting and validation logic.
+
+        Args:
+            task: The task object containing output format specifications.
+        """
         if not (task.output_json or task.output_pydantic):
             self._output_format = None
             self._schema = None
@@ -32,7 +56,14 @@ class LangGraphConverterAdapter(BaseConverterAdapter):
         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"""
+        """Generate an appendix for the system prompt to enforce structured output.
+
+        Creates instructions that are appended to the system prompt to guide
+        the agent in producing properly formatted output.
+
+        Returns:
+            System prompt appendix string, or empty string if no structured output.
+        """
         if not self._output_format or not self._schema:
             return ""
 
@@ -41,19 +72,36 @@ 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. 
+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"""
+        """Add structured output instructions to the system prompt if needed.
+
+        Args:
+            original_prompt: The base system prompt.
+
+        Returns:
+            Enhanced system prompt with structured output instructions.
+        """
         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"""
+        """Post-process the result to ensure it matches the expected format.
+
+        Attempts to extract and validate JSON content from agent responses,
+        handling cases where JSON may be wrapped in markdown or other formatting.
+
+        Args:
+            result: The raw result string from the agent.
+
+        Returns:
+            Processed result string, ideally in valid JSON format.
+        """
         if not self._output_format:
             return result
 
@@ -65,16 +113,16 @@ The output should be raw JSON that exactly matches the specified schema.
                 return result
             except json.JSONDecodeError:
                 # Try to extract JSON from the text
-                import re
-
-                json_match = re.search(r"(\{.*\})", result, re.DOTALL)
+                json_match: re.Match[str] | None = re.search(
+                    r"(\{.*})", result, re.DOTALL
+                )
                 if json_match:
                     try:
-                        extracted = json_match.group(1)
+                        extracted: str = json_match.group(1)
                         # Validate it's proper JSON
                         json.loads(extracted)
                         return extracted
-                    except:
+                    except json.JSONDecodeError:
                         pass
 
         return result

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

@@ -1,78 +1,99 @@
-from typing import Any, List, Optional
+"""OpenAI agents adapter for CrewAI integration.
 
-from pydantic import Field, PrivateAttr
+This module contains the OpenAIAgentAdapter class that integrates OpenAI Assistants
+with CrewAI's agent system, providing tool integration and structured output support.
+"""
+
+from typing import Any, cast
+
+from pydantic import ConfigDict, Field, PrivateAttr
+from typing_extensions import Unpack
 
 from crewai.agents.agent_adapters.base_agent_adapter import BaseAgentAdapter
+from crewai.agents.agent_adapters.openai_agents.openai_agent_tool_adapter import (
+    OpenAIAgentToolAdapter,
+)
+from crewai.agents.agent_adapters.openai_agents.protocols import (
+    AgentKwargs,
+    OpenAIAgentsModule,
+)
+from crewai.agents.agent_adapters.openai_agents.protocols import (
+    OpenAIAgent as OpenAIAgentProtocol,
+)
 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.events.event_bus import crewai_event_bus
 from crewai.events.types.agent_events import (
     AgentExecutionCompletedEvent,
     AgentExecutionErrorEvent,
     AgentExecutionStartedEvent,
 )
+from crewai.tools import BaseTool
+from crewai.tools.agent_tools.agent_tools import AgentTools
+from crewai.utilities import Logger
+from crewai.utilities.import_utils import require
 
-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
+openai_agents_module = cast(
+    OpenAIAgentsModule,
+    require(
+        "agents",
+        purpose="OpenAI agents functionality",
+    ),
+)
+OpenAIAgent = openai_agents_module.Agent
+Runner = openai_agents_module.Runner
+enable_verbose_stdout_logging = openai_agents_module.enable_verbose_stdout_logging
 
 
 class OpenAIAgentAdapter(BaseAgentAdapter):
-    """Adapter for OpenAI Assistants"""
+    """Adapter for OpenAI Assistants.
 
-    model_config = {"arbitrary_types_allowed": True}
+    Integrates OpenAI Assistants API with CrewAI's agent system, providing
+    tool configuration, structured output handling, and task execution.
+    """
 
-    _openai_agent: "OpenAIAgent" = PrivateAttr()
-    _logger: Logger = PrivateAttr(default_factory=lambda: Logger())
-    _active_thread: Optional[str] = PrivateAttr(default=None)
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    _openai_agent: OpenAIAgentProtocol = PrivateAttr()
+    _logger: Logger = PrivateAttr(default_factory=Logger)
+    _active_thread: str | None = PrivateAttr(default=None)
     function_calling_llm: Any = Field(default=None)
     step_callback: Any = Field(default=None)
-    _tool_adapter: "OpenAIAgentToolAdapter" = PrivateAttr()
+    _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)
+        **kwargs: Unpack[AgentKwargs],
+    ) -> None:
+        """Initialize the OpenAI agent adapter.
+
+        Args:
+            **kwargs: All initialization arguments including role, goal, backstory,
+                     model, tools, and agent_config.
+
+        Raises:
+            ImportError: If OpenAI agent dependencies are not installed.
+        """
+        self.llm = kwargs.pop("model", "gpt-4o-mini")
+        super().__init__(**kwargs)
+        self._tool_adapter = OpenAIAgentToolAdapter(tools=kwargs.get("tools"))
+        self._converter_adapter = OpenAIConverterAdapter(agent_adapter=self)
 
     def _build_system_prompt(self) -> str:
-        """Build a system prompt for the OpenAI agent."""
+        """Build a system prompt for the OpenAI agent.
+
+        Creates a prompt containing the agent's role, goal, and backstory,
+        then enhances it with structured output instructions if needed.
+
+        Returns:
+            The complete system prompt string.
+        """
         base_prompt = f"""
             You are {self.role}.
-        
+
             Your goal is: {self.goal}
 
             Your backstory: {self.backstory}
@@ -84,10 +105,25 @@ class OpenAIAgentAdapter(BaseAgentAdapter):
     def execute_task(
         self,
         task: Any,
-        context: Optional[str] = None,
-        tools: Optional[List[BaseTool]] = None,
+        context: str | None = None,
+        tools: list[BaseTool] | None = None,
     ) -> str:
-        """Execute a task using the OpenAI Assistant"""
+        """Execute a task using the OpenAI Assistant.
+
+        Configures the assistant, processes the task, and handles event emission
+        for execution tracking.
+
+        Args:
+            task: The task object to execute.
+            context: Optional context information for the task.
+            tools: Optional additional tools for this execution.
+
+        Returns:
+            The final answer from the task execution.
+
+        Raises:
+            Exception: If task execution fails.
+        """
         self._converter_adapter.configure_structured_output(task)
         self.create_agent_executor(tools)
 
@@ -95,7 +131,7 @@ class OpenAIAgentAdapter(BaseAgentAdapter):
             enable_verbose_stdout_logging()
 
         try:
-            task_prompt = task.prompt()
+            task_prompt: str = task.prompt()
             if context:
                 task_prompt = self.i18n.slice("task_with_context").format(
                     task=task_prompt, context=context
@@ -109,8 +145,8 @@ class OpenAIAgentAdapter(BaseAgentAdapter):
                     task=task,
                 ),
             )
-            result = self.agent_executor.run_sync(self._openai_agent, task_prompt)
-            final_answer = self.handle_execution_result(result)
+            result: Any = self.agent_executor.run_sync(self._openai_agent, task_prompt)
+            final_answer: str = self.handle_execution_result(result)
             crewai_event_bus.emit(
                 self,
                 event=AgentExecutionCompletedEvent(
@@ -120,7 +156,7 @@ class OpenAIAgentAdapter(BaseAgentAdapter):
             return final_answer
 
         except Exception as e:
-            self._logger.log("error", f"Error executing OpenAI task: {str(e)}")
+            self._logger.log("error", f"Error executing OpenAI task: {e!s}")
             crewai_event_bus.emit(
                 self,
                 event=AgentExecutionErrorEvent(
@@ -131,15 +167,22 @@ class OpenAIAgentAdapter(BaseAgentAdapter):
             )
             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 [])
+    def create_agent_executor(self, tools: list[BaseTool] | None = None) -> None:
+        """Configure the OpenAI agent for execution.
 
-        instructions = self._build_system_prompt()
+        While OpenAI handles execution differently through Runner,
+        this method sets up tools and agent configuration.
+
+        Args:
+            tools: Optional tools to configure for the agent.
+
+        Notes:
+            TODO: Properly type agent_executor in BaseAgent to avoid type issues
+            when assigning Runner class to this attribute.
+        """
+        all_tools: list[BaseTool] = list(self.tools or []) + list(tools or [])
+
+        instructions: str = self._build_system_prompt()
         self._openai_agent = OpenAIAgent(
             name=self.role,
             instructions=instructions,
@@ -152,27 +195,48 @@ class OpenAIAgentAdapter(BaseAgentAdapter):
 
         self.agent_executor = Runner
 
-    def configure_tools(self, tools: Optional[List[BaseTool]] = None) -> None:
-        """Configure tools for the OpenAI Assistant"""
+    def configure_tools(self, tools: list[BaseTool] | None = None) -> None:
+        """Configure tools for the OpenAI Assistant.
+
+        Args:
+            tools: Optional tools to configure for the 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"""
+        """Process OpenAI Assistant execution result.
+
+        Converts any structured output to a string through the converter adapter.
+
+        Args:
+            result: The execution result from the OpenAI assistant.
+
+        Returns:
+            Processed result as 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 get_delegation_tools(self, agents: list[BaseAgent]) -> list[BaseTool]:
+        """Implement delegation tools support.
 
-    def configure_structured_output(self, task) -> None:
+        Creates delegation tools that allow this agent to delegate tasks to other agents.
+
+        Args:
+            agents: List of agents available for delegation.
+
+        Returns:
+            List of delegation tools.
+        """
+        agent_tools: AgentTools = AgentTools(agents=agents)
+        return agent_tools.tools()
+
+    def configure_structured_output(self, task: Any) -> None:
         """Configure the structured output for the specific agent implementation.
 
         Args:
-            structured_output: The structured output to be configured
+            task: The task object containing output format specifications.
         """
         self._converter_adapter.configure_structured_output(task)

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

@@ -1,57 +1,125 @@
-import inspect
-from typing import Any, List, Optional
+"""OpenAI agent tool adapter for CrewAI tool integration.
 
-from agents import FunctionTool, Tool
+This module contains the OpenAIAgentToolAdapter class that converts CrewAI tools
+to OpenAI Assistant-compatible format using the agents library.
+"""
+
+import inspect
+import json
+import re
+from collections.abc import Awaitable
+from typing import Any, cast
 
 from crewai.agents.agent_adapters.base_tool_adapter import BaseToolAdapter
+from crewai.agents.agent_adapters.openai_agents.protocols import (
+    OpenAIFunctionTool,
+    OpenAITool,
+)
 from crewai.tools import BaseTool
+from crewai.utilities.import_utils import require
+
+agents_module = cast(
+    Any,
+    require(
+        "agents",
+        purpose="OpenAI agents functionality",
+    ),
+)
+FunctionTool = agents_module.FunctionTool
+Tool = agents_module.Tool
 
 
 class OpenAIAgentToolAdapter(BaseToolAdapter):
-    """Adapter for OpenAI Assistant tools"""
+    """Adapter for OpenAI Assistant tools.
 
-    def __init__(self, tools: Optional[List[BaseTool]] = None):
-        self.original_tools = tools or []
+    Converts CrewAI BaseTool instances to OpenAI Assistant FunctionTool format
+    that can be used by OpenAI agents.
+    """
 
-    def configure_tools(self, tools: List[BaseTool]) -> None:
-        """Configure tools for the OpenAI Assistant"""
+    def __init__(self, tools: list[BaseTool] | None = None) -> None:
+        """Initialize the tool adapter.
+
+        Args:
+            tools: Optional list of CrewAI tools to adapt.
+        """
+        super().__init__()
+        self.original_tools: list[BaseTool] = tools or []
+        self.converted_tools: list[OpenAITool] = []
+
+    def configure_tools(self, tools: list[BaseTool]) -> None:
+        """Configure tools for the OpenAI Assistant.
+
+        Merges provided tools with original tools and converts them to
+        OpenAI Assistant format.
+
+        Args:
+            tools: List of CrewAI tools to configure.
+        """
         if self.original_tools:
-            all_tools = tools + self.original_tools
+            all_tools: list[BaseTool] = tools + self.original_tools
         else:
             all_tools = tools
         if all_tools:
             self.converted_tools = self._convert_tools_to_openai_format(all_tools)
 
+    @staticmethod
     def _convert_tools_to_openai_format(
-        self, tools: Optional[List[BaseTool]]
-    ) -> List[Tool]:
-        """Convert CrewAI tools to OpenAI Assistant tool format"""
+        tools: list[BaseTool] | None,
+    ) -> list[OpenAITool]:
+        """Convert CrewAI tools to OpenAI Assistant tool format.
+
+        Args:
+            tools: List of CrewAI tools to convert.
+
+        Returns:
+            List of OpenAI Assistant FunctionTool instances.
+        """
         if not tools:
             return []
 
         def sanitize_tool_name(name: str) -> str:
-            """Convert tool name to match OpenAI's required pattern"""
-            import re
+            """Convert tool name to match OpenAI's required pattern.
 
-            sanitized = re.sub(r"[^a-zA-Z0-9_-]", "_", name).lower()
-            return sanitized
+            Args:
+                name: Original tool name.
 
-        def create_tool_wrapper(tool: BaseTool):
-            """Create a wrapper function that handles the OpenAI function tool interface"""
+            Returns:
+                Sanitized tool name matching OpenAI requirements.
+            """
+
+            return re.sub(r"[^a-zA-Z0-9_-]", "_", name).lower()
+
+        def create_tool_wrapper(tool: BaseTool) -> Any:
+            """Create a wrapper function that handles the OpenAI function tool interface.
+
+            Args:
+                tool: The CrewAI tool to wrap.
+
+            Returns:
+                Async wrapper function for OpenAI agent integration.
+            """
 
             async def wrapper(context_wrapper: Any, arguments: Any) -> Any:
+                """Wrapper function to adapt CrewAI tool calls to OpenAI format.
+
+                Args:
+                    context_wrapper: OpenAI context wrapper.
+                    arguments: Tool arguments from OpenAI.
+
+                Returns:
+                    Tool execution result.
+                """
                 # Get the parameter name from the schema
-                param_name = list(
-                    tool.args_schema.model_json_schema()["properties"].keys()
-                )[0]
+                param_name: str = next(
+                    iter(tool.args_schema.model_json_schema()["properties"].keys())
+                )
 
                 # Handle different argument types
+                args_dict: dict[str, Any]
                 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}
@@ -59,11 +127,11 @@ class OpenAIAgentToolAdapter(BaseToolAdapter):
                     args_dict = {param_name: str(arguments)}
 
                 # Run the tool with the processed arguments
-                output = tool._run(**args_dict)
+                output: Any | Awaitable[Any] = tool._run(**args_dict)
 
                 # Await if the tool returned a coroutine
                 if inspect.isawaitable(output):
-                    result = await output
+                    result: Any = await output
                 else:
                     result = output
 
@@ -74,17 +142,20 @@ class OpenAIAgentToolAdapter(BaseToolAdapter):
 
             return wrapper
 
-        openai_tools = []
+        openai_tools: list[OpenAITool] = []
         for tool in tools:
-            schema = tool.args_schema.model_json_schema()
+            schema: dict[str, Any] = 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_tool: OpenAIFunctionTool = cast(
+                OpenAIFunctionTool,
+                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)
 

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

@@ -0,0 +1,74 @@
+"""Type protocols for OpenAI agents modules."""
+
+from collections.abc import Callable
+from typing import Any, Protocol, TypedDict, runtime_checkable
+
+from crewai.tools.base_tool import BaseTool
+
+
+class AgentKwargs(TypedDict, total=False):
+    """Typed dict for agent initialization kwargs."""
+
+    role: str
+    goal: str
+    backstory: str
+    model: str
+    tools: list[BaseTool] | None
+    agent_config: dict[str, Any] | None
+
+
+@runtime_checkable
+class OpenAIAgent(Protocol):
+    """Protocol for OpenAI Agent."""
+
+    def __init__(
+        self,
+        name: str,
+        instructions: str,
+        model: str,
+        **kwargs: Any,
+    ) -> None:
+        """Initialize the OpenAI agent."""
+        ...
+
+    tools: list[Any]
+    output_type: Any
+
+
+@runtime_checkable
+class OpenAIRunner(Protocol):
+    """Protocol for OpenAI Runner."""
+
+    @classmethod
+    def run_sync(cls, agent: OpenAIAgent, message: str) -> Any:
+        """Run agent synchronously with a message."""
+        ...
+
+
+@runtime_checkable
+class OpenAIAgentsModule(Protocol):
+    """Protocol for OpenAI agents module."""
+
+    Agent: type[OpenAIAgent]
+    Runner: type[OpenAIRunner]
+    enable_verbose_stdout_logging: Callable[[], None]
+
+
+@runtime_checkable
+class OpenAITool(Protocol):
+    """Protocol for OpenAI Tool."""
+
+
+@runtime_checkable
+class OpenAIFunctionTool(Protocol):
+    """Protocol for OpenAI FunctionTool."""
+
+    def __init__(
+        self,
+        name: str,
+        description: str,
+        params_json_schema: dict[str, Any],
+        on_invoke_tool: Any,
+    ) -> None:
+        """Initialize the function tool."""
+        ...

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

@@ -1,5 +1,12 @@
+"""OpenAI structured output converter for CrewAI task integration.
+
+This module contains the OpenAIConverterAdapter class that handles structured
+output conversion for OpenAI agents, supporting JSON and Pydantic model formats.
+"""
+
 import json
 import re
+from typing import Any, Literal
 
 from crewai.agents.agent_adapters.base_converter_adapter import BaseConverterAdapter
 from crewai.utilities.converter import generate_model_description
@@ -7,8 +14,7 @@ from crewai.utilities.i18n import I18N
 
 
 class OpenAIConverterAdapter(BaseConverterAdapter):
-    """
-    Adapter for handling structured output conversion in OpenAI agents.
+    """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.
@@ -19,19 +25,23 @@ class OpenAIConverterAdapter(BaseConverterAdapter):
         _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.
+    def __init__(self, agent_adapter: Any) -> None:
+        """Initialize the converter adapter with a reference to the agent adapter.
 
         Args:
-            task: The task containing output format requirements
+            agent_adapter: The OpenAI agent adapter instance.
+        """
+        super().__init__(agent_adapter=agent_adapter)
+        self.agent_adapter: Any = agent_adapter
+        self._output_format: Literal["json", "pydantic"] | None = None
+        self._schema: str | None = None
+        self._output_model: Any = None
+
+    def configure_structured_output(self, task: Any) -> 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
@@ -55,19 +65,18 @@ class OpenAIConverterAdapter(BaseConverterAdapter):
             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.
+        """Enhance the base system prompt with structured output requirements if needed.
 
         Args:
-            base_prompt: The original system prompt
+            base_prompt: The original system prompt.
 
         Returns:
-            Enhanced system prompt with output format instructions if needed
+            Enhanced system prompt with output format instructions if needed.
         """
         if not self._output_format:
             return base_prompt
 
-        output_schema = (
+        output_schema: str = (
             I18N()
             .slice("formatted_task_instructions")
             .format(output_format=self._schema)
@@ -76,16 +85,15 @@ class OpenAIConverterAdapter(BaseConverterAdapter):
         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.
+        """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
+            result: The raw result from the agent.
 
         Returns:
-            Processed result conforming to the expected output format
+            Processed result conforming to the expected output format.
         """
         if not self._output_format:
             return result
@@ -97,26 +105,30 @@ class OpenAIConverterAdapter(BaseConverterAdapter):
                 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)
+                code_block_pattern: str = r"```(?:json)?\s*([\s\S]*?)```"
+                code_blocks: list[str] = re.findall(code_block_pattern, result)
 
                 for block in code_blocks:
+                    stripped_block = block.strip()
                     try:
-                        json.loads(block.strip())
-                        return block.strip()
+                        json.loads(stripped_block)
+                        return stripped_block
                     except json.JSONDecodeError:
-                        continue
+                        pass
 
                 # Try to extract any JSON-like structure
-                json_pattern = r"(\{[\s\S]*\})"
-                json_matches = re.findall(json_pattern, result, re.DOTALL)
+                json_pattern: str = r"(\{[\s\S]*\})"
+                json_matches: list[str] = re.findall(json_pattern, result, re.DOTALL)
 
                 for match in json_matches:
+                    is_valid = True
                     try:
                         json.loads(match)
-                        return match
                     except json.JSONDecodeError:
-                        continue
+                        is_valid = False
+
+                    if is_valid:
+                        return match
 
         # If all extraction attempts fail, return the original
         return str(result)

src/crewai/cli/evaluate_crew.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def evaluate_crew(n_iterations: int, model: str) -> None:
     """
@@ -17,7 +19,7 @@ def evaluate_crew(n_iterations: int, model: str) -> None:
         if n_iterations <= 0:
             raise ValueError("The number of iterations must be a positive integer.")
 
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
 
         if result.stderr:
             click.echo(result.stderr, err=True)

src/crewai/cli/git.py

@@ -1,6 +1,8 @@
 import subprocess
 from functools import lru_cache
 
+from crewai.cli.subprocess_utils import run_command
+
 
 class Repository:
     def __init__(self, path="."):
@@ -17,7 +19,7 @@ class Repository:
     def is_git_installed(self) -> bool:
         """Check if Git is installed and available in the system."""
         try:
-            subprocess.run(
+            run_command(
                 ["git", "--version"], capture_output=True, check=True, text=True
             )
             return True
@@ -26,24 +28,29 @@ class Repository:
 
     def fetch(self) -> None:
         """Fetch latest updates from the remote."""
-        subprocess.run(["git", "fetch"], cwd=self.path, check=True)
+        run_command(["git", "fetch"], cwd=self.path, check=True)
 
     def status(self) -> str:
         """Get the git status in porcelain format."""
-        return subprocess.check_output(
+        result = run_command(
             ["git", "status", "--branch", "--porcelain"],
             cwd=self.path,
-            encoding="utf-8",
-        ).strip()
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        return result.stdout.strip()
 
     @lru_cache(maxsize=None)
     def is_git_repo(self) -> bool:
         """Check if the current directory is a git repository."""
         try:
-            subprocess.check_output(
+            run_command(
                 ["git", "rev-parse", "--is-inside-work-tree"],
                 cwd=self.path,
-                encoding="utf-8",
+                capture_output=True,
+                text=True,
+                check=True,
             )
             return True
         except subprocess.CalledProcessError:
@@ -70,7 +77,7 @@ class Repository:
     def origin_url(self) -> str | None:
         """Get the Git repository's remote URL."""
         try:
-            result = subprocess.run(
+            result = run_command(
                 ["git", "remote", "get-url", "origin"],
                 cwd=self.path,
                 capture_output=True,

src/crewai/cli/install_crew.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 # Be mindful about changing this.
 # on some environments we don't use this command but instead uv sync directly
@@ -12,8 +14,8 @@ def install_crew(proxy_options: list[str]) -> None:
     Install the crew by running the UV command to lock and install.
     """
     try:
-        command = ["uv", "sync"] + proxy_options
-        subprocess.run(command, check=True, capture_output=False, text=True)
+        command = ["uv", "sync", *proxy_options]
+        run_command(command, check=True, capture_output=False, text=True)
 
     except subprocess.CalledProcessError as e:
         click.echo(f"An error occurred while running the crew: {e}", err=True)

src/crewai/cli/kickoff_flow.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def kickoff_flow() -> None:
     """
@@ -10,7 +12,7 @@ def kickoff_flow() -> None:
     command = ["uv", "run", "kickoff"]
 
     try:
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
 
         if result.stderr:
             click.echo(result.stderr, err=True)

src/crewai/cli/plot_flow.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def plot_flow() -> None:
     """
@@ -10,7 +12,7 @@ def plot_flow() -> None:
     command = ["uv", "run", "plot"]
 
     try:
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
 
         if result.stderr:
             click.echo(result.stderr, err=True)

src/crewai/cli/replay_from_task.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def replay_task_command(task_id: str) -> None:
     """
@@ -13,7 +15,7 @@ def replay_task_command(task_id: str) -> None:
     command = ["uv", "run", "replay", task_id]
 
     try:
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
         if result.stderr:
             click.echo(result.stderr, err=True)
 

src/crewai/cli/run_crew.py

@@ -1,12 +1,12 @@
 import subprocess
 from enum import Enum
-from typing import List, Optional
 
 import click
 from packaging import version
 
 from crewai.cli.utils import read_toml
 from crewai.cli.version import get_crewai_version
+from crewai.cli.subprocess_utils import run_command
 
 
 class CrewType(Enum):
@@ -57,7 +57,7 @@ def execute_command(crew_type: CrewType) -> None:
     command = ["uv", "run", "kickoff" if crew_type == CrewType.FLOW else "run_crew"]
 
     try:
-        subprocess.run(command, capture_output=False, text=True, check=True)
+        run_command(command, capture_output=False, text=True, check=True)
 
     except subprocess.CalledProcessError as e:
         handle_error(e, crew_type)

src/crewai/cli/subprocess_utils.py

@@ -0,0 +1,60 @@
+import platform
+import subprocess
+from typing import Any
+
+
+def run_command(
+    command: list[str],
+    capture_output: bool = False,
+    text: bool = True,
+    check: bool = True,
+    cwd: str | None = None,
+    env: dict[str, str] | None = None,
+    **kwargs: Any
+) -> subprocess.CompletedProcess:
+    """
+    Cross-platform subprocess execution with Windows compatibility.
+
+    On Windows, uses shell=True to avoid permission issues with restrictive
+    security policies. On other platforms, uses the standard approach.
+
+    Args:
+        command: List of command arguments
+        capture_output: Whether to capture stdout/stderr
+        text: Whether to use text mode
+        check: Whether to raise CalledProcessError on non-zero exit
+        cwd: Working directory
+        env: Environment variables
+        **kwargs: Additional subprocess.run arguments
+
+    Returns:
+        CompletedProcess instance
+
+    Raises:
+        subprocess.CalledProcessError: If check=True and command fails
+    """
+    if platform.system() == "Windows":
+        if isinstance(command, list):
+            command_str = subprocess.list2cmdline(command)
+        else:
+            command_str = command
+
+        return subprocess.run(
+            command_str,
+            shell=True,
+            capture_output=capture_output,
+            text=text,
+            check=check,
+            cwd=cwd,
+            env=env,
+            **kwargs
+        )
+    return subprocess.run(
+        command,
+        capture_output=capture_output,
+        text=text,
+        check=check,
+        cwd=cwd,
+        env=env,
+        **kwargs
+    )

src/crewai/cli/tools/main.py

@@ -1,6 +1,5 @@
 import base64
 import os
-import subprocess
 import tempfile
 from pathlib import Path
 from typing import Any
@@ -11,6 +10,7 @@ from rich.console import Console
 from crewai.cli import git
 from crewai.cli.command import BaseCommand, PlusAPIMixin
 from crewai.cli.config import Settings
+from crewai.cli.subprocess_utils import run_command
 from crewai.cli.utils import (
     extract_available_exports,
     get_project_description,
@@ -56,7 +56,7 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
         os.chdir(project_root)
         try:
             self.login()
-            subprocess.run(["git", "init"], check=True)
+            run_command(["git", "init"], check=True)
             console.print(
                 f"[green]Created custom tool [bold]{folder_name}[/bold]. Run [bold]cd {project_root}[/bold] to start working.[/green]"
             )
@@ -94,7 +94,7 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
         self._print_current_organization()
 
         with tempfile.TemporaryDirectory() as temp_build_dir:
-            subprocess.run(
+            run_command(
                 ["uv", "build", "--sdist", "--out-dir", temp_build_dir],
                 check=True,
                 capture_output=False,
@@ -196,7 +196,7 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
         else:
             add_package_command.extend(["--index", index, tool_handle])
 
-        add_package_result = subprocess.run(
+        add_package_result = run_command(
             add_package_command,
             capture_output=False,
             env=self._build_env_with_credentials(repository_handle),

src/crewai/cli/train_crew.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.subprocess_utils import run_command
+
 
 def train_crew(n_iterations: int, filename: str) -> None:
     """
@@ -19,7 +21,7 @@ def train_crew(n_iterations: int, filename: str) -> None:
         if not filename.endswith(".pkl"):
             raise ValueError("The filename must not end with .pkl")
 
-        result = subprocess.run(command, capture_output=False, text=True, check=True)
+        result = run_command(command, capture_output=False, text=True, check=True)
 
         if result.stderr:
             click.echo(result.stderr, err=True)

src/crewai/context.py

@@ -0,0 +1,25 @@
+import os
+import contextvars
+from typing import Optional
+from contextlib import contextmanager
+
+_platform_integration_token: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
+    "platform_integration_token", default=None
+)
+
+def set_platform_integration_token(integration_token: str) -> None:
+    _platform_integration_token.set(integration_token)
+
+def get_platform_integration_token() -> Optional[str]:
+    token = _platform_integration_token.get()
+    if token is None:
+        token = os.getenv("CREWAI_PLATFORM_INTEGRATION_TOKEN")
+    return token
+
+@contextmanager
+def platform_context(integration_token: str):
+    token = _platform_integration_token.set(integration_token)
+    try:
+        yield
+    finally:
+        _platform_integration_token.reset(token)

src/crewai/memory/storage/mem0_storage.py

@@ -1,10 +1,12 @@
 import os
-from typing import Any, Dict, List
+import re
 from collections import defaultdict
-from mem0 import Memory, MemoryClient
-from crewai.utilities.chromadb import sanitize_collection_name
+from typing import Any, Iterable
+
+from mem0 import Memory, MemoryClient  # type: ignore[import-untyped]
 
 from crewai.memory.storage.interface import Storage
+from crewai.utilities.chromadb import sanitize_collection_name
 
 MAX_AGENT_ID_LENGTH_MEM0 = 255
 
@@ -86,9 +88,28 @@ class Mem0Storage(Storage):
 
         return filter
 
-    def save(self, value: Any, metadata: Dict[str, Any]) -> None:
+    def save(self, value: Any, metadata: dict[str, Any]) -> None:
+        def _last_content(messages: Iterable[dict[str, Any]], role: str) -> str:
+            return next(
+                (m.get("content", "") for m in reversed(list(messages)) if m.get("role") == role),
+                ""
+            )
+        
+        conversations = []
+        messages = metadata.pop("messages", None)
+        if messages:
+            last_user = _last_content(messages, "user")
+            last_assistant = _last_content(messages, "assistant")
+
+            if user_msg := self._get_user_message(last_user):
+                conversations.append({"role": "user", "content": user_msg})
+                
+            if assistant_msg := self._get_assistant_message(last_assistant):
+                conversations.append({"role": "assistant", "content": assistant_msg})
+        else:
+            conversations.append({"role": "assistant", "content": value})
+
         user_id = self.config.get("user_id", "")
-        assistant_message = [{"role" : "assistant","content" : value}]
 
         base_metadata = {
             "short_term": "short_term",
@@ -119,9 +140,9 @@ class Mem0Storage(Storage):
         if agent_id := self.config.get("agent_id", self._get_agent_name()):
             params["agent_id"] = agent_id
 
-        self.memory.add(assistant_message, **params)
+        self.memory.add(conversations, **params)
 
-    def search(self,query: str,limit: int = 3,score_threshold: float = 0.35) -> List[Any]:
+    def search(self,query: str,limit: int = 3,score_threshold: float = 0.35) -> list[Any]:
         params = {
             "query": query,
             "limit": limit,
@@ -160,7 +181,7 @@ class Mem0Storage(Storage):
         # This makes it compatible for Contextual Memory to retrieve
         for result in results["results"]:
             result["context"] = result["memory"]
-        
+
         return [r for r in results["results"]]
 
     def reset(self):
@@ -181,3 +202,16 @@ class Mem0Storage(Storage):
         agents = [self._sanitize_role(agent.role) for agent in agents]
         agents = "_".join(agents)
         return sanitize_collection_name(name=agents, max_collection_length=MAX_AGENT_ID_LENGTH_MEM0)
+
+    def _get_assistant_message(self, text: str) -> str:
+        marker = "Final Answer:"
+        if marker in text:
+            return text.split(marker, 1)[1].strip()
+        return text
+
+    def _get_user_message(self, text: str) -> str:
+        pattern = r"User message:\s*(.*)"
+        match = re.search(pattern, text)
+        if match:
+            return match.group(1).strip()
+        return text

tests/cli/test_subprocess_utils.py

@@ -0,0 +1,140 @@
+import subprocess
+from unittest import mock
+
+import pytest
+
+from crewai.cli.subprocess_utils import run_command
+
+
+class TestRunCommand:
+    """Test the cross-platform subprocess utility."""
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_windows_uses_shell_true(self, mock_subprocess_run, mock_platform):
+        """Test that Windows uses shell=True with proper command conversion."""
+        mock_platform.return_value = "Windows"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args="uv run test", returncode=0
+        )
+
+        command = ["uv", "run", "test"]
+        run_command(command)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[1]["shell"] is True
+        assert isinstance(call_args[0][0], str)
+        assert "uv run test" in call_args[0][0]
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_unix_uses_shell_false(self, mock_subprocess_run, mock_platform):
+        """Test that Unix-like systems use shell=False with list commands."""
+        mock_platform.return_value = "Linux"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args=["uv", "run", "test"], returncode=0
+        )
+
+        command = ["uv", "run", "test"]
+        run_command(command)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[1].get("shell", False) is False
+        assert call_args[0][0] == command
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_windows_command_escaping(self, mock_subprocess_run, mock_platform):
+        """Test that Windows properly escapes command arguments."""
+        mock_platform.return_value = "Windows"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args="test", returncode=0
+        )
+
+        command = ["echo", "hello world", "test&special"]
+        run_command(command)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+        command_str = call_args[0][0]
+
+        assert '"hello world"' in command_str or "'hello world'" in command_str
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_error_handling_preserved(self, mock_subprocess_run, mock_platform):
+        """Test that CalledProcessError is properly raised."""
+        mock_platform.return_value = "Windows"
+        mock_subprocess_run.side_effect = subprocess.CalledProcessError(1, "test")
+
+        with pytest.raises(subprocess.CalledProcessError):
+            run_command(["test"], check=True)
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_all_parameters_passed_through(self, mock_subprocess_run, mock_platform):
+        """Test that all subprocess parameters are properly passed through."""
+        mock_platform.return_value = "Linux"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args=["test"], returncode=0
+        )
+
+        run_command(
+            ["test"],
+            capture_output=True,
+            text=False,
+            check=False,
+            cwd="/home/test",
+            env={"TEST": "value"},
+            timeout=30
+        )
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[1]["capture_output"] is True
+        assert call_args[1]["text"] is False
+        assert call_args[1]["check"] is False
+        assert call_args[1]["cwd"] == "/home/test"
+        assert call_args[1]["env"] == {"TEST": "value"}
+        assert call_args[1]["timeout"] == 30
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_macos_uses_shell_false(self, mock_subprocess_run, mock_platform):
+        """Test that macOS uses shell=False with list commands."""
+        mock_platform.return_value = "Darwin"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args=["uv", "run", "test"], returncode=0
+        )
+
+        command = ["uv", "run", "test"]
+        run_command(command)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[1].get("shell", False) is False
+        assert call_args[0][0] == command
+
+    @mock.patch("platform.system")
+    @mock.patch("subprocess.run")
+    def test_windows_string_passthrough(self, mock_subprocess_run, mock_platform):
+        """Test that Windows passes through string commands unchanged."""
+        mock_platform.return_value = "Windows"
+        mock_subprocess_run.return_value = subprocess.CompletedProcess(
+            args="test command", returncode=0
+        )
+
+        command_str = "test command with spaces"
+        run_command(command_str)
+
+        mock_subprocess_run.assert_called_once()
+        call_args = mock_subprocess_run.call_args
+
+        assert call_args[0][0] == command_str
+        assert call_args[1]["shell"] is True

tests/storage/test_mem0_storage.py

@@ -16,8 +16,7 @@ class MockCrew:
 @pytest.fixture
 def mock_mem0_memory():
     """Fixture to create a mock Memory instance"""
-    mock_memory = MagicMock(spec=Memory)
-    return mock_memory
+    return MagicMock(spec=Memory)
 
 
 @pytest.fixture
@@ -73,8 +72,7 @@ def test_mem0_storage_initialization(mem0_storage_with_mocked_config, mock_mem0_
 @pytest.fixture
 def mock_mem0_memory_client():
     """Fixture to create a mock MemoryClient instance"""
-    mock_memory = MagicMock(spec=MemoryClient)
-    return mock_memory
+    return MagicMock(spec=MemoryClient)
 
 
 @pytest.fixture
@@ -96,8 +94,7 @@ def mem0_storage_with_memory_client_using_config_from_crew(mock_mem0_memory_clie
                     "infer": True
                 }
 
-        mem0_storage = Mem0Storage(type="short_term", crew=crew, config=embedder_config)
-        return mem0_storage
+        return Mem0Storage(type="short_term", crew=crew, config=embedder_config)
 
 
 @pytest.fixture
@@ -111,8 +108,7 @@ def mem0_storage_with_memory_client_using_explictly_config(mock_mem0_memory_clie
         crew = MockCrew()
         new_config = {"provider": "mem0", "config": {"api_key": "new-api-key"}}
 
-        mem0_storage = Mem0Storage(type="short_term", crew=crew, config=new_config)
-        return mem0_storage
+        return Mem0Storage(type="short_term", crew=crew, config=new_config)
 
 
 def test_mem0_storage_with_memory_client_initialization(
@@ -172,14 +168,14 @@ def test_save_method_with_memory_oss(mem0_storage_with_mocked_config):
 
     # Test short_term memory type (already set in fixture)
     test_value = "This is a test memory"
-    test_metadata = {"key": "value"}
+    test_metadata = {'description': 'Respond to user conversation. User message: What do you know about me?', 'messages': [{'role': 'system', 'content': 'You are Friendly chatbot assistant. You are a kind and knowledgeable chatbot assistant. You excel at understanding user needs, providing helpful responses, and maintaining engaging conversations. You remember previous interactions to provide a personalized experience.\nYour personal goal is: Engage in useful and interesting conversations with users while remembering context.\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!'}, {'role': 'user', 'content': '\nCurrent Task: Respond to user conversation. User message: What do you know about me?\n\nThis is the expected criteria for your final answer: Contextually appropriate, helpful, and friendly response.\nyou MUST return the actual complete content as the final answer, not a summary.\n\n# Useful context: \nExternal memories:\n- User is from India\n- User is interested in the solar system\n- User name is Vidit Ostwal\n- User is interested in French cuisine\n\nBegin! This is VERY important to you, use the tools available and give your best Final Answer, your job depends on it!\n\nThought:'}, {'role': 'assistant', 'content': "I now can give a great answer  \nFinal Answer: Hi Vidit! From our previous conversations, I know you're from India and have a great interest in the solar system. It's fascinating to explore the wonders of space, isn't it? Also, I remember you have a passion for French cuisine, which has so many delightful dishes to explore. If there's anything specific you'd like to discuss or learn about—whether it's about the solar system or some great French recipes—feel free to let me know! I'm here to help."}], 'agent': 'Friendly chatbot assistant'}
 
     mem0_storage.save(test_value, test_metadata)
 
     mem0_storage.memory.add.assert_called_once_with(
-        [{"role": "assistant" , "content": test_value}],
+        [{'role': 'user', 'content': 'What do you know about me?'}, {'role': 'assistant', 'content': "Hi Vidit! From our previous conversations, I know you're from India and have a great interest in the solar system. It's fascinating to explore the wonders of space, isn't it? Also, I remember you have a passion for French cuisine, which has so many delightful dishes to explore. If there's anything specific you'd like to discuss or learn about—whether it's about the solar system or some great French recipes—feel free to let me know! I'm here to help."}],
         infer=True,
-        metadata={"type": "short_term", "key": "value"},
+        metadata={'type': 'short_term', 'description': 'Respond to user conversation. User message: What do you know about me?', 'agent': 'Friendly chatbot assistant'},
         run_id="my_run_id",
         user_id="test_user",
         agent_id='Test_Agent'
@@ -191,14 +187,14 @@ def test_save_method_with_multiple_agents(mem0_storage_with_mocked_config):
     mem0_storage.memory.add = MagicMock()
 
     test_value = "This is a test memory"
-    test_metadata = {"key": "value"}
+    test_metadata = {'description': 'Respond to user conversation. User message: What do you know about me?', 'messages': [{'role': 'system', 'content': 'You are Friendly chatbot assistant. You are a kind and knowledgeable chatbot assistant. You excel at understanding user needs, providing helpful responses, and maintaining engaging conversations. You remember previous interactions to provide a personalized experience.\nYour personal goal is: Engage in useful and interesting conversations with users while remembering context.\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!'}, {'role': 'user', 'content': '\nCurrent Task: Respond to user conversation. User message: What do you know about me?\n\nThis is the expected criteria for your final answer: Contextually appropriate, helpful, and friendly response.\nyou MUST return the actual complete content as the final answer, not a summary.\n\n# Useful context: \nExternal memories:\n- User is from India\n- User is interested in the solar system\n- User name is Vidit Ostwal\n- User is interested in French cuisine\n\nBegin! This is VERY important to you, use the tools available and give your best Final Answer, your job depends on it!\n\nThought:'}, {'role': 'assistant', 'content': "I now can give a great answer  \nFinal Answer: Hi Vidit! From our previous conversations, I know you're from India and have a great interest in the solar system. It's fascinating to explore the wonders of space, isn't it? Also, I remember you have a passion for French cuisine, which has so many delightful dishes to explore. If there's anything specific you'd like to discuss or learn about—whether it's about the solar system or some great French recipes—feel free to let me know! I'm here to help."}], 'agent': 'Friendly chatbot assistant'}
 
     mem0_storage.save(test_value, test_metadata)
 
     mem0_storage.memory.add.assert_called_once_with(
-        [{"role": "assistant" , "content": test_value}],
+        [{'role': 'user', 'content': 'What do you know about me?'}, {'role': 'assistant', 'content': "Hi Vidit! From our previous conversations, I know you're from India and have a great interest in the solar system. It's fascinating to explore the wonders of space, isn't it? Also, I remember you have a passion for French cuisine, which has so many delightful dishes to explore. If there's anything specific you'd like to discuss or learn about—whether it's about the solar system or some great French recipes—feel free to let me know! I'm here to help."}],
         infer=True,
-        metadata={"type": "short_term", "key": "value"},
+        metadata={'type': 'short_term', 'description': 'Respond to user conversation. User message: What do you know about me?', 'agent': 'Friendly chatbot assistant'},
         run_id="my_run_id",
         user_id="test_user",
         agent_id='Test_Agent_Test_Agent_2_Test_Agent_3'
@@ -212,14 +208,14 @@ def test_save_method_with_memory_client(mem0_storage_with_memory_client_using_co
 
     # Test short_term memory type (already set in fixture)
     test_value = "This is a test memory"
-    test_metadata = {"key": "value"}
+    test_metadata = {'description': 'Respond to user conversation. User message: What do you know about me?', 'messages': [{'role': 'system', 'content': 'You are Friendly chatbot assistant. You are a kind and knowledgeable chatbot assistant. You excel at understanding user needs, providing helpful responses, and maintaining engaging conversations. You remember previous interactions to provide a personalized experience.\nYour personal goal is: Engage in useful and interesting conversations with users while remembering context.\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!'}, {'role': 'user', 'content': '\nCurrent Task: Respond to user conversation. User message: What do you know about me?\n\nThis is the expected criteria for your final answer: Contextually appropriate, helpful, and friendly response.\nyou MUST return the actual complete content as the final answer, not a summary.\n\n# Useful context: \nExternal memories:\n- User is from India\n- User is interested in the solar system\n- User name is Vidit Ostwal\n- User is interested in French cuisine\n\nBegin! This is VERY important to you, use the tools available and give your best Final Answer, your job depends on it!\n\nThought:'}, {'role': 'assistant', 'content': "I now can give a great answer  \nFinal Answer: Hi Vidit! From our previous conversations, I know you're from India and have a great interest in the solar system. It's fascinating to explore the wonders of space, isn't it? Also, I remember you have a passion for French cuisine, which has so many delightful dishes to explore. If there's anything specific you'd like to discuss or learn about—whether it's about the solar system or some great French recipes—feel free to let me know! I'm here to help."}], 'agent': 'Friendly chatbot assistant'}
 
     mem0_storage.save(test_value, test_metadata)
 
     mem0_storage.memory.add.assert_called_once_with(
-        [{'role': 'assistant' , 'content': test_value}],
+        [{'role': 'user', 'content': 'What do you know about me?'}, {'role': 'assistant', 'content': "Hi Vidit! From our previous conversations, I know you're from India and have a great interest in the solar system. It's fascinating to explore the wonders of space, isn't it? Also, I remember you have a passion for French cuisine, which has so many delightful dishes to explore. If there's anything specific you'd like to discuss or learn about—whether it's about the solar system or some great French recipes—feel free to let me know! I'm here to help."}],
         infer=True,
-        metadata={"type": "short_term", "key": "value"},
+       metadata={'type': 'short_term', 'description': 'Respond to user conversation. User message: What do you know about me?', 'agent': 'Friendly chatbot assistant'},
         version="v2",
         run_id="my_run_id",
         includes="include1",

tests/test_context.py

@@ -0,0 +1,216 @@
+# ruff: noqa: S105
+
+import os
+import pytest
+from unittest.mock import patch
+
+from crewai.context import (
+    set_platform_integration_token,
+    get_platform_integration_token,
+    platform_context,
+    _platform_integration_token,
+)
+
+
+class TestPlatformIntegrationToken:
+    def setup_method(self):
+        _platform_integration_token.set(None)
+
+    def teardown_method(self):
+        _platform_integration_token.set(None)
+
+    def test_set_platform_integration_token(self):
+        test_token = "test-token-123"
+
+        assert get_platform_integration_token() is None
+
+        set_platform_integration_token(test_token)
+
+        assert get_platform_integration_token() == test_token
+
+    def test_get_platform_integration_token_from_context_var(self):
+        test_token = "context-var-token"
+
+        _platform_integration_token.set(test_token)
+
+        assert get_platform_integration_token() == test_token
+
+    @patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-token-456"})
+    def test_get_platform_integration_token_from_env_var(self):
+        assert _platform_integration_token.get() is None
+
+        assert get_platform_integration_token() == "env-token-456"
+
+    @patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-token"})
+    def test_context_var_takes_precedence_over_env_var(self):
+        context_token = "context-token"
+
+        set_platform_integration_token(context_token)
+
+        assert get_platform_integration_token() == context_token
+
+    @patch.dict(os.environ, {}, clear=True)
+    def test_get_platform_integration_token_returns_none_when_not_set(self):
+        assert _platform_integration_token.get() is None
+
+        assert get_platform_integration_token() is None
+
+    def test_platform_context_manager_basic_usage(self):
+        test_token = "context-manager-token"
+
+        assert get_platform_integration_token() is None
+
+        with platform_context(test_token):
+            assert get_platform_integration_token() == test_token
+
+        assert get_platform_integration_token() is None
+
+    def test_platform_context_manager_nested_contexts(self):
+        """Test nested platform_context context managers."""
+        outer_token = "outer-token"
+        inner_token = "inner-token"
+
+        assert get_platform_integration_token() is None
+
+        with platform_context(outer_token):
+            assert get_platform_integration_token() == outer_token
+
+            with platform_context(inner_token):
+                assert get_platform_integration_token() == inner_token
+
+            assert get_platform_integration_token() == outer_token
+
+        assert get_platform_integration_token() is None
+
+    def test_platform_context_manager_preserves_existing_token(self):
+        """Test that platform_context preserves existing token when exiting."""
+        initial_token = "initial-token"
+        context_token = "context-token"
+
+        set_platform_integration_token(initial_token)
+        assert get_platform_integration_token() == initial_token
+
+        with platform_context(context_token):
+            assert get_platform_integration_token() == context_token
+
+        assert get_platform_integration_token() == initial_token
+
+    def test_platform_context_manager_exception_handling(self):
+        """Test that platform_context properly resets token even when exception occurs."""
+        initial_token = "initial-token"
+        context_token = "context-token"
+
+        set_platform_integration_token(initial_token)
+
+        with pytest.raises(ValueError):
+            with platform_context(context_token):
+                assert get_platform_integration_token() == context_token
+                raise ValueError("Test exception")
+
+        assert get_platform_integration_token() == initial_token
+
+    def test_platform_context_manager_with_none_initial_state(self):
+        """Test platform_context when initial state is None."""
+        context_token = "context-token"
+
+        assert get_platform_integration_token() is None
+
+        with pytest.raises(RuntimeError):
+            with platform_context(context_token):
+                assert get_platform_integration_token() == context_token
+                raise RuntimeError("Test exception")
+
+        assert get_platform_integration_token() is None
+
+    @patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": "env-backup"})
+    def test_platform_context_with_env_fallback(self):
+        """Test platform_context interaction with environment variable fallback."""
+        context_token = "context-token"
+
+        assert get_platform_integration_token() == "env-backup"
+
+        with platform_context(context_token):
+            assert get_platform_integration_token() == context_token
+
+        assert get_platform_integration_token() == "env-backup"
+
+    def test_multiple_sequential_context_managers(self):
+        """Test multiple sequential uses of platform_context."""
+        token1 = "token-1"
+        token2 = "token-2"
+        token3 = "token-3"
+
+        with platform_context(token1):
+            assert get_platform_integration_token() == token1
+
+        assert get_platform_integration_token() is None
+
+        with platform_context(token2):
+            assert get_platform_integration_token() == token2
+
+        assert get_platform_integration_token() is None
+
+        with platform_context(token3):
+            assert get_platform_integration_token() == token3
+
+        assert get_platform_integration_token() is None
+
+    def test_empty_string_token(self):
+        empty_token = ""
+
+        set_platform_integration_token(empty_token)
+        assert get_platform_integration_token() == ""
+
+        with platform_context(empty_token):
+            assert get_platform_integration_token() == ""
+
+    def test_special_characters_in_token(self):
+        special_token = "token-with-!@#$%^&*()_+-={}[]|\\:;\"'<>?,./"
+
+        set_platform_integration_token(special_token)
+        assert get_platform_integration_token() == special_token
+
+        with platform_context(special_token):
+            assert get_platform_integration_token() == special_token
+
+    def test_very_long_token(self):
+        long_token = "a" * 10000
+
+        set_platform_integration_token(long_token)
+        assert get_platform_integration_token() == long_token
+
+        with platform_context(long_token):
+            assert get_platform_integration_token() == long_token
+
+    @patch.dict(os.environ, {"CREWAI_PLATFORM_INTEGRATION_TOKEN": ""})
+    def test_empty_env_var(self):
+        assert _platform_integration_token.get() is None
+        assert get_platform_integration_token() == ""
+
+    @patch('crewai.context.os.getenv')
+    def test_env_var_access_error_handling(self, mock_getenv):
+        mock_getenv.side_effect = OSError("Environment access error")
+
+        with pytest.raises(OSError):
+            get_platform_integration_token()
+
+    def test_context_var_isolation_between_tests(self):
+        """Test that context variable changes don't leak between test methods."""
+        test_token = "isolation-test-token"
+
+        assert get_platform_integration_token() is None
+
+        set_platform_integration_token(test_token)
+        assert get_platform_integration_token() == test_token
+
+
+    def test_context_manager_return_value(self):
+        """Test that platform_context can be used in with statement with return value."""
+        test_token = "return-value-token"
+
+        with platform_context(test_token):
+            assert get_platform_integration_token() == test_token
+
+        with platform_context(test_token) as ctx:
+            assert ctx is None
+            assert get_platform_integration_token() == test_token