Fix: replace new_data_artifact with new_artifact (available in a2a-sdk 1.0.0)

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

.gitignore

@@ -30,3 +30,4 @@ chromadb-*.lock
 .crewai/memory
 blogs/*
 secrets/*
+UNKNOWN.egg-info/

docs/ar/changelog.mdx

@@ -4,6 +4,32 @@ description: "تحديثات المنتج والتحسينات وإصلاحات
 icon: "clock"
 mode: "wide"
 ---
+<Update label="23 أبريل 2026">
+  ## v1.14.3a3
+
+  [عرض الإصدار على GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.14.3a3)
+
+  ## ما الذي تغير
+
+  ### الميزات
+  - إضافة دعم لـ e2b
+  - تنفيذ التراجع إلى DefaultAzureCredential عند عدم توفير مفتاح API
+
+  ### إصلاحات الأخطاء
+  - ترقية lxml إلى >=6.1.0 لمعالجة مشكلة الأمان GHSA-vfmq-68hx-4jfw
+
+  ### الوثائق
+  - إزالة الأسئلة الشائعة حول التسعير من صفحة البناء باستخدام الذكاء الاصطناعي عبر جميع اللغات
+
+  ### الأداء
+  - تحسين وقت بدء التشغيل البارد بنسبة ~29% من خلال التحميل الكسول لمجموعة أدوات MCP وأنواع الأحداث
+
+  ## المساهمون
+
+  @alex-clawd, @github-advanced-security[bot], @greysonlalonde, @iris-clawd, @lorenzejay, @mattatcha
+
+</Update>
+
 <Update label="22 أبريل 2026">
   ## v1.14.3a2
 

docs/ar/guides/coding-tools/build-with-ai.mdx

@@ -207,9 +207,6 @@ CrewAI AMP مُصمَّم لفرق الإنتاج. إليك ما تحصل علي
     - **Factory (استضافة ذاتية)** — على بنيتك التحتية لسيطرة كاملة على البيانات
     - **هجين** — دمج السحابة والاستضافة الذاتية حسب حساسية البيانات
   </Accordion>
-  <Accordion title="كيف يعمل التسعير؟">
-    سجّل في [app.crewai.com](https://app.crewai.com) لمعرفة الخطط الحالية. تسعير المؤسسات وFactory متاح عند الطلب.
-  </Accordion>
 </AccordionGroup>
 
 <Card title="استكشف CrewAI AMP →" icon="arrow-right" href="https://app.crewai.com">

docs/en/changelog.mdx

@@ -4,6 +4,32 @@ description: "Product updates, improvements, and bug fixes for CrewAI"
 icon: "clock"
 mode: "wide"
 ---
+<Update label="Apr 23, 2026">
+  ## v1.14.3a3
+
+  [View release on GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.14.3a3)
+
+  ## What's Changed
+
+  ### Features
+  - Add support for e2b
+  - Implement fallback to DefaultAzureCredential when no API key is provided
+
+  ### Bug Fixes
+  - Upgrade lxml to >=6.1.0 to address security issue GHSA-vfmq-68hx-4jfw
+
+  ### Documentation
+  - Remove pricing FAQ from build-with-ai page across all locales
+
+  ### Performance
+  - Improve cold start time by ~29% through lazy-loading of MCP SDK and event types
+
+  ## Contributors
+
+  @alex-clawd, @github-advanced-security[bot], @greysonlalonde, @iris-clawd, @lorenzejay, @mattatcha
+
+</Update>
+
 <Update label="Apr 22, 2026">
   ## v1.14.3a2
 

docs/en/guides/coding-tools/build-with-ai.mdx

@@ -207,9 +207,6 @@ CrewAI AMP is built for production teams. Here's what you get beyond deployment.
     - **Factory (self-hosted)** — run on your own infrastructure for full data control
     - **Hybrid** — mix cloud and self-hosted based on sensitivity requirements
   </Accordion>
-  <Accordion title="How does pricing work?">
-    Sign up at [app.crewai.com](https://app.crewai.com) to see current plans. Enterprise and Factory pricing is available on request.
-  </Accordion>
 </AccordionGroup>
 
 <Card title="Explore CrewAI AMP →" icon="arrow-right" href="https://app.crewai.com">

docs/ko/changelog.mdx

@@ -4,6 +4,32 @@ description: "CrewAI의 제품 업데이트, 개선 사항 및 버그 수정"
 icon: "clock"
 mode: "wide"
 ---
+<Update label="2026년 4월 23일">
+  ## v1.14.3a3
+
+  [GitHub 릴리스 보기](https://github.com/crewAIInc/crewAI/releases/tag/1.14.3a3)
+
+  ## 변경 사항
+
+  ### 기능
+  - e2b 지원 추가
+  - API 키가 제공되지 않을 경우 DefaultAzureCredential로 대체 구현
+
+  ### 버그 수정
+  - 보안 문제 GHSA-vfmq-68hx-4jfw를 해결하기 위해 lxml을 >=6.1.0으로 업그레이드
+
+  ### 문서
+  - 모든 지역에서 build-with-ai 페이지의 가격 FAQ 제거
+
+  ### 성능
+  - MCP SDK 및 이벤트 유형의 지연 로딩을 통해 콜드 스타트 시간을 약 29% 개선
+
+  ## 기여자
+
+  @alex-clawd, @github-advanced-security[bot], @greysonlalonde, @iris-clawd, @lorenzejay, @mattatcha
+
+</Update>
+
 <Update label="2026년 4월 22일">
   ## v1.14.3a2
 

docs/ko/guides/coding-tools/build-with-ai.mdx

@@ -207,9 +207,6 @@ CrewAI AMP는 프로덕션 팀을 위해 만들어졌습니다. 배포 외에
     - **Factory(셀프 호스팅)** — 데이터 통제를 위해 자체 인프라에서 실행
     - **하이브리드** — 민감도에 따라 클라우드와 셀프 호스팅을 혼합
   </Accordion>
-  <Accordion title="가격은 어떻게 되나요?">
-    [app.crewai.com](https://app.crewai.com)에 가입하면 현재 요금제를 확인할 수 있습니다. 엔터프라이즈 및 Factory 가격은 문의 시 안내합니다.
-  </Accordion>
 </AccordionGroup>
 
 <Card title="CrewAI AMP 살펴보기 →" icon="arrow-right" href="https://app.crewai.com">

docs/pt-BR/changelog.mdx

@@ -4,6 +4,32 @@ description: "Atualizações de produto, melhorias e correções do CrewAI"
 icon: "clock"
 mode: "wide"
 ---
+<Update label="23 abr 2026">
+  ## v1.14.3a3
+
+  [Ver release no GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.14.3a3)
+
+  ## O que Mudou
+
+  ### Recursos
+  - Adicionar suporte para e2b
+  - Implementar fallback para DefaultAzureCredential quando nenhuma chave de API for fornecida
+
+  ### Correções de Bugs
+  - Atualizar lxml para >=6.1.0 para resolver problema de segurança GHSA-vfmq-68hx-4jfw
+
+  ### Documentação
+  - Remover FAQ de preços da página build-with-ai em todos os locais
+
+  ### Desempenho
+  - Melhorar o tempo de inicialização a frio em ~29% através do carregamento preguiçoso do SDK MCP e tipos de eventos
+
+  ## Contributors
+
+  @alex-clawd, @github-advanced-security[bot], @greysonlalonde, @iris-clawd, @lorenzejay, @mattatcha
+
+</Update>
+
 <Update label="22 abr 2026">
   ## v1.14.3a2
 

docs/pt-BR/guides/coding-tools/build-with-ai.mdx

@@ -207,9 +207,6 @@ O CrewAI AMP foi feito para equipes em produção. Além da implantação, você
     - **Factory (self-hosted)** — na sua infraestrutura para controle total dos dados
     - **Híbrido** — combine nuvem e self-hosted conforme a sensibilidade dos dados
   </Accordion>
-  <Accordion title="Como funciona o preço?">
-    Cadastre-se em [app.crewai.com](https://app.crewai.com) para ver os planos atuais. Preços enterprise e Factory sob consulta.
-  </Accordion>
 </AccordionGroup>
 
 <Card title="Conheça o CrewAI AMP →" icon="arrow-right" href="https://app.crewai.com">

lib/crewai-files/src/crewai_files/__init__.py

@@ -152,4 +152,4 @@ __all__ = [
     "wrap_file_source",
 ]
 
-__version__ = "1.14.3a2"
+__version__ = "1.14.3a3"

lib/crewai-tools/pyproject.toml

@@ -10,7 +10,7 @@ requires-python = ">=3.10, <3.14"
 dependencies = [
     "pytube~=15.0.0",
     "requests>=2.33.0,<3",
-    "crewai==1.14.3a2",
+    "crewai==1.14.3a3",
     "tiktoken~=0.8.0",
     "beautifulsoup4~=4.13.4",
     "python-docx~=1.2.0",
@@ -112,7 +112,7 @@ github = [
 ]
 rag = [
     "python-docx>=1.1.0",
-    "lxml>=5.3.0,<5.4.0",  # Pin to avoid etree import issues in 5.4.0
+    "lxml>=6.1.0,<7",  # 6.1.0+ required for GHSA-vfmq-68hx-4jfw (XXE in iterparse)
 ]
 xml = [
     "unstructured[local-inference, all-docs]>=0.17.2"
@@ -143,6 +143,11 @@ daytona = [
     "daytona~=0.140.0",
 ]
 
+e2b = [
+    "e2b~=2.20.0",
+    "e2b-code-interpreter~=2.6.0",
+]
+
 
 [tool.uv]
 exclude-newer = "3 days"

lib/crewai-tools/src/crewai_tools/__init__.py

@@ -71,6 +71,11 @@ from crewai_tools.tools.directory_search_tool.directory_search_tool import (
     DirectorySearchTool,
 )
 from crewai_tools.tools.docx_search_tool.docx_search_tool import DOCXSearchTool
+from crewai_tools.tools.e2b_sandbox_tool import (
+    E2BExecTool,
+    E2BFileTool,
+    E2BPythonTool,
+)
 from crewai_tools.tools.exa_tools.exa_search_tool import EXASearchTool
 from crewai_tools.tools.file_read_tool.file_read_tool import FileReadTool
 from crewai_tools.tools.file_writer_tool.file_writer_tool import FileWriterTool
@@ -242,6 +247,9 @@ __all__ = [
     "DaytonaPythonTool",
     "DirectoryReadTool",
     "DirectorySearchTool",
+    "E2BExecTool",
+    "E2BFileTool",
+    "E2BPythonTool",
     "EXASearchTool",
     "EnterpriseActionTool",
     "FileCompressorTool",
@@ -313,4 +321,4 @@ __all__ = [
     "ZapierActionTools",
 ]
 
-__version__ = "1.14.3a2"
+__version__ = "1.14.3a3"

lib/crewai-tools/src/crewai_tools/tools/__init__.py

@@ -60,6 +60,11 @@ from crewai_tools.tools.directory_search_tool.directory_search_tool import (
     DirectorySearchTool,
 )
 from crewai_tools.tools.docx_search_tool.docx_search_tool import DOCXSearchTool
+from crewai_tools.tools.e2b_sandbox_tool import (
+    E2BExecTool,
+    E2BFileTool,
+    E2BPythonTool,
+)
 from crewai_tools.tools.exa_tools.exa_search_tool import EXASearchTool
 from crewai_tools.tools.file_read_tool.file_read_tool import FileReadTool
 from crewai_tools.tools.file_writer_tool.file_writer_tool import FileWriterTool
@@ -227,6 +232,9 @@ __all__ = [
     "DaytonaPythonTool",
     "DirectoryReadTool",
     "DirectorySearchTool",
+    "E2BExecTool",
+    "E2BFileTool",
+    "E2BPythonTool",
     "EXASearchTool",
     "FileCompressorTool",
     "FileReadTool",

lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/README.md

@@ -0,0 +1,120 @@
+# E2B Sandbox Tools
+
+Run shell commands, execute Python, and manage files inside an [E2B](https://e2b.dev/) sandbox. E2B provides isolated, ephemeral VMs suitable for agent-driven code execution, with a Jupyter-style code interpreter for rich Python results.
+
+Three tools are provided so you can pick what the agent actually needs:
+
+- **`E2BExecTool`** — run a shell command (`sandbox.commands.run`).
+- **`E2BPythonTool`** — run a Python cell in the E2B code interpreter (`sandbox.run_code`), returning stdout/stderr and rich results (charts, dataframes).
+- **`E2BFileTool`** — read / write / list / delete files (`sandbox.files.*`).
+
+## Installation
+
+```shell
+uv add "crewai-tools[e2b]"
+# or
+pip install "crewai-tools[e2b]"
+```
+
+Set the API key:
+
+```shell
+export E2B_API_KEY="..."
+```
+
+`E2B_DOMAIN` is also respected if set (for self-hosted or non-default deployments).
+
+## Sandbox lifecycle
+
+All three tools share the same lifecycle controls from `E2BBaseTool`:
+
+| Mode | When the sandbox is created | When it is killed |
+| --- | --- | --- |
+| **Ephemeral** (default, `persistent=False`) | On every `_run` call | At the end of that same call |
+| **Persistent** (`persistent=True`) | Lazily on first use | At process exit (via `atexit`), or manually via `tool.close()` |
+| **Attach** (`sandbox_id="…"`) | Never — the tool attaches to an existing sandbox | Never — the tool will not kill a sandbox it did not create |
+
+Ephemeral mode is the safe default: nothing leaks if the agent forgets to clean up. Use persistent mode when you want filesystem state or installed packages to carry across steps — this is typical when pairing `E2BFileTool` with `E2BExecTool`.
+
+E2B sandboxes also auto-expire after an idle timeout. Tune it via `sandbox_timeout` (seconds, default `300`).
+
+## Examples
+
+### One-shot Python execution (ephemeral)
+
+```python
+from crewai_tools import E2BPythonTool
+
+tool = E2BPythonTool()
+result = tool.run(code="print(sum(range(10)))")
+```
+
+### Multi-step shell session (persistent)
+
+```python
+from crewai_tools import E2BExecTool, E2BFileTool
+
+exec_tool = E2BExecTool(persistent=True)
+file_tool = E2BFileTool(persistent=True)
+
+# Each tool keeps its own persistent sandbox. If you need the *same* sandbox
+# across two tools, create one tool, grab the sandbox id via
+# `tool._persistent_sandbox.sandbox_id`, and pass it to the other via
+# `sandbox_id=...`.
+```
+
+### Attach to an existing sandbox
+
+```python
+from crewai_tools import E2BExecTool
+
+tool = E2BExecTool(sandbox_id="sbx_...")
+```
+
+### Custom create params
+
+```python
+tool = E2BExecTool(
+    persistent=True,
+    template="my-custom-template",
+    sandbox_timeout=600,
+    envs={"MY_FLAG": "1"},
+    metadata={"owner": "crewai-agent"},
+)
+```
+
+## Tool arguments
+
+### `E2BExecTool`
+- `command: str` — shell command to run.
+- `cwd: str | None` — working directory.
+- `envs: dict[str, str] | None` — extra env vars for this command.
+- `timeout: float | None` — seconds.
+
+### `E2BPythonTool`
+- `code: str` — source to execute.
+- `language: str | None` — override kernel language (default: Python).
+- `envs: dict[str, str] | None` — env vars for the run.
+- `timeout: float | None` — seconds.
+
+### `E2BFileTool`
+- `action: "read" | "write" | "append" | "list" | "delete" | "mkdir" | "info" | "exists"`
+- `path: str` — absolute path inside the sandbox.
+- `content: str | None` — required for `append`; optional for `write`.
+- `binary: bool` — if `True`, `content` is base64 on write / returned as base64 on read.
+- `depth: int` — for `list`, how many levels to recurse (default 1).
+
+## Security considerations
+
+These tools hand the LLM arbitrary shell, Python, and filesystem access inside a remote VM. The threat model to keep in mind:
+
+- **Prompt-injection is a code-execution vector.** If the agent ingests untrusted content (web pages, scraped documents, user-supplied files, emails, search results), a malicious instruction hidden in that content can coerce the agent into issuing commands to `E2BExecTool` / `E2BPythonTool`. Treat any pipeline that feeds untrusted text into an agent that also has these tools as equivalent to remote code execution — the LLM is the attacker's shell.
+- **Ephemeral mode (the default) is the main blast-radius control.** A fresh sandbox is created per call and killed at the end, so injected commands cannot persist state, exfiltrate long-lived secrets, or build up tooling across turns. Leave `persistent=False` unless you have a concrete reason to change it.
+- **Avoid this specific combination:**
+  - untrusted content in the agent's context, **plus**
+  - `persistent=True` or an explicit long-lived `sandbox_id`, **plus**
+  - a large `sandbox_timeout` or credentials/secrets seeded into the sandbox via `envs`.
+
+  That stack lets a single injection pivot into a long-running, credentialed shell that survives across turns. If you must run persistently, also keep `sandbox_timeout` short, scope `envs` to the minimum the task needs, and don't feed the same agent untrusted input.
+- **Don't mount production credentials.** Anything you put into `envs`, `metadata`, or files written to the sandbox is reachable from the LLM. Use per-task scoped keys, not your personal API tokens.
+- **E2B's VM isolation is the final backstop**, not a license to relax the above — isolation prevents escape to the host, but everything the sandbox can reach (the public internet, any service whose token you dropped in) is still fair game for an injected command.

lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/__init__.py

@@ -0,0 +1,12 @@
+from crewai_tools.tools.e2b_sandbox_tool.e2b_base_tool import E2BBaseTool
+from crewai_tools.tools.e2b_sandbox_tool.e2b_exec_tool import E2BExecTool
+from crewai_tools.tools.e2b_sandbox_tool.e2b_file_tool import E2BFileTool
+from crewai_tools.tools.e2b_sandbox_tool.e2b_python_tool import E2BPythonTool
+
+
+__all__ = [
+    "E2BBaseTool",
+    "E2BExecTool",
+    "E2BFileTool",
+    "E2BPythonTool",
+]

lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/e2b_base_tool.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+import atexit
+import logging
+import os
+import threading
+from typing import Any, ClassVar
+
+from crewai.tools import BaseTool, EnvVar
+from pydantic import ConfigDict, Field, PrivateAttr, SecretStr
+
+
+logger = logging.getLogger(__name__)
+
+
+class E2BBaseTool(BaseTool):
+    """Shared base for tools that act on an E2B sandbox.
+
+    Lifecycle modes:
+      - persistent=False (default): create a fresh sandbox per `_run` call and
+        kill it when the call returns. Safer and stateless — nothing leaks if
+        the agent forgets cleanup.
+      - persistent=True: lazily create a single sandbox on first use, cache it
+        on the instance, and register an atexit hook to kill it at process
+        exit. Cheaper across many calls and lets files/state carry over.
+      - sandbox_id=<existing>: attach to a sandbox the caller already owns.
+        Never killed by the tool.
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    package_dependencies: list[str] = Field(default_factory=lambda: ["e2b"])
+
+    api_key: SecretStr | None = Field(
+        default_factory=lambda: (
+            SecretStr(val) if (val := os.getenv("E2B_API_KEY")) else None
+        ),
+        description="E2B API key. Falls back to E2B_API_KEY env var.",
+        json_schema_extra={"required": False},
+        repr=False,
+    )
+    domain: str | None = Field(
+        default_factory=lambda: os.getenv("E2B_DOMAIN"),
+        description="E2B API domain override. Falls back to E2B_DOMAIN env var.",
+        json_schema_extra={"required": False},
+    )
+
+    template: str | None = Field(
+        default=None,
+        description=(
+            "Optional template/snapshot name or id to create the sandbox from. "
+            "Defaults to E2B's base template when omitted."
+        ),
+    )
+    persistent: bool = Field(
+        default=False,
+        description=(
+            "If True, reuse one sandbox across all calls to this tool instance "
+            "and kill it at process exit. Default False creates and kills a "
+            "fresh sandbox per call."
+        ),
+    )
+    sandbox_id: str | None = Field(
+        default=None,
+        description=(
+            "Attach to an existing sandbox by id instead of creating a new "
+            "one. The tool will never kill a sandbox it did not create."
+        ),
+    )
+    sandbox_timeout: int = Field(
+        default=300,
+        description=(
+            "Idle timeout in seconds after which E2B auto-kills the sandbox. "
+            "Applied at create time and when attaching via sandbox_id."
+        ),
+    )
+    envs: dict[str, str] | None = Field(
+        default=None,
+        description="Environment variables to set inside the sandbox at create time.",
+    )
+    metadata: dict[str, str] | None = Field(
+        default=None,
+        description="Metadata key-value pairs to attach to the sandbox at create time.",
+    )
+
+    env_vars: list[EnvVar] = Field(
+        default_factory=lambda: [
+            EnvVar(
+                name="E2B_API_KEY",
+                description="API key for E2B sandbox service",
+                required=False,
+            ),
+            EnvVar(
+                name="E2B_DOMAIN",
+                description="E2B API domain (optional)",
+                required=False,
+            ),
+        ]
+    )
+
+    _persistent_sandbox: Any | None = PrivateAttr(default=None)
+    _lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)
+    _cleanup_registered: bool = PrivateAttr(default=False)
+
+    _sdk_cache: ClassVar[dict[str, Any]] = {}
+
+    @classmethod
+    def _import_sandbox_class(cls) -> Any:
+        """Return the Sandbox class used by this tool.
+
+        Subclasses override this to swap in a different SDK (e.g. the code
+        interpreter sandbox). The default uses plain `e2b.Sandbox`.
+        """
+        cached = cls._sdk_cache.get("e2b.Sandbox")
+        if cached is not None:
+            return cached
+        try:
+            from e2b import Sandbox  # type: ignore[import-untyped]
+        except ImportError as exc:
+            raise ImportError(
+                "The 'e2b' package is required for E2B sandbox tools. "
+                "Install it with: uv add e2b  (or) pip install e2b"
+            ) from exc
+        cls._sdk_cache["e2b.Sandbox"] = Sandbox
+        return Sandbox
+
+    def _connect_kwargs(self) -> dict[str, Any]:
+        kwargs: dict[str, Any] = {}
+        if self.api_key is not None:
+            kwargs["api_key"] = self.api_key.get_secret_value()
+        if self.domain:
+            kwargs["domain"] = self.domain
+        if self.sandbox_timeout is not None:
+            kwargs["timeout"] = self.sandbox_timeout
+        return kwargs
+
+    def _create_kwargs(self) -> dict[str, Any]:
+        kwargs: dict[str, Any] = self._connect_kwargs()
+        if self.template is not None:
+            kwargs["template"] = self.template
+        if self.envs is not None:
+            kwargs["envs"] = self.envs
+        if self.metadata is not None:
+            kwargs["metadata"] = self.metadata
+        return kwargs
+
+    def _acquire_sandbox(self) -> tuple[Any, bool]:
+        """Return (sandbox, should_kill_after_use)."""
+        sandbox_cls = self._import_sandbox_class()
+
+        if self.sandbox_id:
+            return (
+                sandbox_cls.connect(self.sandbox_id, **self._connect_kwargs()),
+                False,
+            )
+
+        if self.persistent:
+            with self._lock:
+                if self._persistent_sandbox is None:
+                    self._persistent_sandbox = sandbox_cls.create(
+                        **self._create_kwargs()
+                    )
+                    if not self._cleanup_registered:
+                        atexit.register(self.close)
+                        self._cleanup_registered = True
+                return self._persistent_sandbox, False
+
+        sandbox = sandbox_cls.create(**self._create_kwargs())
+        return sandbox, True
+
+    def _release_sandbox(self, sandbox: Any, should_kill: bool) -> None:
+        if not should_kill:
+            return
+        try:
+            sandbox.kill()
+        except Exception:
+            logger.debug(
+                "Best-effort sandbox cleanup failed after ephemeral use; "
+                "the sandbox may need manual termination.",
+                exc_info=True,
+            )
+
+    def close(self) -> None:
+        """Kill the cached persistent sandbox if one exists."""
+        with self._lock:
+            sandbox = self._persistent_sandbox
+            self._persistent_sandbox = None
+        if sandbox is None:
+            return
+        try:
+            sandbox.kill()
+        except Exception:
+            logger.debug(
+                "Best-effort persistent sandbox cleanup failed at close(); "
+                "the sandbox may need manual termination.",
+                exc_info=True,
+            )

lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/e2b_exec_tool.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from builtins import type as type_
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from crewai_tools.tools.e2b_sandbox_tool.e2b_base_tool import E2BBaseTool
+
+
+class E2BExecToolSchema(BaseModel):
+    command: str = Field(..., description="Shell command to execute in the sandbox.")
+    cwd: str | None = Field(
+        default=None,
+        description="Working directory to run the command in. Defaults to the sandbox home dir.",
+    )
+    envs: dict[str, str] | None = Field(
+        default=None,
+        description="Optional environment variables to set for this command.",
+    )
+    timeout: float | None = Field(
+        default=None,
+        description="Maximum seconds to wait for the command to finish.",
+    )
+
+
+class E2BExecTool(E2BBaseTool):
+    """Run a shell command inside an E2B sandbox."""
+
+    name: str = "E2B Sandbox Exec"
+    description: str = (
+        "Execute a shell command inside an E2B sandbox and return the exit "
+        "code, stdout, and stderr. Use this to run builds, package installs, "
+        "git operations, or any one-off shell command."
+    )
+    args_schema: type_[BaseModel] = E2BExecToolSchema
+
+    def _run(
+        self,
+        command: str,
+        cwd: str | None = None,
+        envs: dict[str, str] | None = None,
+        timeout: float | None = None,
+    ) -> Any:
+        sandbox, should_kill = self._acquire_sandbox()
+        try:
+            run_kwargs: dict[str, Any] = {}
+            if cwd is not None:
+                run_kwargs["cwd"] = cwd
+            if envs is not None:
+                run_kwargs["envs"] = envs
+            if timeout is not None:
+                run_kwargs["timeout"] = timeout
+            result = sandbox.commands.run(command, **run_kwargs)
+            return {
+                "exit_code": getattr(result, "exit_code", None),
+                "stdout": getattr(result, "stdout", None),
+                "stderr": getattr(result, "stderr", None),
+                "error": getattr(result, "error", None),
+            }
+        finally:
+            self._release_sandbox(sandbox, should_kill)

lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/e2b_file_tool.py

@@ -0,0 +1,220 @@
+from __future__ import annotations
+
+import base64
+from builtins import type as type_
+import logging
+import posixpath
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+from crewai_tools.tools.e2b_sandbox_tool.e2b_base_tool import E2BBaseTool
+
+
+logger = logging.getLogger(__name__)
+
+
+FileAction = Literal[
+    "read", "write", "append", "list", "delete", "mkdir", "info", "exists"
+]
+
+
+class E2BFileToolSchema(BaseModel):
+    action: FileAction = Field(
+        ...,
+        description=(
+            "The filesystem action to perform: 'read' (returns file contents), "
+            "'write' (create or replace a file with content), 'append' (append "
+            "content to an existing file — use this for writing large files in "
+            "chunks to avoid hitting tool-call size limits), 'list' (lists a "
+            "directory), 'delete' (removes a file/dir), 'mkdir' (creates a "
+            "directory), 'info' (returns file metadata), 'exists' (returns a "
+            "boolean for whether the path exists)."
+        ),
+    )
+    path: str = Field(..., description="Absolute path inside the sandbox.")
+    content: str | None = Field(
+        default=None,
+        description=(
+            "Content to write or append. If omitted for 'write', an empty file "
+            "is created. For files larger than a few KB, prefer one 'write' "
+            "with empty content followed by multiple 'append' calls of ~4KB "
+            "each to stay within tool-call payload limits."
+        ),
+    )
+    binary: bool = Field(
+        default=False,
+        description=(
+            "For 'write'/'append': treat content as base64 and upload raw "
+            "bytes. For 'read': return contents as base64 instead of decoded "
+            "utf-8."
+        ),
+    )
+    depth: int = Field(
+        default=1,
+        description="For action='list': how many levels deep to recurse (default 1).",
+    )
+
+    @model_validator(mode="after")
+    def _validate_action_args(self) -> E2BFileToolSchema:
+        if self.action == "append" and self.content is None:
+            raise ValueError(
+                "action='append' requires 'content'. Pass the chunk to append "
+                "in the 'content' field."
+            )
+        return self
+
+
+class E2BFileTool(E2BBaseTool):
+    """Read, write, and manage files inside an E2B sandbox.
+
+    Notes:
+      - Most useful with `persistent=True` or an explicit `sandbox_id`. With
+        the default ephemeral mode, files disappear when this tool call
+        finishes.
+    """
+
+    name: str = "E2B Sandbox Files"
+    description: str = (
+        "Perform filesystem operations inside an E2B sandbox: read a file, "
+        "write content to a path, append content to an existing file, list a "
+        "directory, delete a path, make a directory, fetch file metadata, or "
+        "check whether a path exists. For files larger than a few KB, create "
+        "the file with action='write' and empty content, then send the body "
+        "via multiple 'append' calls of ~4KB each to stay within tool-call "
+        "payload limits."
+    )
+    args_schema: type_[BaseModel] = E2BFileToolSchema
+
+    def _run(
+        self,
+        action: FileAction,
+        path: str,
+        content: str | None = None,
+        binary: bool = False,
+        depth: int = 1,
+    ) -> Any:
+        sandbox, should_kill = self._acquire_sandbox()
+        try:
+            if action == "read":
+                return self._read(sandbox, path, binary=binary)
+            if action == "write":
+                return self._write(sandbox, path, content or "", binary=binary)
+            if action == "append":
+                return self._append(sandbox, path, content or "", binary=binary)
+            if action == "list":
+                return self._list(sandbox, path, depth=depth)
+            if action == "delete":
+                sandbox.files.remove(path)
+                return {"status": "deleted", "path": path}
+            if action == "mkdir":
+                created = sandbox.files.make_dir(path)
+                return {"status": "created", "path": path, "created": bool(created)}
+            if action == "info":
+                return self._info(sandbox, path)
+            if action == "exists":
+                return {"path": path, "exists": bool(sandbox.files.exists(path))}
+            raise ValueError(f"Unknown action: {action}")
+        finally:
+            self._release_sandbox(sandbox, should_kill)
+
+    def _read(self, sandbox: Any, path: str, *, binary: bool) -> dict[str, Any]:
+        if binary:
+            data: bytes = sandbox.files.read(path, format="bytes")
+            return {
+                "path": path,
+                "encoding": "base64",
+                "content": base64.b64encode(data).decode("ascii"),
+            }
+        try:
+            content: str = sandbox.files.read(path)
+            return {"path": path, "encoding": "utf-8", "content": content}
+        except UnicodeDecodeError:
+            data = sandbox.files.read(path, format="bytes")
+            return {
+                "path": path,
+                "encoding": "base64",
+                "content": base64.b64encode(data).decode("ascii"),
+                "note": "File was not valid utf-8; returned as base64.",
+            }
+
+    def _write(
+        self, sandbox: Any, path: str, content: str, *, binary: bool
+    ) -> dict[str, Any]:
+        payload: str | bytes = base64.b64decode(content) if binary else content
+        self._ensure_parent_dir(sandbox, path)
+        sandbox.files.write(path, payload)
+        size = (
+            len(payload)
+            if isinstance(payload, (bytes, bytearray))
+            else len(payload.encode("utf-8"))
+        )
+        return {"status": "written", "path": path, "bytes": size}
+
+    def _append(
+        self, sandbox: Any, path: str, content: str, *, binary: bool
+    ) -> dict[str, Any]:
+        chunk: bytes = base64.b64decode(content) if binary else content.encode("utf-8")
+        self._ensure_parent_dir(sandbox, path)
+        try:
+            existing: bytes = sandbox.files.read(path, format="bytes")
+        except Exception:
+            existing = b""
+        payload = existing + chunk
+        sandbox.files.write(path, payload)
+        return {
+            "status": "appended",
+            "path": path,
+            "appended_bytes": len(chunk),
+            "total_bytes": len(payload),
+        }
+
+    @staticmethod
+    def _ensure_parent_dir(sandbox: Any, path: str) -> None:
+        parent = posixpath.dirname(path)
+        if not parent or parent in ("/", "."):
+            return
+        try:
+            sandbox.files.make_dir(parent)
+        except Exception:
+            logger.debug(
+                "Best-effort parent-directory create failed for %s; "
+                "assuming it already exists and proceeding with the write.",
+                parent,
+                exc_info=True,
+            )
+
+    def _list(self, sandbox: Any, path: str, *, depth: int) -> dict[str, Any]:
+        entries = sandbox.files.list(path, depth=depth)
+        return {
+            "path": path,
+            "entries": [self._entry_to_dict(e) for e in entries],
+        }
+
+    def _info(self, sandbox: Any, path: str) -> dict[str, Any]:
+        return self._entry_to_dict(sandbox.files.get_info(path))
+
+    @staticmethod
+    def _entry_to_dict(entry: Any) -> dict[str, Any]:
+        fields = (
+            "name",
+            "path",
+            "type",
+            "size",
+            "mode",
+            "permissions",
+            "owner",
+            "group",
+            "modified_time",
+            "symlink_target",
+        )
+        result: dict[str, Any] = {}
+        for field in fields:
+            value = getattr(entry, field, None)
+            if value is not None and field == "modified_time":
+                result[field] = (
+                    value.isoformat() if hasattr(value, "isoformat") else str(value)
+                )
+            else:
+                result[field] = value
+        return result

lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/e2b_python_tool.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+from builtins import type as type_
+from typing import Any, ClassVar
+
+from pydantic import BaseModel, Field
+
+from crewai_tools.tools.e2b_sandbox_tool.e2b_base_tool import E2BBaseTool
+
+
+class E2BPythonToolSchema(BaseModel):
+    code: str = Field(
+        ...,
+        description="Python source to execute inside the sandbox.",
+    )
+    language: str | None = Field(
+        default=None,
+        description=(
+            "Override the execution language (e.g. 'python', 'r', 'javascript'). "
+            "Defaults to Python when omitted."
+        ),
+    )
+    envs: dict[str, str] | None = Field(
+        default=None,
+        description="Optional environment variables for the run.",
+    )
+    timeout: float | None = Field(
+        default=None,
+        description="Maximum seconds to wait for the code to finish.",
+    )
+
+
+class E2BPythonTool(E2BBaseTool):
+    """Run Python code inside an E2B code interpreter sandbox.
+
+    Uses `e2b_code_interpreter`, which runs cells in a persistent Jupyter-style
+    kernel so state (imports, variables) carries across calls when
+    `persistent=True`.
+    """
+
+    name: str = "E2B Sandbox Python"
+    description: str = (
+        "Execute a block of Python code inside an E2B code interpreter sandbox "
+        "and return captured stdout, stderr, the final expression value, and "
+        "any rich results (charts, dataframes). Use this for data processing, "
+        "quick scripts, or analysis that should run in an isolated environment."
+    )
+    args_schema: type_[BaseModel] = E2BPythonToolSchema
+
+    package_dependencies: list[str] = Field(
+        default_factory=lambda: ["e2b_code_interpreter"],
+    )
+
+    _ci_cache: ClassVar[dict[str, Any]] = {}
+
+    @classmethod
+    def _import_sandbox_class(cls) -> Any:
+        cached = cls._ci_cache.get("Sandbox")
+        if cached is not None:
+            return cached
+        try:
+            from e2b_code_interpreter import Sandbox  # type: ignore[import-untyped]
+        except ImportError as exc:
+            raise ImportError(
+                "The 'e2b_code_interpreter' package is required for the E2B "
+                "Python tool. Install it with: "
+                "uv add e2b-code-interpreter  (or) "
+                "pip install e2b-code-interpreter"
+            ) from exc
+        cls._ci_cache["Sandbox"] = Sandbox
+        return Sandbox
+
+    def _run(
+        self,
+        code: str,
+        language: str | None = None,
+        envs: dict[str, str] | None = None,
+        timeout: float | None = None,
+    ) -> Any:
+        sandbox, should_kill = self._acquire_sandbox()
+        try:
+            run_kwargs: dict[str, Any] = {}
+            if language is not None:
+                run_kwargs["language"] = language
+            if envs is not None:
+                run_kwargs["envs"] = envs
+            if timeout is not None:
+                run_kwargs["timeout"] = timeout
+            execution = sandbox.run_code(code, **run_kwargs)
+            return self._serialize_execution(execution)
+        finally:
+            self._release_sandbox(sandbox, should_kill)
+
+    @staticmethod
+    def _serialize_execution(execution: Any) -> dict[str, Any]:
+        logs = getattr(execution, "logs", None)
+        error = getattr(execution, "error", None)
+        results = getattr(execution, "results", None) or []
+        return {
+            "text": getattr(execution, "text", None),
+            "stdout": list(getattr(logs, "stdout", []) or []) if logs else [],
+            "stderr": list(getattr(logs, "stderr", []) or []) if logs else [],
+            "error": (
+                {
+                    "name": getattr(error, "name", None),
+                    "value": getattr(error, "value", None),
+                    "traceback": getattr(error, "traceback", None),
+                }
+                if error
+                else None
+            ),
+            "results": [E2BPythonTool._serialize_result(r) for r in results],
+            "execution_count": getattr(execution, "execution_count", None),
+        }
+
+    @staticmethod
+    def _serialize_result(result: Any) -> dict[str, Any]:
+        fields = (
+            "text",
+            "html",
+            "markdown",
+            "svg",
+            "png",
+            "jpeg",
+            "pdf",
+            "latex",
+            "json",
+            "javascript",
+            "data",
+            "is_main_result",
+            "extra",
+        )
+        return {field: getattr(result, field, None) for field in fields}

lib/crewai-tools/tool.specs.json

@@ -8734,6 +8734,668 @@
         "type": "object"
       }
     },
+    {
+      "description": "Execute a shell command inside an E2B sandbox and return the exit code, stdout, and stderr. Use this to run builds, package installs, git operations, or any one-off shell command.",
+      "env_vars": [
+        {
+          "default": null,
+          "description": "API key for E2B sandbox service",
+          "name": "E2B_API_KEY",
+          "required": false
+        },
+        {
+          "default": null,
+          "description": "E2B API domain (optional)",
+          "name": "E2B_DOMAIN",
+          "required": false
+        }
+      ],
+      "humanized_name": "E2B Sandbox Exec",
+      "init_params_schema": {
+        "$defs": {
+          "EnvVar": {
+            "properties": {
+              "default": {
+                "anyOf": [
+                  {
+                    "type": "string"
+                  },
+                  {
+                    "type": "null"
+                  }
+                ],
+                "default": null,
+                "title": "Default"
+              },
+              "description": {
+                "title": "Description",
+                "type": "string"
+              },
+              "name": {
+                "title": "Name",
+                "type": "string"
+              },
+              "required": {
+                "default": true,
+                "title": "Required",
+                "type": "boolean"
+              }
+            },
+            "required": [
+              "name",
+              "description"
+            ],
+            "title": "EnvVar",
+            "type": "object"
+          }
+        },
+        "description": "Run a shell command inside an E2B sandbox.",
+        "properties": {
+          "api_key": {
+            "anyOf": [
+              {
+                "format": "password",
+                "type": "string",
+                "writeOnly": true
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "E2B API key. Falls back to E2B_API_KEY env var.",
+            "required": false,
+            "title": "Api Key"
+          },
+          "domain": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "E2B API domain override. Falls back to E2B_DOMAIN env var.",
+            "required": false,
+            "title": "Domain"
+          },
+          "envs": {
+            "anyOf": [
+              {
+                "additionalProperties": {
+                  "type": "string"
+                },
+                "type": "object"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Environment variables to set inside the sandbox at create time.",
+            "title": "Envs"
+          },
+          "metadata": {
+            "anyOf": [
+              {
+                "additionalProperties": {
+                  "type": "string"
+                },
+                "type": "object"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Metadata key-value pairs to attach to the sandbox at create time.",
+            "title": "Metadata"
+          },
+          "persistent": {
+            "default": false,
+            "description": "If True, reuse one sandbox across all calls to this tool instance and kill it at process exit. Default False creates and kills a fresh sandbox per call.",
+            "title": "Persistent",
+            "type": "boolean"
+          },
+          "sandbox_id": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Attach to an existing sandbox by id instead of creating a new one. The tool will never kill a sandbox it did not create.",
+            "title": "Sandbox Id"
+          },
+          "sandbox_timeout": {
+            "default": 300,
+            "description": "Idle timeout in seconds after which E2B auto-kills the sandbox. Applied at create time and when attaching via sandbox_id.",
+            "title": "Sandbox Timeout",
+            "type": "integer"
+          },
+          "template": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Optional template/snapshot name or id to create the sandbox from. Defaults to E2B's base template when omitted.",
+            "title": "Template"
+          }
+        },
+        "required": [],
+        "title": "E2BExecTool",
+        "type": "object"
+      },
+      "name": "E2BExecTool",
+      "package_dependencies": [
+        "e2b"
+      ],
+      "run_params_schema": {
+        "properties": {
+          "command": {
+            "description": "Shell command to execute in the sandbox.",
+            "title": "Command",
+            "type": "string"
+          },
+          "cwd": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Working directory to run the command in. Defaults to the sandbox home dir.",
+            "title": "Cwd"
+          },
+          "envs": {
+            "anyOf": [
+              {
+                "additionalProperties": {
+                  "type": "string"
+                },
+                "type": "object"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Optional environment variables to set for this command.",
+            "title": "Envs"
+          },
+          "timeout": {
+            "anyOf": [
+              {
+                "type": "number"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Maximum seconds to wait for the command to finish.",
+            "title": "Timeout"
+          }
+        },
+        "required": [
+          "command"
+        ],
+        "title": "E2BExecToolSchema",
+        "type": "object"
+      }
+    },
+    {
+      "description": "Perform filesystem operations inside an E2B sandbox: read a file, write content to a path, append content to an existing file, list a directory, delete a path, make a directory, fetch file metadata, or check whether a path exists. For files larger than a few KB, create the file with action='write' and empty content, then send the body via multiple 'append' calls of ~4KB each to stay within tool-call payload limits.",
+      "env_vars": [
+        {
+          "default": null,
+          "description": "API key for E2B sandbox service",
+          "name": "E2B_API_KEY",
+          "required": false
+        },
+        {
+          "default": null,
+          "description": "E2B API domain (optional)",
+          "name": "E2B_DOMAIN",
+          "required": false
+        }
+      ],
+      "humanized_name": "E2B Sandbox Files",
+      "init_params_schema": {
+        "$defs": {
+          "EnvVar": {
+            "properties": {
+              "default": {
+                "anyOf": [
+                  {
+                    "type": "string"
+                  },
+                  {
+                    "type": "null"
+                  }
+                ],
+                "default": null,
+                "title": "Default"
+              },
+              "description": {
+                "title": "Description",
+                "type": "string"
+              },
+              "name": {
+                "title": "Name",
+                "type": "string"
+              },
+              "required": {
+                "default": true,
+                "title": "Required",
+                "type": "boolean"
+              }
+            },
+            "required": [
+              "name",
+              "description"
+            ],
+            "title": "EnvVar",
+            "type": "object"
+          }
+        },
+        "description": "Read, write, and manage files inside an E2B sandbox.\n\nNotes:\n  - Most useful with `persistent=True` or an explicit `sandbox_id`. With\n    the default ephemeral mode, files disappear when this tool call\n    finishes.",
+        "properties": {
+          "api_key": {
+            "anyOf": [
+              {
+                "format": "password",
+                "type": "string",
+                "writeOnly": true
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "E2B API key. Falls back to E2B_API_KEY env var.",
+            "required": false,
+            "title": "Api Key"
+          },
+          "domain": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "E2B API domain override. Falls back to E2B_DOMAIN env var.",
+            "required": false,
+            "title": "Domain"
+          },
+          "envs": {
+            "anyOf": [
+              {
+                "additionalProperties": {
+                  "type": "string"
+                },
+                "type": "object"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Environment variables to set inside the sandbox at create time.",
+            "title": "Envs"
+          },
+          "metadata": {
+            "anyOf": [
+              {
+                "additionalProperties": {
+                  "type": "string"
+                },
+                "type": "object"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Metadata key-value pairs to attach to the sandbox at create time.",
+            "title": "Metadata"
+          },
+          "persistent": {
+            "default": false,
+            "description": "If True, reuse one sandbox across all calls to this tool instance and kill it at process exit. Default False creates and kills a fresh sandbox per call.",
+            "title": "Persistent",
+            "type": "boolean"
+          },
+          "sandbox_id": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Attach to an existing sandbox by id instead of creating a new one. The tool will never kill a sandbox it did not create.",
+            "title": "Sandbox Id"
+          },
+          "sandbox_timeout": {
+            "default": 300,
+            "description": "Idle timeout in seconds after which E2B auto-kills the sandbox. Applied at create time and when attaching via sandbox_id.",
+            "title": "Sandbox Timeout",
+            "type": "integer"
+          },
+          "template": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Optional template/snapshot name or id to create the sandbox from. Defaults to E2B's base template when omitted.",
+            "title": "Template"
+          }
+        },
+        "required": [],
+        "title": "E2BFileTool",
+        "type": "object"
+      },
+      "name": "E2BFileTool",
+      "package_dependencies": [
+        "e2b"
+      ],
+      "run_params_schema": {
+        "properties": {
+          "action": {
+            "description": "The filesystem action to perform: 'read' (returns file contents), 'write' (create or replace a file with content), 'append' (append content to an existing file \u2014 use this for writing large files in chunks to avoid hitting tool-call size limits), 'list' (lists a directory), 'delete' (removes a file/dir), 'mkdir' (creates a directory), 'info' (returns file metadata), 'exists' (returns a boolean for whether the path exists).",
+            "enum": [
+              "read",
+              "write",
+              "append",
+              "list",
+              "delete",
+              "mkdir",
+              "info",
+              "exists"
+            ],
+            "title": "Action",
+            "type": "string"
+          },
+          "binary": {
+            "default": false,
+            "description": "For 'write'/'append': treat content as base64 and upload raw bytes. For 'read': return contents as base64 instead of decoded utf-8.",
+            "title": "Binary",
+            "type": "boolean"
+          },
+          "content": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Content to write or append. If omitted for 'write', an empty file is created. For files larger than a few KB, prefer one 'write' with empty content followed by multiple 'append' calls of ~4KB each to stay within tool-call payload limits.",
+            "title": "Content"
+          },
+          "depth": {
+            "default": 1,
+            "description": "For action='list': how many levels deep to recurse (default 1).",
+            "title": "Depth",
+            "type": "integer"
+          },
+          "path": {
+            "description": "Absolute path inside the sandbox.",
+            "title": "Path",
+            "type": "string"
+          }
+        },
+        "required": [
+          "action",
+          "path"
+        ],
+        "title": "E2BFileToolSchema",
+        "type": "object"
+      }
+    },
+    {
+      "description": "Execute a block of Python code inside an E2B code interpreter sandbox and return captured stdout, stderr, the final expression value, and any rich results (charts, dataframes). Use this for data processing, quick scripts, or analysis that should run in an isolated environment.",
+      "env_vars": [
+        {
+          "default": null,
+          "description": "API key for E2B sandbox service",
+          "name": "E2B_API_KEY",
+          "required": false
+        },
+        {
+          "default": null,
+          "description": "E2B API domain (optional)",
+          "name": "E2B_DOMAIN",
+          "required": false
+        }
+      ],
+      "humanized_name": "E2B Sandbox Python",
+      "init_params_schema": {
+        "$defs": {
+          "EnvVar": {
+            "properties": {
+              "default": {
+                "anyOf": [
+                  {
+                    "type": "string"
+                  },
+                  {
+                    "type": "null"
+                  }
+                ],
+                "default": null,
+                "title": "Default"
+              },
+              "description": {
+                "title": "Description",
+                "type": "string"
+              },
+              "name": {
+                "title": "Name",
+                "type": "string"
+              },
+              "required": {
+                "default": true,
+                "title": "Required",
+                "type": "boolean"
+              }
+            },
+            "required": [
+              "name",
+              "description"
+            ],
+            "title": "EnvVar",
+            "type": "object"
+          }
+        },
+        "description": "Run Python code inside an E2B code interpreter sandbox.\n\nUses `e2b_code_interpreter`, which runs cells in a persistent Jupyter-style\nkernel so state (imports, variables) carries across calls when\n`persistent=True`.",
+        "properties": {
+          "api_key": {
+            "anyOf": [
+              {
+                "format": "password",
+                "type": "string",
+                "writeOnly": true
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "E2B API key. Falls back to E2B_API_KEY env var.",
+            "required": false,
+            "title": "Api Key"
+          },
+          "domain": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "E2B API domain override. Falls back to E2B_DOMAIN env var.",
+            "required": false,
+            "title": "Domain"
+          },
+          "envs": {
+            "anyOf": [
+              {
+                "additionalProperties": {
+                  "type": "string"
+                },
+                "type": "object"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Environment variables to set inside the sandbox at create time.",
+            "title": "Envs"
+          },
+          "metadata": {
+            "anyOf": [
+              {
+                "additionalProperties": {
+                  "type": "string"
+                },
+                "type": "object"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Metadata key-value pairs to attach to the sandbox at create time.",
+            "title": "Metadata"
+          },
+          "persistent": {
+            "default": false,
+            "description": "If True, reuse one sandbox across all calls to this tool instance and kill it at process exit. Default False creates and kills a fresh sandbox per call.",
+            "title": "Persistent",
+            "type": "boolean"
+          },
+          "sandbox_id": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Attach to an existing sandbox by id instead of creating a new one. The tool will never kill a sandbox it did not create.",
+            "title": "Sandbox Id"
+          },
+          "sandbox_timeout": {
+            "default": 300,
+            "description": "Idle timeout in seconds after which E2B auto-kills the sandbox. Applied at create time and when attaching via sandbox_id.",
+            "title": "Sandbox Timeout",
+            "type": "integer"
+          },
+          "template": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Optional template/snapshot name or id to create the sandbox from. Defaults to E2B's base template when omitted.",
+            "title": "Template"
+          }
+        },
+        "required": [],
+        "title": "E2BPythonTool",
+        "type": "object"
+      },
+      "name": "E2BPythonTool",
+      "package_dependencies": [
+        "e2b_code_interpreter"
+      ],
+      "run_params_schema": {
+        "properties": {
+          "code": {
+            "description": "Python source to execute inside the sandbox.",
+            "title": "Code",
+            "type": "string"
+          },
+          "envs": {
+            "anyOf": [
+              {
+                "additionalProperties": {
+                  "type": "string"
+                },
+                "type": "object"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Optional environment variables for the run.",
+            "title": "Envs"
+          },
+          "language": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Override the execution language (e.g. 'python', 'r', 'javascript'). Defaults to Python when omitted.",
+            "title": "Language"
+          },
+          "timeout": {
+            "anyOf": [
+              {
+                "type": "number"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "description": "Maximum seconds to wait for the code to finish.",
+            "title": "Timeout"
+          }
+        },
+        "required": [
+          "code"
+        ],
+        "title": "E2BPythonToolSchema",
+        "type": "object"
+      }
+    },
     {
       "description": "Search the internet using Exa",
       "env_vars": [

lib/crewai/pyproject.toml

@@ -55,7 +55,7 @@ Repository = "https://github.com/crewAIInc/crewAI"
 
 [project.optional-dependencies]
 tools = [
-    "crewai-tools==1.14.3a2",
+    "crewai-tools==1.14.3a3",
 ]
 embeddings = [
     "tiktoken~=0.8.0"
@@ -94,12 +94,13 @@ google-genai = [
 ]
 azure-ai-inference = [
     "azure-ai-inference~=1.0.0b9",
+    "azure-identity>=1.17.0,<2",
 ]
 anthropic = [
     "anthropic~=0.73.0",
 ]
 a2a = [
-     "a2a-sdk~=0.3.10",
+     "a2a-sdk>=1.0.0,<2",
      "httpx-auth~=0.23.1",
      "httpx-sse~=0.4.0",
      "aiocache[redis,memcached]~=0.12.3",

lib/crewai/src/crewai/__init__.py

@@ -48,7 +48,7 @@ def _suppress_pydantic_deprecation_warnings() -> None:
 
 _suppress_pydantic_deprecation_warnings()
 
-__version__ = "1.14.3a2"
+__version__ = "1.14.3a3"
 
 _LAZY_IMPORTS: dict[str, tuple[str, str]] = {
     "Memory": ("crewai.memory.unified_memory", "Memory"),

lib/crewai/src/crewai/a2a/_compat.py

@@ -0,0 +1,318 @@
+"""Compatibility layer for a2a-sdk v0.3 → v1.0 migration.
+
+Centralizes import aliases and helper functions so the rest of the
+a2a module can use a single import regardless of SDK version.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from a2a.client.errors import A2AClientError
+
+
+# ---------------------------------------------------------------------------
+# Error re-exports
+# In v0.3 the class was called A2AClientHTTPError; v1.0 renamed it to
+# A2AClientError.  We expose the new name *and* an alias used across the
+# codebase so callers can migrate incrementally.
+# ---------------------------------------------------------------------------
+A2AClientHTTPError = A2AClientError  # back-compat alias
+
+# ---------------------------------------------------------------------------
+# Type helpers - Protobuf Part access
+# In v0.3 Part was a Pydantic discriminated-union with ``part.root.kind``
+# and ``part.root.text``; in v1.0 Part is a protobuf message with a
+# ``content`` oneof.
+# ---------------------------------------------------------------------------
+
+from a2a.types import Part  # noqa: E402
+
+
+def part_is_text(part: Part) -> bool:
+    """Return True when the Part carries text content."""
+    return part.HasField("text")
+
+
+def part_text(part: Part) -> str:
+    """Return the text payload of a Part (assumes text content)."""
+    return part.text
+
+
+def part_has_data(part: Part) -> bool:
+    """Return True when the Part carries structured data."""
+    return part.HasField("data")
+
+
+def part_has_file(part: Part) -> bool:
+    """Return True when the Part carries a file (url or raw bytes)."""
+    return part.HasField("url") or part.HasField("raw")
+
+
+# ---------------------------------------------------------------------------
+# Enum value aliases
+# v0.3: TaskState.completed, Role.user  (lower snake_case strings)
+# v1.0: TaskState.TASK_STATE_COMPLETED, Role.ROLE_USER  (SCREAMING_SNAKE_CASE)
+# ---------------------------------------------------------------------------
+
+from a2a.types import Role, TaskState  # noqa: E402
+
+
+# TaskState aliases
+TASK_STATE_SUBMITTED = TaskState.TASK_STATE_SUBMITTED
+TASK_STATE_WORKING = TaskState.TASK_STATE_WORKING
+TASK_STATE_COMPLETED = TaskState.TASK_STATE_COMPLETED
+TASK_STATE_FAILED = TaskState.TASK_STATE_FAILED
+TASK_STATE_CANCELED = TaskState.TASK_STATE_CANCELED
+TASK_STATE_INPUT_REQUIRED = TaskState.TASK_STATE_INPUT_REQUIRED
+TASK_STATE_AUTH_REQUIRED = TaskState.TASK_STATE_AUTH_REQUIRED
+TASK_STATE_REJECTED = TaskState.TASK_STATE_REJECTED
+
+# Role aliases
+ROLE_USER = Role.ROLE_USER
+ROLE_AGENT = Role.ROLE_AGENT
+
+
+# ---------------------------------------------------------------------------
+# Protobuf object helpers
+# Protobuf objects don't have model_dump() / model_copy().
+# ---------------------------------------------------------------------------
+
+from google.protobuf.json_format import (  # noqa: E402
+    MessageToDict,
+)
+
+
+def proto_to_json(msg: Any) -> str:
+    """Serialize a protobuf message to a JSON string.
+
+    Replaces ``msg.model_dump_json(...)`` from v0.3 Pydantic models.
+    """
+    from google.protobuf.json_format import MessageToJson
+
+    return MessageToJson(msg, preserving_proto_field_name=True, indent=2)
+
+
+def agent_card_to_dict(agent_card: Any, *, exclude_none: bool = True) -> dict[str, Any]:
+    """Serialize a protobuf AgentCard to a plain dict.
+
+    Works like ``agent_card.model_dump(exclude_none=True)`` did in v0.3.
+    """
+    return MessageToDict(
+        agent_card,
+        preserving_proto_field_name=True,
+        always_print_fields_with_no_presence=not exclude_none,
+    )
+
+
+def proto_copy(msg: Any) -> Any:
+    """Return a deep copy of a protobuf message (replaces model_copy)."""
+    new = type(msg)()
+    new.CopyFrom(msg)
+    return new
+
+
+# ---------------------------------------------------------------------------
+# Message / Part construction helpers
+# v0.3: Message(role=Role.user, parts=[Part(root=TextPart(text=...))])
+# v1.0: Message(role=Role.ROLE_USER, parts=[Part(text=...)])
+# ---------------------------------------------------------------------------
+
+from a2a.types import Message  # noqa: E402
+
+
+def new_text_part(text: str, **kwargs: Any) -> Part:
+    """Create a Part with text content (v1.0 style)."""
+    return Part(text=text, **kwargs)
+
+
+def new_text_message(
+    text: str,
+    *,
+    role: Any = ROLE_AGENT,
+    message_id: str | None = None,
+    context_id: str | None = None,
+    task_id: str | None = None,
+    **kwargs: Any,
+) -> Message:
+    """Create a Message with a single text Part."""
+    import uuid as _uuid
+
+    return Message(
+        role=role,
+        message_id=message_id or str(_uuid.uuid4()),
+        parts=[Part(text=text)],
+        context_id=context_id or "",
+        task_id=task_id or "",
+        **kwargs,
+    )
+
+
+def make_send_request(message: Message) -> Any:
+    """Wrap a Message in a SendMessageRequest (v1.0 API).
+
+    In v0.3, ``client.send_message(message)`` accepted a bare ``Message``.
+    In v1.0, it expects ``SendMessageRequest(message=message)``.
+    """
+    from a2a.types import SendMessageRequest
+
+    return SendMessageRequest(message=message)
+
+
+# ---------------------------------------------------------------------------
+# AgentCard field access helpers
+# v0.3: agent_card.url, agent_card.preferred_transport, agent_card.additional_interfaces
+# v1.0: agent_card.supported_interfaces, interface.url, interface.protocol_binding
+# ---------------------------------------------------------------------------
+
+from a2a.types import AgentCard, AgentInterface  # noqa: E402
+
+
+def agent_card_url(agent_card: AgentCard) -> str:
+    """Get the primary URL from an AgentCard.
+
+    In v0.3 this was ``agent_card.url``.
+    In v1.0 the URL lives inside ``supported_interfaces``.
+    """
+    if agent_card.supported_interfaces:
+        return agent_card.supported_interfaces[0].url
+    return ""
+
+
+def agent_card_preferred_transport(agent_card: AgentCard) -> str:
+    """Get the preferred transport protocol from an AgentCard.
+
+    In v0.3 this was ``agent_card.preferred_transport``.
+    In v1.0 it's the protocol_binding of the first supported_interface.
+    """
+    if agent_card.supported_interfaces:
+        return agent_card.supported_interfaces[0].protocol_binding
+    return "JSONRPC"
+
+
+def agent_card_interfaces(agent_card: AgentCard) -> list[AgentInterface]:
+    """Get all interfaces from an AgentCard.
+
+    In v0.3 these were split between the primary url and
+    ``agent_card.additional_interfaces``.
+    In v1.0 everything is in ``supported_interfaces``.
+    """
+    return (
+        list(agent_card.supported_interfaces) if agent_card.supported_interfaces else []
+    )
+
+
+def agent_card_protocol_version(agent_card: AgentCard) -> str:
+    """Get the protocol version from an AgentCard.
+
+    In v0.3 this was ``agent_card.protocol_version``.
+    In v1.0 it's per-interface in ``interface.protocol_version``.
+    """
+    if agent_card.supported_interfaces:
+        return agent_card.supported_interfaces[0].protocol_version or ""
+    return ""
+
+
+# ---------------------------------------------------------------------------
+# StreamResponse helpers
+# v0.3: send_message returned AsyncIterator[tuple[Task, Update] | Message]
+# v1.0: send_message returns AsyncIterator[StreamResponse]
+# ---------------------------------------------------------------------------
+from a2a.types import (  # noqa: E402
+    StreamResponse,
+    TaskStatusUpdateEvent,
+)
+
+
+def is_stream_message(chunk: StreamResponse) -> bool:
+    """Check if a StreamResponse contains a Message."""
+    return chunk.HasField("message")
+
+
+def is_stream_task(chunk: StreamResponse) -> bool:
+    """Check if a StreamResponse contains a Task."""
+    return chunk.HasField("task")
+
+
+def is_stream_status_update(chunk: StreamResponse) -> bool:
+    """Check if a StreamResponse contains a TaskStatusUpdateEvent."""
+    return chunk.HasField("status_update")
+
+
+def is_stream_artifact_update(chunk: StreamResponse) -> bool:
+    """Check if a StreamResponse contains a TaskArtifactUpdateEvent."""
+    return chunk.HasField("artifact_update")
+
+
+# ---------------------------------------------------------------------------
+# Client configuration helpers
+# v0.3: ClientConfig.supported_transports, push_notification_configs (list)
+# v1.0: ClientConfig.supported_protocol_bindings, push_notification_config (singular)
+# ---------------------------------------------------------------------------
+
+from a2a.client import ClientConfig  # noqa: E402
+from a2a.types import TaskPushNotificationConfig  # noqa: E402
+
+
+def create_client_config(
+    *,
+    httpx_client: Any = None,
+    supported_transports: list[str] | None = None,
+    streaming: bool = True,
+    polling: bool = False,
+    accepted_output_modes: list[str] | None = None,
+    push_notification_config: TaskPushNotificationConfig | None = None,
+    grpc_channel_factory: Any = None,
+) -> ClientConfig:
+    """Create a ClientConfig compatible with a2a-sdk v1.0."""
+    return ClientConfig(
+        httpx_client=httpx_client,
+        supported_protocol_bindings=supported_transports or ["JSONRPC"],
+        streaming=streaming,
+        polling=polling,
+        accepted_output_modes=accepted_output_modes
+        or ["text/plain", "application/json"],
+        push_notification_config=push_notification_config,
+        grpc_channel_factory=grpc_channel_factory,
+    )
+
+
+# ---------------------------------------------------------------------------
+# GetTaskRequest / SubscribeToTaskRequest
+# v0.3: TaskQueryParams, TaskIdParams
+# v1.0: GetTaskRequest, SubscribeToTaskRequest
+# ---------------------------------------------------------------------------
+
+from a2a.types import GetTaskRequest, SubscribeToTaskRequest  # noqa: E402
+
+
+# Expose v0.3 names as aliases for the v1.0 types
+TaskQueryParams = GetTaskRequest
+TaskIdParams = SubscribeToTaskRequest
+
+
+# ---------------------------------------------------------------------------
+# Task status helpers
+# v1.0 TaskStatusUpdateEvent no longer has a `final` field.  Finality is
+# determined by the task state being terminal.
+# ---------------------------------------------------------------------------
+
+TERMINAL_STATES: frozenset[int] = frozenset(
+    {
+        TASK_STATE_COMPLETED,
+        TASK_STATE_FAILED,
+        TASK_STATE_REJECTED,
+        TASK_STATE_CANCELED,
+    }
+)
+
+
+def is_status_update_final(update: TaskStatusUpdateEvent) -> bool:
+    """Determine if a status update is final.
+
+    In v0.3 this was ``update.final``.  In v1.0 finality is inferred from
+    the task state being terminal.
+    """
+    if update.status and update.status.state:
+        return update.status.state in TERMINAL_STATES
+    return False

lib/crewai/src/crewai/a2a/auth/utils.py

@@ -11,12 +11,13 @@ import re
 import threading
 from typing import Final, Literal, cast
 
-from a2a.client.errors import A2AClientHTTPError
+from a2a.client.errors import A2AClientError
 from a2a.types import (
     APIKeySecurityScheme,
     AgentCard,
     HTTPAuthSecurityScheme,
     OAuth2SecurityScheme,
+    SecurityScheme,
 )
 from httpx import AsyncClient, Response
 
@@ -112,7 +113,7 @@ def _raise_auth_mismatch(
         f"AgentCard requires {required} authentication, "
         f"but {type(provided_auth).__name__} was provided"
     )
-    raise A2AClientHTTPError(401, msg)
+    raise A2AClientError(msg)
 
 
 def parse_www_authenticate(header_value: str) -> dict[str, dict[str, str]]:
@@ -159,25 +160,44 @@ def validate_auth_against_agent_card(
         A2AClientHTTPError: If auth doesn't match AgentCard requirements (status_code=401).
     """
 
-    if not agent_card.security or not agent_card.security_schemes:
+    if not agent_card.security_requirements or not agent_card.security_schemes:
         return
 
     if not auth:
         msg = "AgentCard requires authentication but no auth scheme provided"
-        raise A2AClientHTTPError(401, msg)
+        raise A2AClientError(msg)
 
-    first_security_req = agent_card.security[0] if agent_card.security else {}
+    first_security_req = (
+        agent_card.security_requirements[0]
+        if agent_card.security_requirements
+        else None
+    )
+    if first_security_req is None:
+        return
 
-    for scheme_name in first_security_req.keys():
-        security_scheme_wrapper = agent_card.security_schemes.get(scheme_name)
+    for scheme_name in first_security_req.schemes.keys():
+        security_scheme_wrapper: SecurityScheme | None = (
+            agent_card.security_schemes.get(scheme_name)
+        )
         if not security_scheme_wrapper:
             continue
 
-        scheme = security_scheme_wrapper.root
+        scheme_field = security_scheme_wrapper.WhichOneof("scheme")
+        if scheme_field is None:
+            continue
 
-        if allowed_classes := _SCHEME_AUTH_MAPPING.get(type(scheme)):
-            if not isinstance(auth, allowed_classes):
-                _raise_auth_mismatch(allowed_classes, auth)
+        scheme = getattr(security_scheme_wrapper, scheme_field)
+
+        if isinstance(scheme, OAuth2SecurityScheme):
+            allowed = _SCHEME_AUTH_MAPPING.get(OAuth2SecurityScheme)
+            if allowed and not isinstance(auth, allowed):
+                _raise_auth_mismatch(allowed, auth)
+            return
+
+        if isinstance(scheme, APIKeySecurityScheme):
+            allowed = _SCHEME_AUTH_MAPPING.get(APIKeySecurityScheme)
+            if allowed and not isinstance(auth, allowed):
+                _raise_auth_mismatch(allowed, auth)
             return
 
         if isinstance(scheme, HTTPAuthSecurityScheme):
@@ -188,7 +208,7 @@ def validate_auth_against_agent_card(
             return
 
     msg = "Could not validate auth against AgentCard security requirements"
-    raise A2AClientHTTPError(401, msg)
+    raise A2AClientError(msg)
 
 
 async def retry_on_401(

lib/crewai/src/crewai/a2a/config.py

@@ -568,7 +568,9 @@ class A2AServerConfig(BaseModel):
         auth: Authentication scheme for A2A endpoints.
     """
 
-    model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
+    model_config: ClassVar[ConfigDict] = ConfigDict(
+        extra="forbid", arbitrary_types_allowed=True
+    )
 
     name: str | None = Field(
         default=None,

lib/crewai/src/crewai/a2a/extensions/registry.py

@@ -10,9 +10,7 @@ See: https://a2a-protocol.org/latest/topics/extensions/
 
 from __future__ import annotations
 
-from typing import Any
-
-from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
+from a2a.client.interceptors import BeforeArgs, ClientCallInterceptor
 from a2a.extensions.common import (
     HTTP_EXTENSION_HEADER,
 )
@@ -63,30 +61,15 @@ class ExtensionsMiddleware(ClientCallInterceptor):
         """
         self._extensions = extensions
 
-    async def intercept(
-        self,
-        method_name: str,
-        request_payload: dict[str, Any],
-        http_kwargs: dict[str, Any],
-        agent_card: AgentCard | None,
-        context: ClientCallContext | None,
-    ) -> tuple[dict[str, Any], dict[str, Any]]:
-        """Add extensions header to the request.
+    async def before(self, args: BeforeArgs) -> None:
+        """Add extensions header before the request is sent.
 
         Args:
-            method_name: The A2A method being called.
-            request_payload: The JSON-RPC request payload.
-            http_kwargs: HTTP request kwargs (headers, etc).
-            agent_card: The target agent's card.
-            context: Optional call context.
-
-        Returns:
-            Tuple of (request_payload, modified_http_kwargs).
+            args: The BeforeArgs containing method, input, agent_card, etc.
         """
-        if self._extensions:
-            headers = http_kwargs.setdefault("headers", {})
-            headers[HTTP_EXTENSION_HEADER] = ",".join(self._extensions)
-        return request_payload, http_kwargs
+        if self._extensions and isinstance(args.input, dict):
+            metadata = args.input.setdefault("metadata", {})
+            metadata[HTTP_EXTENSION_HEADER] = ",".join(self._extensions)
 
 
 def validate_required_extensions(

lib/crewai/src/crewai/a2a/task_helpers.py

@@ -4,22 +4,32 @@ from __future__ import annotations
 
 from collections.abc import AsyncIterator
 from typing import TYPE_CHECKING, Any
-import uuid
 
-from a2a.client.errors import A2AClientHTTPError
+from a2a.client.errors import A2AClientError
 from a2a.types import (
     AgentCard,
     Message,
-    Part,
-    Role,
-    Task,
-    TaskArtifactUpdateEvent,
-    TaskState,
-    TaskStatusUpdateEvent,
-    TextPart,
+    StreamResponse,
 )
 from typing_extensions import NotRequired, TypedDict
 
+from crewai.a2a._compat import (
+    ROLE_AGENT,
+    TASK_STATE_AUTH_REQUIRED,
+    TASK_STATE_CANCELED,
+    TASK_STATE_COMPLETED,
+    TASK_STATE_FAILED,
+    TASK_STATE_INPUT_REQUIRED,
+    TASK_STATE_REJECTED,
+    TASK_STATE_SUBMITTED,
+    TASK_STATE_WORKING,
+    agent_card_to_dict,
+    is_stream_message,
+    is_stream_task,
+    new_text_message,
+    part_is_text,
+    part_text,
+)
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.a2a_events import (
     A2AConnectionErrorEvent,
@@ -30,31 +40,29 @@ from crewai.events.types.a2a_events import (
 if TYPE_CHECKING:
     from a2a.types import Task as A2ATask
 
-SendMessageEvent = (
-    tuple[Task, TaskStatusUpdateEvent | TaskArtifactUpdateEvent | None] | Message
-)
+SendMessageEvent = StreamResponse
 
 
-TERMINAL_STATES: frozenset[TaskState] = frozenset(
+TERMINAL_STATES: frozenset[int] = frozenset(
     {
-        TaskState.completed,
-        TaskState.failed,
-        TaskState.rejected,
-        TaskState.canceled,
+        TASK_STATE_COMPLETED,
+        TASK_STATE_FAILED,
+        TASK_STATE_REJECTED,
+        TASK_STATE_CANCELED,
     }
 )
 
-ACTIONABLE_STATES: frozenset[TaskState] = frozenset(
+ACTIONABLE_STATES: frozenset[int] = frozenset(
     {
-        TaskState.input_required,
-        TaskState.auth_required,
+        TASK_STATE_INPUT_REQUIRED,
+        TASK_STATE_AUTH_REQUIRED,
     }
 )
 
-PENDING_STATES: frozenset[TaskState] = frozenset(
+PENDING_STATES: frozenset[int] = frozenset(
     {
-        TaskState.submitted,
-        TaskState.working,
+        TASK_STATE_SUBMITTED,
+        TASK_STATE_WORKING,
     }
 )
 
@@ -62,7 +70,7 @@ PENDING_STATES: frozenset[TaskState] = frozenset(
 class TaskStateResult(TypedDict):
     """Result dictionary from processing A2A task state."""
 
-    status: TaskState
+    status: int
     history: list[Message]
     result: NotRequired[str]
     error: NotRequired[str]
@@ -83,26 +91,22 @@ def extract_task_result_parts(a2a_task: A2ATask) -> list[str]:
 
     if a2a_task.status and a2a_task.status.message:
         msg = a2a_task.status.message
-        result_parts.extend(
-            part.root.text for part in msg.parts if part.root.kind == "text"
-        )
+        result_parts.extend(part_text(part) for part in msg.parts if part_is_text(part))
 
     if not result_parts and a2a_task.history:
         for history_msg in reversed(a2a_task.history):
-            if history_msg.role == Role.agent:
+            if history_msg.role == ROLE_AGENT:
                 result_parts.extend(
-                    part.root.text
-                    for part in history_msg.parts
-                    if part.root.kind == "text"
+                    part_text(part) for part in history_msg.parts if part_is_text(part)
                 )
                 break
 
     if a2a_task.artifacts:
         result_parts.extend(
-            part.root.text
+            part_text(part)
             for artifact in a2a_task.artifacts
             for part in artifact.parts
-            if part.root.kind == "text"
+            if part_is_text(part)
         )
 
     return result_parts
@@ -122,15 +126,15 @@ def extract_error_message(a2a_task: A2ATask, default: str) -> str:
         msg = a2a_task.status.message
         if msg:
             for part in msg.parts:
-                if part.root.kind == "text":
-                    return str(part.root.text)
-        return str(msg)
+                if part_is_text(part):
+                    return str(part_text(part))
+            return str(msg)
 
     if a2a_task.history:
         for history_msg in reversed(a2a_task.history):
             for part in history_msg.parts:
-                if part.root.kind == "text":
-                    return str(part.root.text)
+                if part_is_text(part):
+                    return str(part_text(part))
 
     return default
 
@@ -174,7 +178,7 @@ def process_task_state(
     if result_parts is None:
         result_parts = []
 
-    if a2a_task.status.state == TaskState.completed:
+    if a2a_task.status.state == TASK_STATE_COMPLETED:
         if not result_parts:
             extracted_parts = extract_task_result_parts(a2a_task)
             result_parts.extend(extracted_parts)
@@ -204,22 +208,21 @@ def process_task_state(
         )
 
         return TaskStateResult(
-            status=TaskState.completed,
-            agent_card=agent_card.model_dump(exclude_none=True),
+            status=TASK_STATE_COMPLETED,
+            agent_card=agent_card_to_dict(agent_card),
             result=response_text,
             history=new_messages,
         )
 
-    if a2a_task.status.state == TaskState.input_required:
+    if a2a_task.status.state == TASK_STATE_INPUT_REQUIRED:
         if a2a_task.history:
             new_messages.extend(a2a_task.history)
 
         response_text = extract_error_message(a2a_task, "Additional input required")
         if response_text and not a2a_task.history:
-            agent_message = Message(
-                role=Role.agent,
-                message_id=str(uuid.uuid4()),
-                parts=[Part(root=TextPart(text=response_text))],
+            agent_message = new_text_message(
+                response_text,
+                role=ROLE_AGENT,
                 context_id=a2a_task.context_id,
                 task_id=a2a_task.id,
             )
@@ -247,34 +250,34 @@ def process_task_state(
         )
 
         return TaskStateResult(
-            status=TaskState.input_required,
+            status=TASK_STATE_INPUT_REQUIRED,
             error=response_text,
             history=new_messages,
-            agent_card=agent_card.model_dump(exclude_none=True),
+            agent_card=agent_card_to_dict(agent_card),
         )
 
-    if a2a_task.status.state in {TaskState.failed, TaskState.rejected}:
+    if a2a_task.status.state in {TASK_STATE_FAILED, TASK_STATE_REJECTED}:
         error_msg = extract_error_message(a2a_task, "Task failed without error message")
         if a2a_task.history:
             new_messages.extend(a2a_task.history)
         return TaskStateResult(
-            status=TaskState.failed,
+            status=TASK_STATE_FAILED,
             error=error_msg,
             history=new_messages,
         )
 
-    if a2a_task.status.state == TaskState.auth_required:
+    if a2a_task.status.state == TASK_STATE_AUTH_REQUIRED:
         error_msg = extract_error_message(a2a_task, "Authentication required")
         return TaskStateResult(
-            status=TaskState.auth_required,
+            status=TASK_STATE_AUTH_REQUIRED,
             error=error_msg,
             history=new_messages,
         )
 
-    if a2a_task.status.state == TaskState.canceled:
+    if a2a_task.status.state == TASK_STATE_CANCELED:
         error_msg = extract_error_message(a2a_task, "Task was canceled")
         return TaskStateResult(
-            status=TaskState.canceled,
+            status=TASK_STATE_CANCELED,
             error=error_msg,
             history=new_messages,
         )
@@ -286,7 +289,7 @@ def process_task_state(
 
 
 async def send_message_and_get_task_id(
-    event_stream: AsyncIterator[SendMessageEvent],
+    event_stream: AsyncIterator[StreamResponse],
     new_messages: list[Message],
     agent_card: AgentCard,
     turn_number: int,
@@ -321,11 +324,12 @@ async def send_message_and_get_task_id(
         Task ID string if agent needs polling/waiting, or TaskStateResult if done.
     """
     try:
-        async for event in event_stream:
-            if isinstance(event, Message):
+        async for chunk in event_stream:
+            if is_stream_message(chunk):
+                event = chunk.message
                 new_messages.append(event)
                 result_parts = [
-                    part.root.text for part in event.parts if part.root.kind == "text"
+                    part_text(part) for part in event.parts if part_is_text(part)
                 ]
                 response_text = " ".join(result_parts) if result_parts else ""
 
@@ -348,14 +352,14 @@ async def send_message_and_get_task_id(
                 )
 
                 return TaskStateResult(
-                    status=TaskState.completed,
+                    status=TASK_STATE_COMPLETED,
                     result=response_text,
                     history=new_messages,
-                    agent_card=agent_card.model_dump(exclude_none=True),
+                    agent_card=agent_card_to_dict(agent_card),
                 )
 
-            if isinstance(event, tuple):
-                a2a_task, _ = event
+            if is_stream_task(chunk):
+                a2a_task = chunk.task
 
                 if a2a_task.status.state in TERMINAL_STATES | ACTIONABLE_STATES:
                     result = process_task_state(
@@ -376,18 +380,17 @@ async def send_message_and_get_task_id(
                 return a2a_task.id
 
         return TaskStateResult(
-            status=TaskState.failed,
+            status=TASK_STATE_FAILED,
             error="No task ID received from initial message",
             history=new_messages,
         )
 
-    except A2AClientHTTPError as e:
-        error_msg = f"HTTP Error {e.status_code}: {e!s}"
+    except A2AClientError as e:
+        error_msg = f"A2A Client Error: {e!s}"
 
-        error_message = Message(
-            role=Role.agent,
-            message_id=str(uuid.uuid4()),
-            parts=[Part(root=TextPart(text=error_msg))],
+        error_message = new_text_message(
+            error_msg,
+            role=ROLE_AGENT,
             context_id=context_id,
         )
         new_messages.append(error_message)
@@ -397,8 +400,7 @@ async def send_message_and_get_task_id(
             A2AConnectionErrorEvent(
                 endpoint=endpoint or "",
                 error=str(e),
-                error_type="http_error",
-                status_code=e.status_code,
+                error_type="client_error",
                 a2a_agent_name=a2a_agent_name,
                 operation="send_message",
                 context_id=context_id,
@@ -423,7 +425,7 @@ async def send_message_and_get_task_id(
             ),
         )
         return TaskStateResult(
-            status=TaskState.failed,
+            status=TASK_STATE_FAILED,
             error=error_msg,
             history=new_messages,
         )
@@ -431,10 +433,9 @@ async def send_message_and_get_task_id(
     except Exception as e:
         error_msg = f"Unexpected error during send_message: {e!s}"
 
-        error_message = Message(
-            role=Role.agent,
-            message_id=str(uuid.uuid4()),
-            parts=[Part(root=TextPart(text=error_msg))],
+        error_message = new_text_message(
+            error_msg,
+            role=ROLE_AGENT,
             context_id=context_id,
         )
         new_messages.append(error_message)
@@ -469,7 +470,7 @@ async def send_message_and_get_task_id(
             ),
         )
         return TaskStateResult(
-            status=TaskState.failed,
+            status=TASK_STATE_FAILED,
             error=error_msg,
             history=new_messages,
         )

lib/crewai/src/crewai/a2a/updates/polling/handler.py

@@ -5,21 +5,21 @@ from __future__ import annotations
 import asyncio
 import time
 from typing import TYPE_CHECKING, Any
-import uuid
 
 from a2a.client import Client
-from a2a.client.errors import A2AClientHTTPError
+from a2a.client.errors import A2AClientError
 from a2a.types import (
     AgentCard,
+    GetTaskRequest,
     Message,
-    Part,
-    Role,
-    TaskQueryParams,
-    TaskState,
-    TextPart,
 )
 from typing_extensions import Unpack
 
+from crewai.a2a._compat import (
+    ROLE_AGENT,
+    TASK_STATE_FAILED,
+    new_text_message,
+)
 from crewai.a2a.errors import A2APollingTimeoutError
 from crewai.a2a.task_helpers import (
     ACTIONABLE_STATES,
@@ -84,7 +84,7 @@ async def _poll_task_until_complete(
     while True:
         poll_count += 1
         task = await client.get_task(
-            TaskQueryParams(id=task_id, history_length=history_length)
+            GetTaskRequest(id=task_id, history_length=history_length)
         )
 
         elapsed = time.monotonic() - start_time
@@ -94,7 +94,7 @@ async def _poll_task_until_complete(
             A2APollingStatusEvent(
                 task_id=task_id,
                 context_id=effective_context_id,
-                state=str(task.status.state.value),
+                state=str(task.status.state),
                 elapsed_seconds=elapsed,
                 poll_count=poll_count,
                 endpoint=endpoint,
@@ -158,9 +158,11 @@ class PollingHandler:
         from_task = kwargs.get("from_task")
         from_agent = kwargs.get("from_agent")
 
+        from crewai.a2a._compat import make_send_request
+
         try:
             result_or_task_id = await send_message_and_get_task_id(
-                event_stream=client.send_message(message),
+                event_stream=client.send_message(make_send_request(message)),
                 new_messages=new_messages,
                 agent_card=agent_card,
                 turn_number=turn_number,
@@ -222,7 +224,7 @@ class PollingHandler:
                 return result
 
             return TaskStateResult(
-                status=TaskState.failed,
+                status=TASK_STATE_FAILED,
                 error=f"Unexpected task state: {final_task.status.state}",
                 history=new_messages,
             )
@@ -230,10 +232,9 @@ class PollingHandler:
         except A2APollingTimeoutError as e:
             error_msg = str(e)
 
-            error_message = Message(
-                role=Role.agent,
-                message_id=str(uuid.uuid4()),
-                parts=[Part(root=TextPart(text=error_msg))],
+            error_message = new_text_message(
+                error_msg,
+                role=ROLE_AGENT,
                 context_id=context_id,
                 task_id=task_id,
             )
@@ -256,18 +257,17 @@ class PollingHandler:
                 ),
             )
             return TaskStateResult(
-                status=TaskState.failed,
+                status=TASK_STATE_FAILED,
                 error=error_msg,
                 history=new_messages,
             )
 
-        except A2AClientHTTPError as e:
-            error_msg = f"HTTP Error {e.status_code}: {e!s}"
+        except A2AClientError as e:
+            error_msg = f"A2A Client Error: {e!s}"
 
-            error_message = Message(
-                role=Role.agent,
-                message_id=str(uuid.uuid4()),
-                parts=[Part(root=TextPart(text=error_msg))],
+            error_message = new_text_message(
+                error_msg,
+                role=ROLE_AGENT,
                 context_id=context_id,
                 task_id=task_id,
             )
@@ -278,8 +278,7 @@ class PollingHandler:
                 A2AConnectionErrorEvent(
                     endpoint=endpoint,
                     error=str(e),
-                    error_type="http_error",
-                    status_code=e.status_code,
+                    error_type="client_error",
                     a2a_agent_name=a2a_agent_name,
                     operation="polling",
                     context_id=context_id,
@@ -305,7 +304,7 @@ class PollingHandler:
                 ),
             )
             return TaskStateResult(
-                status=TaskState.failed,
+                status=TASK_STATE_FAILED,
                 error=error_msg,
                 history=new_messages,
             )
@@ -313,10 +312,9 @@ class PollingHandler:
         except Exception as e:
             error_msg = f"Unexpected error during polling: {e!s}"
 
-            error_message = Message(
-                role=Role.agent,
-                message_id=str(uuid.uuid4()),
-                parts=[Part(root=TextPart(text=error_msg))],
+            error_message = new_text_message(
+                error_msg,
+                role=ROLE_AGENT,
                 context_id=context_id,
                 task_id=task_id,
             )
@@ -353,7 +351,7 @@ class PollingHandler:
                 ),
             )
             return TaskStateResult(
-                status=TaskState.failed,
+                status=TASK_STATE_FAILED,
                 error=error_msg,
                 history=new_messages,
             )

lib/crewai/src/crewai/a2a/updates/push_notifications/config.py

@@ -2,9 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Annotated
+from typing import Annotated, Any
 
-from a2a.types import PushNotificationAuthenticationInfo
 from pydantic import AnyHttpUrl, BaseModel, BeforeValidator, Field
 
 from crewai.a2a.updates.base import PushNotificationResultStore
@@ -46,7 +45,7 @@ class PushNotificationConfig(BaseModel):
     url: AnyHttpUrl = Field(description="Callback URL for push notifications")
     id: str | None = Field(default=None, description="Unique config identifier")
     token: str | None = Field(default=None, description="Validation token")
-    authentication: PushNotificationAuthenticationInfo | None = Field(
+    authentication: Any | None = Field(
         default=None, description="Auth info for agent to use when calling webhook"
     )
     timeout: float | None = Field(

lib/crewai/src/crewai/a2a/updates/push_notifications/handler.py

@@ -4,20 +4,20 @@ from __future__ import annotations
 
 import logging
 from typing import TYPE_CHECKING, Any
-import uuid
 
 from a2a.client import Client
-from a2a.client.errors import A2AClientHTTPError
+from a2a.client.errors import A2AClientError
 from a2a.types import (
     AgentCard,
     Message,
-    Part,
-    Role,
-    TaskState,
-    TextPart,
 )
 from typing_extensions import Unpack
 
+from crewai.a2a._compat import (
+    ROLE_AGENT,
+    TASK_STATE_FAILED,
+    new_text_message,
+)
 from crewai.a2a.task_helpers import (
     TaskStateResult,
     process_task_state,
@@ -69,10 +69,9 @@ def _handle_push_error(
     Returns:
         TaskStateResult with failed status.
     """
-    error_message = Message(
-        role=Role.agent,
-        message_id=str(uuid.uuid4()),
-        parts=[Part(root=TextPart(text=error_msg))],
+    error_message = new_text_message(
+        error_msg,
+        role=ROLE_AGENT,
         context_id=params.context_id,
         task_id=task_id,
     )
@@ -110,7 +109,7 @@ def _handle_push_error(
         ),
     )
     return TaskStateResult(
-        status=TaskState.failed,
+        status=TASK_STATE_FAILED,
         error=error_msg,
         history=new_messages,
     )
@@ -221,7 +220,7 @@ class PushNotificationHandler:
                 ),
             )
             return TaskStateResult(
-                status=TaskState.failed,
+                status=TASK_STATE_FAILED,
                 error=error_msg,
                 history=new_messages,
             )
@@ -245,14 +244,16 @@ class PushNotificationHandler:
                 ),
             )
             return TaskStateResult(
-                status=TaskState.failed,
+                status=TASK_STATE_FAILED,
                 error=error_msg,
                 history=new_messages,
             )
 
+        from crewai.a2a._compat import make_send_request
+
         try:
             result_or_task_id = await send_message_and_get_task_id(
-                event_stream=client.send_message(message),
+                event_stream=client.send_message(make_send_request(message)),
                 new_messages=new_messages,
                 agent_card=agent_card,
                 turn_number=params.turn_number,
@@ -304,7 +305,7 @@ class PushNotificationHandler:
 
             if final_task is None:
                 return TaskStateResult(
-                    status=TaskState.failed,
+                    status=TASK_STATE_FAILED,
                     error=f"Push notification timeout after {polling_timeout}s",
                     history=new_messages,
                 )
@@ -325,21 +326,20 @@ class PushNotificationHandler:
                 return result
 
             return TaskStateResult(
-                status=TaskState.failed,
+                status=TASK_STATE_FAILED,
                 error=f"Unexpected task state: {final_task.status.state}",
                 history=new_messages,
             )
 
-        except A2AClientHTTPError as e:
+        except A2AClientError as e:
             return _handle_push_error(
                 error=e,
-                error_msg=f"HTTP Error {e.status_code}: {e!s}",
-                error_type="http_error",
+                error_msg=f"A2A Client Error: {e!s}",
+                error_type="client_error",
                 new_messages=new_messages,
                 agent_branch=agent_branch,
                 params=params,
                 task_id=task_id,
-                status_code=e.status_code,
             )
 
         except Exception as e:

lib/crewai/src/crewai/a2a/updates/streaming/handler.py

@@ -5,25 +5,30 @@ from __future__ import annotations
 import asyncio
 import logging
 from typing import Final
-import uuid
 
 from a2a.client import Client
-from a2a.client.errors import A2AClientHTTPError
+from a2a.client.errors import A2AClientError
 from a2a.types import (
     AgentCard,
+    GetTaskRequest,
     Message,
-    Part,
-    Role,
+    SubscribeToTaskRequest,
     Task,
     TaskArtifactUpdateEvent,
-    TaskIdParams,
-    TaskQueryParams,
-    TaskState,
-    TaskStatusUpdateEvent,
-    TextPart,
 )
 from typing_extensions import Unpack
 
+from crewai.a2a._compat import (
+    ROLE_AGENT,
+    TASK_STATE_FAILED,
+    is_stream_artifact_update,
+    is_stream_message,
+    is_stream_status_update,
+    is_stream_task,
+    new_text_message,
+    part_is_text,
+    part_text,
+)
 from crewai.a2a.task_helpers import (
     ACTIONABLE_STATES,
     TERMINAL_STATES,
@@ -50,6 +55,16 @@ MAX_RESUBSCRIBE_ATTEMPTS: Final[int] = 3
 RESUBSCRIBE_BACKOFF_BASE: Final[float] = 1.0
 
 
+def _extract_text_from_artifact(artifact: TaskArtifactUpdateEvent) -> list[str]:
+    """Extract text parts from an artifact update event."""
+    parts: list[str] = []
+    if artifact.artifact and artifact.artifact.parts:
+        parts.extend(
+            part_text(part) for part in artifact.artifact.parts if part_is_text(part)
+        )
+    return parts
+
+
 class StreamingHandler:
     """SSE streaming-based update handler."""
 
@@ -86,7 +101,7 @@ class StreamingHandler:
         params = extract_common_params(kwargs)  # type: ignore[arg-type]
 
         try:
-            a2a_task: Task = await client.get_task(TaskQueryParams(id=task_id))
+            a2a_task: Task = await client.get_task(GetTaskRequest(id=task_id))
 
             if a2a_task.status.state in TERMINAL_STATES:
                 logger.info(
@@ -138,26 +153,18 @@ class StreamingHandler:
                     if attempt > 0:
                         await asyncio.sleep(backoff)
 
-                    event_stream = client.resubscribe(TaskIdParams(id=task_id))
+                    event_stream = client.subscribe(SubscribeToTaskRequest(id=task_id))
 
-                    async for event in event_stream:
-                        if isinstance(event, tuple):
-                            resubscribed_task, update = event
+                    async for chunk in event_stream:
+                        if is_stream_task(chunk):
+                            resubscribed_task = chunk.task
 
-                            is_final_update = (
-                                process_status_update(update, result_parts)
-                                if isinstance(update, TaskStatusUpdateEvent)
-                                else False
+                        if is_stream_status_update(chunk):
+                            update = chunk.status_update
+                            is_final_update = process_status_update(
+                                update, result_parts
                             )
 
-                            if isinstance(update, TaskArtifactUpdateEvent):
-                                artifact = update.artifact
-                                result_parts.extend(
-                                    part.root.text
-                                    for part in artifact.parts
-                                    if part.root.kind == "text"
-                                )
-
                             if (
                                 is_final_update
                                 or resubscribed_task.status.state
@@ -178,15 +185,20 @@ class StreamingHandler:
                                     is_final=is_final_update,
                                 )
 
-                        elif isinstance(event, Message):
-                            new_messages.append(event)
+                        if is_stream_artifact_update(chunk):
+                            artifact = chunk.artifact_update
+                            result_parts.extend(_extract_text_from_artifact(artifact))
+
+                        if is_stream_message(chunk):
+                            msg = chunk.message
+                            new_messages.append(msg)
                             result_parts.extend(
-                                part.root.text
-                                for part in event.parts
-                                if part.root.kind == "text"
+                                part_text(part)
+                                for part in msg.parts
+                                if part_is_text(part)
                             )
 
-                    final_task = await client.get_task(TaskQueryParams(id=task_id))
+                    final_task = await client.get_task(GetTaskRequest(id=task_id))
                     return process_task_state(
                         a2a_task=final_task,
                         new_messages=new_messages,
@@ -258,9 +270,12 @@ class StreamingHandler:
 
         result_parts: list[str] = []
         final_result: TaskStateResult | None = None
-        event_stream = client.send_message(message)
+        from crewai.a2a._compat import make_send_request
+
+        event_stream = client.send_message(make_send_request(message))
         chunk_index = 0
         current_task_id: str | None = task_id
+        current_task: Task | None = None
 
         crewai_event_bus.emit(
             agent_branch,
@@ -278,22 +293,25 @@ class StreamingHandler:
         )
 
         try:
-            async for event in event_stream:
-                if isinstance(event, tuple):
-                    a2a_task, _ = event
-                    current_task_id = a2a_task.id
+            async for chunk in event_stream:
+                # Extract task from task payload
+                if is_stream_task(chunk):
+                    current_task = chunk.task
+                    current_task_id = current_task.id
 
-                if isinstance(event, Message):
-                    new_messages.append(event)
-                    message_context_id = event.context_id or params.context_id
-                    for part in event.parts:
-                        if part.root.kind == "text":
-                            text = part.root.text
+                # Handle standalone message responses
+                if is_stream_message(chunk):
+                    msg = chunk.message
+                    new_messages.append(msg)
+                    message_context_id = msg.context_id or params.context_id
+                    for part in msg.parts:
+                        if part_is_text(part):
+                            text = part_text(part)
                             result_parts.append(text)
                             crewai_event_bus.emit(
                                 agent_branch,
                                 A2AStreamingChunkEvent(
-                                    task_id=event.task_id or task_id,
+                                    task_id=msg.task_id or task_id,
                                     context_id=message_context_id,
                                     chunk=text,
                                     chunk_index=chunk_index,
@@ -307,38 +325,40 @@ class StreamingHandler:
                             )
                             chunk_index += 1
 
-                elif isinstance(event, tuple):
-                    a2a_task, update = event
-
-                    if isinstance(update, TaskArtifactUpdateEvent):
-                        artifact = update.artifact
+                # Handle artifact updates
+                elif is_stream_artifact_update(chunk):
+                    artifact_update = chunk.artifact_update
+                    artifact = artifact_update.artifact
+                    if artifact and artifact.parts:
                         result_parts.extend(
-                            part.root.text
+                            part_text(part)
                             for part in artifact.parts
-                            if part.root.kind == "text"
+                            if part_is_text(part)
                         )
                         artifact_size = None
                         if artifact.parts:
                             artifact_size = sum(
-                                len(p.root.text.encode())
-                                if p.root.kind == "text"
-                                else len(getattr(p.root, "data", b""))
+                                len(part_text(p).encode())
+                                if part_is_text(p)
+                                else len(getattr(p, "raw", b""))
                                 for p in artifact.parts
                             )
-                        effective_context_id = a2a_task.context_id or params.context_id
+                        effective_context_id = (
+                            current_task.context_id if current_task else None
+                        ) or params.context_id
                         crewai_event_bus.emit(
                             agent_branch,
                             A2AArtifactReceivedEvent(
-                                task_id=a2a_task.id,
+                                task_id=artifact_update.task_id or current_task_id,
                                 artifact_id=artifact.artifact_id,
                                 artifact_name=artifact.name,
                                 artifact_description=artifact.description,
-                                mime_type=artifact.parts[0].root.kind
-                                if artifact.parts
+                                mime_type="text"
+                                if artifact.parts and part_is_text(artifact.parts[0])
                                 else None,
                                 size_bytes=artifact_size,
-                                append=update.append or False,
-                                last_chunk=update.last_chunk or False,
+                                append=artifact_update.append or False,
+                                last_chunk=artifact_update.last_chunk or False,
                                 endpoint=params.endpoint,
                                 a2a_agent_name=params.a2a_agent_name,
                                 context_id=effective_context_id,
@@ -349,86 +369,63 @@ class StreamingHandler:
                             ),
                         )
 
-                    is_final_update = (
-                        process_status_update(update, result_parts)
-                        if isinstance(update, TaskStatusUpdateEvent)
-                        else False
-                    )
+                # Handle status updates
+                elif is_stream_status_update(chunk):
+                    update = chunk.status_update
+                    is_final_update = process_status_update(update, result_parts)
 
-                    if (
-                        not is_final_update
-                        and a2a_task.status.state
-                        not in TERMINAL_STATES | ACTIONABLE_STATES
+                    if current_task and (
+                        is_final_update
+                        or current_task.status.state
+                        in TERMINAL_STATES | ACTIONABLE_STATES
                     ):
+                        final_result = process_task_state(
+                            a2a_task=current_task,
+                            new_messages=new_messages,
+                            agent_card=agent_card,
+                            turn_number=params.turn_number,
+                            is_multiturn=params.is_multiturn,
+                            agent_role=params.agent_role,
+                            result_parts=result_parts,
+                            endpoint=params.endpoint,
+                            a2a_agent_name=params.a2a_agent_name,
+                            from_task=params.from_task,
+                            from_agent=params.from_agent,
+                            is_final=is_final_update,
+                        )
+                    elif not current_task and is_final_update:
+                        pass
+                    else:
                         continue
 
-                    final_result = process_task_state(
-                        a2a_task=a2a_task,
-                        new_messages=new_messages,
-                        agent_card=agent_card,
-                        turn_number=params.turn_number,
-                        is_multiturn=params.is_multiturn,
-                        agent_role=params.agent_role,
-                        result_parts=result_parts,
-                        endpoint=params.endpoint,
-                        a2a_agent_name=params.a2a_agent_name,
-                        from_task=params.from_task,
-                        from_agent=params.from_agent,
-                        is_final=is_final_update,
-                    )
-                    if final_result:
-                        break
+        except A2AClientError as e:
+            logger.warning(
+                "Stream interrupted",
+                extra={
+                    "task_id": current_task_id,
+                    "error": str(e),
+                    "error_type": type(e).__name__,
+                },
+            )
 
-        except A2AClientHTTPError as e:
             if current_task_id:
-                logger.info(
-                    "Stream interrupted with HTTP error, attempting recovery",
-                    extra={
-                        "task_id": current_task_id,
-                        "error": str(e),
-                        "status_code": e.status_code,
-                    },
+                recovery_result = await StreamingHandler._try_recover_from_interruption(
+                    client=client,
+                    task_id=current_task_id,
+                    new_messages=new_messages,
+                    agent_card=agent_card,
+                    result_parts=result_parts,
+                    **kwargs,
                 )
-                recovery_kwargs = {k: v for k, v in kwargs.items() if k != "task_id"}
-                recovered_result = (
-                    await StreamingHandler._try_recover_from_interruption(
-                        client=client,
-                        task_id=current_task_id,
-                        new_messages=new_messages,
-                        agent_card=agent_card,
-                        result_parts=result_parts,
-                        **recovery_kwargs,
-                    )
-                )
-                if recovered_result:
-                    logger.info(
-                        "Successfully recovered task after HTTP error",
-                        extra={
-                            "task_id": current_task_id,
-                            "status": str(recovered_result.get("status")),
-                        },
-                    )
-                    return recovered_result
+                if recovery_result:
+                    return recovery_result
 
-                logger.warning(
-                    "Failed to recover from HTTP error, returning failure",
-                    extra={
-                        "task_id": current_task_id,
-                        "status_code": e.status_code,
-                        "original_error": str(e),
-                    },
-                )
-
-            error_msg = f"HTTP Error {e.status_code}: {e!s}"
-            error_type = "http_error"
-            status_code = e.status_code
-
-            error_message = Message(
-                role=Role.agent,
-                message_id=str(uuid.uuid4()),
-                parts=[Part(root=TextPart(text=error_msg))],
+            error_msg = f"A2A Client Error: {e!s}"
+            error_message = new_text_message(
+                error_msg,
+                role=ROLE_AGENT,
                 context_id=params.context_id,
-                task_id=task_id,
+                task_id=current_task_id,
             )
             new_messages.append(error_message)
 
@@ -437,12 +434,11 @@ class StreamingHandler:
                 A2AConnectionErrorEvent(
                     endpoint=params.endpoint,
                     error=str(e),
-                    error_type=error_type,
-                    status_code=status_code,
+                    error_type="client_error",
                     a2a_agent_name=params.a2a_agent_name,
                     operation="streaming",
                     context_id=params.context_id,
-                    task_id=task_id,
+                    task_id=current_task_id,
                     from_task=params.from_task,
                     from_agent=params.from_agent,
                 ),
@@ -464,116 +460,40 @@ class StreamingHandler:
                 ),
             )
             return TaskStateResult(
-                status=TaskState.failed,
-                error=error_msg,
-                history=new_messages,
-            )
-
-        except (asyncio.TimeoutError, asyncio.CancelledError, ConnectionError) as e:
-            error_type = type(e).__name__.lower()
-            if current_task_id:
-                logger.info(
-                    f"Stream interrupted with {error_type}, attempting recovery",
-                    extra={"task_id": current_task_id, "error": str(e)},
-                )
-                recovery_kwargs = {k: v for k, v in kwargs.items() if k != "task_id"}
-                recovered_result = (
-                    await StreamingHandler._try_recover_from_interruption(
-                        client=client,
-                        task_id=current_task_id,
-                        new_messages=new_messages,
-                        agent_card=agent_card,
-                        result_parts=result_parts,
-                        **recovery_kwargs,
-                    )
-                )
-                if recovered_result:
-                    logger.info(
-                        f"Successfully recovered task after {error_type}",
-                        extra={
-                            "task_id": current_task_id,
-                            "status": str(recovered_result.get("status")),
-                        },
-                    )
-                    return recovered_result
-
-                logger.warning(
-                    f"Failed to recover from {error_type}, returning failure",
-                    extra={
-                        "task_id": current_task_id,
-                        "error_type": error_type,
-                        "original_error": str(e),
-                    },
-                )
-
-            error_msg = f"Connection error during streaming: {e!s}"
-            status_code = None
-
-            error_message = Message(
-                role=Role.agent,
-                message_id=str(uuid.uuid4()),
-                parts=[Part(root=TextPart(text=error_msg))],
-                context_id=params.context_id,
-                task_id=task_id,
-            )
-            new_messages.append(error_message)
-
-            crewai_event_bus.emit(
-                agent_branch,
-                A2AConnectionErrorEvent(
-                    endpoint=params.endpoint,
-                    error=str(e),
-                    error_type=error_type,
-                    status_code=status_code,
-                    a2a_agent_name=params.a2a_agent_name,
-                    operation="streaming",
-                    context_id=params.context_id,
-                    task_id=task_id,
-                    from_task=params.from_task,
-                    from_agent=params.from_agent,
-                ),
-            )
-            crewai_event_bus.emit(
-                agent_branch,
-                A2AResponseReceivedEvent(
-                    response=error_msg,
-                    turn_number=params.turn_number,
-                    context_id=params.context_id,
-                    is_multiturn=params.is_multiturn,
-                    status="failed",
-                    final=True,
-                    agent_role=params.agent_role,
-                    endpoint=params.endpoint,
-                    a2a_agent_name=params.a2a_agent_name,
-                    from_task=params.from_task,
-                    from_agent=params.from_agent,
-                ),
-            )
-            return TaskStateResult(
-                status=TaskState.failed,
+                status=TASK_STATE_FAILED,
                 error=error_msg,
                 history=new_messages,
             )
 
         except Exception as e:
-            logger.exception(
-                "Unexpected error during streaming",
+            logger.warning(
+                "Unexpected stream error",
                 extra={
                     "task_id": current_task_id,
+                    "error": str(e),
                     "error_type": type(e).__name__,
-                    "endpoint": params.endpoint,
                 },
+                exc_info=True,
             )
-            error_msg = f"Unexpected error during streaming: {type(e).__name__}: {e!s}"
-            error_type = "unexpected_error"
-            status_code = None
 
-            error_message = Message(
-                role=Role.agent,
-                message_id=str(uuid.uuid4()),
-                parts=[Part(root=TextPart(text=error_msg))],
+            if current_task_id:
+                recovery_result = await StreamingHandler._try_recover_from_interruption(
+                    client=client,
+                    task_id=current_task_id,
+                    new_messages=new_messages,
+                    agent_card=agent_card,
+                    result_parts=result_parts,
+                    **kwargs,
+                )
+                if recovery_result:
+                    return recovery_result
+
+            error_msg = f"Unexpected error during streaming: {e!s}"
+            error_message = new_text_message(
+                error_msg,
+                role=ROLE_AGENT,
                 context_id=params.context_id,
-                task_id=task_id,
+                task_id=current_task_id,
             )
             new_messages.append(error_message)
 
@@ -582,12 +502,11 @@ class StreamingHandler:
                 A2AConnectionErrorEvent(
                     endpoint=params.endpoint,
                     error=str(e),
-                    error_type=error_type,
-                    status_code=status_code,
+                    error_type="unexpected_error",
                     a2a_agent_name=params.a2a_agent_name,
                     operation="streaming",
                     context_id=params.context_id,
-                    task_id=task_id,
+                    task_id=current_task_id,
                     from_task=params.from_task,
                     from_agent=params.from_agent,
                 ),
@@ -609,38 +528,33 @@ class StreamingHandler:
                 ),
             )
             return TaskStateResult(
-                status=TaskState.failed,
+                status=TASK_STATE_FAILED,
                 error=error_msg,
                 history=new_messages,
             )
 
-        finally:
-            aclose = getattr(event_stream, "aclose", None)
-            if aclose:
-                try:
-                    await aclose()
-                except Exception as close_error:
-                    crewai_event_bus.emit(
-                        agent_branch,
-                        A2AConnectionErrorEvent(
-                            endpoint=params.endpoint,
-                            error=str(close_error),
-                            error_type="stream_close_error",
-                            a2a_agent_name=params.a2a_agent_name,
-                            operation="stream_close",
-                            context_id=params.context_id,
-                            task_id=task_id,
-                            from_task=params.from_task,
-                            from_agent=params.from_agent,
-                        ),
-                    )
-
         if final_result:
             return final_result
 
-        return TaskStateResult(
-            status=TaskState.completed,
-            result=" ".join(result_parts) if result_parts else "",
-            history=new_messages,
-            agent_card=agent_card.model_dump(exclude_none=True),
+        response_text = " ".join(result_parts) if result_parts else ""
+        crewai_event_bus.emit(
+            agent_branch,
+            A2AResponseReceivedEvent(
+                response=response_text,
+                turn_number=params.turn_number,
+                context_id=params.context_id,
+                is_multiturn=params.is_multiturn,
+                status="completed",
+                final=True,
+                agent_role=params.agent_role,
+                endpoint=params.endpoint,
+                a2a_agent_name=params.a2a_agent_name,
+                from_task=params.from_task,
+                from_agent=params.from_agent,
+            ),
+        )
+        return TaskStateResult(
+            status=TASK_STATE_FAILED,
+            error="Stream ended without terminal state",
+            history=new_messages,
         )

lib/crewai/src/crewai/a2a/updates/streaming/params.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 
 from a2a.types import TaskStatusUpdateEvent
 
+from crewai.a2a._compat import is_status_update_final, part_is_text, part_text
+
 
 def process_status_update(
     update: TaskStatusUpdateEvent,
@@ -18,11 +20,11 @@ def process_status_update(
     Returns:
         True if this is a final update, False otherwise.
     """
-    is_final = update.final
+    is_final = is_status_update_final(update)
     if update.status and update.status.message and update.status.message.parts:
         result_parts.extend(
-            part.root.text
+            part_text(part)
             for part in update.status.message.parts
-            if part.root.kind == "text" and part.root.text
+            if part_is_text(part) and part_text(part)
         )
     return is_final

lib/crewai/src/crewai/a2a/utils/agent_card.py

@@ -12,12 +12,18 @@ import time
 from types import MethodType
 from typing import TYPE_CHECKING
 
-from a2a.client.errors import A2AClientHTTPError
-from a2a.types import AgentCapabilities, AgentCard, AgentSkill
+from a2a.client.errors import A2AClientError
+from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
 from aiocache import cached  # type: ignore[import-untyped]
 from aiocache.serializers import PickleSerializer  # type: ignore[import-untyped]
+from google.protobuf.json_format import ParseDict
 import httpx
 
+from crewai.a2a._compat import (
+    agent_card_protocol_version,
+    agent_card_to_dict,
+    proto_copy,
+)
 from crewai.a2a.auth.client_schemes import APIKeyAuth, HTTPDigestAuth
 from crewai.a2a.auth.utils import (
     _auth_store,
@@ -277,9 +283,9 @@ async def _afetch_agent_card_impl(
             )
             response.raise_for_status()
 
-            agent_card = AgentCard.model_validate(response.json())
+            agent_card = ParseDict(response.json(), AgentCard())
             fetch_time_ms = (time.perf_counter() - start_time) * 1000
-            agent_card_dict = agent_card.model_dump(exclude_none=True)
+            agent_card_dict = agent_card_to_dict(agent_card)
 
             crewai_event_bus.emit(
                 None,
@@ -287,7 +293,7 @@ async def _afetch_agent_card_impl(
                     endpoint=endpoint,
                     a2a_agent_name=agent_card.name,
                     agent_card=agent_card_dict,
-                    protocol_version=agent_card.protocol_version,
+                    protocol_version=agent_card_protocol_version(agent_card),
                     provider=agent_card_dict.get("provider"),
                     cached=False,
                     fetch_time_ms=fetch_time_ms,
@@ -326,7 +332,7 @@ async def _afetch_agent_card_impl(
                     ),
                 )
 
-                raise A2AClientHTTPError(401, msg) from e
+                raise A2AClientError(msg) from e
 
             crewai_event_bus.emit(
                 None,
@@ -470,7 +476,9 @@ def _crew_to_agent_card(crew: Crew, url: str) -> AgentCard:
     return AgentCard(
         name=crew_name,
         description=" ".join(description_parts),
-        url=url,
+        supported_interfaces=[
+            AgentInterface(url=url, protocol_binding="JSONRPC"),
+        ],
         version="1.0.0",
         capabilities=AgentCapabilities(
             streaming=True,
@@ -540,28 +548,43 @@ def _agent_to_agent_card(agent: Agent, url: str) -> AgentCard:
             if ext.uri not in existing_uris:
                 existing_exts.append(ext)
 
-        capabilities = capabilities.model_copy(update={"extensions": existing_exts})
+        capabilities = proto_copy(capabilities)
+        del capabilities.extensions[:]
+        capabilities.extensions.extend(existing_exts)
+
+    primary_interface = AgentInterface(
+        url=server_config.url or url,
+        protocol_binding=server_config.transport.preferred or "JSONRPC",
+        protocol_version=server_config.protocol_version or "",
+    )
+    interfaces = [primary_interface]
+    if server_config.additional_interfaces:
+        interfaces.extend(server_config.additional_interfaces)
 
     card = AgentCard(
         name=name,
         description=description,
-        url=server_config.url or url,
+        supported_interfaces=interfaces,
         version=server_config.version,
         capabilities=capabilities,
         default_input_modes=server_config.default_input_modes,
         default_output_modes=server_config.default_output_modes,
         skills=skills,
-        preferred_transport=server_config.transport.preferred,
-        protocol_version=server_config.protocol_version,
-        provider=server_config.provider,
-        documentation_url=server_config.documentation_url,
-        icon_url=server_config.icon_url,
-        additional_interfaces=server_config.additional_interfaces,
-        security=server_config.security,
-        security_schemes=server_config.security_schemes,
-        supports_authenticated_extended_card=server_config.supports_authenticated_extended_card,
+        documentation_url=server_config.documentation_url or "",
+        icon_url=server_config.icon_url or "",
     )
 
+    if server_config.provider:
+        card.provider.CopyFrom(server_config.provider)
+
+    if server_config.security_schemes:
+        for k, v in server_config.security_schemes.items():
+            card.security_schemes[k].CopyFrom(v)
+
+    if server_config.security:
+        for req in server_config.security:
+            card.security_requirements.append(req)
+
     if server_config.signing_config:
         signature = sign_agent_card(
             card,
@@ -569,9 +592,11 @@ def _agent_to_agent_card(agent: Agent, url: str) -> AgentCard:
             key_id=server_config.signing_config.key_id,
             algorithm=server_config.signing_config.algorithm,
         )
-        card = card.model_copy(update={"signatures": [signature]})
+        del card.signatures[:]
+        card.signatures.append(signature)
     elif server_config.signatures:
-        card = card.model_copy(update={"signatures": server_config.signatures})
+        del card.signatures[:]
+        card.signatures.extend(server_config.signatures)
 
     return card
 

lib/crewai/src/crewai/a2a/utils/agent_card_signing.py

@@ -18,6 +18,7 @@ import logging
 from typing import Any, Literal
 
 from a2a.types import AgentCard, AgentCardSignature
+from google.protobuf.json_format import MessageToDict
 import jwt
 from pydantic import SecretStr
 
@@ -58,7 +59,11 @@ def _serialize_agent_card(agent_card: AgentCard) -> str:
     Returns:
         Canonical JSON string representation.
     """
-    card_dict = agent_card.model_dump(exclude={"signatures"}, exclude_none=True)
+    card_dict = MessageToDict(
+        agent_card,
+        preserving_proto_field_name=True,
+    )
+    card_dict.pop("signatures", None)
     return json.dumps(card_dict, sort_keys=True, separators=(",", ":"))
 
 

lib/crewai/src/crewai/a2a/utils/content_type.py

@@ -264,7 +264,7 @@ def negotiate_content_types(
         crewai_event_bus.emit(
             None,
             A2AContentTypeNegotiatedEvent(
-                endpoint=endpoint or agent_card.url,
+                endpoint=endpoint or "",
                 a2a_agent_name=a2a_agent_name or agent_card.name,
                 skill_name=skill_name,
                 client_input_modes=client_input_modes,
@@ -303,22 +303,21 @@ def get_part_content_type(part: Part) -> str:
     """Extract MIME type from an A2A Part.
 
     Args:
-        part: A Part object containing TextPart, DataPart, or FilePart.
+        part: A Part object (protobuf oneof: text, data, raw, url).
 
     Returns:
         The MIME type string for this part.
     """
-    root = part.root
-    if root.kind == "text":
+    if part.HasField("text"):
         return TEXT_PLAIN
-    if root.kind == "data":
-        metadata = root.metadata or {}
+    if part.HasField("data"):
+        metadata = dict(part.metadata) if part.metadata else {}
         mime = metadata.get("mimeType", "")
         if mime == APPLICATION_A2UI_JSON:
             return APPLICATION_A2UI_JSON
         return APPLICATION_JSON
-    if root.kind == "file":
-        return root.file.mime_type or APPLICATION_OCTET_STREAM
+    if part.HasField("raw") or part.HasField("url"):
+        return part.media_type or APPLICATION_OCTET_STREAM
     return APPLICATION_OCTET_STREAM
 
 

lib/crewai/src/crewai/a2a/utils/delegation.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import asyncio
-import base64
 from collections.abc import AsyncIterator, Callable, MutableMapping
 import concurrent.futures
 from contextlib import asynccontextmanager
@@ -12,20 +11,27 @@ import logging
 from typing import TYPE_CHECKING, Any, Final, Literal
 import uuid
 
-from a2a.client import Client, ClientConfig, ClientFactory
+from a2a.client import Client, ClientFactory
 from a2a.types import (
     AgentCard,
-    FilePart,
-    FileWithBytes,
     Message,
     Part,
-    PushNotificationConfig as A2APushNotificationConfig,
     Role,
-    TextPart,
+    TaskPushNotificationConfig,
 )
 import httpx
 from pydantic import BaseModel
 
+from crewai.a2a._compat import (
+    ROLE_USER,
+    agent_card_interfaces,
+    agent_card_protocol_version,
+    agent_card_to_dict,
+    agent_card_url,
+    create_client_config,
+    new_text_part,
+    proto_copy,
+)
 from crewai.a2a.auth.client_schemes import APIKeyAuth, HTTPDigestAuth
 from crewai.a2a.auth.utils import (
     _auth_store,
@@ -41,8 +47,6 @@ from crewai.a2a.task_helpers import TaskStateResult
 from crewai.a2a.types import (
     HANDLER_REGISTRY,
     HandlerType,
-    PartsDict,
-    PartsMetadataDict,
     TransportType,
 )
 from crewai.a2a.updates import (
@@ -107,13 +111,13 @@ def _create_file_parts(input_files: dict[str, Any] | None) -> list[Part]:
     parts: list[Part] = []
     for name, file_input in input_files.items():
         content_bytes = file_input.read()
-        content_base64 = base64.b64encode(content_bytes).decode()
-        file_with_bytes = FileWithBytes(
-            bytes=content_base64,
-            mimeType=file_input.content_type,
-            name=file_input.filename or name,
+        parts.append(
+            Part(
+                raw=content_bytes,
+                media_type=file_input.content_type or "application/octet-stream",
+                filename=file_input.filename or name,
+            )
         )
-        parts.append(Part(root=FilePart(file=file_with_bytes)))
 
     return parts
 
@@ -301,7 +305,7 @@ async def aexecute_a2a_delegation(
 
     is_multiturn = len(conversation_history) > 0
     if turn_number is None:
-        turn_number = len([m for m in conversation_history if m.role == Role.user]) + 1
+        turn_number = len([m for m in conversation_history if m.role == ROLE_USER]) + 1
 
     try:
         result = await _aexecute_a2a_delegation_impl(
@@ -349,7 +353,7 @@ async def aexecute_a2a_delegation(
         )
         raise
 
-    agent_card_data = result.get("agent_card")
+    agent_card_data: dict[str, Any] | None = result.get("agent_card")
     crewai_event_bus.emit(
         agent_branch,
         A2ADelegationCompletedEvent(
@@ -423,14 +427,14 @@ async def _aexecute_a2a_delegation_impl(
 
     unsupported_exts = validate_required_extensions(agent_card, client_extensions)
     if unsupported_exts:
-        ext_uris = [ext.uri for ext in unsupported_exts]
+        ext_uris = [e.uri for e in unsupported_exts]
         raise ValueError(
             f"Agent requires extensions not supported by client: {ext_uris}"
         )
 
     negotiated: NegotiatedTransport | None = None
     effective_transport: TransportType = transport.preferred or _DEFAULT_TRANSPORT
-    effective_url = endpoint
+    effective_url = agent_card_url(agent_card) or endpoint
 
     client_transports: list[str] = (
         list(transport.supported) if transport.supported else [_DEFAULT_TRANSPORT]
@@ -456,9 +460,9 @@ async def _aexecute_a2a_delegation_impl(
                 "endpoint": endpoint,
                 "client_transports": client_transports,
                 "server_transports": [
-                    iface.transport for iface in agent_card.additional_interfaces or []
-                ]
-                + [agent_card.preferred_transport or "JSONRPC"],
+                    iface.protocol_binding
+                    for iface in agent_card_interfaces(agent_card)
+                ],
             },
         )
 
@@ -476,11 +480,9 @@ async def _aexecute_a2a_delegation_impl(
 
     headers, _ = await _prepare_auth_headers(auth, timeout)
 
-    a2a_agent_name = None
-    if agent_card.name:
-        a2a_agent_name = agent_card.name
+    a2a_agent_name = agent_card.name or None
 
-    agent_card_dict = agent_card.model_dump(exclude_none=True)
+    agent_card_dict = agent_card_to_dict(agent_card)
     crewai_event_bus.emit(
         agent_branch,
         A2ADelegationStartedEvent(
@@ -492,7 +494,7 @@ async def _aexecute_a2a_delegation_impl(
             turn_number=turn_number,
             a2a_agent_name=a2a_agent_name,
             agent_card=agent_card_dict,
-            protocol_version=agent_card.protocol_version,
+            protocol_version=agent_card_protocol_version(agent_card),
             provider=agent_card_dict.get("provider"),
             skill_id=skill_id,
             metadata=metadata,
@@ -512,7 +514,7 @@ async def _aexecute_a2a_delegation_impl(
                 context_id=context_id,
                 a2a_agent_name=a2a_agent_name,
                 agent_card=agent_card_dict,
-                protocol_version=agent_card.protocol_version,
+                protocol_version=agent_card_protocol_version(agent_card),
                 provider=agent_card_dict.get("provider"),
                 skill_id=skill_id,
                 reference_task_ids=reference_task_ids,
@@ -534,26 +536,18 @@ async def _aexecute_a2a_delegation_impl(
         if first_task_id := conversation_history[0].task_id:
             task_id = first_task_id
 
-    parts: PartsDict = {"text": message_text}
-    if response_model:
-        parts.update(
-            {
-                "metadata": PartsMetadataDict(
-                    mimeType="application/json",
-                    schema=response_model.model_json_schema(),
-                )
-            }
-        )
-
     message_metadata = metadata.copy() if metadata else {}
     if skill_id:
         message_metadata["skill_id"] = skill_id
+    if response_model:
+        message_metadata["mimeType"] = "application/json"
+        message_metadata["schema"] = response_model.model_json_schema()
 
-    parts_list: list[Part] = [Part(root=TextPart(**parts))]
+    parts_list: list[Part] = [new_text_part(message_text)]
     parts_list.extend(_create_file_parts(input_files))
 
     message = Message(
-        role=Role.user,
+        role=ROLE_USER,
         message_id=str(uuid.uuid4()),
         parts=parts_list,
         context_id=context_id,
@@ -625,8 +619,11 @@ async def _aexecute_a2a_delegation_impl(
     use_streaming = not use_polling and push_config_for_client is None
 
     client_agent_card = agent_card
-    if effective_url != agent_card.url:
-        client_agent_card = agent_card.model_copy(update={"url": effective_url})
+    card_url = agent_card_url(agent_card)
+    if effective_url != card_url:
+        client_agent_card = proto_copy(agent_card)
+        if client_agent_card.supported_interfaces:
+            client_agent_card.supported_interfaces[0].url = effective_url
 
     async with _create_a2a_client(
         agent_card=client_agent_card,
@@ -649,7 +646,7 @@ async def _aexecute_a2a_delegation_impl(
             **handler_kwargs,
         )
         result["a2a_agent_name"] = a2a_agent_name
-        result["agent_card"] = agent_card.model_dump(exclude_none=True)
+        result["agent_card"] = agent_card_to_dict(agent_card)
         return result
 
 
@@ -933,15 +930,12 @@ async def _create_a2a_client(
         if auth and isinstance(auth, (HTTPDigestAuth, APIKeyAuth)):
             configure_auth_client(auth, httpx_client)
 
-        push_configs: list[A2APushNotificationConfig] = []
+        push_config: TaskPushNotificationConfig | None = None
         if push_notification_config is not None:
-            push_configs.append(
-                A2APushNotificationConfig(
-                    url=str(push_notification_config.url),
-                    id=push_notification_config.id,
-                    token=push_notification_config.token,
-                    authentication=push_notification_config.authentication,
-                )
+            push_config = TaskPushNotificationConfig(
+                url=str(push_notification_config.url),
+                id=push_notification_config.id or "",
+                token=push_notification_config.token or "",
             )
 
         grpc_channel_factory = None
@@ -951,13 +945,14 @@ async def _create_a2a_client(
                 auth=auth,
             )
 
-        config = ClientConfig(
+        config = create_client_config(
             httpx_client=httpx_client,
             supported_transports=[transport_protocol],
             streaming=streaming and not use_polling,
             polling=use_polling,
-            accepted_output_modes=accepted_output_modes or DEFAULT_CLIENT_OUTPUT_MODES,  # type: ignore[arg-type]
-            push_notification_configs=push_configs,
+            accepted_output_modes=accepted_output_modes
+            or list(DEFAULT_CLIENT_OUTPUT_MODES),
+            push_notification_config=push_config,
             grpc_channel_factory=grpc_channel_factory,
         )
 
@@ -965,6 +960,6 @@ async def _create_a2a_client(
         client = factory.create(agent_card)
 
         if client_extensions:
-            await client.add_request_middleware(ExtensionsMiddleware(client_extensions))
+            await client.add_interceptor(ExtensionsMiddleware(client_extensions))
 
         yield client

lib/crewai/src/crewai/a2a/utils/task.py

@@ -13,13 +13,15 @@ import os
 from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast
 from urllib.parse import urlparse
 
+from a2a.helpers.proto_helpers import (
+    new_artifact,
+    new_text_artifact,
+    new_text_message as new_agent_text_message,
+)
 from a2a.server.agent_execution import RequestContext
 from a2a.server.events import EventQueue
 from a2a.types import (
     Artifact,
-    FileWithBytes,
-    FileWithUri,
-    InternalError,
     InvalidParamsError,
     Message,
     Part,
@@ -28,18 +30,20 @@ from a2a.types import (
     TaskStatus,
     TaskStatusUpdateEvent,
 )
-from a2a.utils import (
-    get_data_parts,
-    get_file_parts,
-    new_agent_text_message,
-    new_data_artifact,
-    new_text_artifact,
-)
-from a2a.utils.errors import ServerError
+from a2a.utils.errors import A2AError as ServerError
 from aiocache import SimpleMemoryCache, caches  # type: ignore[import-untyped]
 from pydantic import BaseModel
 from typing_extensions import TypedDict
 
+from crewai.a2a._compat import (
+    TASK_STATE_CANCELED,
+    TASK_STATE_COMPLETED,
+    TASK_STATE_FAILED,
+    part_has_data,
+    part_has_file,
+    part_is_text,
+    proto_copy,
+)
 from crewai.a2a.utils.agent_card import _get_server_config
 from crewai.a2a.utils.content_type import validate_message_parts
 from crewai.events.event_bus import crewai_event_bus
@@ -64,6 +68,28 @@ P = ParamSpec("P")
 T = TypeVar("T")
 
 
+def _get_data_parts(parts: list[Part]) -> list[dict[str, Any]]:
+    """Extract data parts from a list of protobuf Parts.
+
+    In a2a-sdk v1.0, data is stored via the ``data`` oneof field on Part
+    (a ``google.protobuf.Value``).
+    """
+    result: list[dict[str, Any]] = []
+    for part in parts:
+        if part_has_data(part):
+            from google.protobuf.json_format import MessageToDict
+
+            val = MessageToDict(part.data)
+            if isinstance(val, dict):
+                result.append(val)
+    return result
+
+
+def _get_file_parts(parts: list[Part]) -> list[Part]:
+    """Return parts that carry file content (raw bytes or url)."""
+    return [p for p in parts if part_has_file(p)]
+
+
 class RedisCacheConfig(TypedDict, total=False):
     """Configuration for aiocache Redis backend."""
 
@@ -196,12 +222,12 @@ def cancellable(
 
 
 def _convert_a2a_files_to_file_inputs(
-    a2a_files: list[FileWithBytes | FileWithUri],
+    a2a_files: list[Part],
 ) -> dict[str, Any]:
-    """Convert a2a file types to crewai FileInput dict.
+    """Convert a2a file parts to crewai FileInput dict.
 
     Args:
-        a2a_files: List of FileWithBytes or FileWithUri from a2a SDK.
+        a2a_files: List of Parts that carry file content (raw or url).
 
     Returns:
         Dictionary mapping file names to FileInput objects.
@@ -213,15 +239,13 @@ def _convert_a2a_files_to_file_inputs(
         return {}
 
     file_dict: dict[str, Any] = {}
-    for idx, a2a_file in enumerate(a2a_files):
-        if isinstance(a2a_file, FileWithBytes):
-            file_bytes = base64.b64decode(a2a_file.bytes)
-            name = a2a_file.name or f"file_{idx}"
-            file_source = FileBytes(data=file_bytes, filename=a2a_file.name)
+    for idx, part in enumerate(a2a_files):
+        name = part.filename or f"file_{idx}"
+        if part.HasField("raw"):
+            file_source = FileBytes(data=part.raw, filename=name)
             file_dict[name] = File(source=file_source)
-        elif isinstance(a2a_file, FileWithUri):
-            name = a2a_file.name or f"file_{idx}"
-            file_dict[name] = File(source=a2a_file.uri)
+        elif part.HasField("url"):
+            file_dict[name] = File(source=part.url)
 
     return file_dict
 
@@ -239,8 +263,9 @@ def _extract_response_schema(parts: list[Part]) -> dict[str, Any] | None:
         JSON schema dict if found, None otherwise.
     """
     for part in parts:
-        if part.root.kind == "text" and part.root.metadata:
-            schema = part.root.metadata.get("schema")
+        if part_is_text(part) and part.metadata:
+            metadata_dict = dict(part.metadata)
+            schema = metadata_dict.get("schema")
             if schema and isinstance(schema, dict):
                 return schema  # type: ignore[no-any-return]
     return None
@@ -261,9 +286,17 @@ def _create_result_artifact(
     """
     artifact_name = f"result_{task_id}"
     if isinstance(result, dict):
-        return new_data_artifact(artifact_name, result)
+        from google.protobuf import struct_pb2
+
+        val = struct_pb2.Value()
+        val.struct_value.update(result)
+        return new_artifact([Part(data=val)], artifact_name)
     if isinstance(result, BaseModel):
-        return new_data_artifact(artifact_name, result.model_dump())
+        from google.protobuf import struct_pb2
+
+        val = struct_pb2.Value()
+        val.struct_value.update(result.model_dump())
+        return new_artifact([Part(data=val)], artifact_name)
     return new_text_artifact(artifact_name, str(result))
 
 
@@ -330,7 +363,7 @@ async def _execute_impl(
 
     response_model: type[BaseModel] | None = None
     structured_inputs: list[dict[str, Any]] = []
-    a2a_files: list[FileWithBytes | FileWithUri] = []
+    a2a_files: list[Part] = []
 
     if context.message and context.message.parts:
         schema = _extract_response_schema(context.message.parts)
@@ -343,8 +376,8 @@ async def _execute_impl(
                     extra={"error": str(e), "schema_title": schema.get("title")},
                 )
 
-        structured_inputs = get_data_parts(context.message.parts)
-        a2a_files = get_file_parts(context.message.parts)
+        structured_inputs = _get_data_parts(context.message.parts)
+        a2a_files = _get_file_parts(context.message.parts)
 
     task_id = context.task_id
     context_id = context.context_id
@@ -387,12 +420,14 @@ async def _execute_impl(
             )
         result_str = str(result)
         history: list[Message] = [context.message] if context.message else []
-        history.append(new_agent_text_message(result_str, context_id, task_id))
+        history.append(
+            new_agent_text_message(result_str, context_id=context_id, task_id=task_id)
+        )
         await event_queue.enqueue_event(
             A2ATask(
                 id=task_id,
                 context_id=context_id,
-                status=TaskStatus(state=TaskState.completed),
+                status=TaskStatus(state=TASK_STATE_COMPLETED),
                 artifacts=[_create_result_artifact(result, task_id)],
                 history=history,
             )
@@ -429,9 +464,7 @@ async def _execute_impl(
                 from_agent=agent,
             ),
         )
-        raise ServerError(
-            error=InternalError(message=f"Task execution failed: {e}")
-        ) from e
+        raise ServerError(f"Task execution failed: {e}") from e
 
 
 async def execute_with_extensions(
@@ -476,9 +509,9 @@ async def cancel(
         raise ServerError(InvalidParamsError(message="task_id and context_id required"))
 
     if context.current_task and context.current_task.status.state in (
-        TaskState.completed,
-        TaskState.failed,
-        TaskState.canceled,
+        TASK_STATE_COMPLETED,
+        TASK_STATE_FAILED,
+        TASK_STATE_CANCELED,
     ):
         return context.current_task
 
@@ -492,13 +525,12 @@ async def cancel(
         TaskStatusUpdateEvent(
             task_id=task_id,
             context_id=context_id,
-            status=TaskStatus(state=TaskState.canceled),
-            final=True,
+            status=TaskStatus(state=TASK_STATE_CANCELED),
         )
     )
 
     if context.current_task:
-        context.current_task.status = TaskStatus(state=TaskState.canceled)
+        context.current_task.status.CopyFrom(TaskStatus(state=TASK_STATE_CANCELED))
         return context.current_task
     return None
 
@@ -571,7 +603,7 @@ def list_tasks(
 
     result: list[A2ATask] = []
     for task in page:
-        task = task.model_copy(deep=True)
+        task = proto_copy(task)
         if history_length is not None and task.history:
             task.history = task.history[-history_length:]
         if not include_artifacts:

lib/crewai/src/crewai/a2a/utils/transport.py

@@ -12,6 +12,11 @@ from typing import Final, Literal
 
 from a2a.types import AgentCard, AgentInterface
 
+from crewai.a2a._compat import (
+    agent_card_interfaces,
+    agent_card_preferred_transport,
+    agent_card_url,
+)
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.a2a_events import A2ATransportNegotiatedEvent
 
@@ -85,23 +90,21 @@ def _get_server_interfaces(agent_card: AgentCard) -> list[AgentInterface]:
         List of AgentInterface objects representing all available endpoints.
     """
     interfaces: list[AgentInterface] = []
-
-    primary_transport = agent_card.preferred_transport or JSONRPC_TRANSPORT
-    interfaces.append(
-        AgentInterface(
-            transport=primary_transport,
-            url=agent_card.url,
+    for interface in agent_card_interfaces(agent_card):
+        is_duplicate = any(
+            i.url == interface.url and i.protocol_binding == interface.protocol_binding
+            for i in interfaces
         )
-    )
+        if not is_duplicate:
+            interfaces.append(interface)
 
-    if agent_card.additional_interfaces:
-        for interface in agent_card.additional_interfaces:
-            is_duplicate = any(
-                i.url == interface.url and i.transport == interface.transport
-                for i in interfaces
+    if not interfaces:
+        interfaces.append(
+            AgentInterface(
+                url=agent_card_url(agent_card),
+                protocol_binding=JSONRPC_TRANSPORT,
             )
-            if not is_duplicate:
-                interfaces.append(interface)
+        )
 
     return interfaces
 
@@ -149,11 +152,11 @@ def negotiate_transport(
     )
 
     server_interfaces = _get_server_interfaces(agent_card)
-    server_transports = [i.transport.upper() for i in server_interfaces]
+    server_transports = [i.protocol_binding.upper() for i in server_interfaces]
 
     transport_to_interface: dict[str, AgentInterface] = {}
     for interface in server_interfaces:
-        transport_upper = interface.transport.upper()
+        transport_upper = interface.protocol_binding.upper()
         if transport_upper not in transport_to_interface:
             transport_to_interface[transport_upper] = interface
 
@@ -162,19 +165,21 @@ def negotiate_transport(
     if client_preferred and client_preferred in transport_to_interface:
         interface = transport_to_interface[client_preferred]
         result = NegotiatedTransport(
-            transport=interface.transport,
+            transport=interface.protocol_binding,
             url=interface.url,
             source="client_preferred",
         )
     else:
-        server_preferred = (agent_card.preferred_transport or JSONRPC_TRANSPORT).upper()
+        server_preferred = (
+            agent_card_preferred_transport(agent_card) or JSONRPC_TRANSPORT
+        ).upper()
         if (
             server_preferred in client_transports
             and server_preferred in transport_to_interface
         ):
             interface = transport_to_interface[server_preferred]
             result = NegotiatedTransport(
-                transport=interface.transport,
+                transport=interface.protocol_binding,
                 url=interface.url,
                 source="server_preferred",
             )
@@ -183,7 +188,7 @@ def negotiate_transport(
                 if transport in transport_to_interface:
                     interface = transport_to_interface[transport]
                     result = NegotiatedTransport(
-                        transport=interface.transport,
+                        transport=interface.protocol_binding,
                         url=interface.url,
                         source="fallback",
                     )
@@ -199,14 +204,14 @@ def negotiate_transport(
         crewai_event_bus.emit(
             None,
             A2ATransportNegotiatedEvent(
-                endpoint=endpoint or agent_card.url,
+                endpoint=endpoint or agent_card_url(agent_card),
                 a2a_agent_name=a2a_agent_name or agent_card.name,
                 negotiated_transport=result.transport,
                 negotiated_url=result.url,
                 source=result.source,
                 client_supported_transports=client_transports,
                 server_supported_transports=server_transports,
-                server_preferred_transport=agent_card.preferred_transport
+                server_preferred_transport=agent_card_preferred_transport(agent_card)
                 or JSONRPC_TRANSPORT,
                 client_preferred_transport=client_preferred,
             ),

lib/crewai/src/crewai/a2a/wrapper.py

@@ -14,9 +14,18 @@ import json
 from types import MethodType
 from typing import TYPE_CHECKING, Any, NamedTuple
 
-from a2a.types import Role, TaskState
 from pydantic import BaseModel, ValidationError
 
+from crewai.a2a._compat import (
+    ROLE_AGENT,
+    ROLE_USER,
+    TASK_STATE_COMPLETED,
+    TASK_STATE_INPUT_REQUIRED,
+    agent_card_to_dict,
+    part_is_text,
+    part_text,
+    proto_to_json,
+)
 from crewai.a2a.config import A2AClientConfig, A2AConfig
 from crewai.a2a.extensions.base import (
     A2AExtension,
@@ -681,7 +690,7 @@ def _augment_prompt_with_a2a(
                 }
                 agents_text += f"\n{json.dumps(filtered, indent=2)}\n"
             else:
-                agents_text += f"\n{card.model_dump_json(indent=2, exclude_none=True, include={'description', 'url', 'skills'})}\n"
+                agents_text += f"\n{proto_to_json(card)}\n"
 
     failed_agents = failed_agents or {}
     if failed_agents:
@@ -695,7 +704,7 @@ def _augment_prompt_with_a2a(
 
     if conversation_history:
         for msg in conversation_history:
-            history_text += f"\n{msg.model_dump_json(indent=2, exclude_none=True, exclude={'message_id'})}\n"
+            history_text += f"\n{proto_to_json(msg)}\n"
 
     history_text = PREVIOUS_A2A_CONVERSATION_TEMPLATE.substitute(
         previous_a2a_conversation=history_text
@@ -780,9 +789,9 @@ def _handle_max_turns_exceeded(
     """
     if conversation_history:
         for msg in reversed(conversation_history):
-            if msg.role == Role.agent:
+            if msg.role == ROLE_AGENT:
                 text_parts = [
-                    part.root.text for part in msg.parts if part.root.kind == "text"
+                    part_text(part) for part in msg.parts if part_is_text(part)
                 ]
                 final_message = (
                     " ".join(text_parts) if text_parts else "Conversation completed"
@@ -985,7 +994,9 @@ def _init_delegation_state(
         reference_task_ids=list(ctx.reference_task_ids),
         conversation_history=[],
         agent_card=current_agent_card,
-        agent_card_dict=current_agent_card.model_dump() if current_agent_card else None,
+        agent_card_dict=agent_card_to_dict(current_agent_card)
+        if current_agent_card
+        else None,
         agent_name=current_agent_card.name if current_agent_card else None,
     )
 
@@ -1110,7 +1121,7 @@ def _handle_task_completion(
         - remote_notice: Template notice about remote agent response
     """
     remote_notice = ""
-    if a2a_result["status"] == TaskState.completed:
+    if a2a_result["status"] == TASK_STATE_COMPLETED:
         remote_notice = REMOTE_AGENT_RESPONSE_NOTICE
 
         if task_id_config is not None and task_id_config not in reference_task_ids:
@@ -1294,7 +1305,7 @@ def _delegate_to_a2a(
                 extensions=ctx.extensions,
                 conversation_history=conversation_history,
                 agent_id=ctx.agent_id,
-                agent_role=Role.user,
+                agent_role=ROLE_USER,
                 agent_branch=agent_branch,
                 response_model=ctx.agent_config.response_model,
                 turn_number=turn_num + 1,
@@ -1316,7 +1327,10 @@ def _delegate_to_a2a(
                 if latest_message.context_id is not None:
                     context_id = latest_message.context_id
 
-            if a2a_result["status"] in [TaskState.completed, TaskState.input_required]:
+            if a2a_result["status"] in [
+                TASK_STATE_COMPLETED,
+                TASK_STATE_INPUT_REQUIRED,
+            ]:
                 trusted_result, task_id, reference_task_ids, remote_notice = (
                     _handle_task_completion(
                         a2a_result,
@@ -1649,7 +1663,7 @@ async def _adelegate_to_a2a(
                 extensions=ctx.extensions,
                 conversation_history=conversation_history,
                 agent_id=ctx.agent_id,
-                agent_role=Role.user,
+                agent_role=ROLE_USER,
                 agent_branch=agent_branch,
                 response_model=ctx.agent_config.response_model,
                 turn_number=turn_num + 1,
@@ -1671,7 +1685,10 @@ async def _adelegate_to_a2a(
                 if latest_message.context_id is not None:
                     context_id = latest_message.context_id
 
-            if a2a_result["status"] in [TaskState.completed, TaskState.input_required]:
+            if a2a_result["status"] in [
+                TASK_STATE_COMPLETED,
+                TASK_STATE_INPUT_REQUIRED,
+            ]:
                 trusted_result, task_id, reference_task_ids, remote_notice = (
                     _handle_task_completion(
                         a2a_result,

lib/crewai/src/crewai/agent/core.py

@@ -78,8 +78,7 @@ from crewai.knowledge.knowledge import Knowledge
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
 from crewai.lite_agent_output import LiteAgentOutput
 from crewai.llms.base_llm import BaseLLM
-from crewai.mcp import MCPServerConfig
-from crewai.mcp.tool_resolver import MCPToolResolver
+from crewai.mcp.config import MCPServerConfig
 from crewai.rag.embeddings.types import EmbedderConfig
 from crewai.security.fingerprint import Fingerprint
 from crewai.skills.loader import activate_skill, discover_skills
@@ -119,6 +118,7 @@ if TYPE_CHECKING:
 
     from crewai.a2a.config import A2AClientConfig, A2AConfig, A2AServerConfig
     from crewai.agents.agent_builder.base_agent import PlatformAppOrAction
+    from crewai.mcp.tool_resolver import MCPToolResolver
     from crewai.task import Task
     from crewai.tools.base_tool import BaseTool
     from crewai.tools.structured_tool import CrewStructuredTool
@@ -394,15 +394,17 @@ class Agent(BaseAgent):
         self,
         resolved_crew_skills: list[SkillModel] | None = None,
     ) -> None:
-        """Resolve skill paths and activate skills to INSTRUCTIONS level.
+        """Resolve skill paths while preserving explicit disclosure levels.
 
-        Path entries trigger discovery and activation. Pre-loaded Skill objects
-        below INSTRUCTIONS level are activated. Crew-level skills are merged in
-        with event emission so observability is consistent regardless of origin.
+        Path entries trigger discovery and activation because directory-based
+        skills opt into eager loading. Pre-loaded Skill objects keep their
+        current disclosure level so callers can attach METADATA-only skills and
+        progressively activate them later. Crew-level skills are merged in with
+        event emission so observability is consistent regardless of origin.
 
         Args:
-            resolved_crew_skills: Pre-resolved crew skills (already discovered
-                and activated). When provided, avoids redundant discovery per agent.
+            resolved_crew_skills: Pre-resolved crew skills. When provided,
+                avoids redundant discovery per agent.
         """
         from crewai.crew import Crew
 
@@ -443,8 +445,7 @@ class Agent(BaseAgent):
             elif isinstance(item, SkillModel):
                 if item.name not in seen:
                     seen.add(item.name)
-                    activated = activate_skill(item, source=self)
-                    if activated is item and item.disclosure_level >= INSTRUCTIONS:
+                    if item.disclosure_level >= INSTRUCTIONS:
                         crewai_event_bus.emit(
                             self,
                             event=SkillActivatedEvent(
@@ -454,7 +455,7 @@ class Agent(BaseAgent):
                                 disclosure_level=item.disclosure_level,
                             ),
                         )
-                    resolved.append(activated)
+                    resolved.append(item)
 
         self.skills = resolved if resolved else None
 
@@ -1120,6 +1121,8 @@ class Agent(BaseAgent):
         Delegates to :class:`~crewai.mcp.tool_resolver.MCPToolResolver`.
         """
         self._cleanup_mcp_clients()
+        from crewai.mcp.tool_resolver import MCPToolResolver
+
         self._mcp_resolver = MCPToolResolver(agent=self, logger=self._logger)
         return self._mcp_resolver.resolve(mcps)
 

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

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

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

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

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

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

lib/crewai/src/crewai/events/__init__.py

@@ -6,111 +6,20 @@ This module provides the event infrastructure that allows users to:
 - Build custom logging and analytics
 - Extend CrewAI with custom event handlers
 - Declare handler dependencies for ordered execution
+
+Event type classes are lazy-loaded on first access to avoid importing
+~12 Pydantic model modules (and their transitive deps) at package init time.
 """
 
 from __future__ import annotations
 
+import importlib
 from typing import TYPE_CHECKING, Any
 
 from crewai.events.base_event_listener import BaseEventListener
 from crewai.events.depends import Depends
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.handler_graph import CircularDependencyError
-from crewai.events.types.crew_events import (
-    CrewKickoffCompletedEvent,
-    CrewKickoffFailedEvent,
-    CrewKickoffStartedEvent,
-    CrewTestCompletedEvent,
-    CrewTestFailedEvent,
-    CrewTestResultEvent,
-    CrewTestStartedEvent,
-    CrewTrainCompletedEvent,
-    CrewTrainFailedEvent,
-    CrewTrainStartedEvent,
-)
-from crewai.events.types.flow_events import (
-    FlowCreatedEvent,
-    FlowEvent,
-    FlowFinishedEvent,
-    FlowPlotEvent,
-    FlowStartedEvent,
-    HumanFeedbackReceivedEvent,
-    HumanFeedbackRequestedEvent,
-    MethodExecutionFailedEvent,
-    MethodExecutionFinishedEvent,
-    MethodExecutionStartedEvent,
-)
-from crewai.events.types.knowledge_events import (
-    KnowledgeQueryCompletedEvent,
-    KnowledgeQueryFailedEvent,
-    KnowledgeQueryStartedEvent,
-    KnowledgeRetrievalCompletedEvent,
-    KnowledgeRetrievalStartedEvent,
-    KnowledgeSearchQueryFailedEvent,
-)
-from crewai.events.types.llm_events import (
-    LLMCallCompletedEvent,
-    LLMCallFailedEvent,
-    LLMCallStartedEvent,
-    LLMStreamChunkEvent,
-)
-from crewai.events.types.llm_guardrail_events import (
-    LLMGuardrailCompletedEvent,
-    LLMGuardrailStartedEvent,
-)
-from crewai.events.types.logging_events import (
-    AgentLogsExecutionEvent,
-    AgentLogsStartedEvent,
-)
-from crewai.events.types.mcp_events import (
-    MCPConfigFetchFailedEvent,
-    MCPConnectionCompletedEvent,
-    MCPConnectionFailedEvent,
-    MCPConnectionStartedEvent,
-    MCPToolExecutionCompletedEvent,
-    MCPToolExecutionFailedEvent,
-    MCPToolExecutionStartedEvent,
-)
-from crewai.events.types.memory_events import (
-    MemoryQueryCompletedEvent,
-    MemoryQueryFailedEvent,
-    MemoryQueryStartedEvent,
-    MemoryRetrievalCompletedEvent,
-    MemoryRetrievalFailedEvent,
-    MemoryRetrievalStartedEvent,
-    MemorySaveCompletedEvent,
-    MemorySaveFailedEvent,
-    MemorySaveStartedEvent,
-)
-from crewai.events.types.reasoning_events import (
-    AgentReasoningCompletedEvent,
-    AgentReasoningFailedEvent,
-    AgentReasoningStartedEvent,
-    ReasoningEvent,
-)
-from crewai.events.types.skill_events import (
-    SkillActivatedEvent,
-    SkillDiscoveryCompletedEvent,
-    SkillDiscoveryStartedEvent,
-    SkillEvent,
-    SkillLoadFailedEvent,
-    SkillLoadedEvent,
-)
-from crewai.events.types.task_events import (
-    TaskCompletedEvent,
-    TaskEvaluationEvent,
-    TaskFailedEvent,
-    TaskStartedEvent,
-)
-from crewai.events.types.tool_usage_events import (
-    ToolExecutionErrorEvent,
-    ToolSelectionErrorEvent,
-    ToolUsageErrorEvent,
-    ToolUsageEvent,
-    ToolUsageFinishedEvent,
-    ToolUsageStartedEvent,
-    ToolValidateInputErrorEvent,
-)
 
 
 if TYPE_CHECKING:
@@ -125,6 +34,250 @@ if TYPE_CHECKING:
         LiteAgentExecutionErrorEvent,
         LiteAgentExecutionStartedEvent,
     )
+    from crewai.events.types.checkpoint_events import (
+        CheckpointBaseEvent,
+        CheckpointCompletedEvent,
+        CheckpointFailedEvent,
+        CheckpointForkBaseEvent,
+        CheckpointForkCompletedEvent,
+        CheckpointForkStartedEvent,
+        CheckpointPrunedEvent,
+        CheckpointRestoreBaseEvent,
+        CheckpointRestoreCompletedEvent,
+        CheckpointRestoreFailedEvent,
+        CheckpointRestoreStartedEvent,
+        CheckpointStartedEvent,
+    )
+    from crewai.events.types.crew_events import (
+        CrewKickoffCompletedEvent,
+        CrewKickoffFailedEvent,
+        CrewKickoffStartedEvent,
+        CrewTestCompletedEvent,
+        CrewTestFailedEvent,
+        CrewTestResultEvent,
+        CrewTestStartedEvent,
+        CrewTrainCompletedEvent,
+        CrewTrainFailedEvent,
+        CrewTrainStartedEvent,
+    )
+    from crewai.events.types.flow_events import (
+        FlowCreatedEvent,
+        FlowEvent,
+        FlowFinishedEvent,
+        FlowPlotEvent,
+        FlowStartedEvent,
+        HumanFeedbackReceivedEvent,
+        HumanFeedbackRequestedEvent,
+        MethodExecutionFailedEvent,
+        MethodExecutionFinishedEvent,
+        MethodExecutionStartedEvent,
+    )
+    from crewai.events.types.knowledge_events import (
+        KnowledgeQueryCompletedEvent,
+        KnowledgeQueryFailedEvent,
+        KnowledgeQueryStartedEvent,
+        KnowledgeRetrievalCompletedEvent,
+        KnowledgeRetrievalStartedEvent,
+        KnowledgeSearchQueryFailedEvent,
+    )
+    from crewai.events.types.llm_events import (
+        LLMCallCompletedEvent,
+        LLMCallFailedEvent,
+        LLMCallStartedEvent,
+        LLMStreamChunkEvent,
+    )
+    from crewai.events.types.llm_guardrail_events import (
+        LLMGuardrailCompletedEvent,
+        LLMGuardrailStartedEvent,
+    )
+    from crewai.events.types.logging_events import (
+        AgentLogsExecutionEvent,
+        AgentLogsStartedEvent,
+    )
+    from crewai.events.types.mcp_events import (
+        MCPConfigFetchFailedEvent,
+        MCPConnectionCompletedEvent,
+        MCPConnectionFailedEvent,
+        MCPConnectionStartedEvent,
+        MCPToolExecutionCompletedEvent,
+        MCPToolExecutionFailedEvent,
+        MCPToolExecutionStartedEvent,
+    )
+    from crewai.events.types.memory_events import (
+        MemoryQueryCompletedEvent,
+        MemoryQueryFailedEvent,
+        MemoryQueryStartedEvent,
+        MemoryRetrievalCompletedEvent,
+        MemoryRetrievalFailedEvent,
+        MemoryRetrievalStartedEvent,
+        MemorySaveCompletedEvent,
+        MemorySaveFailedEvent,
+        MemorySaveStartedEvent,
+    )
+    from crewai.events.types.reasoning_events import (
+        AgentReasoningCompletedEvent,
+        AgentReasoningFailedEvent,
+        AgentReasoningStartedEvent,
+        ReasoningEvent,
+    )
+    from crewai.events.types.skill_events import (
+        SkillActivatedEvent,
+        SkillDiscoveryCompletedEvent,
+        SkillDiscoveryStartedEvent,
+        SkillEvent,
+        SkillLoadFailedEvent,
+        SkillLoadedEvent,
+    )
+    from crewai.events.types.task_events import (
+        TaskCompletedEvent,
+        TaskEvaluationEvent,
+        TaskFailedEvent,
+        TaskStartedEvent,
+    )
+    from crewai.events.types.tool_usage_events import (
+        ToolExecutionErrorEvent,
+        ToolSelectionErrorEvent,
+        ToolUsageErrorEvent,
+        ToolUsageEvent,
+        ToolUsageFinishedEvent,
+        ToolUsageStartedEvent,
+        ToolValidateInputErrorEvent,
+    )
+
+# Map every event class name → its module path for lazy loading
+_LAZY_EVENT_MAPPING: dict[str, str] = {
+    # agent_events
+    "AgentEvaluationCompletedEvent": "crewai.events.types.agent_events",
+    "AgentEvaluationFailedEvent": "crewai.events.types.agent_events",
+    "AgentEvaluationStartedEvent": "crewai.events.types.agent_events",
+    "AgentExecutionCompletedEvent": "crewai.events.types.agent_events",
+    "AgentExecutionErrorEvent": "crewai.events.types.agent_events",
+    "AgentExecutionStartedEvent": "crewai.events.types.agent_events",
+    "LiteAgentExecutionCompletedEvent": "crewai.events.types.agent_events",
+    "LiteAgentExecutionErrorEvent": "crewai.events.types.agent_events",
+    "LiteAgentExecutionStartedEvent": "crewai.events.types.agent_events",
+    # checkpoint_events
+    "CheckpointBaseEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointCompletedEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointFailedEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointForkBaseEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointForkCompletedEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointForkStartedEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointPrunedEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointRestoreBaseEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointRestoreCompletedEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointRestoreFailedEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointRestoreStartedEvent": "crewai.events.types.checkpoint_events",
+    "CheckpointStartedEvent": "crewai.events.types.checkpoint_events",
+    # crew_events
+    "CrewKickoffCompletedEvent": "crewai.events.types.crew_events",
+    "CrewKickoffFailedEvent": "crewai.events.types.crew_events",
+    "CrewKickoffStartedEvent": "crewai.events.types.crew_events",
+    "CrewTestCompletedEvent": "crewai.events.types.crew_events",
+    "CrewTestFailedEvent": "crewai.events.types.crew_events",
+    "CrewTestResultEvent": "crewai.events.types.crew_events",
+    "CrewTestStartedEvent": "crewai.events.types.crew_events",
+    "CrewTrainCompletedEvent": "crewai.events.types.crew_events",
+    "CrewTrainFailedEvent": "crewai.events.types.crew_events",
+    "CrewTrainStartedEvent": "crewai.events.types.crew_events",
+    # flow_events
+    "FlowCreatedEvent": "crewai.events.types.flow_events",
+    "FlowEvent": "crewai.events.types.flow_events",
+    "FlowFinishedEvent": "crewai.events.types.flow_events",
+    "FlowPlotEvent": "crewai.events.types.flow_events",
+    "FlowStartedEvent": "crewai.events.types.flow_events",
+    "HumanFeedbackReceivedEvent": "crewai.events.types.flow_events",
+    "HumanFeedbackRequestedEvent": "crewai.events.types.flow_events",
+    "MethodExecutionFailedEvent": "crewai.events.types.flow_events",
+    "MethodExecutionFinishedEvent": "crewai.events.types.flow_events",
+    "MethodExecutionStartedEvent": "crewai.events.types.flow_events",
+    # knowledge_events
+    "KnowledgeQueryCompletedEvent": "crewai.events.types.knowledge_events",
+    "KnowledgeQueryFailedEvent": "crewai.events.types.knowledge_events",
+    "KnowledgeQueryStartedEvent": "crewai.events.types.knowledge_events",
+    "KnowledgeRetrievalCompletedEvent": "crewai.events.types.knowledge_events",
+    "KnowledgeRetrievalStartedEvent": "crewai.events.types.knowledge_events",
+    "KnowledgeSearchQueryFailedEvent": "crewai.events.types.knowledge_events",
+    # llm_events
+    "LLMCallCompletedEvent": "crewai.events.types.llm_events",
+    "LLMCallFailedEvent": "crewai.events.types.llm_events",
+    "LLMCallStartedEvent": "crewai.events.types.llm_events",
+    "LLMStreamChunkEvent": "crewai.events.types.llm_events",
+    # llm_guardrail_events
+    "LLMGuardrailCompletedEvent": "crewai.events.types.llm_guardrail_events",
+    "LLMGuardrailStartedEvent": "crewai.events.types.llm_guardrail_events",
+    # logging_events
+    "AgentLogsExecutionEvent": "crewai.events.types.logging_events",
+    "AgentLogsStartedEvent": "crewai.events.types.logging_events",
+    # mcp_events
+    "MCPConfigFetchFailedEvent": "crewai.events.types.mcp_events",
+    "MCPConnectionCompletedEvent": "crewai.events.types.mcp_events",
+    "MCPConnectionFailedEvent": "crewai.events.types.mcp_events",
+    "MCPConnectionStartedEvent": "crewai.events.types.mcp_events",
+    "MCPToolExecutionCompletedEvent": "crewai.events.types.mcp_events",
+    "MCPToolExecutionFailedEvent": "crewai.events.types.mcp_events",
+    "MCPToolExecutionStartedEvent": "crewai.events.types.mcp_events",
+    # memory_events
+    "MemoryQueryCompletedEvent": "crewai.events.types.memory_events",
+    "MemoryQueryFailedEvent": "crewai.events.types.memory_events",
+    "MemoryQueryStartedEvent": "crewai.events.types.memory_events",
+    "MemoryRetrievalCompletedEvent": "crewai.events.types.memory_events",
+    "MemoryRetrievalFailedEvent": "crewai.events.types.memory_events",
+    "MemoryRetrievalStartedEvent": "crewai.events.types.memory_events",
+    "MemorySaveCompletedEvent": "crewai.events.types.memory_events",
+    "MemorySaveFailedEvent": "crewai.events.types.memory_events",
+    "MemorySaveStartedEvent": "crewai.events.types.memory_events",
+    # reasoning_events
+    "AgentReasoningCompletedEvent": "crewai.events.types.reasoning_events",
+    "AgentReasoningFailedEvent": "crewai.events.types.reasoning_events",
+    "AgentReasoningStartedEvent": "crewai.events.types.reasoning_events",
+    "ReasoningEvent": "crewai.events.types.reasoning_events",
+    # skill_events
+    "SkillActivatedEvent": "crewai.events.types.skill_events",
+    "SkillDiscoveryCompletedEvent": "crewai.events.types.skill_events",
+    "SkillDiscoveryStartedEvent": "crewai.events.types.skill_events",
+    "SkillEvent": "crewai.events.types.skill_events",
+    "SkillLoadFailedEvent": "crewai.events.types.skill_events",
+    "SkillLoadedEvent": "crewai.events.types.skill_events",
+    # task_events
+    "TaskCompletedEvent": "crewai.events.types.task_events",
+    "TaskEvaluationEvent": "crewai.events.types.task_events",
+    "TaskFailedEvent": "crewai.events.types.task_events",
+    "TaskStartedEvent": "crewai.events.types.task_events",
+    # tool_usage_events
+    "ToolExecutionErrorEvent": "crewai.events.types.tool_usage_events",
+    "ToolSelectionErrorEvent": "crewai.events.types.tool_usage_events",
+    "ToolUsageErrorEvent": "crewai.events.types.tool_usage_events",
+    "ToolUsageEvent": "crewai.events.types.tool_usage_events",
+    "ToolUsageFinishedEvent": "crewai.events.types.tool_usage_events",
+    "ToolUsageStartedEvent": "crewai.events.types.tool_usage_events",
+    "ToolValidateInputErrorEvent": "crewai.events.types.tool_usage_events",
+}
+
+_extension_exports: dict[str, Any] = {}
+
+
+def __getattr__(name: str) -> Any:
+    """Lazy import for event types and registered extensions."""
+    if name in _LAZY_EVENT_MAPPING:
+        module_path = _LAZY_EVENT_MAPPING[name]
+        module = importlib.import_module(module_path)
+        val = getattr(module, name)
+        globals()[name] = val  # cache for subsequent access
+        return val
+
+    if name in _extension_exports:
+        value = _extension_exports[name]
+        if isinstance(value, str):
+            module_path, _, attr_name = value.rpartition(".")
+            if module_path:
+                module = importlib.import_module(module_path)
+                return getattr(module, attr_name)
+            return importlib.import_module(value)
+        return value
+
+    msg = f"module {__name__!r} has no attribute {name!r}"
+    raise AttributeError(msg)
 
 
 __all__ = [
@@ -140,6 +293,18 @@ __all__ = [
     "AgentReasoningFailedEvent",
     "AgentReasoningStartedEvent",
     "BaseEventListener",
+    "CheckpointBaseEvent",
+    "CheckpointCompletedEvent",
+    "CheckpointFailedEvent",
+    "CheckpointForkBaseEvent",
+    "CheckpointForkCompletedEvent",
+    "CheckpointForkStartedEvent",
+    "CheckpointPrunedEvent",
+    "CheckpointRestoreBaseEvent",
+    "CheckpointRestoreCompletedEvent",
+    "CheckpointRestoreFailedEvent",
+    "CheckpointRestoreStartedEvent",
+    "CheckpointStartedEvent",
     "CircularDependencyError",
     "CrewKickoffCompletedEvent",
     "CrewKickoffFailedEvent",
@@ -214,42 +379,3 @@ __all__ = [
     "_extension_exports",
     "crewai_event_bus",
 ]
-
-_AGENT_EVENT_MAPPING = {
-    "AgentEvaluationCompletedEvent": "crewai.events.types.agent_events",
-    "AgentEvaluationFailedEvent": "crewai.events.types.agent_events",
-    "AgentEvaluationStartedEvent": "crewai.events.types.agent_events",
-    "AgentExecutionCompletedEvent": "crewai.events.types.agent_events",
-    "AgentExecutionErrorEvent": "crewai.events.types.agent_events",
-    "AgentExecutionStartedEvent": "crewai.events.types.agent_events",
-    "LiteAgentExecutionCompletedEvent": "crewai.events.types.agent_events",
-    "LiteAgentExecutionErrorEvent": "crewai.events.types.agent_events",
-    "LiteAgentExecutionStartedEvent": "crewai.events.types.agent_events",
-}
-
-_extension_exports: dict[str, Any] = {}
-
-
-def __getattr__(name: str) -> Any:
-    """Lazy import for agent events and registered extensions."""
-    if name in _AGENT_EVENT_MAPPING:
-        import importlib
-
-        module_path = _AGENT_EVENT_MAPPING[name]
-        module = importlib.import_module(module_path)
-        return getattr(module, name)
-
-    if name in _extension_exports:
-        import importlib
-
-        value = _extension_exports[name]
-        if isinstance(value, str):
-            module_path, _, attr_name = value.rpartition(".")
-            if module_path:
-                module = importlib.import_module(module_path)
-                return getattr(module, attr_name)
-            return importlib.import_module(value)
-        return value
-
-    msg = f"module {__name__!r} has no attribute {name!r}"
-    raise AttributeError(msg)

lib/crewai/src/crewai/events/event_bus.py

@@ -64,6 +64,22 @@ P = ParamSpec("P")
 R = TypeVar("R")
 
 
+_replaying: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "crewai_event_replaying", default=False
+)
+
+
+def is_replaying() -> bool:
+    """Return True if the current context is dispatching a replayed event.
+
+    Listeners with side effects (checkpoint writes, external API calls that
+    should not be repeated) should early-return when this is true. Listeners
+    whose purpose is reconstructing timeline state (trace batch, console
+    formatter) should ignore the flag and process replayed events normally.
+    """
+    return _replaying.get()
+
+
 class CrewAIEventsBus:
     """Singleton event bus for handling events in CrewAI.
 
@@ -261,6 +277,11 @@ class CrewAIEventsBus:
             self._runtime_state = state
             self._registered_entity_ids = {id(e) for e in state.root}
 
+    @property
+    def runtime_state(self) -> RuntimeState | None:
+        """The RuntimeState currently attached to the bus, if any."""
+        return self._runtime_state
+
     def register_entity(self, entity: Any) -> None:
         """Add an entity to the RuntimeState, creating it if needed.
 
@@ -568,6 +589,87 @@ class CrewAIEventsBus:
 
         return None
 
+    async def _acall_handlers_replaying(
+        self,
+        source: Any,
+        event: BaseEvent,
+        handlers: AsyncHandlerSet,
+    ) -> None:
+        """Call async handlers with the replaying flag set on the loop thread."""
+        token = _replaying.set(True)
+        try:
+            await self._acall_handlers(source, event, handlers)
+        finally:
+            _replaying.reset(token)
+
+    async def _emit_with_dependencies_replaying(
+        self, source: Any, event: BaseEvent
+    ) -> None:
+        """Dependency-aware dispatch with the replaying flag set."""
+        token = _replaying.set(True)
+        try:
+            await self._emit_with_dependencies(source, event)
+        finally:
+            _replaying.reset(token)
+
+    def replay(self, source: Any, event: BaseEvent) -> Future[None] | None:
+        """Dispatch a previously-recorded event without mutating its fields.
+
+        Unlike :meth:`emit`, this does not run ``_prepare_event`` (so stored
+        event ids and ``emission_sequence`` are preserved) and does not
+        re-record the event. Listeners can call :func:`is_replaying` to
+        opt out of side-effectful processing.
+
+        Args:
+            source: The emitting object.
+            event: The previously-recorded event to dispatch.
+
+        Returns:
+            Future that completes when handlers finish, or None if no handlers.
+        """
+        event_type = type(event)
+
+        with self._rwlock.r_locked():
+            if self._shutting_down:
+                return None
+            has_dependencies = event_type in self._handler_dependencies
+            sync_handlers = self._sync_handlers.get(event_type, frozenset())
+            async_handlers = self._async_handlers.get(event_type, frozenset())
+
+        if not sync_handlers and not async_handlers:
+            return None
+
+        self._ensure_executor_initialized()
+        self._has_pending_events = True
+
+        token = _replaying.set(True)
+        try:
+            if has_dependencies:
+                return self._track_future(
+                    asyncio.run_coroutine_threadsafe(
+                        self._emit_with_dependencies_replaying(source, event),
+                        self._loop,
+                    )
+                )
+
+            if sync_handlers:
+                ctx = contextvars.copy_context()
+                sync_future = self._sync_executor.submit(
+                    ctx.run, self._call_handlers, source, event, sync_handlers
+                )
+                self._track_future(sync_future)
+                if not async_handlers:
+                    return sync_future
+
+            return self._track_future(
+                asyncio.run_coroutine_threadsafe(
+                    self._acall_handlers_replaying(source, event, async_handlers),
+                    self._loop,
+                )
+            )
+        finally:
+            _replaying.reset(token)
+
     def flush(self, timeout: float | None = 30.0) -> bool:
         """Block until all pending event handlers complete.
 

lib/crewai/src/crewai/events/event_types.py

@@ -30,6 +30,17 @@ from crewai.events.types.agent_events import (
     AgentExecutionStartedEvent,
     LiteAgentExecutionCompletedEvent,
 )
+from crewai.events.types.checkpoint_events import (
+    CheckpointCompletedEvent,
+    CheckpointFailedEvent,
+    CheckpointForkCompletedEvent,
+    CheckpointForkStartedEvent,
+    CheckpointPrunedEvent,
+    CheckpointRestoreCompletedEvent,
+    CheckpointRestoreFailedEvent,
+    CheckpointRestoreStartedEvent,
+    CheckpointStartedEvent,
+)
 from crewai.events.types.crew_events import (
     CrewKickoffCompletedEvent,
     CrewKickoffFailedEvent,
@@ -183,4 +194,13 @@ EventTypes = (
     | MCPToolExecutionCompletedEvent
     | MCPToolExecutionFailedEvent
     | MCPConfigFetchFailedEvent
+    | CheckpointStartedEvent
+    | CheckpointCompletedEvent
+    | CheckpointFailedEvent
+    | CheckpointForkStartedEvent
+    | CheckpointForkCompletedEvent
+    | CheckpointRestoreStartedEvent
+    | CheckpointRestoreCompletedEvent
+    | CheckpointRestoreFailedEvent
+    | CheckpointPrunedEvent
 )

lib/crewai/src/crewai/events/types/checkpoint_events.py

@@ -0,0 +1,97 @@
+"""Event family for automatic state checkpointing and forking."""
+
+from typing import Literal
+
+from crewai.events.base_events import BaseEvent
+
+
+class CheckpointBaseEvent(BaseEvent):
+    """Base event for checkpoint lifecycle operations."""
+
+    type: str
+    location: str
+    provider: str
+    trigger: str | None = None
+    branch: str | None = None
+    parent_id: str | None = None
+
+
+class CheckpointStartedEvent(CheckpointBaseEvent):
+    """Event emitted immediately before a checkpoint is written."""
+
+    type: Literal["checkpoint_started"] = "checkpoint_started"
+
+
+class CheckpointCompletedEvent(CheckpointBaseEvent):
+    """Event emitted when a checkpoint has been written successfully."""
+
+    type: Literal["checkpoint_completed"] = "checkpoint_completed"
+    checkpoint_id: str
+    duration_ms: float
+
+
+class CheckpointFailedEvent(CheckpointBaseEvent):
+    """Event emitted when a checkpoint write fails."""
+
+    type: Literal["checkpoint_failed"] = "checkpoint_failed"
+    error: str
+
+
+class CheckpointPrunedEvent(CheckpointBaseEvent):
+    """Event emitted after pruning old checkpoints from a branch."""
+
+    type: Literal["checkpoint_pruned"] = "checkpoint_pruned"
+    removed_count: int
+    max_checkpoints: int
+
+
+class CheckpointForkBaseEvent(BaseEvent):
+    """Base event for fork lifecycle operations on a RuntimeState."""
+
+    type: str
+    branch: str
+    parent_branch: str | None = None
+    parent_checkpoint_id: str | None = None
+
+
+class CheckpointForkStartedEvent(CheckpointForkBaseEvent):
+    """Event emitted immediately before a fork relabels the branch."""
+
+    type: Literal["checkpoint_fork_started"] = "checkpoint_fork_started"
+
+
+class CheckpointForkCompletedEvent(CheckpointForkBaseEvent):
+    """Event emitted after a fork has established the new branch."""
+
+    type: Literal["checkpoint_fork_completed"] = "checkpoint_fork_completed"
+
+
+class CheckpointRestoreBaseEvent(BaseEvent):
+    """Base event for checkpoint restore lifecycle operations."""
+
+    type: str
+    location: str
+    provider: str | None = None
+
+
+class CheckpointRestoreStartedEvent(CheckpointRestoreBaseEvent):
+    """Event emitted immediately before a checkpoint restore begins."""
+
+    type: Literal["checkpoint_restore_started"] = "checkpoint_restore_started"
+
+
+class CheckpointRestoreCompletedEvent(CheckpointRestoreBaseEvent):
+    """Event emitted when a checkpoint has been restored successfully."""
+
+    type: Literal["checkpoint_restore_completed"] = "checkpoint_restore_completed"
+    checkpoint_id: str
+    branch: str | None = None
+    parent_id: str | None = None
+    duration_ms: float
+
+
+class CheckpointRestoreFailedEvent(CheckpointRestoreBaseEvent):
+    """Event emitted when a checkpoint restore fails."""
+
+    type: Literal["checkpoint_restore_failed"] = "checkpoint_restore_failed"
+    error: str

lib/crewai/src/crewai/flow/flow.py

@@ -45,6 +45,7 @@ from pydantic import (
     BeforeValidator,
     ConfigDict,
     Field,
+    PlainSerializer,
     PrivateAttr,
     SerializeAsAny,
     ValidationError,
@@ -58,6 +59,7 @@ from crewai.events.event_bus import crewai_event_bus
 from crewai.events.event_context import (
     get_current_parent_id,
     reset_last_event_id,
+    restore_event_scope,
     triggered_by_scope,
 )
 from crewai.events.listeners.tracing.trace_listener import (
@@ -157,6 +159,37 @@ def _resolve_persistence(value: Any) -> Any:
     return value
 
 
+_INITIAL_STATE_CLASS_MARKER = "__crewai_pydantic_class_schema__"
+
+
+def _serialize_initial_state(value: Any) -> Any:
+    """Make ``initial_state`` safe for JSON checkpoint serialization.
+
+    ``BaseModel`` class refs are emitted as their JSON schema under a sentinel
+    marker key so deserialization can round-trip them back to a class.
+    ``BaseModel`` instances are dumped to JSON (round-trip as plain dicts,
+    which ``_create_initial_state`` accepts). Bare ``type`` values that are
+    not ``BaseModel`` subclasses (e.g. ``dict``) are dropped since they
+    can't be represented in JSON.
+    """
+    if isinstance(value, type):
+        if issubclass(value, BaseModel):
+            return {_INITIAL_STATE_CLASS_MARKER: value.model_json_schema()}
+        return None
+    if isinstance(value, BaseModel):
+        return value.model_dump(mode="json")
+    return value
+
+
+def _deserialize_initial_state(value: Any) -> Any:
+    """Rehydrate a class ref serialized by :func:`_serialize_initial_state`."""
+    if isinstance(value, dict) and _INITIAL_STATE_CLASS_MARKER in value:
+        from crewai.utilities.pydantic_schema_utils import create_model_from_schema
+
+        return create_model_from_schema(value[_INITIAL_STATE_CLASS_MARKER])
+    return value
+
+
 class FlowState(BaseModel):
     """Base model for all flow states, ensuring each state has a unique ID."""
 
@@ -908,7 +941,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
 
     entity_type: Literal["flow"] = "flow"
 
-    initial_state: Any = Field(default=None)
+    initial_state: Annotated[  # type: ignore[type-arg]
+        type[BaseModel] | type[dict] | dict[str, Any] | BaseModel | None,
+        BeforeValidator(_deserialize_initial_state),
+        PlainSerializer(_serialize_initial_state, return_type=Any, when_used="json"),
+    ] = Field(default=None)
     name: str | None = Field(default=None)
     tracing: bool | None = Field(default=None)
     stream: bool = Field(default=False)
@@ -980,13 +1017,18 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             A Flow instance on the new branch. Call kickoff() to run.
         """
         flow = cls.from_checkpoint(config)
-        state = crewai_event_bus._runtime_state
+        state = crewai_event_bus.runtime_state
         if state is None:
             raise RuntimeError(
                 "Cannot fork: no runtime state on the event bus. "
                 "Ensure from_checkpoint() succeeded before calling fork()."
             )
         state.fork(branch)
+        new_id = str(uuid4())
+        if isinstance(flow._state, dict):
+            flow._state["id"] = new_id
+        else:
+            object.__setattr__(flow._state, "id", new_id)
         return flow
 
     checkpoint_completed_methods: set[str] | None = Field(default=None)
@@ -1008,6 +1050,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             }
         if self.checkpoint_state is not None:
             self._restore_state(self.checkpoint_state)
+        restore_event_scope(())
+        reset_last_event_id()
 
     _methods: dict[FlowMethodName, FlowMethod[Any, Any]] = PrivateAttr(
         default_factory=dict
@@ -1030,6 +1074,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
     _human_feedback_method_outputs: dict[str, Any] = PrivateAttr(default_factory=dict)
     _input_history: list[InputHistoryEntry] = PrivateAttr(default_factory=list)
     _state: Any = PrivateAttr(default=None)
+    _execution_id: str = PrivateAttr(default_factory=lambda: str(uuid4()))
 
     def __class_getitem__(cls: type[Flow[T]], item: type[T]) -> type[Flow[T]]:  # type: ignore[override]
         class _FlowGeneric(cls):  # type: ignore[valid-type,misc]
@@ -1820,6 +1865,27 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         except (AttributeError, TypeError):
             return ""  # Safely handle any unexpected attribute access issues
 
+    @property
+    def execution_id(self) -> str:
+        """Stable identifier for this flow execution.
+
+        Separate from ``flow_id`` / ``state.id``, which consumers may
+        override via ``kickoff(inputs={"id": ...})`` to resume a persisted
+        flow. ``execution_id`` is never affected by ``inputs`` and stays
+        stable for the lifetime of a single run, so it is the correct key
+        for telemetry, tracing, and any external correlation that must
+        uniquely identify a single execution even when callers pass an
+        ``id`` in ``inputs``.
+
+        Defaults to a fresh ``uuid4`` per ``Flow`` instance; assign to
+        override when an outer system already has an execution identity.
+        """
+        return self._execution_id
+
+    @execution_id.setter
+    def execution_id(self, value: str) -> None:
+        self._execution_id = value
+
     def _initialize_state(self, inputs: dict[str, Any]) -> None:
         """Initialize or update flow state with new inputs.
 
@@ -2133,9 +2199,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         flow_id_token = None
         request_id_token = None
         if current_flow_id.get() is None:
-            flow_id_token = current_flow_id.set(self.flow_id)
+            flow_id_token = current_flow_id.set(self.execution_id)
         if current_flow_request_id.get() is None:
-            request_id_token = current_flow_request_id.set(self.flow_id)
+            request_id_token = current_flow_request_id.set(self.execution_id)
 
         try:
             # Reset flow state for fresh execution unless restoring from persistence
@@ -2214,6 +2280,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             if inputs is not None and "id" not in inputs:
                 self._initialize_state(inputs)
 
+            if self._is_execution_resuming:
+                await self._replay_recorded_events()
+
             try:
                 # Determine which start methods to execute at kickoff
                 # Conditional start methods (with __trigger_methods__) are only triggered by their conditions
@@ -2361,6 +2430,44 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         """
         return await self.kickoff_async(inputs, input_files, from_checkpoint)
 
+    async def _replay_recorded_events(self) -> None:
+        """Dispatch recorded ``MethodExecution*`` events from the event record."""
+        state = crewai_event_bus.runtime_state
+        if state is None:
+            return
+        record = state.event_record
+        if len(record) == 0:
+            return
+
+        replayable = (
+            MethodExecutionStartedEvent,
+            MethodExecutionFinishedEvent,
+            MethodExecutionFailedEvent,
+        )
+        flow_name = self.name or self.__class__.__name__
+        nodes = sorted(
+            (
+                n
+                for n in record.all_nodes()
+                if isinstance(n.event, replayable)
+                and n.event.flow_name == flow_name
+                and n.event.method_name in self._completed_methods
+            ),
+            key=lambda n: n.event.emission_sequence or 0,
+        )
+
+        for node in nodes:
+            future = crewai_event_bus.replay(self, node.event)
+            if future is not None:
+                try:
+                    await asyncio.wrap_future(future)
+                except Exception:
+                    logger.warning(
+                        "Replayed event handler failed: %s",
+                        node.event.type,
+                        exc_info=True,
+                    )
+
     async def _execute_start_method(self, start_method_name: FlowMethodName) -> None:
         """Executes a flow's start method and its triggered listeners.
 

lib/crewai/src/crewai/llms/providers/azure/completion.py

@@ -183,11 +183,6 @@ class AzureCompletion(BaseLLM):
                     AzureCompletion._is_azure_openai_endpoint(self.endpoint)
                 )
 
-        if not self.api_key:
-            raise ValueError(
-                "Azure API key is required. Set AZURE_API_KEY environment "
-                "variable or pass api_key parameter."
-            )
         if not self.endpoint:
             raise ValueError(
                 "Azure endpoint is required. Set AZURE_ENDPOINT environment "
@@ -195,12 +190,39 @@ class AzureCompletion(BaseLLM):
             )
         client_kwargs: dict[str, Any] = {
             "endpoint": self.endpoint,
-            "credential": AzureKeyCredential(self.api_key),
+            "credential": self._resolve_credential(),
         }
         if self.api_version:
             client_kwargs["api_version"] = self.api_version
         return client_kwargs
 
+    def _resolve_credential(self) -> Any:
+        """Return an Azure credential, preferring the API key when set.
+
+        Without an API key, fall back to ``DefaultAzureCredential`` from
+        ``azure-identity``. That chain auto-detects the standard keyless
+        paths the customer's environment may provide — OIDC Workload
+        Identity Federation (``AZURE_FEDERATED_TOKEN_FILE`` +
+        ``AZURE_TENANT_ID`` + ``AZURE_CLIENT_ID``), Managed Identity on
+        AKS/Azure VMs, environment-configured service principals, and
+        developer tools like the Azure CLI. Installing ``azure-identity``
+        is what enables these paths; without it we raise the existing
+        API-key error.
+        """
+        if self.api_key:
+            return AzureKeyCredential(self.api_key)
+
+        try:
+            from azure.identity import DefaultAzureCredential
+        except ImportError:
+            raise ValueError(
+                "Azure API key is required when azure-identity is not "
+                "installed. Set AZURE_API_KEY, or install azure-identity "
+                'for keyless auth: uv add "crewai[azure-ai-inference]"'
+            ) from None
+
+        return DefaultAzureCredential()
+
     def _get_sync_client(self) -> Any:
         if self._client is None:
             self._client = self._build_sync_client()

lib/crewai/src/crewai/mcp/__init__.py

@@ -2,9 +2,17 @@
 
 This module provides native MCP client functionality, allowing CrewAI agents
 to connect to any MCP-compliant server using various transport types.
+
+Heavy imports (MCPClient, MCPToolResolver, BaseTransport, TransportType) are
+lazy-loaded on first access to avoid pulling in the ``mcp`` SDK (~400ms)
+when only lightweight config/filter types are needed.
 """
 
-from crewai.mcp.client import MCPClient
+from __future__ import annotations
+
+import importlib
+from typing import TYPE_CHECKING, Any
+
 from crewai.mcp.config import (
     MCPServerConfig,
     MCPServerHTTP,
@@ -18,8 +26,28 @@ from crewai.mcp.filters import (
     create_dynamic_tool_filter,
     create_static_tool_filter,
 )
-from crewai.mcp.tool_resolver import MCPToolResolver
-from crewai.mcp.transports.base import BaseTransport, TransportType
+
+if TYPE_CHECKING:
+    from crewai.mcp.client import MCPClient
+    from crewai.mcp.tool_resolver import MCPToolResolver
+    from crewai.mcp.transports.base import BaseTransport, TransportType
+
+_LAZY: dict[str, tuple[str, str]] = {
+    "MCPClient": ("crewai.mcp.client", "MCPClient"),
+    "MCPToolResolver": ("crewai.mcp.tool_resolver", "MCPToolResolver"),
+    "BaseTransport": ("crewai.mcp.transports.base", "BaseTransport"),
+    "TransportType": ("crewai.mcp.transports.base", "TransportType"),
+}
+
+
+def __getattr__(name: str) -> Any:
+    if name in _LAZY:
+        mod_path, attr = _LAZY[name]
+        mod = importlib.import_module(mod_path)
+        val = getattr(mod, attr)
+        globals()[name] = val  # cache for subsequent access
+        return val
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
 
 __all__ = [

lib/crewai/src/crewai/state/checkpoint_listener.py

@@ -10,12 +10,22 @@ from __future__ import annotations
 import json
 import logging
 import threading
+import time
 from typing import Any
 
 from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.crew import Crew
 from crewai.events.base_events import BaseEvent
-from crewai.events.event_bus import CrewAIEventsBus, crewai_event_bus
+from crewai.events.event_bus import CrewAIEventsBus, crewai_event_bus, is_replaying
+from crewai.events.types.checkpoint_events import (
+    CheckpointBaseEvent,
+    CheckpointCompletedEvent,
+    CheckpointFailedEvent,
+    CheckpointForkBaseEvent,
+    CheckpointPrunedEvent,
+    CheckpointRestoreBaseEvent,
+    CheckpointStartedEvent,
+)
 from crewai.flow.flow import Flow
 from crewai.state.checkpoint_config import CheckpointConfig
 from crewai.state.runtime import RuntimeState, _prepare_entities
@@ -53,12 +63,26 @@ def _resolve(value: CheckpointConfig | bool | None) -> CheckpointConfig | None |
     if isinstance(value, CheckpointConfig):
         _ensure_handlers_registered()
         return value
-    if value is True:
+    if value:
         _ensure_handlers_registered()
         return CheckpointConfig()
     if value is False:
         return _SENTINEL
-    return None  # None = inherit
+    return None
+
+
+def _resolve_from_agent(agent: BaseAgent) -> CheckpointConfig | None:
+    """Resolve a checkpoint config starting from an agent, walking to its crew."""
+    result = _resolve(agent.checkpoint)
+    if isinstance(result, CheckpointConfig):
+        return result
+    if result is _SENTINEL:
+        return None
+    crew = agent.crew
+    if isinstance(crew, Crew):
+        crew_result = _resolve(crew.checkpoint)
+        return crew_result if isinstance(crew_result, CheckpointConfig) else None
+    return None
 
 
 def _find_checkpoint(source: Any) -> CheckpointConfig | None:
@@ -77,28 +101,11 @@ def _find_checkpoint(source: Any) -> CheckpointConfig | None:
         result = _resolve(source.checkpoint)
         return result if isinstance(result, CheckpointConfig) else None
     if isinstance(source, BaseAgent):
-        result = _resolve(source.checkpoint)
-        if isinstance(result, CheckpointConfig):
-            return result
-        if result is _SENTINEL:
-            return None
-        crew = source.crew
-        if isinstance(crew, Crew):
-            result = _resolve(crew.checkpoint)
-            return result if isinstance(result, CheckpointConfig) else None
-        return None
+        return _resolve_from_agent(source)
     if isinstance(source, Task):
         agent = source.agent
         if isinstance(agent, BaseAgent):
-            result = _resolve(agent.checkpoint)
-            if isinstance(result, CheckpointConfig):
-                return result
-            if result is _SENTINEL:
-                return None
-            crew = agent.crew
-            if isinstance(crew, Crew):
-                result = _resolve(crew.checkpoint)
-                return result if isinstance(result, CheckpointConfig) else None
+            return _resolve_from_agent(agent)
         return None
     return None
 
@@ -107,27 +114,106 @@ def _do_checkpoint(
     state: RuntimeState, cfg: CheckpointConfig, event: BaseEvent | None = None
 ) -> None:
     """Write a checkpoint and prune old ones if configured."""
-    _prepare_entities(state.root)
-    payload = state.model_dump(mode="json")
-    if event is not None:
-        payload["trigger"] = event.type
-    data = json.dumps(payload)
-    location = cfg.provider.checkpoint(
-        data,
-        cfg.location,
-        parent_id=state._parent_id,
-        branch=state._branch,
-    )
-    state._chain_lineage(cfg.provider, location)
+    provider_name: str = type(cfg.provider).__name__
+    trigger: str | None = event.type if event is not None else None
+    context: dict[str, Any] = {
+        "task_id": event.task_id if event is not None else None,
+        "task_name": event.task_name if event is not None else None,
+        "agent_id": event.agent_id if event is not None else None,
+        "agent_role": event.agent_role if event is not None else None,
+    }
 
-    checkpoint_id: str = cfg.provider.extract_id(location)
+    parent_id_snapshot: str | None = state._parent_id
+    branch_snapshot: str = state._branch
+
+    crewai_event_bus.emit(
+        cfg,
+        CheckpointStartedEvent(
+            location=cfg.location,
+            provider=provider_name,
+            trigger=trigger,
+            branch=branch_snapshot,
+            parent_id=parent_id_snapshot,
+            **context,
+        ),
+    )
+
+    start: float = time.perf_counter()
+    try:
+        _prepare_entities(state.root)
+        payload = state.model_dump(mode="json")
+        if event is not None:
+            payload["trigger"] = event.type
+        data = json.dumps(payload)
+        location = cfg.provider.checkpoint(
+            data,
+            cfg.location,
+            parent_id=parent_id_snapshot,
+            branch=branch_snapshot,
+        )
+        state._chain_lineage(cfg.provider, location)
+        checkpoint_id: str = cfg.provider.extract_id(location)
+    except Exception as exc:
+        crewai_event_bus.emit(
+            cfg,
+            CheckpointFailedEvent(
+                location=cfg.location,
+                provider=provider_name,
+                trigger=trigger,
+                branch=branch_snapshot,
+                parent_id=parent_id_snapshot,
+                error=str(exc),
+                **context,
+            ),
+        )
+        raise
+
+    duration_ms: float = (time.perf_counter() - start) * 1000.0
     msg: str = (
         f"Checkpoint saved. Resume with: crewai checkpoint resume {checkpoint_id}"
     )
     logger.info(msg)
 
+    crewai_event_bus.emit(
+        cfg,
+        CheckpointCompletedEvent(
+            location=location,
+            provider=provider_name,
+            trigger=trigger,
+            branch=branch_snapshot,
+            parent_id=parent_id_snapshot,
+            checkpoint_id=checkpoint_id,
+            duration_ms=duration_ms,
+            **context,
+        ),
+    )
+
     if cfg.max_checkpoints is not None:
-        cfg.provider.prune(cfg.location, cfg.max_checkpoints, branch=state._branch)
+        try:
+            removed_count: int = cfg.provider.prune(
+                cfg.location, cfg.max_checkpoints, branch=branch_snapshot
+            )
+        except Exception:
+            logger.warning(
+                "Checkpoint prune failed for %s (branch=%s)",
+                cfg.location,
+                branch_snapshot,
+                exc_info=True,
+            )
+            return
+        crewai_event_bus.emit(
+            cfg,
+            CheckpointPrunedEvent(
+                location=cfg.location,
+                provider=provider_name,
+                trigger=trigger,
+                branch=branch_snapshot,
+                parent_id=parent_id_snapshot,
+                removed_count=removed_count,
+                max_checkpoints=cfg.max_checkpoints,
+                **context,
+            ),
+        )
 
 
 def _should_checkpoint(source: Any, event: BaseEvent) -> CheckpointConfig | None:
@@ -142,6 +228,13 @@ def _should_checkpoint(source: Any, event: BaseEvent) -> CheckpointConfig | None
 
 def _on_any_event(source: Any, event: BaseEvent, state: Any) -> None:
     """Sync handler registered on every event class."""
+    if is_replaying():
+        return
+    if isinstance(
+        event,
+        (CheckpointBaseEvent, CheckpointForkBaseEvent, CheckpointRestoreBaseEvent),
+    ):
+        return
     cfg = _should_checkpoint(source, event)
     if cfg is None:
         return
@@ -161,7 +254,8 @@ def _register_all_handlers(event_bus: CrewAIEventsBus) -> None:
     seen: set[type] = set()
 
     def _collect(cls: type[BaseEvent]) -> None:
-        for sub in cls.__subclasses__():
+        subclasses: list[type[BaseEvent]] = cls.__subclasses__()
+        for sub in subclasses:
             if sub not in seen:
                 seen.add(sub)
                 type_field = sub.model_fields.get("type")

lib/crewai/src/crewai/state/event_record.py

@@ -39,7 +39,8 @@ def _build_event_type_map() -> None:
     """Populate _event_type_map from all BaseEvent subclasses."""
 
     def _collect(cls: type[BaseEvent]) -> None:
-        for sub in cls.__subclasses__():
+        subclasses: list[type[BaseEvent]] = cls.__subclasses__()
+        for sub in subclasses:
             type_field = sub.model_fields.get("type")
             if type_field and type_field.default:
                 _event_type_map[type_field.default] = sub
@@ -196,6 +197,21 @@ class EventRecord(BaseModel):
                 node for node in self.nodes.values() if not node.neighbors("parent")
             ]
 
+    def all_nodes(self) -> list[EventNode]:
+        """Return a snapshot of every node under the read lock.
+
+        Returns:
+            A list copy of the current nodes, safe to iterate without holding
+            the lock.
+        """
+        with self._lock.r_locked():
+            return list(self.nodes.values())
+
+    def clear(self) -> None:
+        """Remove all nodes from the record under the write lock."""
+        with self._lock.w_locked():
+            self.nodes.clear()
+
     def __len__(self) -> int:
         with self._lock.r_locked():
             return len(self.nodes)

lib/crewai/src/crewai/state/provider/core.py

@@ -61,13 +61,16 @@ class BaseProvider(BaseModel, ABC):
         ...
 
     @abstractmethod
-    def prune(self, location: str, max_keep: int, *, branch: str = "main") -> None:
+    def prune(self, location: str, max_keep: int, *, branch: str = "main") -> int:
         """Remove old checkpoints, keeping at most *max_keep* per branch.
 
         Args:
             location: The storage destination passed to ``checkpoint``.
             max_keep: Maximum number of checkpoints to retain.
             branch: Only prune checkpoints on this branch.
+
+        Returns:
+            The number of checkpoints removed.
         """
         ...
 

lib/crewai/src/crewai/state/provider/json_provider.py

@@ -95,17 +95,20 @@ class JsonProvider(BaseProvider):
             await f.write(data)
         return str(file_path)
 
-    def prune(self, location: str, max_keep: int, *, branch: str = "main") -> None:
+    def prune(self, location: str, max_keep: int, *, branch: str = "main") -> int:
         """Remove oldest checkpoint files beyond *max_keep* on a branch."""
         _safe_branch(location, branch)
         branch_dir = os.path.join(location, branch)
         pattern = os.path.join(branch_dir, "*.json")
         files = sorted(glob.glob(pattern), key=os.path.getmtime)
+        removed = 0
         for path in files if max_keep == 0 else files[:-max_keep]:
             try:
                 os.remove(path)
+                removed += 1
             except OSError:  # noqa: PERF203
                 logger.debug("Failed to remove %s", path, exc_info=True)
+        return removed
 
     def extract_id(self, location: str) -> str:
         """Extract the checkpoint ID from a file path.

lib/crewai/src/crewai/state/provider/sqlite_provider.py

@@ -111,11 +111,13 @@ class SqliteProvider(BaseProvider):
             await db.commit()
         return f"{location}#{checkpoint_id}"
 
-    def prune(self, location: str, max_keep: int, *, branch: str = "main") -> None:
+    def prune(self, location: str, max_keep: int, *, branch: str = "main") -> int:
         """Remove oldest checkpoint rows beyond *max_keep* on a branch."""
         with sqlite3.connect(location) as conn:
-            conn.execute(_PRUNE, (branch, branch, max_keep))
+            cursor = conn.execute(_PRUNE, (branch, branch, max_keep))
+            removed: int = cursor.rowcount
             conn.commit()
+        return max(removed, 0)
 
     def extract_id(self, location: str) -> str:
         """Extract the checkpoint ID from a ``db_path#id`` string."""

lib/crewai/src/crewai/state/runtime.py

@@ -10,6 +10,7 @@ via ``RuntimeState.model_rebuild()``.
 from __future__ import annotations
 
 import logging
+import time
 from typing import TYPE_CHECKING, Any
 import uuid
 
@@ -23,6 +24,17 @@ from pydantic import (
 )
 
 from crewai.context import capture_execution_context
+from crewai.events.event_bus import crewai_event_bus
+from crewai.events.types.checkpoint_events import (
+    CheckpointCompletedEvent,
+    CheckpointFailedEvent,
+    CheckpointForkCompletedEvent,
+    CheckpointForkStartedEvent,
+    CheckpointRestoreCompletedEvent,
+    CheckpointRestoreFailedEvent,
+    CheckpointRestoreStartedEvent,
+    CheckpointStartedEvent,
+)
 from crewai.state.checkpoint_config import CheckpointConfig
 from crewai.state.event_record import EventRecord
 from crewai.state.provider.core import BaseProvider
@@ -89,7 +101,7 @@ def _migrate(data: dict[str, Any]) -> dict[str, Any]:
     """
     raw = data.get("crewai_version")
     current = Version(get_crewai_version())
-    stored = Version(raw) if raw else Version("0.0.0")
+    stored = Version(raw) if isinstance(raw, str) and raw else Version("0.0.0")
 
     if raw is None:
         logger.warning("Checkpoint has no crewai_version — treating as 0.0.0")
@@ -159,6 +171,63 @@ class RuntimeState(RootModel):  # type: ignore[type-arg]
         self._checkpoint_id = provider.extract_id(location)
         self._parent_id = self._checkpoint_id
 
+    def _begin_checkpoint(self, location: str) -> tuple[str, str | None, str, float]:
+        """Emit the start event and return the invariant context for a checkpoint."""
+        provider_name: str = type(self._provider).__name__
+        parent_id_snapshot: str | None = self._parent_id
+        branch_snapshot: str = self._branch
+        crewai_event_bus.emit(
+            self,
+            CheckpointStartedEvent(
+                location=location,
+                provider=provider_name,
+                branch=branch_snapshot,
+                parent_id=parent_id_snapshot,
+            ),
+        )
+        return provider_name, parent_id_snapshot, branch_snapshot, time.perf_counter()
+
+    def _emit_checkpoint_failed(
+        self,
+        location: str,
+        provider_name: str,
+        branch_snapshot: str,
+        parent_id_snapshot: str | None,
+        exc: Exception,
+    ) -> None:
+        """Emit the failure event for a checkpoint write."""
+        crewai_event_bus.emit(
+            self,
+            CheckpointFailedEvent(
+                location=location,
+                provider=provider_name,
+                branch=branch_snapshot,
+                parent_id=parent_id_snapshot,
+                error=str(exc),
+            ),
+        )
+
+    def _emit_checkpoint_completed(
+        self,
+        result: str,
+        provider_name: str,
+        branch_snapshot: str,
+        parent_id_snapshot: str | None,
+        start: float,
+    ) -> None:
+        """Emit the completion event for a successful checkpoint write."""
+        crewai_event_bus.emit(
+            self,
+            CheckpointCompletedEvent(
+                location=result,
+                provider=provider_name,
+                branch=branch_snapshot,
+                parent_id=parent_id_snapshot,
+                checkpoint_id=self._provider.extract_id(result),
+                duration_ms=(time.perf_counter() - start) * 1000.0,
+            ),
+        )
+
     def checkpoint(self, location: str) -> str:
         """Write a checkpoint.
 
@@ -169,14 +238,27 @@ class RuntimeState(RootModel):  # type: ignore[type-arg]
         Returns:
             A location identifier for the saved checkpoint.
         """
-        _prepare_entities(self.root)
-        result = self._provider.checkpoint(
-            self.model_dump_json(),
-            location,
-            parent_id=self._parent_id,
-            branch=self._branch,
+        provider_name, parent_id_snapshot, branch_snapshot, start = (
+            self._begin_checkpoint(location)
+        )
+        try:
+            _prepare_entities(self.root)
+            result = self._provider.checkpoint(
+                self.model_dump_json(),
+                location,
+                parent_id=parent_id_snapshot,
+                branch=branch_snapshot,
+            )
+            self._chain_lineage(self._provider, result)
+        except Exception as exc:
+            self._emit_checkpoint_failed(
+                location, provider_name, branch_snapshot, parent_id_snapshot, exc
+            )
+            raise
+
+        self._emit_checkpoint_completed(
+            result, provider_name, branch_snapshot, parent_id_snapshot, start
         )
-        self._chain_lineage(self._provider, result)
         return result
 
     async def acheckpoint(self, location: str) -> str:
@@ -189,14 +271,27 @@ class RuntimeState(RootModel):  # type: ignore[type-arg]
         Returns:
             A location identifier for the saved checkpoint.
         """
-        _prepare_entities(self.root)
-        result = await self._provider.acheckpoint(
-            self.model_dump_json(),
-            location,
-            parent_id=self._parent_id,
-            branch=self._branch,
+        provider_name, parent_id_snapshot, branch_snapshot, start = (
+            self._begin_checkpoint(location)
+        )
+        try:
+            _prepare_entities(self.root)
+            result = await self._provider.acheckpoint(
+                self.model_dump_json(),
+                location,
+                parent_id=parent_id_snapshot,
+                branch=branch_snapshot,
+            )
+            self._chain_lineage(self._provider, result)
+        except Exception as exc:
+            self._emit_checkpoint_failed(
+                location, provider_name, branch_snapshot, parent_id_snapshot, exc
+            )
+            raise
+
+        self._emit_checkpoint_completed(
+            result, provider_name, branch_snapshot, parent_id_snapshot, start
         )
-        self._chain_lineage(self._provider, result)
         return result
 
     def fork(self, branch: str | None = None) -> None:
@@ -211,11 +306,32 @@ class RuntimeState(RootModel):  # type: ignore[type-arg]
                 times without collisions.
         """
         if branch:
-            self._branch = branch
+            new_branch = branch
         elif self._checkpoint_id:
-            self._branch = f"fork/{self._checkpoint_id}_{uuid.uuid4().hex[:6]}"
+            new_branch = f"fork/{self._checkpoint_id}_{uuid.uuid4().hex[:6]}"
         else:
-            self._branch = f"fork/{uuid.uuid4().hex[:8]}"
+            new_branch = f"fork/{uuid.uuid4().hex[:8]}"
+
+        parent_branch: str | None = self._branch
+        parent_checkpoint_id: str | None = self._checkpoint_id
+
+        crewai_event_bus.emit(
+            self,
+            CheckpointForkStartedEvent(
+                branch=new_branch,
+                parent_branch=parent_branch,
+                parent_checkpoint_id=parent_checkpoint_id,
+            ),
+        )
+        self._branch = new_branch
+        crewai_event_bus.emit(
+            self,
+            CheckpointForkCompletedEvent(
+                branch=new_branch,
+                parent_branch=parent_branch,
+                parent_checkpoint_id=parent_checkpoint_id,
+            ),
+        )
 
     @classmethod
     def from_checkpoint(cls, config: CheckpointConfig, **kwargs: Any) -> RuntimeState:
@@ -233,13 +349,41 @@ class RuntimeState(RootModel):  # type: ignore[type-arg]
         if config.restore_from is None:
             raise ValueError("CheckpointConfig.restore_from must be set")
         location = str(config.restore_from)
-        provider = detect_provider(location)
-        raw = provider.from_checkpoint(location)
-        state = cls.model_validate_json(raw, **kwargs)
-        state._provider = provider
-        checkpoint_id = provider.extract_id(location)
-        state._checkpoint_id = checkpoint_id
-        state._parent_id = checkpoint_id
+
+        crewai_event_bus.emit(config, CheckpointRestoreStartedEvent(location=location))
+        start: float = time.perf_counter()
+        provider_name: str | None = None
+        try:
+            provider = detect_provider(location)
+            provider_name = type(provider).__name__
+            raw = provider.from_checkpoint(location)
+            state = cls.model_validate_json(raw, **kwargs)
+            state._provider = provider
+            checkpoint_id = provider.extract_id(location)
+            state._checkpoint_id = checkpoint_id
+            state._parent_id = checkpoint_id
+        except Exception as exc:
+            crewai_event_bus.emit(
+                config,
+                CheckpointRestoreFailedEvent(
+                    location=location,
+                    provider=provider_name,
+                    error=str(exc),
+                ),
+            )
+            raise
+
+        crewai_event_bus.emit(
+            config,
+            CheckpointRestoreCompletedEvent(
+                location=location,
+                provider=provider_name,
+                checkpoint_id=checkpoint_id,
+                branch=state._branch,
+                parent_id=state._parent_id,
+                duration_ms=(time.perf_counter() - start) * 1000.0,
+            ),
+        )
         return state
 
     @classmethod
@@ -260,13 +404,41 @@ class RuntimeState(RootModel):  # type: ignore[type-arg]
         if config.restore_from is None:
             raise ValueError("CheckpointConfig.restore_from must be set")
         location = str(config.restore_from)
-        provider = detect_provider(location)
-        raw = await provider.afrom_checkpoint(location)
-        state = cls.model_validate_json(raw, **kwargs)
-        state._provider = provider
-        checkpoint_id = provider.extract_id(location)
-        state._checkpoint_id = checkpoint_id
-        state._parent_id = checkpoint_id
+
+        crewai_event_bus.emit(config, CheckpointRestoreStartedEvent(location=location))
+        start: float = time.perf_counter()
+        provider_name: str | None = None
+        try:
+            provider = detect_provider(location)
+            provider_name = type(provider).__name__
+            raw = await provider.afrom_checkpoint(location)
+            state = cls.model_validate_json(raw, **kwargs)
+            state._provider = provider
+            checkpoint_id = provider.extract_id(location)
+            state._checkpoint_id = checkpoint_id
+            state._parent_id = checkpoint_id
+        except Exception as exc:
+            crewai_event_bus.emit(
+                config,
+                CheckpointRestoreFailedEvent(
+                    location=location,
+                    provider=provider_name,
+                    error=str(exc),
+                ),
+            )
+            raise
+
+        crewai_event_bus.emit(
+            config,
+            CheckpointRestoreCompletedEvent(
+                location=location,
+                provider=provider_name,
+                checkpoint_id=checkpoint_id,
+                branch=state._branch,
+                parent_id=state._parent_id,
+                duration_ms=(time.perf_counter() - start) * 1000.0,
+            ),
+        )
         return state
 
 

lib/crewai/tests/a2a/test_a2a_integration.py

@@ -3,12 +3,23 @@ from __future__ import annotations
 import os
 import uuid
 
+import httpx
 import pytest
 import pytest_asyncio
 
-from a2a.client import ClientFactory
-from a2a.types import AgentCard, Message, Part, Role, TaskState, TextPart
+from a2a.client import A2ACardResolver, ClientFactory
+from a2a.types import AgentCapabilities, AgentCard, AgentInterface, Message, Part, Role, TaskState
 
+from crewai.a2a._compat import (
+    ROLE_AGENT,
+    ROLE_USER,
+    TASK_STATE_COMPLETED,
+    TASK_STATE_FAILED,
+    agent_card_url,
+    make_send_request,
+    new_text_message,
+    new_text_part,
+)
 from crewai.a2a.updates.polling.handler import PollingHandler
 from crewai.a2a.updates.streaming.handler import StreamingHandler
 
@@ -17,27 +28,31 @@ A2A_TEST_ENDPOINT = os.getenv("A2A_TEST_ENDPOINT", "http://localhost:9999")
 
 
 @pytest_asyncio.fixture
-async def a2a_client():
+async def card_resolver():
+    """Create an A2ACardResolver for the test server."""
+    async with httpx.AsyncClient() as http_client:
+        resolver = A2ACardResolver(http_client, A2A_TEST_ENDPOINT)
+        yield resolver
+
+
+@pytest_asyncio.fixture
+async def agent_card(card_resolver) -> AgentCard:
+    """Fetch the real agent card from the server."""
+    return await card_resolver.get_agent_card()
+
+
+@pytest_asyncio.fixture
+async def a2a_client(agent_card):
     """Create A2A client for test server."""
-    client = await ClientFactory.connect(A2A_TEST_ENDPOINT)
+    factory = ClientFactory()
+    client = factory.create(agent_card)
     yield client
-    await client.close()
 
 
 @pytest.fixture
 def test_message() -> Message:
     """Create a simple test message."""
-    return Message(
-        role=Role.user,
-        parts=[Part(root=TextPart(text="What is 2 + 2?"))],
-        message_id=str(uuid.uuid4()),
-    )
-
-
-@pytest_asyncio.fixture
-async def agent_card(a2a_client) -> AgentCard:
-    """Fetch the real agent card from the server."""
-    return await a2a_client.get_card()
+    return new_text_message("What is 2 + 2?", role=ROLE_USER)
 
 
 class TestA2AAgentCardFetching:
@@ -45,13 +60,13 @@ class TestA2AAgentCardFetching:
 
     @pytest.mark.vcr()
     @pytest.mark.asyncio
-    async def test_fetch_agent_card(self, a2a_client) -> None:
+    async def test_fetch_agent_card(self, card_resolver) -> None:
         """Test fetching an agent card from the server."""
-        card = await a2a_client.get_card()
+        card = await card_resolver.get_agent_card()
 
         assert card is not None
         assert card.name == "GPT Assistant"
-        assert card.url is not None
+        assert card.supported_interfaces is not None
         assert card.capabilities is not None
         assert card.capabilities.streaming is True
 
@@ -80,7 +95,7 @@ class TestA2APollingIntegration:
         )
 
         assert isinstance(result, dict)
-        assert result["status"] == TaskState.completed
+        assert result["status"] == TASK_STATE_COMPLETED
         assert result.get("result") is not None
         assert "4" in result["result"]
 
@@ -104,11 +119,11 @@ class TestA2AStreamingIntegration:
             message=test_message,
             new_messages=new_messages,
             agent_card=agent_card,
-            endpoint=agent_card.url,
+            endpoint=agent_card_url(agent_card),
         )
 
         assert isinstance(result, dict)
-        assert result["status"] == TaskState.completed
+        assert result["status"] == TASK_STATE_COMPLETED
         assert result.get("result") is not None
 
 
@@ -123,19 +138,19 @@ class TestA2ATaskOperations:
         test_message: Message,
     ) -> None:
         """Test sending a message and getting a response."""
-        from a2a.types import Task
+        from a2a.types import StreamResponse, Task
+
+        from crewai.a2a._compat import is_stream_task
 
         final_task: Task | None = None
-        async for event in a2a_client.send_message(test_message):
-            if isinstance(event, tuple) and len(event) >= 1:
-                task, _ = event
-                if isinstance(task, Task):
-                    final_task = task
+        async for event in a2a_client.send_message(make_send_request(test_message)):
+            if isinstance(event, StreamResponse) and is_stream_task(event):
+                final_task = event.task
 
         assert final_task is not None
-        assert final_task.id is not None
+        assert final_task.id != ""
         assert final_task.status is not None
-        assert final_task.status.state == TaskState.completed
+        assert final_task.status.state == TaskState.TASK_STATE_COMPLETED
 
 
 class TestA2APushNotificationHandler:
@@ -148,17 +163,19 @@ class TestA2APushNotificationHandler:
     @pytest.fixture
     def mock_agent_card(self) -> AgentCard:
         """Create a minimal valid agent card for testing."""
-        from a2a.types import AgentCapabilities
-
         return AgentCard(
             name="Test Agent",
             description="Test agent for push notification tests",
-            url="http://localhost:9999",
+            supported_interfaces=[
+                AgentInterface(
+                    url="http://localhost:9999",
+                    protocol_binding="JSONRPC",
+                ),
+            ],
             version="1.0.0",
             capabilities=AgentCapabilities(streaming=True, push_notifications=True),
             default_input_modes=["text"],
             default_output_modes=["text"],
-            skills=[],
         )
 
     @pytest.fixture
@@ -169,7 +186,7 @@ class TestA2APushNotificationHandler:
         return Task(
             id="task-123",
             context_id="ctx-123",
-            status=TaskStatus(state=TaskState.working),
+            status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
         )
 
     @pytest.mark.asyncio
@@ -181,7 +198,7 @@ class TestA2APushNotificationHandler:
         """Test that push handler waits for result from store."""
         from unittest.mock import AsyncMock, MagicMock
 
-        from a2a.types import Task, TaskStatus
+        from a2a.types import StreamResponse, Task, TaskStatus
         from pydantic import AnyHttpUrl
 
         from crewai.a2a.updates.push_notifications.config import PushNotificationConfig
@@ -190,15 +207,14 @@ class TestA2APushNotificationHandler:
         completed_task = Task(
             id="task-123",
             context_id="ctx-123",
-            status=TaskStatus(state=TaskState.completed),
-            history=[],
+            status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
         )
 
         mock_store = MagicMock()
         mock_store.wait_for_result = AsyncMock(return_value=completed_task)
 
         async def mock_send_message(*args, **kwargs):
-            yield (mock_task, None)
+            yield StreamResponse(task=mock_task)
 
         mock_client = MagicMock()
         mock_client.send_message = mock_send_message
@@ -209,11 +225,7 @@ class TestA2APushNotificationHandler:
             result_store=mock_store,
         )
 
-        test_msg = Message(
-            role=Role.user,
-            parts=[Part(root=TextPart(text="What is 2+2?"))],
-            message_id="msg-001",
-        )
+        test_msg = new_text_message("What is 2+2?", role=ROLE_USER)
 
         new_messages: list[Message] = []
 
@@ -226,7 +238,7 @@ class TestA2APushNotificationHandler:
             result_store=mock_store,
             polling_timeout=30.0,
             polling_interval=1.0,
-            endpoint=mock_agent_card.url,
+            endpoint=agent_card_url(mock_agent_card),
         )
 
         mock_store.wait_for_result.assert_called_once_with(
@@ -235,7 +247,7 @@ class TestA2APushNotificationHandler:
             poll_interval=1.0,
         )
 
-        assert result["status"] == TaskState.completed
+        assert result["status"] == TASK_STATE_COMPLETED
 
     @pytest.mark.asyncio
     async def test_push_handler_returns_failure_on_timeout(
@@ -245,7 +257,7 @@ class TestA2APushNotificationHandler:
         """Test that push handler returns failure when result store times out."""
         from unittest.mock import AsyncMock, MagicMock
 
-        from a2a.types import Task, TaskStatus
+        from a2a.types import StreamResponse, Task, TaskStatus
         from pydantic import AnyHttpUrl
 
         from crewai.a2a.updates.push_notifications.config import PushNotificationConfig
@@ -257,11 +269,11 @@ class TestA2APushNotificationHandler:
         working_task = Task(
             id="task-456",
             context_id="ctx-456",
-            status=TaskStatus(state=TaskState.working),
+            status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
         )
 
         async def mock_send_message(*args, **kwargs):
-            yield (working_task, None)
+            yield StreamResponse(task=working_task)
 
         mock_client = MagicMock()
         mock_client.send_message = mock_send_message
@@ -272,11 +284,7 @@ class TestA2APushNotificationHandler:
             result_store=mock_store,
         )
 
-        test_msg = Message(
-            role=Role.user,
-            parts=[Part(root=TextPart(text="test"))],
-            message_id="msg-002",
-        )
+        test_msg = new_text_message("test", role=ROLE_USER)
 
         new_messages: list[Message] = []
 
@@ -289,10 +297,10 @@ class TestA2APushNotificationHandler:
             result_store=mock_store,
             polling_timeout=5.0,
             polling_interval=0.5,
-            endpoint=mock_agent_card.url,
+            endpoint=agent_card_url(mock_agent_card),
         )
 
-        assert result["status"] == TaskState.failed
+        assert result["status"] == TASK_STATE_FAILED
         assert "timeout" in result.get("error", "").lower()
 
     @pytest.mark.asyncio
@@ -307,11 +315,7 @@ class TestA2APushNotificationHandler:
 
         mock_client = MagicMock()
 
-        test_msg = Message(
-            role=Role.user,
-            parts=[Part(root=TextPart(text="test"))],
-            message_id="msg-003",
-        )
+        test_msg = new_text_message("test", role=ROLE_USER)
 
         new_messages: list[Message] = []
 
@@ -320,8 +324,8 @@ class TestA2APushNotificationHandler:
             message=test_msg,
             new_messages=new_messages,
             agent_card=mock_agent_card,
-            endpoint=mock_agent_card.url,
+            endpoint=agent_card_url(mock_agent_card),
         )
 
-        assert result["status"] == TaskState.failed
+        assert result["status"] == TASK_STATE_FAILED
         assert "config" in result.get("error", "").lower()

lib/crewai/tests/a2a/test_a2a_sdk_v1_compat.py

@@ -0,0 +1,517 @@
+"""Tests for a2a-sdk v1.0 compatibility.
+
+These tests validate that crewai.a2a modules correctly import and work with
+a2a-sdk v1.0.x (protobuf-based types). They cover the core issue described
+in https://github.com/crewAIInc/crewAI/issues/5607:
+
+    ImportError: cannot import name 'A2AClientHTTPError' from 'a2a.client.errors'
+
+The migration from a2a-sdk ~0.3.10 to >=1.0.0,<2 introduced major breaking
+changes including protobuf-based types, renamed error classes, and new enum
+value conventions.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import pytest
+
+
+class TestSdkV1Imports:
+    """Verify that old v0.3 names no longer exist, and our compat layer works."""
+
+    def test_a2a_client_error_importable(self) -> None:
+        """A2AClientError (renamed from A2AClientHTTPError) should be importable."""
+        from a2a.client.errors import A2AClientError
+
+        assert A2AClientError is not None
+
+    def test_old_a2a_client_http_error_removed(self) -> None:
+        """A2AClientHTTPError no longer exists in a2a-sdk v1.0."""
+        with pytest.raises(ImportError):
+            from a2a.client.errors import A2AClientHTTPError  # noqa: F401
+
+    def test_compat_alias_maps_to_new_error(self) -> None:
+        """Our _compat alias should map to the new error class."""
+        from a2a.client.errors import A2AClientError
+
+        from crewai.a2a._compat import A2AClientHTTPError
+
+        assert A2AClientHTTPError is A2AClientError
+
+    def test_text_part_removed_in_v1(self) -> None:
+        """TextPart no longer exists as a separate type in a2a-sdk v1.0."""
+        with pytest.raises(ImportError):
+            from a2a.types import TextPart  # noqa: F401
+
+    def test_protobuf_types_importable(self) -> None:
+        """Key protobuf types should be importable from a2a.types."""
+        from a2a.types import (  # noqa: F401
+            AgentCapabilities,
+            AgentCard,
+            AgentInterface,
+            GetTaskRequest,
+            Message,
+            Part,
+            Role,
+            StreamResponse,
+            SubscribeToTaskRequest,
+            Task,
+            TaskPushNotificationConfig,
+            TaskState,
+            TaskStatusUpdateEvent,
+        )
+
+
+class TestCompatLayer:
+    """Tests for the crewai.a2a._compat compatibility layer."""
+
+    def test_role_constants(self) -> None:
+        """ROLE_USER and ROLE_AGENT should be valid Role enum values."""
+        from a2a.types import Role
+
+        from crewai.a2a._compat import ROLE_AGENT, ROLE_USER
+
+        assert ROLE_USER == Role.ROLE_USER
+        assert ROLE_AGENT == Role.ROLE_AGENT
+
+    def test_task_state_constants(self) -> None:
+        """TASK_STATE_* should be valid TaskState enum values."""
+        from a2a.types import TaskState
+
+        from crewai.a2a._compat import (
+            TASK_STATE_CANCELED,
+            TASK_STATE_COMPLETED,
+            TASK_STATE_FAILED,
+            TASK_STATE_INPUT_REQUIRED,
+            TASK_STATE_REJECTED,
+            TASK_STATE_SUBMITTED,
+            TASK_STATE_WORKING,
+        )
+
+        assert TASK_STATE_SUBMITTED == TaskState.TASK_STATE_SUBMITTED
+        assert TASK_STATE_WORKING == TaskState.TASK_STATE_WORKING
+        assert TASK_STATE_COMPLETED == TaskState.TASK_STATE_COMPLETED
+        assert TASK_STATE_FAILED == TaskState.TASK_STATE_FAILED
+        assert TASK_STATE_CANCELED == TaskState.TASK_STATE_CANCELED
+        assert TASK_STATE_INPUT_REQUIRED == TaskState.TASK_STATE_INPUT_REQUIRED
+        assert TASK_STATE_REJECTED == TaskState.TASK_STATE_REJECTED
+
+    def test_terminal_states(self) -> None:
+        """TERMINAL_STATES should include completed, failed, rejected, canceled."""
+        from crewai.a2a._compat import (
+            TASK_STATE_CANCELED,
+            TASK_STATE_COMPLETED,
+            TASK_STATE_FAILED,
+            TASK_STATE_REJECTED,
+            TERMINAL_STATES,
+        )
+
+        assert TASK_STATE_COMPLETED in TERMINAL_STATES
+        assert TASK_STATE_FAILED in TERMINAL_STATES
+        assert TASK_STATE_REJECTED in TERMINAL_STATES
+        assert TASK_STATE_CANCELED in TERMINAL_STATES
+
+
+class TestPartHelpers:
+    """Tests for protobuf Part helpers."""
+
+    def test_new_text_part(self) -> None:
+        """new_text_part should create a Part with text field set."""
+        from crewai.a2a._compat import new_text_part, part_is_text, part_text
+
+        part = new_text_part("hello world")
+        assert part_is_text(part)
+        assert part_text(part) == "hello world"
+
+    def test_part_is_text_false_for_non_text(self) -> None:
+        """part_is_text should return False for non-text parts."""
+        from a2a.types import Part
+        from google.protobuf.struct_pb2 import Value
+
+        from crewai.a2a._compat import part_is_text
+
+        v = Value()
+        v.string_value = "test"
+        part = Part(data=v)
+        assert not part_is_text(part)
+
+    def test_part_has_data(self) -> None:
+        """part_has_data should detect data parts."""
+        from a2a.types import Part
+        from google.protobuf.struct_pb2 import Value
+
+        from crewai.a2a._compat import part_has_data
+
+        v = Value()
+        v.string_value = "test"
+        part = Part(data=v)
+        assert part_has_data(part)
+
+    def test_part_has_file(self) -> None:
+        """part_has_file should detect raw/url file parts."""
+        from a2a.types import Part
+
+        from crewai.a2a._compat import part_has_file
+
+        raw_part = Part(raw=b"file content", media_type="application/pdf")
+        assert part_has_file(raw_part)
+
+        url_part = Part(url="https://example.com/file.pdf", media_type="application/pdf")
+        assert part_has_file(url_part)
+
+
+class TestMessageHelpers:
+    """Tests for protobuf Message helpers."""
+
+    def test_new_text_message(self) -> None:
+        """new_text_message should create a Message with a text Part."""
+        from crewai.a2a._compat import (
+            ROLE_USER,
+            new_text_message,
+            part_is_text,
+            part_text,
+        )
+
+        msg = new_text_message("test message", role=ROLE_USER)
+        assert msg.role == ROLE_USER
+        assert len(msg.parts) == 1
+        assert part_is_text(msg.parts[0])
+        assert part_text(msg.parts[0]) == "test message"
+
+    def test_new_text_message_with_context_and_task(self) -> None:
+        """new_text_message should accept context_id and task_id."""
+        from crewai.a2a._compat import ROLE_AGENT, new_text_message
+
+        msg = new_text_message(
+            "response",
+            role=ROLE_AGENT,
+            context_id="ctx-123",
+            task_id="task-456",
+        )
+        assert msg.context_id == "ctx-123"
+        assert msg.task_id == "task-456"
+
+
+class TestAgentCardHelpers:
+    """Tests for protobuf AgentCard helpers."""
+
+    def test_agent_card_to_dict(self) -> None:
+        """agent_card_to_dict should serialize an AgentCard to a plain dict."""
+        from a2a.types import AgentCard, AgentInterface
+
+        from crewai.a2a._compat import agent_card_to_dict
+
+        card = AgentCard(
+            name="Test Agent",
+            description="A test agent",
+            supported_interfaces=[
+                AgentInterface(url="http://localhost:9999", protocol_binding="JSONRPC"),
+            ],
+            version="1.0.0",
+        )
+        result = agent_card_to_dict(card)
+        assert isinstance(result, dict)
+        assert result["name"] == "Test Agent"
+        assert result["description"] == "A test agent"
+
+    def test_agent_card_url(self) -> None:
+        """agent_card_url should return the URL from the first interface."""
+        from a2a.types import AgentCard, AgentInterface
+
+        from crewai.a2a._compat import agent_card_url
+
+        card = AgentCard(
+            name="Test",
+            supported_interfaces=[
+                AgentInterface(url="http://localhost:9999", protocol_binding="JSONRPC"),
+            ],
+        )
+        assert agent_card_url(card) == "http://localhost:9999"
+
+    def test_agent_card_url_empty_when_no_interfaces(self) -> None:
+        """agent_card_url should return empty string if no interfaces."""
+        from a2a.types import AgentCard
+
+        from crewai.a2a._compat import agent_card_url
+
+        card = AgentCard(name="No Interfaces")
+        assert agent_card_url(card) == ""
+
+    def test_agent_card_preferred_transport(self) -> None:
+        """agent_card_preferred_transport should return protocol_binding."""
+        from a2a.types import AgentCard, AgentInterface
+
+        from crewai.a2a._compat import agent_card_preferred_transport
+
+        card = AgentCard(
+            name="Test",
+            supported_interfaces=[
+                AgentInterface(url="http://localhost", protocol_binding="GRPC"),
+            ],
+        )
+        assert agent_card_preferred_transport(card) == "GRPC"
+
+    def test_agent_card_interfaces(self) -> None:
+        """agent_card_interfaces should return all interfaces."""
+        from a2a.types import AgentCard, AgentInterface
+
+        from crewai.a2a._compat import agent_card_interfaces
+
+        card = AgentCard(
+            name="Test",
+            supported_interfaces=[
+                AgentInterface(url="http://a.com", protocol_binding="JSONRPC"),
+                AgentInterface(url="http://b.com", protocol_binding="GRPC"),
+            ],
+        )
+        interfaces = agent_card_interfaces(card)
+        assert len(interfaces) == 2
+
+    def test_agent_card_protocol_version(self) -> None:
+        """agent_card_protocol_version should return protocol version from first interface."""
+        from a2a.types import AgentCard, AgentInterface
+
+        from crewai.a2a._compat import agent_card_protocol_version
+
+        card = AgentCard(
+            name="Test",
+            supported_interfaces=[
+                AgentInterface(
+                    url="http://localhost",
+                    protocol_binding="JSONRPC",
+                    protocol_version="0.3",
+                ),
+            ],
+        )
+        assert agent_card_protocol_version(card) == "0.3"
+
+
+class TestProtoCopy:
+    """Tests for protobuf deep copy helper."""
+
+    def test_proto_copy_creates_independent_copy(self) -> None:
+        """proto_copy should create a deep copy of a protobuf message."""
+        from a2a.types import AgentCard, AgentInterface
+
+        from crewai.a2a._compat import proto_copy
+
+        original = AgentCard(
+            name="Original",
+            supported_interfaces=[
+                AgentInterface(url="http://original.com", protocol_binding="JSONRPC"),
+            ],
+        )
+        copy = proto_copy(original)
+        copy.name = "Modified"
+
+        assert original.name == "Original"
+        assert copy.name == "Modified"
+
+
+class TestStreamResponseHelpers:
+    """Tests for StreamResponse event helpers."""
+
+    def test_is_stream_message(self) -> None:
+        """is_stream_message should detect messages in StreamResponse."""
+        from a2a.types import Message, StreamResponse
+
+        from crewai.a2a._compat import ROLE_AGENT, is_stream_message, new_text_part
+
+        msg = Message(
+            role=ROLE_AGENT,
+            parts=[new_text_part("hello")],
+            message_id=str(uuid.uuid4()),
+        )
+        sr = StreamResponse(message=msg)
+        assert is_stream_message(sr)
+
+    def test_is_stream_task(self) -> None:
+        """is_stream_task should detect tasks in StreamResponse."""
+        from a2a.types import StreamResponse, Task, TaskState, TaskStatus
+
+        from crewai.a2a._compat import is_stream_task
+
+        task = Task(
+            id="task-1",
+            status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
+        )
+        sr = StreamResponse(task=task)
+        assert is_stream_task(sr)
+
+    def test_is_stream_status_update(self) -> None:
+        """is_stream_status_update should detect status updates."""
+        from a2a.types import StreamResponse, TaskState, TaskStatus, TaskStatusUpdateEvent
+
+        from crewai.a2a._compat import is_stream_status_update
+
+        update = TaskStatusUpdateEvent(
+            task_id="task-1",
+            context_id="ctx-1",
+            status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+        )
+        sr = StreamResponse(status_update=update)
+        assert is_stream_status_update(sr)
+
+
+class TestStatusUpdateFinality:
+    """Tests for status update finality detection."""
+
+    def test_completed_is_final(self) -> None:
+        """Completed status should be final."""
+        from a2a.types import TaskState, TaskStatus, TaskStatusUpdateEvent
+
+        from crewai.a2a._compat import is_status_update_final
+
+        update = TaskStatusUpdateEvent(
+            task_id="t1",
+            context_id="c1",
+            status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
+        )
+        assert is_status_update_final(update) is True
+
+    def test_working_is_not_final(self) -> None:
+        """Working status should not be final."""
+        from a2a.types import TaskState, TaskStatus, TaskStatusUpdateEvent
+
+        from crewai.a2a._compat import is_status_update_final
+
+        update = TaskStatusUpdateEvent(
+            task_id="t1",
+            context_id="c1",
+            status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+        )
+        assert is_status_update_final(update) is False
+
+    def test_failed_is_final(self) -> None:
+        """Failed status should be final."""
+        from a2a.types import TaskState, TaskStatus, TaskStatusUpdateEvent
+
+        from crewai.a2a._compat import is_status_update_final
+
+        update = TaskStatusUpdateEvent(
+            task_id="t1",
+            context_id="c1",
+            status=TaskStatus(state=TaskState.TASK_STATE_FAILED),
+        )
+        assert is_status_update_final(update) is True
+
+
+class TestClientConfigHelper:
+    """Tests for client configuration helper."""
+
+    def test_create_client_config(self) -> None:
+        """create_client_config should produce a valid ClientConfig."""
+        from crewai.a2a._compat import create_client_config
+
+        config = create_client_config(
+            supported_transports=["JSONRPC", "GRPC"],
+            streaming=True,
+            polling=False,
+        )
+        assert config.supported_protocol_bindings == ["JSONRPC", "GRPC"]
+        assert config.streaming is True
+        assert config.polling is False
+
+
+class TestProtoToJson:
+    """Tests for proto_to_json serialization."""
+
+    def test_proto_to_json(self) -> None:
+        """proto_to_json should serialize a protobuf to JSON string."""
+        from a2a.types import AgentCard, AgentInterface
+
+        from crewai.a2a._compat import proto_to_json
+
+        card = AgentCard(
+            name="Test Agent",
+            supported_interfaces=[
+                AgentInterface(url="http://localhost:9999", protocol_binding="JSONRPC"),
+            ],
+        )
+        json_str = proto_to_json(card)
+        assert isinstance(json_str, str)
+        assert "Test Agent" in json_str
+
+
+class TestModuleImports:
+    """Verify all crewai.a2a submodules import without error under v1.0."""
+
+    def test_import_compat(self) -> None:
+        from crewai.a2a._compat import A2AClientHTTPError  # noqa: F401
+
+    def test_import_task_helpers(self) -> None:
+        from crewai.a2a.task_helpers import process_task_state  # noqa: F401
+
+    def test_import_polling_handler(self) -> None:
+        from crewai.a2a.updates.polling.handler import PollingHandler  # noqa: F401
+
+    def test_import_streaming_handler(self) -> None:
+        from crewai.a2a.updates.streaming.handler import StreamingHandler  # noqa: F401
+
+    def test_import_push_handler(self) -> None:
+        from crewai.a2a.updates.push_notifications.handler import PushNotificationHandler  # noqa: F401
+
+    def test_import_auth_utils(self) -> None:
+        from crewai.a2a.auth.utils import validate_auth_against_agent_card  # noqa: F401
+
+    def test_import_delegation(self) -> None:
+        from crewai.a2a.utils.delegation import execute_a2a_delegation  # noqa: F401
+
+    def test_import_transport(self) -> None:
+        from crewai.a2a.utils.transport import negotiate_transport  # noqa: F401
+
+    def test_import_agent_card(self) -> None:
+        from crewai.a2a.utils.agent_card import afetch_agent_card  # noqa: F401
+
+    def test_import_agent_card_signing(self) -> None:
+        from crewai.a2a.utils.agent_card_signing import sign_agent_card  # noqa: F401
+
+    def test_import_wrapper(self) -> None:
+        from crewai.a2a.wrapper import wrap_agent_with_a2a_instance  # noqa: F401
+
+    def test_import_extensions_registry(self) -> None:
+        from crewai.a2a.extensions.registry import ExtensionsMiddleware  # noqa: F401
+
+    def test_import_content_type(self) -> None:
+        from crewai.a2a.utils.content_type import get_part_content_type  # noqa: F401
+
+
+class TestGetPartContentType:
+    """Tests for get_part_content_type with v1.0 protobuf Parts."""
+
+    def test_text_part_returns_text_plain(self) -> None:
+        from a2a.types import Part
+
+        from crewai.a2a.utils.content_type import get_part_content_type
+
+        part = Part(text="hello")
+        assert get_part_content_type(part) == "text/plain"
+
+    def test_data_part_returns_application_json(self) -> None:
+        from a2a.types import Part
+        from google.protobuf.struct_pb2 import Value
+
+        from crewai.a2a.utils.content_type import get_part_content_type
+
+        v = Value()
+        v.string_value = "test"
+        part = Part(data=v)
+        assert get_part_content_type(part) == "application/json"
+
+    def test_raw_part_returns_media_type(self) -> None:
+        from a2a.types import Part
+
+        from crewai.a2a.utils.content_type import get_part_content_type
+
+        part = Part(raw=b"pdf content", media_type="application/pdf")
+        assert get_part_content_type(part) == "application/pdf"
+
+    def test_url_part_returns_media_type(self) -> None:
+        from a2a.types import Part
+
+        from crewai.a2a.utils.content_type import get_part_content_type
+
+        part = Part(url="https://example.com/image.png", media_type="image/png")
+        assert get_part_content_type(part) == "image/png"

lib/crewai/tests/a2a/utils/test_agent_card.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from a2a.types import AgentCard, AgentSkill
 
 from crewai import Agent
+from crewai.a2a._compat import agent_card_to_dict, agent_card_url, proto_to_json
 from crewai.a2a.config import A2AClientConfig, A2AServerConfig
 from crewai.a2a.utils.agent_card import inject_a2a_server_methods
 
@@ -154,7 +155,7 @@ class TestToAgentCard:
 
         card = agent.to_agent_card("http://my-server.com:9000")
 
-        assert card.url == "http://my-server.com:9000"
+        assert agent_card_url(card) == "http://my-server.com:9000"
 
     def test_uses_server_config_url(self) -> None:
         """AgentCard url should prefer A2AServerConfig.url over provided URL."""
@@ -167,7 +168,8 @@ class TestToAgentCard:
 
         card = agent.to_agent_card("http://fallback-url.com")
 
-        assert card.url == "http://configured-url.com/"
+        url = agent_card_url(card)
+        assert url.rstrip("/") == "http://configured-url.com"
 
     def test_generates_default_skill(self) -> None:
         """AgentCard should have at least one skill based on agent role."""
@@ -246,16 +248,16 @@ class TestAgentCardJsonStructure:
         )
 
         card = agent.to_agent_card("http://localhost:8000")
-        json_data = card.model_dump()
+        json_data = agent_card_to_dict(card)
 
         assert "name" in json_data
         assert "description" in json_data
-        assert "url" in json_data
+        assert "supported_interfaces" in json_data
         assert "version" in json_data
         assert "skills" in json_data
         assert "capabilities" in json_data
-        assert "defaultInputModes" in json_data
-        assert "defaultOutputModes" in json_data
+        assert "default_input_modes" in json_data
+        assert "default_output_modes" in json_data
 
     def test_json_skills_structure(self) -> None:
         """Each skill in JSON should have required fields."""
@@ -267,7 +269,7 @@ class TestAgentCardJsonStructure:
         )
 
         card = agent.to_agent_card("http://localhost:8000")
-        json_data = card.model_dump()
+        json_data = agent_card_to_dict(card)
 
         assert len(json_data["skills"]) >= 1
         skill = json_data["skills"][0]
@@ -286,11 +288,11 @@ class TestAgentCardJsonStructure:
         )
 
         card = agent.to_agent_card("http://localhost:8000")
-        json_data = card.model_dump()
+        json_data = agent_card_to_dict(card)
 
         capabilities = json_data["capabilities"]
         assert "streaming" in capabilities
-        assert "pushNotifications" in capabilities
+        assert "push_notifications" in capabilities
 
     def test_json_serializable(self) -> None:
         """AgentCard should be JSON serializable."""
@@ -302,14 +304,14 @@ class TestAgentCardJsonStructure:
         )
 
         card = agent.to_agent_card("http://localhost:8000")
-        json_str = card.model_dump_json()
+        json_str = proto_to_json(card)
 
         assert isinstance(json_str, str)
         assert "Test Agent" in json_str
         assert "http://localhost:8000" in json_str
 
-    def test_json_excludes_none_values(self) -> None:
-        """AgentCard JSON with exclude_none should omit None fields."""
+    def test_json_excludes_unset_fields(self) -> None:
+        """AgentCard JSON should omit fields that were not explicitly set."""
         agent = Agent(
             role="Test Agent",
             goal="Test goal",
@@ -318,8 +320,6 @@ class TestAgentCardJsonStructure:
         )
 
         card = agent.to_agent_card("http://localhost:8000")
-        json_data = card.model_dump(exclude_none=True)
+        json_data = agent_card_to_dict(card)
 
         assert "provider" not in json_data
-        assert "documentationUrl" not in json_data
-        assert "iconUrl" not in json_data

lib/crewai/tests/a2a/utils/test_task.py

@@ -12,6 +12,7 @@ from a2a.server.agent_execution import RequestContext
 from a2a.server.events import EventQueue
 from a2a.types import Message, Task as A2ATask, TaskState, TaskStatus
 
+from crewai.a2a._compat import TASK_STATE_CANCELED, TASK_STATE_WORKING
 from crewai.a2a.utils.task import cancel, cancellable, execute
 
 
@@ -38,12 +39,13 @@ def mock_task(mock_context: MagicMock) -> MagicMock:
 @pytest.fixture
 def mock_context() -> MagicMock:
     """Create a mock RequestContext."""
+    from crewai.a2a._compat import ROLE_USER, new_text_message
+
     context = MagicMock(spec=RequestContext)
     context.task_id = "test-task-123"
     context.context_id = "test-context-456"
     context.get_user_input.return_value = "Test user message"
-    context.message = MagicMock(spec=Message)
-    context.message.parts = []
+    context.message = new_text_message("Test user message", role=ROLE_USER)
     context.current_task = None
     return context
 
@@ -291,8 +293,7 @@ class TestCancel:
 
         assert event.task_id == mock_context.task_id
         assert event.context_id == mock_context.context_id
-        assert event.status.state == TaskState.canceled
-        assert event.final is True
+        assert event.status.state == TASK_STATE_CANCELED
 
     @pytest.mark.asyncio
     async def test_returns_none_when_no_current_task(
@@ -315,13 +316,13 @@ class TestCancel:
     ) -> None:
         """Cancel returns updated task when context has current_task."""
         current_task = MagicMock(spec=A2ATask)
-        current_task.status = TaskStatus(state=TaskState.working)
+        current_task.status = TaskStatus(state=TASK_STATE_WORKING)
         mock_context.current_task = current_task
 
         result = await cancel(mock_context, mock_event_queue)
 
         assert result is current_task
-        assert result.status.state == TaskState.canceled
+        assert result.status.state == TASK_STATE_CANCELED
 
     @pytest.mark.asyncio
     async def test_cleanup_after_cancel(

lib/crewai/tests/events/test_event_replay.py

@@ -0,0 +1,165 @@
+"""Tests for event bus replay dispatch and is_replaying flag."""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import patch
+
+from crewai.events.event_bus import _replaying, crewai_event_bus, is_replaying
+from crewai.events.types.flow_events import (
+    MethodExecutionFinishedEvent,
+    MethodExecutionStartedEvent,
+)
+
+
+def _make_started(method: str, event_id: str, sequence: int) -> MethodExecutionStartedEvent:
+    """Build a MethodExecutionStartedEvent with explicit ids/sequence."""
+    ev = MethodExecutionStartedEvent(
+        method_name=method,
+        flow_name="F",
+        params={},
+        state={},
+    )
+    ev.event_id = event_id
+    ev.emission_sequence = sequence
+    return ev
+
+
+class TestReplayPreservesFields:
+    """replay() must not overwrite event_id, parent_event_id, or emission_sequence."""
+
+    def test_preserves_ids_and_sequence(self) -> None:
+        captured: list[MethodExecutionStartedEvent] = []
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(MethodExecutionStartedEvent)
+            def _capture(_: Any, event: MethodExecutionStartedEvent) -> None:
+                captured.append(event)
+
+            ev = _make_started("outline", "orig-id-1", 42)
+            ev.parent_event_id = "parent-abc"
+
+            future = crewai_event_bus.replay(object(), ev)
+            if future is not None:
+                future.result(timeout=5.0)
+
+        assert len(captured) == 1
+        assert captured[0].event_id == "orig-id-1"
+        assert captured[0].parent_event_id == "parent-abc"
+        assert captured[0].emission_sequence == 42
+
+
+class TestIsReplayingFlag:
+    """is_replaying() must be True inside handlers dispatched via replay()."""
+
+    def test_flag_true_during_replay(self) -> None:
+        seen: list[bool] = []
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(MethodExecutionStartedEvent)
+            def _capture(_: Any, __: MethodExecutionStartedEvent) -> None:
+                seen.append(is_replaying())
+
+            ev = _make_started("m", "id-1", 1)
+            future = crewai_event_bus.replay(object(), ev)
+            if future is not None:
+                future.result(timeout=5.0)
+
+        assert seen == [True]
+        assert is_replaying() is False
+
+    def test_flag_false_during_emit(self) -> None:
+        seen: list[bool] = []
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(MethodExecutionStartedEvent)
+            def _capture(_: Any, __: MethodExecutionStartedEvent) -> None:
+                seen.append(is_replaying())
+
+            ev = _make_started("m", "id-1", 1)
+            future = crewai_event_bus.emit(object(), ev)
+            if future is not None:
+                future.result(timeout=5.0)
+
+        assert seen == [False]
+
+
+class TestCheckpointListenerOptsOut:
+    """CheckpointListener must early-return during replay."""
+
+    def test_checkpoint_not_written_on_replay(self) -> None:
+        from crewai.state.checkpoint_config import CheckpointConfig
+        from crewai.state.checkpoint_listener import _on_any_event
+
+        class FlowLike:
+            entity_type = "flow"
+            checkpoint = CheckpointConfig(trigger_all=True)
+
+        ev = _make_started("m", "id-1", 1)
+
+        with patch("crewai.state.checkpoint_listener._do_checkpoint") as do_cp:
+            token = _replaying.set(True)
+            try:
+                _on_any_event(FlowLike(), ev, state=None)
+            finally:
+                _replaying.reset(token)
+            assert do_cp.call_count == 0
+
+
+class TestFlowResumeReplaysEvents:
+    """End-to-end: a resumed flow emits MethodExecution* events for completed methods."""
+
+    def test_resume_dispatches_completed_method_events(self, tmp_path) -> None:
+        from crewai.flow.flow import Flow, listen, start
+        from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
+
+        db_path = tmp_path / "flows.db"
+        persistence = SQLiteFlowPersistence(str(db_path))
+
+        class ThreeStepFlow(Flow[dict]):
+            @start()
+            def step_a(self) -> str:
+                return "a"
+
+            @listen(step_a)
+            def step_b(self) -> str:
+                return "b"
+
+            @listen(step_b)
+            def step_c(self) -> str:
+                return "c"
+
+        if crewai_event_bus.runtime_state is not None:
+            crewai_event_bus.runtime_state.event_record.clear()
+
+        flow1 = ThreeStepFlow(persistence=persistence)
+        flow1.kickoff()
+        flow_id = flow1.state["id"]
+
+        captured_started: list[str] = []
+        captured_finished: list[str] = []
+
+        flow2 = ThreeStepFlow(persistence=persistence)
+        flow2._completed_methods = {"step_a", "step_b"}
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(MethodExecutionStartedEvent)
+            def _cs(_: Any, event: MethodExecutionStartedEvent) -> None:
+                captured_started.append(event.method_name)
+
+            @crewai_event_bus.on(MethodExecutionFinishedEvent)
+            def _cf(_: Any, event: MethodExecutionFinishedEvent) -> None:
+                captured_finished.append(event.method_name)
+
+            flow2.kickoff(inputs={"id": flow_id})
+
+        assert captured_started.count("step_a") == 1
+        assert captured_started.count("step_b") == 1
+        assert captured_started.count("step_c") == 1
+        assert captured_finished.count("step_a") == 1
+        assert captured_finished.count("step_b") == 1
+        assert captured_finished.count("step_c") == 1

lib/crewai/tests/llms/azure/test_azure.py

@@ -389,17 +389,41 @@ def test_azure_raises_error_when_endpoint_missing():
             llm._get_sync_client()
 
 
-def test_azure_raises_error_when_api_key_missing():
-    """Credentials are validated lazily: construction succeeds, first
+def test_azure_raises_error_when_api_key_missing_without_azure_identity():
+    """Without an API key AND without ``azure-identity`` installed,
     client build raises the descriptive error."""
     from crewai.llms.providers.azure.completion import AzureCompletion
 
     with patch.dict(os.environ, {}, clear=True):
-        llm = AzureCompletion(
-            model="gpt-4", endpoint="https://test.openai.azure.com"
-        )
-        with pytest.raises(ValueError, match="Azure API key is required"):
-            llm._get_sync_client()
+        with patch.dict("sys.modules", {"azure.identity": None}):
+            llm = AzureCompletion(
+                model="gpt-4", endpoint="https://test.openai.azure.com"
+            )
+            with pytest.raises(ValueError, match="Azure API key is required"):
+                llm._get_sync_client()
+
+
+def test_azure_uses_default_credential_when_api_key_missing():
+    """With ``azure-identity`` installed, a missing API key falls back to
+    ``DefaultAzureCredential`` instead of raising. This is the path that
+    enables keyless auth (OIDC WIF on EKS/AKS, Managed Identity, Azure
+    CLI) without any crewAI-specific config."""
+    from unittest.mock import MagicMock
+
+    from crewai.llms.providers.azure.completion import AzureCompletion
+
+    sentinel = MagicMock(name="DefaultAzureCredential()")
+    with patch.dict(os.environ, {}, clear=True):
+        with patch(
+            "azure.identity.DefaultAzureCredential", return_value=sentinel
+        ) as mock_cls:
+            llm = AzureCompletion(
+                model="gpt-4",
+                endpoint="https://test-ai.services.example.com",
+            )
+            kwargs = llm._make_client_kwargs()
+            assert kwargs["credential"] is sentinel
+            mock_cls.assert_called()
 
 
 @pytest.mark.asyncio

lib/crewai/tests/skills/test_integration.py

@@ -4,6 +4,8 @@ from pathlib import Path
 
 import pytest
 
+from crewai import Agent
+from crewai.agent.utils import append_skill_context
 from crewai.skills.loader import activate_skill, discover_skills, format_skill_context
 from crewai.skills.models import INSTRUCTIONS, METADATA
 
@@ -76,3 +78,23 @@ class TestSkillDiscoveryAndActivation:
             all_skills.extend(discover_skills(search_path))
         names = {s.name for s in all_skills}
         assert names == {"skill-a", "skill-b"}
+
+    def test_agent_preserves_metadata_for_discovered_skills(self, tmp_path: Path) -> None:
+        _create_skill_dir(tmp_path, "travel", body="Use this skill for travel planning.")
+        discovered = discover_skills(tmp_path)
+
+        agent = Agent(
+            role="Travel Advisor",
+            goal="Provide personalized travel suggestions.",
+            backstory="An experienced travel consultant.",
+            skills=discovered,
+        )
+
+        assert agent.skills is not None
+        assert agent.skills[0].disclosure_level == METADATA
+        assert agent.skills[0].instructions is None
+
+        prompt = append_skill_context(agent, "Plan a 10-day Japan itinerary.")
+        assert "## Skill: travel" in prompt
+        assert "Skill travel" in prompt
+        assert "Use this skill for travel planning." not in prompt

lib/crewai/tests/test_checkpoint.py

@@ -11,11 +11,12 @@ from typing import Any
 from unittest.mock import MagicMock, patch
 
 import pytest
+from pydantic import BaseModel
 
 from crewai.agent.core import Agent
 from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.crew import Crew
-from crewai.flow.flow import Flow, start
+from crewai.flow.flow import _INITIAL_STATE_CLASS_MARKER, Flow, start
 from crewai.state.checkpoint_config import CheckpointConfig
 from crewai.state.checkpoint_listener import (
     _find_checkpoint,
@@ -310,6 +311,65 @@ class TestRuntimeStateLineage:
         assert state._branch != first
 
 
+class TestFlowInitialStateSerialization:
+    """Regression tests for checkpoint serialization of ``Flow.initial_state``."""
+
+    def test_class_ref_serializes_as_schema(self) -> None:
+        class MyState(BaseModel):
+            id: str = "x"
+            foo: str = "bar"
+
+        flow = Flow(initial_state=MyState)
+        state = RuntimeState(root=[flow])
+        dumped = json.loads(state.model_dump_json())
+        entity = dumped["entities"][0]
+        wrapped = entity["initial_state"]
+        assert isinstance(wrapped, dict)
+        assert _INITIAL_STATE_CLASS_MARKER in wrapped
+        assert wrapped[_INITIAL_STATE_CLASS_MARKER].get("title") == "MyState"
+
+    def test_class_ref_round_trips_to_basemodel_subclass(self) -> None:
+        class MyState(BaseModel):
+            id: str = "x"
+            foo: str = "bar"
+
+        flow = Flow(initial_state=MyState)
+        raw = RuntimeState(root=[flow]).model_dump_json()
+        restored = RuntimeState.model_validate_json(
+            raw, context={"from_checkpoint": True}
+        )
+        rehydrated = restored.root[0].initial_state
+        assert isinstance(rehydrated, type)
+        assert issubclass(rehydrated, BaseModel)
+        assert set(rehydrated.model_fields.keys()) == {"id", "foo"}
+
+    def test_instance_serializes_as_values(self) -> None:
+        class MyState(BaseModel):
+            id: str = "x"
+            foo: str = "bar"
+
+        flow = Flow(initial_state=MyState(foo="baz"))
+        state = RuntimeState(root=[flow])
+        dumped = json.loads(state.model_dump_json())
+        entity = dumped["entities"][0]
+        assert entity["initial_state"] == {"id": "x", "foo": "baz"}
+
+    def test_dict_passthrough(self) -> None:
+        flow = Flow(initial_state={"id": "x", "foo": "bar"})
+        state = RuntimeState(root=[flow])
+        dumped = json.loads(state.model_dump_json())
+        entity = dumped["entities"][0]
+        assert entity["initial_state"] == {"id": "x", "foo": "bar"}
+
+    def test_dict_round_trips_as_dict(self) -> None:
+        flow = Flow(initial_state={"id": "x", "foo": "bar"})
+        raw = RuntimeState(root=[flow]).model_dump_json()
+        restored = RuntimeState.model_validate_json(
+            raw, context={"from_checkpoint": True}
+        )
+        assert restored.root[0].initial_state == {"id": "x", "foo": "bar"}
+
+
 # ---------- JsonProvider forking ----------
 
 

lib/crewai/tests/test_crew.py

@@ -4519,8 +4519,8 @@ def test_sets_flow_context_when_using_crewbase_pattern_inside_flow():
     flow.kickoff()
 
     assert captured_crew is not None
-    assert captured_crew._flow_id == flow.flow_id  # type: ignore[attr-defined]
-    assert captured_crew._request_id == flow.flow_id  # type: ignore[attr-defined]
+    assert captured_crew._flow_id == flow.execution_id  # type: ignore[attr-defined]
+    assert captured_crew._request_id == flow.execution_id  # type: ignore[attr-defined]
 
 
 def test_sets_flow_context_when_outside_flow(researcher, writer):
@@ -4554,8 +4554,8 @@ def test_sets_flow_context_when_inside_flow(researcher, writer):
 
     flow = MyFlow()
     result = flow.kickoff()
-    assert result._flow_id == flow.flow_id  # type: ignore[attr-defined]
-    assert result._request_id == flow.flow_id  # type: ignore[attr-defined]
+    assert result._flow_id == flow.execution_id  # type: ignore[attr-defined]
+    assert result._request_id == flow.execution_id  # type: ignore[attr-defined]
 
 
 def test_reset_knowledge_with_no_crew_knowledge(researcher, writer):

lib/crewai/tests/test_flow_execution_id.py

@@ -0,0 +1,127 @@
+"""Regression tests for ``Flow.execution_id``.
+
+``execution_id`` is the stable tracking identifier for a single flow run.
+It must stay independent of ``state.id`` so that consumers passing an
+``id`` in ``inputs`` (used for persistence restore) cannot destabilize
+the identity used by telemetry, tracing, and external correlation.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from crewai.flow.flow import Flow, FlowState, start
+from crewai.flow.flow_context import current_flow_id, current_flow_request_id
+
+
+class _CaptureState(FlowState):
+    captured_flow_id: str = ""
+    captured_state_id: str = ""
+    captured_current_flow_id: str = ""
+    captured_execution_id: str = ""
+
+
+class _IdentityCaptureFlow(Flow[_CaptureState]):
+    initial_state = _CaptureState
+
+    @start()
+    def capture(self) -> None:
+        self.state.captured_flow_id = self.flow_id
+        self.state.captured_state_id = self.state.id
+        self.state.captured_current_flow_id = current_flow_id.get() or ""
+        self.state.captured_execution_id = self.execution_id
+
+
+def test_execution_id_defaults_to_fresh_uuid_per_instance() -> None:
+    a = _IdentityCaptureFlow()
+    b = _IdentityCaptureFlow()
+
+    assert a.execution_id
+    assert b.execution_id
+    assert a.execution_id != b.execution_id
+
+
+def test_execution_id_survives_consumer_id_in_inputs() -> None:
+    flow = _IdentityCaptureFlow()
+    original_execution_id = flow.execution_id
+
+    flow.kickoff(inputs={"id": "consumer-supplied-id"})
+
+    assert flow.state.id == "consumer-supplied-id"
+    assert flow.flow_id == "consumer-supplied-id"
+    assert flow.execution_id == original_execution_id
+    assert flow.execution_id != "consumer-supplied-id"
+
+
+def test_two_runs_with_same_consumer_id_have_distinct_execution_ids() -> None:
+    flow_a = _IdentityCaptureFlow()
+    flow_b = _IdentityCaptureFlow()
+
+    colliding_id = "shared-consumer-id"
+    flow_a.kickoff(inputs={"id": colliding_id})
+    flow_b.kickoff(inputs={"id": colliding_id})
+
+    assert flow_a.state.id == colliding_id
+    assert flow_b.state.id == colliding_id
+    assert flow_a.execution_id != flow_b.execution_id
+
+
+def test_execution_id_is_writable() -> None:
+    flow = _IdentityCaptureFlow()
+    flow.execution_id = "external-task-id"
+
+    assert flow.execution_id == "external-task-id"
+
+    flow.kickoff(inputs={"id": "consumer-supplied-id"})
+    assert flow.execution_id == "external-task-id"
+    assert flow.state.id == "consumer-supplied-id"
+
+
+def test_current_flow_id_context_var_matches_execution_id() -> None:
+    flow = _IdentityCaptureFlow()
+    flow.execution_id = "external-task-id"
+
+    flow.kickoff(inputs={"id": "consumer-supplied-id"})
+
+    assert flow.state.captured_current_flow_id == "external-task-id"
+    assert flow.state.captured_flow_id == "consumer-supplied-id"
+    assert flow.state.captured_execution_id == "external-task-id"
+
+
+def test_execution_id_not_included_in_serialized_state() -> None:
+    flow = _IdentityCaptureFlow()
+    flow.execution_id = "external-task-id"
+    flow.kickoff()
+
+    dumped = flow.state.model_dump()
+    assert "execution_id" not in dumped
+    assert "_execution_id" not in dumped
+    assert dumped["id"] == flow.state.id
+
+
+def test_dict_state_flow_also_exposes_stable_execution_id() -> None:
+    class DictFlow(Flow[dict[str, Any]]):
+        initial_state = dict  # type: ignore[assignment]
+
+        @start()
+        def noop(self) -> None:
+            pass
+
+    flow = DictFlow()
+    original = flow.execution_id
+    flow.kickoff(inputs={"id": "consumer-supplied-id"})
+
+    assert flow.state["id"] == "consumer-supplied-id"
+    assert flow.execution_id == original
+
+
+@pytest.fixture(autouse=True)
+def _reset_flow_context_vars():
+    yield
+    for var in (current_flow_id, current_flow_request_id):
+        try:
+            var.set(None)
+        except LookupError:
+            # ContextVar was never set in this context; nothing to reset.
+            pass

lib/devtools/src/crewai_devtools/__init__.py

@@ -1,3 +1,3 @@
 """CrewAI development tools."""
 
-__version__ = "1.14.3a2"
+__version__ = "1.14.3a3"

pyproject.toml

@@ -164,7 +164,7 @@ info = "Commits must follow Conventional Commits 1.0.0."
 [tool.uv]
 # Pinned to include the security patch releases (authlib 1.6.11,
 # langchain-text-splitters 1.1.2) uploaded on 2026-04-16.
-exclude-newer = "2026-04-17"
+exclude-newer = "2026-04-22"
 
 # composio-core pins rich<14 but textual requires rich>=14.
 # onnxruntime 1.24+ dropped Python 3.10 wheels; cap it so qdrant[fastembed] resolves on 3.10.