andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-05 15:09:22 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Merge branch 'main' into lorenze/imp/memory-prompt-influence

Browse Source
This commit is contained in:
Lorenze Jay
2026-04-22 15:10:35 -07:00
committed by GitHub
parent 15bf60fa29 bc2fb71560
commit d1b35d8897
63 changed files with 8207 additions and 811 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
lib/crewai-files/src/crewai_files/__init__.py
View File

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

12
lib/crewai-tools/pyproject.toml
View File

@@ -10,7 +10,7 @@ requires-python = ">=3.10, <3.14"
dependencies = [
"pytube~=15.0.0",
"requests>=2.33.0,<3",
"crewai==1.14.2rc1",
"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"
@@ -139,6 +139,14 @@ contextual = [
"contextual-client>=0.1.0",
"nest-asyncio>=1.6.0",
]
daytona = [
"daytona~=0.140.0",
]
e2b = [
"e2b~=2.20.0",
"e2b-code-interpreter~=2.6.0",
]
[tool.uv]

18
lib/crewai-tools/src/crewai_tools/__init__.py
View File

@@ -59,6 +59,11 @@ from crewai_tools.tools.dalle_tool.dalle_tool import DallETool
from crewai_tools.tools.databricks_query_tool.databricks_query_tool import (
DatabricksQueryTool,
)
from crewai_tools.tools.daytona_sandbox_tool import (
DaytonaExecTool,
DaytonaFileTool,
DaytonaPythonTool,
)
from crewai_tools.tools.directory_read_tool.directory_read_tool import (
DirectoryReadTool,
)
@@ -66,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
@@ -232,8 +242,14 @@ __all__ = [
"DOCXSearchTool",
"DallETool",
"DatabricksQueryTool",
"DaytonaExecTool",
"DaytonaFileTool",
"DaytonaPythonTool",
"DirectoryReadTool",
"DirectorySearchTool",
"E2BExecTool",
"E2BFileTool",
"E2BPythonTool",
"EXASearchTool",
"EnterpriseActionTool",
"FileCompressorTool",
@@ -305,4 +321,4 @@ __all__ = [
"ZapierActionTools",
]
__version__ = "1.14.2rc1"
__version__ = "1.14.3a3"

16
lib/crewai-tools/src/crewai_tools/tools/__init__.py
View File

@@ -48,6 +48,11 @@ from crewai_tools.tools.dalle_tool.dalle_tool import DallETool
from crewai_tools.tools.databricks_query_tool.databricks_query_tool import (
DatabricksQueryTool,
)
from crewai_tools.tools.daytona_sandbox_tool import (
DaytonaExecTool,
DaytonaFileTool,
DaytonaPythonTool,
)
from crewai_tools.tools.directory_read_tool.directory_read_tool import (
DirectoryReadTool,
)
@@ -55,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
@@ -217,8 +227,14 @@ __all__ = [
"DOCXSearchTool",
"DallETool",
"DatabricksQueryTool",
"DaytonaExecTool",
"DaytonaFileTool",
"DaytonaPythonTool",
"DirectoryReadTool",
"DirectorySearchTool",
"E2BExecTool",
"E2BFileTool",
"E2BPythonTool",
"EXASearchTool",
"FileCompressorTool",
"FileReadTool",

107
lib/crewai-tools/src/crewai_tools/tools/daytona_sandbox_tool/README.md Normal file
View File

@@ -0,0 +1,107 @@
# Daytona Sandbox Tools
Run shell commands, execute Python, and manage files inside a [Daytona](https://www.daytona.io/) sandbox. Daytona provides isolated, ephemeral compute environments suitable for agent-driven code execution.
Three tools are provided so you can pick what the agent actually needs:
- **`DaytonaExecTool`** — run a shell command (`sandbox.process.exec`).
- **`DaytonaPythonTool`** — run a Python script (`sandbox.process.code_run`).
- **`DaytonaFileTool`** — read / write / list / delete files (`sandbox.fs.*`).
## Installation
```shell
uv add "crewai-tools[daytona]"
# or
pip install "crewai-tools[daytona]"
```
Set the API key:
```shell
export DAYTONA_API_KEY="..."
```
`DAYTONA_API_URL` and `DAYTONA_TARGET` are also respected if set.
## Sandbox lifecycle
All three tools share the same lifecycle controls from `DaytonaBaseTool`:
| Mode | When the sandbox is created | When it is deleted |
| --- | --- | --- |
| **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 delete 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 `DaytonaFileTool` with `DaytonaExecTool`.
## Examples
### One-shot Python execution (ephemeral)
```python
from crewai_tools import DaytonaPythonTool
tool = DaytonaPythonTool()
result = tool.run(code="print(sum(range(10)))")
```
### Multi-step shell session (persistent)
```python
from crewai_tools import DaytonaExecTool, DaytonaFileTool
exec_tool = DaytonaExecTool(persistent=True)
file_tool = DaytonaFileTool(persistent=True)
# Agent writes a script, then runs it — both share the same sandbox instance
# because they each keep their own persistent sandbox. If you need the *same*
# sandbox across two tools, create one tool, grab the sandbox id via
# `tool._persistent_sandbox.id`, and pass it to the other via `sandbox_id=...`.
```
### Attach to an existing sandbox
```python
from crewai_tools import DaytonaExecTool
tool = DaytonaExecTool(sandbox_id="my-long-lived-sandbox")
```
### Custom create params
Pass Daytona's `CreateSandboxFromSnapshotParams` kwargs via `create_params`:
```python
tool = DaytonaExecTool(
persistent=True,
create_params={
"language": "python",
"env_vars": {"MY_FLAG": "1"},
"labels": {"owner": "crewai-agent"},
},
)
```
## Tool arguments
### `DaytonaExecTool`
- `command: str` — shell command to run.
- `cwd: str | None` — working directory.
- `env: dict[str, str] | None` — extra env vars for this command.
- `timeout: int | None` — seconds.
### `DaytonaPythonTool`
- `code: str` — Python source to execute.
- `argv: list[str] | None` — argv forwarded via `CodeRunParams`.
- `env: dict[str, str] | None` — env vars forwarded via `CodeRunParams`.
- `timeout: int | None` — seconds.
### `DaytonaFileTool`
- `action: "read" | "write" | "list" | "delete" | "mkdir" | "info"`
- `path: str` — absolute path inside the sandbox.
- `content: str | None` — required for `write`.
- `binary: bool` — if `True`, `content` is base64 on write / returned as base64 on read.
- `recursive: bool` — for `delete`, removes directories recursively.
- `mode: str` — for `mkdir`, octal permission string (default `"0755"`).

13
lib/crewai-tools/src/crewai_tools/tools/daytona_sandbox_tool/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
from crewai_tools.tools.daytona_sandbox_tool.daytona_base_tool import DaytonaBaseTool
from crewai_tools.tools.daytona_sandbox_tool.daytona_exec_tool import DaytonaExecTool
from crewai_tools.tools.daytona_sandbox_tool.daytona_file_tool import DaytonaFileTool
from crewai_tools.tools.daytona_sandbox_tool.daytona_python_tool import (
DaytonaPythonTool,
)
__all__ = [
"DaytonaBaseTool",
"DaytonaExecTool",
"DaytonaFileTool",
"DaytonaPythonTool",
]

198
lib/crewai-tools/src/crewai_tools/tools/daytona_sandbox_tool/daytona_base_tool.py Normal file
View File

@@ -0,0 +1,198 @@
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
logger = logging.getLogger(__name__)
class DaytonaBaseTool(BaseTool):
"""Shared base for tools that act on a Daytona sandbox.
Lifecycle modes:
- persistent=False (default): create a fresh sandbox per `_run` call and
delete 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 delete 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 deleted by the tool.
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
package_dependencies: list[str] = Field(default_factory=lambda: ["daytona"])
api_key: str | None = Field(
default_factory=lambda: os.getenv("DAYTONA_API_KEY"),
description="Daytona API key. Falls back to DAYTONA_API_KEY env var.",
json_schema_extra={"required": False},
)
api_url: str | None = Field(
default_factory=lambda: os.getenv("DAYTONA_API_URL"),
description="Daytona API URL override. Falls back to DAYTONA_API_URL env var.",
json_schema_extra={"required": False},
)
target: str | None = Field(
default_factory=lambda: os.getenv("DAYTONA_TARGET"),
description="Daytona target region. Falls back to DAYTONA_TARGET env var.",
json_schema_extra={"required": False},
)
persistent: bool = Field(
default=False,
description=(
"If True, reuse one sandbox across all calls to this tool instance "
"and delete it at process exit. Default False creates and deletes a "
"fresh sandbox per call."
),
)
sandbox_id: str | None = Field(
default=None,
description=(
"Attach to an existing sandbox by id or name instead of creating a "
"new one. The tool will never delete a sandbox it did not create."
),
)
create_params: dict[str, Any] | None = Field(
default=None,
description=(
"Optional kwargs forwarded to CreateSandboxFromSnapshotParams when "
"creating a sandbox (e.g. language, snapshot, env_vars, labels)."
),
)
sandbox_timeout: float = Field(
default=60.0,
description="Timeout in seconds for sandbox create/delete operations.",
)
env_vars: list[EnvVar] = Field(
default_factory=lambda: [
EnvVar(
name="DAYTONA_API_KEY",
description="API key for Daytona sandbox service",
required=False,
),
EnvVar(
name="DAYTONA_API_URL",
description="Daytona API base URL (optional)",
required=False,
),
EnvVar(
name="DAYTONA_TARGET",
description="Daytona target region (optional)",
required=False,
),
]
)
_client: Any | None = PrivateAttr(default=None)
_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_sdk(cls) -> dict[str, Any]:
if cls._sdk_cache:
return cls._sdk_cache
try:
from daytona import (
CreateSandboxFromSnapshotParams,
Daytona,
DaytonaConfig,
)
except ImportError as exc:
raise ImportError(
"The 'daytona' package is required for Daytona sandbox tools. "
"Install it with: uv add daytona (or) pip install daytona"
) from exc
cls._sdk_cache = {
"Daytona": Daytona,
"DaytonaConfig": DaytonaConfig,
"CreateSandboxFromSnapshotParams": CreateSandboxFromSnapshotParams,
}
return cls._sdk_cache
def _get_client(self) -> Any:
if self._client is not None:
return self._client
sdk = self._import_sdk()
config_kwargs: dict[str, Any] = {}
if self.api_key:
config_kwargs["api_key"] = self.api_key
if self.api_url:
config_kwargs["api_url"] = self.api_url
if self.target:
config_kwargs["target"] = self.target
config = sdk["DaytonaConfig"](**config_kwargs) if config_kwargs else None
self._client = sdk["Daytona"](config) if config else sdk["Daytona"]()
return self._client
def _build_create_params(self) -> Any | None:
if not self.create_params:
return None
sdk = self._import_sdk()
return sdk["CreateSandboxFromSnapshotParams"](**self.create_params)
def _acquire_sandbox(self) -> tuple[Any, bool]:
"""Return (sandbox, should_delete_after_use)."""
client = self._get_client()
if self.sandbox_id:
return client.get(self.sandbox_id), False
if self.persistent:
with self._lock:
if self._persistent_sandbox is None:
self._persistent_sandbox = client.create(
self._build_create_params(),
timeout=self.sandbox_timeout,
)
if not self._cleanup_registered:
atexit.register(self.close)
self._cleanup_registered = True
return self._persistent_sandbox, False
sandbox = client.create(
self._build_create_params(),
timeout=self.sandbox_timeout,
)
return sandbox, True
def _release_sandbox(self, sandbox: Any, should_delete: bool) -> None:
if not should_delete:
return
try:
sandbox.delete(timeout=self.sandbox_timeout)
except Exception:
logger.debug(
"Best-effort sandbox cleanup failed after ephemeral use; "
"the sandbox may need manual deletion.",
exc_info=True,
)
def close(self) -> None:
"""Delete 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.delete(timeout=self.sandbox_timeout)
except Exception:
logger.debug(
"Best-effort persistent sandbox cleanup failed at close(); "
"the sandbox may need manual deletion.",
exc_info=True,
)

59
lib/crewai-tools/src/crewai_tools/tools/daytona_sandbox_tool/daytona_exec_tool.py Normal file
View File

@@ -0,0 +1,59 @@
from __future__ import annotations
from builtins import type as type_
from typing import Any
from pydantic import BaseModel, Field
from crewai_tools.tools.daytona_sandbox_tool.daytona_base_tool import DaytonaBaseTool
class DaytonaExecToolSchema(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 work dir.",
)
env: dict[str, str] | None = Field(
default=None,
description="Optional environment variables to set for this command.",
)
timeout: int | None = Field(
default=None,
description="Maximum seconds to wait for the command to finish.",
)
class DaytonaExecTool(DaytonaBaseTool):
"""Run a shell command inside a Daytona sandbox."""
name: str = "Daytona Sandbox Exec"
description: str = (
"Execute a shell command inside a Daytona sandbox and return the exit "
"code and combined output. Use this to run builds, package installs, "
"git operations, or any one-off shell command."
)
args_schema: type_[BaseModel] = DaytonaExecToolSchema
def _run(
self,
command: str,
cwd: str | None = None,
env: dict[str, str] | None = None,
timeout: int | None = None,
) -> Any:
sandbox, should_delete = self._acquire_sandbox()
try:
response = sandbox.process.exec(
command,
cwd=cwd,
env=env,
timeout=timeout,
)
return {
"exit_code": getattr(response, "exit_code", None),
"result": getattr(response, "result", None),
"artifacts": getattr(response, "artifacts", None),
}
finally:
self._release_sandbox(sandbox, should_delete)

205
lib/crewai-tools/src/crewai_tools/tools/daytona_sandbox_tool/daytona_file_tool.py Normal file
View File

@@ -0,0 +1,205 @@
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.daytona_sandbox_tool.daytona_base_tool import DaytonaBaseTool
logger = logging.getLogger(__name__)
FileAction = Literal["read", "write", "append", "list", "delete", "mkdir", "info"]
class DaytonaFileToolSchema(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)."
),
)
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': treat content as base64 and upload raw bytes. "
"For 'read': return contents as base64 instead of decoded utf-8."
),
)
recursive: bool = Field(
default=False,
description="For action='delete': remove directories recursively.",
)
mode: str = Field(
default="0755",
description="For action='mkdir': octal permission string (default 0755).",
)
@model_validator(mode="after")
def _validate_action_args(self) -> DaytonaFileToolSchema:
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 DaytonaFileTool(DaytonaBaseTool):
"""Read, write, and manage files inside a Daytona 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 = "Daytona Sandbox Files"
description: str = (
"Perform filesystem operations inside a Daytona sandbox: read a file, "
"write content to a path, append content to an existing file, list a "
"directory, delete a path, make a directory, or fetch file metadata. "
"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] = DaytonaFileToolSchema
def _run(
self,
action: FileAction,
path: str,
content: str | None = None,
binary: bool = False,
recursive: bool = False,
mode: str = "0755",
) -> Any:
sandbox, should_delete = 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)
if action == "delete":
sandbox.fs.delete_file(path, recursive=recursive)
return {"status": "deleted", "path": path}
if action == "mkdir":
sandbox.fs.create_folder(path, mode)
return {"status": "created", "path": path, "mode": mode}
if action == "info":
return self._info(sandbox, path)
raise ValueError(f"Unknown action: {action}")
finally:
self._release_sandbox(sandbox, should_delete)
def _read(self, sandbox: Any, path: str, *, binary: bool) -> dict[str, Any]:
data: bytes = sandbox.fs.download_file(path)
if binary:
return {
"path": path,
"encoding": "base64",
"content": base64.b64encode(data).decode("ascii"),
}
try:
return {"path": path, "encoding": "utf-8", "content": data.decode("utf-8")}
except UnicodeDecodeError:
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 = base64.b64decode(content) if binary else content.encode("utf-8")
self._ensure_parent_dir(sandbox, path)
sandbox.fs.upload_file(payload, path)
return {"status": "written", "path": path, "bytes": len(payload)}
def _append(
self, sandbox: Any, path: str, content: str, *, binary: bool
) -> dict[str, Any]:
chunk = base64.b64decode(content) if binary else content.encode("utf-8")
self._ensure_parent_dir(sandbox, path)
try:
existing: bytes = sandbox.fs.download_file(path)
except Exception:
existing = b""
payload = existing + chunk
sandbox.fs.upload_file(payload, path)
return {
"status": "appended",
"path": path,
"appended_bytes": len(chunk),
"total_bytes": len(payload),
}
@staticmethod
def _ensure_parent_dir(sandbox: Any, path: str) -> None:
"""Make sure the parent directory of `path` exists.
Daytona's upload returns 400 if the parent directory is missing. We
best-effort mkdir the parent; any error (e.g. already exists) is
swallowed because `create_folder` is not idempotent on the server.
"""
parent = posixpath.dirname(path)
if not parent or parent in ("/", "."):
return
try:
sandbox.fs.create_folder(parent, "0755")
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) -> dict[str, Any]:
entries = sandbox.fs.list_files(path)
return {
"path": path,
"entries": [self._file_info_to_dict(entry) for entry in entries],
}
def _info(self, sandbox: Any, path: str) -> dict[str, Any]:
return self._file_info_to_dict(sandbox.fs.get_file_info(path))
@staticmethod
def _file_info_to_dict(info: Any) -> dict[str, Any]:
fields = (
"name",
"size",
"mode",
"permissions",
"is_dir",
"mod_time",
"owner",
"group",
)
return {field: getattr(info, field, None) for field in fields}

82
lib/crewai-tools/src/crewai_tools/tools/daytona_sandbox_tool/daytona_python_tool.py Normal file
View File

@@ -0,0 +1,82 @@
from __future__ import annotations
from builtins import type as type_
from typing import Any
from pydantic import BaseModel, Field
from crewai_tools.tools.daytona_sandbox_tool.daytona_base_tool import DaytonaBaseTool
class DaytonaPythonToolSchema(BaseModel):
code: str = Field(
...,
description="Python source to execute inside the sandbox.",
)
argv: list[str] | None = Field(
default=None,
description="Optional argv passed to the script (forwarded as params.argv).",
)
env: dict[str, str] | None = Field(
default=None,
description="Optional environment variables for the run (forwarded as params.env).",
)
timeout: int | None = Field(
default=None,
description="Maximum seconds to wait for the code to finish.",
)
class DaytonaPythonTool(DaytonaBaseTool):
"""Run Python source inside a Daytona sandbox."""
name: str = "Daytona Sandbox Python"
description: str = (
"Execute a block of Python code inside a Daytona sandbox and return the "
"exit code, captured stdout, and any produced artifacts. Use this for "
"data processing, quick scripts, or analysis that should run in an "
"isolated environment."
)
args_schema: type_[BaseModel] = DaytonaPythonToolSchema
def _run(
self,
code: str,
argv: list[str] | None = None,
env: dict[str, str] | None = None,
timeout: int | None = None,
) -> Any:
sandbox, should_delete = self._acquire_sandbox()
try:
params = self._build_code_run_params(argv=argv, env=env)
response = sandbox.process.code_run(code, params=params, timeout=timeout)
return {
"exit_code": getattr(response, "exit_code", None),
"result": getattr(response, "result", None),
"artifacts": getattr(response, "artifacts", None),
}
finally:
self._release_sandbox(sandbox, should_delete)
def _build_code_run_params(
self,
argv: list[str] | None,
env: dict[str, str] | None,
) -> Any | None:
if argv is None and env is None:
return None
try:
from daytona import CodeRunParams
except ImportError as exc:
raise ImportError(
"Could not import daytona.CodeRunParams while building "
"argv/env for sandbox.process.code_run. This usually means the "
"installed 'daytona' SDK is too old or incompatible. Upgrade "
"with: pip install -U 'crewai-tools[daytona]'"
) from exc
kwargs: dict[str, Any] = {}
if argv is not None:
kwargs["argv"] = argv
if env is not None:
kwargs["env"] = env
return CodeRunParams(**kwargs)

120
lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/README.md Normal file
View File

@@ -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.

12
lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/__init__.py Normal file
View File

@@ -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",
]

197
lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/e2b_base_tool.py Normal file
View File

@@ -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,
)

62
lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/e2b_exec_tool.py Normal file
View File

@@ -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)

220
lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/e2b_file_tool.py Normal file
View File

@@ -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

133
lib/crewai-tools/src/crewai_tools/tools/e2b_sandbox_tool/e2b_python_tool.py Normal file
View File

@@ -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}

1290
lib/crewai-tools/tool.specs.json
View File

File diff suppressed because it is too large Load Diff

5
lib/crewai/pyproject.toml
View File

@@ -24,7 +24,7 @@ dependencies = [
"tokenizers>=0.21,<1",
"openpyxl~=3.1.5",
# Authentication and Security
"python-dotenv~=1.1.1",
"python-dotenv>=1.2.2,<2",
"pyjwt>=2.9.0,<3",
# TUI
"textual>=7.5.0",
@@ -55,7 +55,7 @@ Repository = "https://github.com/crewAIInc/crewAI"
[project.optional-dependencies]
tools = [
"crewai-tools==1.14.2rc1",
"crewai-tools==1.14.3a3",
]
embeddings = [
"tiktoken~=0.8.0"
@@ -94,6 +94,7 @@ google-genai = [
]
azure-ai-inference = [
"azure-ai-inference~=1.0.0b9",
"azure-identity>=1.17.0,<2",
]
anthropic = [
"anthropic~=0.73.0",

55
lib/crewai/src/crewai/__init__.py
View File

@@ -1,10 +1,9 @@
import contextvars
import threading
from typing import Any
import urllib.request
import importlib
import sys
from typing import TYPE_CHECKING, Annotated, Any
import warnings
from pydantic import PydanticUserError
from pydantic import Field, PydanticUserError
from crewai.agent.core import Agent
from crewai.agent.planning_config import PlanningConfig
@@ -20,7 +19,10 @@ from crewai.state.checkpoint_config import CheckpointConfig # noqa: F401
from crewai.task import Task
from crewai.tasks.llm_guardrail import LLMGuardrail
from crewai.tasks.task_output import TaskOutput
from crewai.telemetry.telemetry import Telemetry
if TYPE_CHECKING:
from crewai.memory.unified_memory import Memory
def _suppress_pydantic_deprecation_warnings() -> None:
@@ -46,38 +48,7 @@ def _suppress_pydantic_deprecation_warnings() -> None:
_suppress_pydantic_deprecation_warnings()
__version__ = "1.14.2rc1"
_telemetry_submitted = False
def _track_install() -> None:
"""Track package installation/first-use via Scarf analytics."""
global _telemetry_submitted
if _telemetry_submitted or Telemetry._is_telemetry_disabled():
return
try:
pixel_url = "https://api.scarf.sh/v2/packages/CrewAI/crewai/docs/00f2dad1-8334-4a39-934e-003b2e1146db"
req = urllib.request.Request(pixel_url) # noqa: S310
req.add_header("User-Agent", f"CrewAI-Python/{__version__}")
with urllib.request.urlopen(req, timeout=2): # noqa: S310
_telemetry_submitted = True
except Exception: # noqa: S110
pass
def _track_install_async() -> None:
"""Track installation in background thread to avoid blocking imports."""
if not Telemetry._is_telemetry_disabled():
ctx = contextvars.copy_context()
thread = threading.Thread(target=ctx.run, args=(_track_install,), daemon=True)
thread.start()
_track_install_async()
__version__ = "1.14.3a3"
_LAZY_IMPORTS: dict[str, tuple[str, str]] = {
"Memory": ("crewai.memory.unified_memory", "Memory"),
@@ -89,8 +60,6 @@ def __getattr__(name: str) -> Any:
"""Lazily import heavy modules (e.g. Memory → lancedb) on first access."""
if name in _LAZY_IMPORTS:
module_path, attr = _LAZY_IMPORTS[name]
import importlib
mod = importlib.import_module(module_path)
val = getattr(mod, attr)
globals()[name] = val
@@ -148,8 +117,6 @@ try:
except ImportError:
pass
import sys
_full_namespace = {
**_base_namespace,
"ToolsHandler": _ToolsHandler,
@@ -192,10 +159,6 @@ try:
Flow.model_rebuild(force=True, _types_namespace=_full_namespace)
_AgentExecutor.model_rebuild(force=True, _types_namespace=_full_namespace)
from typing import Annotated
from pydantic import Field
from crewai.state.runtime import RuntimeState
Entity = Annotated[

95
lib/crewai/src/crewai/agent/core.py
View File

@@ -29,7 +29,7 @@ from pydantic import (
model_validator,
)
from pydantic.functional_serializers import PlainSerializer
from typing_extensions import Self
from typing_extensions import Self, TypeIs
from crewai.agent.planning_config import PlanningConfig
from crewai.agent.utils import (
@@ -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
@@ -133,6 +133,13 @@ _EXECUTOR_CLASS_MAP: dict[str, type] = {
}
def _is_resuming_agent_executor(
executor: CrewAgentExecutor | AgentExecutor | None,
) -> TypeIs[AgentExecutor]:
"""Type guard: True when the executor is resuming from a checkpoint."""
return isinstance(executor, AgentExecutor) and executor._resuming
def _validate_executor_class(value: Any) -> Any:
if isinstance(value, str):
cls = _EXECUTOR_CLASS_MAP.get(value)
@@ -1113,6 +1120,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)
@@ -1366,24 +1375,42 @@ class Agent(BaseAgent):
prompt, stop_words, rpm_limit_fn = self._build_execution_prompt(raw_tools)
executor = AgentExecutor(
llm=cast(BaseLLM, self.llm),
agent=self,
prompt=prompt,
max_iter=self.max_iter,
tools=parsed_tools,
tools_names=get_tool_names(parsed_tools),
stop_words=stop_words,
tools_description=render_text_description_and_args(parsed_tools),
tools_handler=self.tools_handler,
original_tools=raw_tools,
step_callback=self.step_callback,
function_calling_llm=self.function_calling_llm,
respect_context_window=self.respect_context_window,
request_within_rpm_limit=rpm_limit_fn,
callbacks=[TokenCalcHandler(self._token_process)],
response_model=response_format,
)
if _is_resuming_agent_executor(self.agent_executor):
executor = self.agent_executor
executor.tools = parsed_tools
executor.tools_names = get_tool_names(parsed_tools)
executor.tools_description = render_text_description_and_args(parsed_tools)
executor.original_tools = raw_tools
executor.prompt = prompt
executor.response_model = response_format
executor.stop_words = stop_words
executor.tools_handler = self.tools_handler
executor.step_callback = self.step_callback
executor.function_calling_llm = cast(
BaseLLM | None, self.function_calling_llm
)
executor.respect_context_window = self.respect_context_window
executor.request_within_rpm_limit = rpm_limit_fn
executor.callbacks = [TokenCalcHandler(self._token_process)]
else:
executor = AgentExecutor(
llm=cast(BaseLLM, self.llm),
agent=self,
prompt=prompt,
max_iter=self.max_iter,
tools=parsed_tools,
tools_names=get_tool_names(parsed_tools),
stop_words=stop_words,
tools_description=render_text_description_and_args(parsed_tools),
tools_handler=self.tools_handler,
original_tools=raw_tools,
step_callback=self.step_callback,
function_calling_llm=self.function_calling_llm,
respect_context_window=self.respect_context_window,
request_within_rpm_limit=rpm_limit_fn,
callbacks=[TokenCalcHandler(self._token_process)],
response_model=response_format,
)
all_files: dict[str, Any] = {}
if isinstance(messages, str):
@@ -1504,14 +1531,17 @@ class Agent(BaseAgent):
)
try:
crewai_event_bus.emit(
self,
event=LiteAgentExecutionStartedEvent(
if self.checkpoint_kickoff_event_id is not None:
self._kickoff_event_id = self.checkpoint_kickoff_event_id
self.checkpoint_kickoff_event_id = None
else:
started_event = LiteAgentExecutionStartedEvent(
agent_info=agent_info,
tools=parsed_tools,
messages=messages,
),
)
)
crewai_event_bus.emit(self, event=started_event)
self._kickoff_event_id = started_event.event_id
output = self._execute_and_build_output(executor, inputs, response_format)
return self._finalize_kickoff(
@@ -1808,14 +1838,17 @@ class Agent(BaseAgent):
)
try:
crewai_event_bus.emit(
self,
event=LiteAgentExecutionStartedEvent(
if self.checkpoint_kickoff_event_id is not None:
self._kickoff_event_id = self.checkpoint_kickoff_event_id
self.checkpoint_kickoff_event_id = None
else:
started_event = LiteAgentExecutionStartedEvent(
agent_info=agent_info,
tools=parsed_tools,
messages=messages,
),
)
)
crewai_event_bus.emit(self, event=started_event)
self._kickoff_event_id = started_event.event_id
output = await self._execute_and_build_output_async(
executor, inputs, response_format

75
lib/crewai/src/crewai/agents/agent_builder/base_agent.py
View File

@@ -28,6 +28,9 @@ from crewai.agents.agent_builder.base_agent_executor import BaseAgentExecutor
from crewai.agents.agent_builder.utilities.base_token_process import TokenProcess
from crewai.agents.cache.cache_handler import CacheHandler
from crewai.agents.tools_handler import ToolsHandler
from crewai.events.base_events import set_emission_counter
from crewai.events.event_bus import crewai_event_bus
from crewai.events.event_context import restore_event_scope, set_last_event_id
from crewai.knowledge.knowledge import Knowledge
from crewai.knowledge.knowledge_config import KnowledgeConfig
from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
@@ -51,6 +54,7 @@ from crewai.utilities.string_utils import interpolate_only
if TYPE_CHECKING:
from crewai.context import ExecutionContext
from crewai.crew import Crew
from crewai.state.runtime import RuntimeState
def _validate_crew_ref(value: Any) -> Any:
@@ -219,6 +223,7 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
_original_goal: str | None = PrivateAttr(default=None)
_original_backstory: str | None = PrivateAttr(default=None)
_token_process: TokenProcess = PrivateAttr(default_factory=TokenProcess)
_kickoff_event_id: str | None = PrivateAttr(default=None)
id: UUID4 = Field(default_factory=uuid.uuid4, frozen=True)
role: str = Field(description="Role of the agent")
goal: str = Field(description="Objective of the agent")
@@ -335,30 +340,90 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
min_length=1,
)
execution_context: ExecutionContext | None = Field(default=None)
checkpoint_kickoff_event_id: str | None = Field(default=None)
@classmethod
def from_checkpoint(cls, config: CheckpointConfig) -> Self:
"""Restore an Agent from a checkpoint.
"""Restore an Agent from a checkpoint, ready to resume via kickoff().
Args:
config: Checkpoint configuration with ``restore_from`` set.
config: Checkpoint configuration with ``restore_from`` set to
the path of the checkpoint to load.
Returns:
An Agent instance. Call kickoff() to resume execution.
"""
from crewai.context import apply_execution_context
from crewai.state.runtime import RuntimeState
state = RuntimeState.from_checkpoint(config, context={"from_checkpoint": True})
crewai_event_bus.set_runtime_state(state)
for entity in state.root:
if isinstance(entity, cls):
if entity.execution_context is not None:
apply_execution_context(entity.execution_context)
if entity.agent_executor is not None:
entity.agent_executor.agent = entity
entity.agent_executor._resuming = True
entity._restore_runtime(state)
return entity
raise ValueError(
f"No {cls.__name__} found in checkpoint: {config.restore_from}"
)
@classmethod
def fork(cls, config: CheckpointConfig, branch: str | None = None) -> Self:
"""Fork an Agent from a checkpoint, creating a new execution branch.
Args:
config: Checkpoint configuration with ``restore_from`` set.
branch: Branch label for the fork. Auto-generated if not provided.
Returns:
An Agent instance on the new branch. Call kickoff() to run.
"""
agent = cls.from_checkpoint(config)
state = crewai_event_bus._runtime_state
if state is None:
raise RuntimeError("Cannot fork: no runtime state on the event bus.")
state.fork(branch)
return agent
def _restore_runtime(self, state: RuntimeState) -> None:
"""Re-create runtime objects after restoring from a checkpoint.
Args:
state: The RuntimeState containing the event record.
"""
if self.agent_executor is not None:
self.agent_executor.agent = self
self.agent_executor._resuming = True
if self.checkpoint_kickoff_event_id is not None:
self._kickoff_event_id = self.checkpoint_kickoff_event_id
self._restore_event_scope(state)
def _restore_event_scope(self, state: RuntimeState) -> None:
"""Rebuild the event scope stack from the checkpoint's event record.
Args:
state: The RuntimeState containing the event record.
"""
stack: list[tuple[str, str]] = []
kickoff_id = self._kickoff_event_id
if kickoff_id:
stack.append((kickoff_id, "lite_agent_execution_started"))
restore_event_scope(tuple(stack))
last_event_id: str | None = None
max_seq = 0
for node in state.event_record.nodes.values():
seq = node.event.emission_sequence or 0
if seq > max_seq:
max_seq = seq
last_event_id = node.event.event_id
if last_event_id is not None:
set_last_event_id(last_event_id)
if max_seq > 0:
set_emission_counter(max_seq)
@model_validator(mode="before")
@classmethod
def process_model_config(cls, values: Any) -> dict[str, Any]:

48
lib/crewai/src/crewai/cli/checkpoint_cli.py
View File

@@ -106,17 +106,50 @@ def _parse_checkpoint_json(raw: str, source: str) -> dict[str, Any]:
"name": entity.get("name"),
"id": entity.get("id"),
}
raw_agents = entity.get("agents", [])
agents_by_id: dict[str, dict[str, Any]] = {}
parsed_agents: list[dict[str, Any]] = []
for ag in raw_agents:
agent_info: dict[str, Any] = {
"id": ag.get("id", ""),
"role": ag.get("role", ""),
"goal": ag.get("goal", ""),
}
parsed_agents.append(agent_info)
if ag.get("id"):
agents_by_id[str(ag["id"])] = agent_info
if parsed_agents:
info["agents"] = parsed_agents
if tasks:
info["tasks_completed"] = completed
info["tasks_total"] = len(tasks)
info["tasks"] = [
{
parsed_tasks: list[dict[str, Any]] = []
for t in tasks:
task_info: dict[str, Any] = {
"description": t.get("description", ""),
"completed": t.get("output") is not None,
"output": (t.get("output") or {}).get("raw", ""),
}
for t in tasks
]
task_agent = t.get("agent")
if isinstance(task_agent, dict):
task_info["agent_role"] = task_agent.get("role", "")
task_info["agent_id"] = task_agent.get("id", "")
elif isinstance(task_agent, str) and task_agent in agents_by_id:
task_info["agent_role"] = agents_by_id[task_agent].get("role", "")
task_info["agent_id"] = task_agent
parsed_tasks.append(task_info)
info["tasks"] = parsed_tasks
if entity.get("entity_type") == "flow":
completed_methods = entity.get("checkpoint_completed_methods")
if completed_methods:
info["completed_methods"] = sorted(completed_methods)
state = entity.get("checkpoint_state")
if isinstance(state, dict):
info["flow_state"] = state
parsed_entities.append(info)
inputs: dict[str, Any] = {}
@@ -439,6 +472,8 @@ def _entity_type_from_meta(meta: dict[str, Any]) -> str:
for ent in meta.get("entities", []):
if ent.get("type") == "flow":
return "flow"
if ent.get("type") == "agent":
return "agent"
return "crew"
@@ -472,6 +507,11 @@ def resume_checkpoint(location: str, checkpoint_id: str | None) -> None:
flow = Flow.from_checkpoint(config)
result = asyncio.run(flow.kickoff_async(inputs=inputs))
elif entity_type == "agent":
from crewai.agent import Agent
agent = Agent.from_checkpoint(config)
result = asyncio.run(agent.akickoff(messages="Resume execution."))
else:
from crewai.crew import Crew

627
lib/crewai/src/crewai/cli/checkpoint_tui.py
View File

@@ -3,17 +3,20 @@
from __future__ import annotations
from collections import defaultdict
from datetime import datetime
from typing import Any, ClassVar, Literal
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.widgets import (
Button,
Collapsible,
Footer,
Header,
Input,
Static,
TabPane,
TabbedContent,
TextArea,
Tree,
)
@@ -32,6 +35,22 @@ _TERTIARY = "#ffffff"
_DIM = "#888888"
_BG_DARK = "#0d1117"
_BG_PANEL = "#161b22"
_ACCENT = "#c9a227"
_SUCCESS = "#3fb950"
_PENDING = "#e3b341"
_ENTITY_ICONS: dict[str, str] = {
"flow": "◆",
"crew": "●",
"agent": "◈",
"unknown": "○",
}
_ENTITY_COLORS: dict[str, str] = {
"flow": _ACCENT,
"crew": _SECONDARY,
"agent": _PRIMARY,
"unknown": _DIM,
}
def _load_entries(location: str) -> list[dict[str, Any]]:
@@ -40,8 +59,27 @@ def _load_entries(location: str) -> list[dict[str, Any]]:
return _list_json(location)
def _human_ts(ts: str) -> str:
"""Turn '2026-04-17 17:05:00' into a short relative label."""
try:
dt = datetime.strptime(ts, "%Y-%m-%d %H:%M:%S")
except ValueError:
return ts
now = datetime.now()
delta = now.date() - dt.date()
hour = dt.hour % 12 or 12
ampm = "am" if dt.hour < 12 else "pm"
time_str = f"{hour}:{dt.minute:02d}{ampm}"
if delta.days == 0:
return time_str
if delta.days == 1:
return f"yest {time_str}"
if delta.days < 7:
return f"{dt.strftime('%a').lower()} {time_str}"
return f"{dt.strftime('%b')} {dt.day}"
def _short_id(name: str) -> str:
"""Shorten a checkpoint name for tree display."""
if len(name) > 30:
return name[:27] + "..."
return name
@@ -63,29 +101,29 @@ def _entry_id(entry: dict[str, Any]) -> str:
return name
def _build_entity_header(ent: dict[str, Any]) -> str:
"""Build rich text header for an entity (progress bar only)."""
lines: list[str] = []
tasks = ent.get("tasks")
if isinstance(tasks, list):
completed = ent.get("tasks_completed", 0)
total = ent.get("tasks_total", 0)
pct = int(completed / total * 100) if total else 0
bar_len = 20
filled = int(bar_len * completed / total) if total else 0
bar = f"[{_PRIMARY}]{'█' * filled}[/][{_DIM}]{'░' * (bar_len - filled)}[/]"
lines.append(f"{bar} {completed}/{total} tasks ({pct}%)")
return "\n".join(lines)
def _build_progress_bar(completed: int, total: int, width: int = 20) -> str:
if total == 0:
return f"[{_DIM}]{'░' * width}[/] 0/0"
pct = int(completed / total * 100)
filled = int(width * completed / total)
color = _SUCCESS if completed == total else _PRIMARY
bar = f"[{color}]{'█' * filled}[/][{_DIM}]{'░' * (width - filled)}[/]"
return f"{bar} {completed}/{total} ({pct}%)"
def _entity_icon(etype: str) -> str:
icon = _ENTITY_ICONS.get(etype, _ENTITY_ICONS["unknown"])
color = _ENTITY_COLORS.get(etype, _DIM)
return f"[{color}]{icon}[/]"
# Return type: (location, action, inputs, task_output_overrides, entity_type)
_TuiResult = (
tuple[
str,
str,
dict[str, Any] | None,
dict[int, str] | None,
Literal["crew", "flow"],
Literal["crew", "flow", "agent"],
]
| None
)
@@ -122,7 +160,7 @@ class CheckpointTUI(App[_TuiResult]):
height: 1fr;
}}
#tree-panel {{
width: 45%;
width: 40%;
background: {_BG_PANEL};
border: round {_SECONDARY};
padding: 0 1;
@@ -132,41 +170,81 @@ class CheckpointTUI(App[_TuiResult]):
border: round {_PRIMARY};
}}
#detail-container {{
width: 55%;
width: 60%;
height: 1fr;
}}
#detail-scroll {{
height: 1fr;
background: {_BG_PANEL};
border: round {_SECONDARY};
padding: 1 2;
scrollbar-color: {_PRIMARY};
}}
#detail-scroll:focus-within {{
border: round {_PRIMARY};
}}
#detail-header {{
margin-bottom: 1;
}}
#status {{
height: 1;
padding: 0 2;
color: {_DIM};
}}
#inputs-section {{
display: none;
height: auto;
max-height: 8;
padding: 0 1;
#detail-tabs {{
height: 1fr;
}}
#inputs-section.visible {{
display: block;
TabbedContent > ContentSwitcher {{
background: {_BG_PANEL};
height: 1fr;
}}
#inputs-label {{
height: 1;
TabPane {{
padding: 0;
}}
Tabs {{
background: {_BG_DARK};
}}
Tab {{
background: {_BG_DARK};
color: {_DIM};
padding: 0 2;
}}
Tab.-active {{
background: {_BG_PANEL};
color: {_PRIMARY};
}}
Tab:hover {{
color: {_TERTIARY};
}}
Underline > .underline--bar {{
color: {_SECONDARY};
background: {_BG_DARK};
}}
.tab-scroll {{
background: {_BG_PANEL};
height: 1fr;
padding: 1 2;
scrollbar-color: {_PRIMARY};
}}
.section-header {{
padding: 0 0 0 1;
margin: 1 0 0 0;
}}
.detail-line {{
padding: 0 0 0 1;
}}
.task-label {{
padding: 0 1;
}}
.task-output-editor {{
height: auto;
max-height: 10;
margin: 0 1 1 3;
border: round {_DIM};
}}
.task-output-editor:focus {{
border: round {_PRIMARY};
}}
Collapsible {{
background: {_BG_PANEL};
padding: 0;
margin: 0 0 1 1;
}}
CollapsibleTitle {{
background: {_BG_DARK};
color: {_TERTIARY};
padding: 0 1;
}}
CollapsibleTitle:hover {{
background: {_SECONDARY};
}}
.input-row {{
height: 3;
padding: 0 1;
@@ -180,55 +258,9 @@ class CheckpointTUI(App[_TuiResult]):
.input-row Input {{
width: 1fr;
}}
#no-inputs-label {{
height: 1;
.empty-state {{
color: {_DIM};
padding: 0 1;
}}
#action-buttons {{
height: 3;
align: right middle;
padding: 0 1;
display: none;
}}
#action-buttons.visible {{
display: block;
}}
#action-buttons Button {{
margin: 0 0 0 1;
min-width: 10;
}}
#btn-resume {{
background: {_SECONDARY};
color: {_TERTIARY};
}}
#btn-resume:hover {{
background: {_PRIMARY};
}}
#btn-fork {{
background: {_PRIMARY};
color: {_TERTIARY};
}}
#btn-fork:hover {{
background: {_SECONDARY};
}}
.entity-title {{
padding: 1 1 0 1;
}}
.entity-detail {{
padding: 0 1;
}}
.task-output-editor {{
height: auto;
max-height: 10;
margin: 0 1 1 1;
border: round {_DIM};
}}
.task-output-editor:focus {{
border: round {_PRIMARY};
}}
.task-label {{
padding: 0 1;
padding: 1;
}}
Tree {{
background: {_BG_PANEL};
@@ -242,6 +274,8 @@ class CheckpointTUI(App[_TuiResult]):
BINDINGS: ClassVar[list[Binding | tuple[str, str] | tuple[str, str, str]]] = [
("q", "quit", "Quit"),
("r", "refresh", "Refresh"),
("e", "resume", "Resume"),
("f", "fork", "Fork"),
]
def __init__(self, location: str = "./.checkpoints") -> None:
@@ -256,27 +290,49 @@ class CheckpointTUI(App[_TuiResult]):
yield Header(show_clock=False)
with Horizontal(id="main-layout"):
tree: Tree[dict[str, Any]] = Tree("Checkpoints", id="tree-panel")
tree.show_root = True
tree.show_root = False
tree.guide_depth = 3
yield tree
with Vertical(id="detail-container"):
yield Static("", id="status")
with VerticalScroll(id="detail-scroll"):
yield Static(
f"[{_DIM}]Select a checkpoint from the tree[/]", # noqa: S608
id="detail-header",
)
with Vertical(id="inputs-section"):
yield Static("Inputs", id="inputs-label")
with Horizontal(id="action-buttons"):
yield Button("Resume", id="btn-resume")
yield Button("Fork", id="btn-fork")
with TabbedContent(id="detail-tabs"):
with TabPane("Overview", id="tab-overview"):
with VerticalScroll(classes="tab-scroll"):
yield Static(
f"[{_DIM}]Select a checkpoint from the tree[/]", # noqa: S608
id="overview-empty",
)
with TabPane("Tasks", id="tab-tasks"):
with VerticalScroll(classes="tab-scroll"):
yield Static(
f"[{_DIM}]Select a checkpoint to view tasks[/]",
id="tasks-empty",
)
with TabPane("Inputs", id="tab-inputs"):
with VerticalScroll(classes="tab-scroll"):
yield Static(
f"[{_DIM}]Select a checkpoint to view inputs[/]",
id="inputs-empty",
)
yield Footer()
async def on_mount(self) -> None:
self._refresh_tree()
self.query_one("#tree-panel", Tree).root.expand()
# ── Tree building ──────────────────────────────────────────────
@staticmethod
def _top_level_entity(entry: dict[str, Any]) -> tuple[str, str]:
etype, ename = "unknown", ""
for ent in entry.get("entities", []):
t = ent.get("type", "unknown")
if t == "flow":
return "flow", ent.get("name") or ""
if t == "crew" and etype != "crew":
etype, ename = "crew", ent.get("name") or ""
return etype, ename
def _refresh_tree(self) -> None:
self._entries = _load_entries(self._location)
self._selected_entry = None
@@ -285,45 +341,57 @@ class CheckpointTUI(App[_TuiResult]):
tree.clear()
if not self._entries:
self.query_one("#detail-header", Static).update(
f"[{_DIM}]No checkpoints in {self._location}[/]"
)
self.query_one("#status", Static).update("")
self.sub_title = self._location
self.query_one("#status", Static).update("")
return
# Group by branch
branches: dict[str, list[dict[str, Any]]] = defaultdict(list)
grouped: dict[tuple[str, str], dict[str, list[dict[str, Any]]]] = defaultdict(
lambda: defaultdict(list)
)
for entry in self._entries:
key = self._top_level_entity(entry)
branch = entry.get("branch", "main")
branches[branch].append(entry)
# Index checkpoint names to tree nodes so forks can attach
node_by_name: dict[str, Any] = {}
grouped[key][branch].append(entry)
def _make_label(e: dict[str, Any]) -> str:
name = e.get("name", "")
ts = e.get("ts") or ""
trigger = e.get("trigger") or ""
parts = [f"[bold]{_short_id(name)}[/]"]
if ts:
time_part = ts.split(" ")[-1] if " " in ts else ts
time_part = ts.split(" ")[-1] if " " in ts else ts
total_c, total_t = 0, 0
for ent in e.get("entities", []):
c = ent.get("tasks_completed")
t = ent.get("tasks_total")
if c is not None and t is not None:
total_c += c
total_t += t
parts: list[str] = []
if time_part:
parts.append(f"[{_DIM}]{time_part}[/]")
if trigger:
parts.append(f"[{_PRIMARY}]{trigger}[/]")
return " ".join(parts)
if total_t:
display_c = total_c
if trigger == "task_started" and total_c < total_t:
display_c = total_c + 1
color = _SUCCESS if total_c == total_t else _DIM
parts.append(f"[{color}]{display_c}/{total_t}[/]")
return " ".join(parts) if parts else _short_id(e.get("name", ""))
fork_parents: set[str] = set()
for branch_name, entries in branches.items():
if branch_name == "main" or not entries:
continue
oldest = min(entries, key=lambda e: str(e.get("name", "")))
first_parent = oldest.get("parent_id")
if first_parent:
fork_parents.add(str(first_parent))
for branches in grouped.values():
for branch_name, entries in branches.items():
if branch_name == "main" or not entries:
continue
oldest = min(entries, key=lambda e: str(e.get("name", "")))
first_parent = oldest.get("parent_id")
if first_parent:
fork_parents.add(str(first_parent))
node_by_name: dict[str, Any] = {}
def _add_checkpoint(parent_node: Any, e: dict[str, Any]) -> None:
"""Add a checkpoint node — expandable only if a fork attaches to it."""
cp_id = _entry_id(e)
if cp_id in fork_parents:
node = parent_node.add(
@@ -333,67 +401,97 @@ class CheckpointTUI(App[_TuiResult]):
node = parent_node.add_leaf(_make_label(e), data=e)
node_by_name[cp_id] = node
if "main" in branches:
for entry in reversed(branches["main"]):
_add_checkpoint(tree.root, entry)
type_order = {"flow": 0, "crew": 1}
sorted_keys = sorted(
grouped.keys(), key=lambda k: (type_order.get(k[0], 9), k[1])
)
for etype, ename in sorted_keys:
branches = grouped[(etype, ename)]
icon = _entity_icon(etype)
color = _ENTITY_COLORS.get(etype, _DIM)
total = sum(len(v) for v in branches.values())
label_parts = [f"{icon} [bold {color}]{etype.upper()}[/]"]
if ename:
label_parts.append(f"[bold]{ename}[/]")
label_parts.append(f"[{_DIM}]({total})[/]")
all_entries = [e for bl in branches.values() for e in bl]
timestamps = [str(e.get("ts", "")) for e in all_entries if e.get("ts")]
if timestamps:
latest = max(timestamps)
label_parts.append(f"[{_DIM}]{_human_ts(latest)}[/]")
entity_label = " ".join(label_parts)
entity_node = tree.root.add(entity_label, expand=True)
if "main" in branches:
for entry in reversed(branches["main"]):
_add_checkpoint(entity_node, entry)
fork_branches = [
(name, sorted(entries, key=lambda e: str(e.get("name", ""))))
for name, entries in branches.items()
if name != "main"
]
remaining = fork_branches
max_passes = len(remaining) + 1
while remaining and max_passes > 0:
max_passes -= 1
deferred = []
made_progress = False
for branch_name, entries in remaining:
first_parent = entries[0].get("parent_id") if entries else None
if first_parent and str(first_parent) not in node_by_name:
deferred.append((branch_name, entries))
continue
attach_to: Any = entity_node
if first_parent:
attach_to = node_by_name.get(str(first_parent), entity_node)
branch_label = (
f"[bold {_SECONDARY}]{branch_name}[/] "
f"[{_DIM}]({len(entries)})[/]"
)
branch_node = attach_to.add(branch_label, expand=False)
for entry in entries:
_add_checkpoint(branch_node, entry)
made_progress = True
remaining = deferred
if not made_progress:
break
fork_branches = [
(name, sorted(entries, key=lambda e: str(e.get("name", ""))))
for name, entries in branches.items()
if name != "main"
]
remaining = fork_branches
max_passes = len(remaining) + 1
while remaining and max_passes > 0:
max_passes -= 1
deferred = []
made_progress = False
for branch_name, entries in remaining:
first_parent = entries[0].get("parent_id") if entries else None
if first_parent and str(first_parent) not in node_by_name:
deferred.append((branch_name, entries))
continue
attach_to: Any = tree.root
if first_parent:
attach_to = node_by_name.get(str(first_parent), tree.root)
branch_label = (
f"[bold {_SECONDARY}]{branch_name}[/] [{_DIM}]({len(entries)})[/]"
f"[bold {_SECONDARY}]{branch_name}[/] "
f"[{_DIM}]({len(entries)})[/] [{_DIM}](orphaned)[/]"
)
branch_node = attach_to.add(branch_label, expand=False)
branch_node = entity_node.add(branch_label, expand=False)
for entry in entries:
_add_checkpoint(branch_node, entry)
made_progress = True
remaining = deferred
if not made_progress:
break
for branch_name, entries in remaining:
branch_label = (
f"[bold {_SECONDARY}]{branch_name}[/] "
f"[{_DIM}]({len(entries)})[/] [{_DIM}](orphaned)[/]"
)
branch_node = tree.root.add(branch_label, expand=False)
for entry in entries:
_add_checkpoint(branch_node, entry)
count = len(self._entries)
storage = "SQLite" if _is_sqlite(self._location) else "JSON"
self.sub_title = self._location
self.query_one("#status", Static).update(f" {count} checkpoint(s) | {storage}")
async def _show_detail(self, entry: dict[str, Any]) -> None:
"""Update the detail panel for a checkpoint entry."""
self._selected_entry = entry
self.query_one("#action-buttons").add_class("visible")
# ── Detail panel ───────────────────────────────────────────────
detail_scroll = self.query_one("#detail-scroll", VerticalScroll)
# Remove all dynamic children except the header — await so IDs are freed
to_remove = [c for c in detail_scroll.children if c.id != "detail-header"]
for child in to_remove:
async def _clear_scroll(self, tab_id: str) -> VerticalScroll:
tab = self.query_one(f"#{tab_id}", TabPane)
scroll = tab.query_one(VerticalScroll)
for child in list(scroll.children):
await child.remove()
return scroll
async def _show_detail(self, entry: dict[str, Any]) -> None:
self._selected_entry = entry
await self._render_overview(entry)
await self._render_tasks(entry)
await self._render_inputs(entry.get("inputs", {}))
async def _render_overview(self, entry: dict[str, Any]) -> None:
scroll = await self._clear_scroll("tab-overview")
# Header
name = entry.get("name", "")
ts = entry.get("ts") or "unknown"
trigger = entry.get("trigger") or ""
@@ -414,42 +512,115 @@ class CheckpointTUI(App[_TuiResult]):
header_lines.append(f" [bold]Branch[/] [{_SECONDARY}]{branch}[/]")
if parent_id:
header_lines.append(f" [bold]Parent[/] [{_DIM}]{parent_id}[/]")
if "path" in entry:
header_lines.append(f" [bold]Path[/] [{_DIM}]{entry['path']}[/]")
if "db" in entry:
header_lines.append(f" [bold]Database[/] [{_DIM}]{entry['db']}[/]")
self.query_one("#detail-header", Static).update("\n".join(header_lines))
await scroll.mount(Static("\n".join(header_lines)))
for ent in entry.get("entities", []):
etype = ent.get("type", "unknown")
ename = ent.get("name", "unnamed")
icon = _entity_icon(etype)
color = _ENTITY_COLORS.get(etype, _DIM)
eid = str(ent.get("id", ""))[:8]
entity_title = (
f"\n{icon} [bold {color}]{etype.upper()}[/] [bold]{ename}[/]"
)
if eid:
entity_title += f" [{_DIM}]{eid}…[/]"
await scroll.mount(Static(entity_title, classes="section-header"))
await scroll.mount(Static(f"[{_DIM}]{'─' * 46}[/]", classes="detail-line"))
if etype == "flow":
methods = ent.get("completed_methods", [])
if methods:
method_list = ", ".join(f"[{_SUCCESS}]{m}[/]" for m in methods)
await scroll.mount(
Static(
f" [bold]Methods[/] {method_list}",
classes="detail-line",
)
)
flow_state = ent.get("flow_state")
if isinstance(flow_state, dict) and flow_state:
state_parts: list[str] = []
for k, v in list(flow_state.items())[:5]:
sv = str(v)
if len(sv) > 40:
sv = sv[:37] + "..."
state_parts.append(f"[{_DIM}]{k}[/]={sv}")
await scroll.mount(
Static(
f" [bold]State[/] {', '.join(state_parts)}",
classes="detail-line",
)
)
agents = ent.get("agents", [])
if agents:
agent_lines: list[Static] = []
for ag in agents:
role = ag.get("role", "unnamed")
goal = ag.get("goal", "")
if len(goal) > 60:
goal = goal[:57] + "..."
agent_line = f" {_entity_icon('agent')} [bold]{role}[/]"
if goal:
agent_line += f"\n [{_DIM}]{goal}[/]"
agent_lines.append(Static(agent_line))
collapsible = Collapsible(
*agent_lines,
title=f"Agents ({len(agents)})",
collapsed=len(agents) > 3,
)
await scroll.mount(collapsible)
async def _render_tasks(self, entry: dict[str, Any]) -> None:
scroll = await self._clear_scroll("tab-tasks")
# Entity details and editable task outputs — mounted flat for scrolling
self._task_output_ids = []
flat_task_idx = 0
has_tasks = False
for ent_idx, ent in enumerate(entry.get("entities", [])):
etype = ent.get("type", "unknown")
ename = ent.get("name", "unnamed")
completed = ent.get("tasks_completed")
total = ent.get("tasks_total")
entity_title = f"[bold {_SECONDARY}]{etype}: {ename}[/]"
if completed is not None and total is not None:
entity_title += f" [{_DIM}]{completed}/{total} tasks[/]"
await detail_scroll.mount(Static(entity_title, classes="entity-title"))
await detail_scroll.mount(
Static(_build_entity_header(ent), classes="entity-detail")
)
icon = _entity_icon(etype)
color = _ENTITY_COLORS.get(etype, _DIM)
tasks = ent.get("tasks", [])
if not tasks:
continue
has_tasks = True
completed = ent.get("tasks_completed", 0)
total = ent.get("tasks_total", 0)
await scroll.mount(
Static(
f"{icon} [bold {color}]{ename}[/] "
f"{_build_progress_bar(completed, total, width=16)}",
classes="section-header",
)
)
for i, task in enumerate(tasks):
desc = str(task.get("description", ""))
if len(desc) > 55:
desc = desc[:52] + "..."
if len(desc) > 50:
desc = desc[:47] + "..."
agent_role = task.get("agent_role", "")
if task.get("completed"):
icon = "[green]✓[/]"
await detail_scroll.mount(
Static(f" {icon} {i + 1}. {desc}", classes="task-label")
)
status_icon = f"[{_SUCCESS}]✓[/]"
task_line = f" {status_icon} {i + 1}. {desc}"
if agent_role:
task_line += (
f" [{_DIM}]→ {_entity_icon('agent')} {agent_role}[/]"
)
await scroll.mount(Static(task_line, classes="task-label"))
output_text = task.get("output", "")
editor_id = f"task-output-{ent_idx}-{i}"
await detail_scroll.mount(
await scroll.mount(
TextArea(
str(output_text),
classes="task-output-editor",
@@ -460,28 +631,25 @@ class CheckpointTUI(App[_TuiResult]):
(flat_task_idx, editor_id, str(output_text))
)
else:
icon = "[yellow]○[/]"
await detail_scroll.mount(
Static(f" {icon} {i + 1}. {desc}", classes="task-label")
)
status_icon = f"[{_PENDING}]○[/]"
task_line = f" {status_icon} {i + 1}. {desc}"
if agent_role:
task_line += (
f" [{_DIM}]→ {_entity_icon('agent')} {agent_role}[/]"
)
await scroll.mount(Static(task_line, classes="task-label"))
flat_task_idx += 1
# Build input fields
await self._build_input_fields(entry.get("inputs", {}))
if not has_tasks:
await scroll.mount(Static(f"[{_DIM}]No tasks[/]", classes="empty-state"))
async def _build_input_fields(self, inputs: dict[str, Any]) -> None:
"""Rebuild the inputs section with one field per input key."""
section = self.query_one("#inputs-section")
# Remove old dynamic children — await so IDs are freed
for widget in list(section.query(".input-row, .no-inputs")):
await widget.remove()
async def _render_inputs(self, inputs: dict[str, Any]) -> None:
scroll = await self._clear_scroll("tab-inputs")
self._input_keys = []
if not inputs:
await section.mount(Static(f"[{_DIM}]No inputs[/]", classes="no-inputs"))
section.add_class("visible")
await scroll.mount(Static(f"[{_DIM}]No inputs[/]", classes="empty-state"))
return
for key, value in inputs.items():
@@ -491,12 +659,11 @@ class CheckpointTUI(App[_TuiResult]):
row.compose_add_child(
Input(value=str(value), placeholder=key, id=f"input-{key}")
)
await section.mount(row)
await scroll.mount(row)
section.add_class("visible")
# ── Data collection ────────────────────────────────────────────
def _collect_inputs(self) -> dict[str, Any] | None:
"""Collect current values from input fields."""
if not self._input_keys:
return None
result: dict[str, Any] = {}
@@ -506,7 +673,6 @@ class CheckpointTUI(App[_TuiResult]):
return result
def _collect_task_overrides(self) -> dict[int, str] | None:
"""Collect edited task outputs. Returns only changed values."""
if not self._task_output_ids or self._selected_entry is None:
return None
overrides: dict[int, str] = {}
@@ -516,38 +682,48 @@ class CheckpointTUI(App[_TuiResult]):
overrides[task_idx] = editor.text
return overrides or None
def _detect_entity_type(self, entry: dict[str, Any]) -> Literal["crew", "flow"]:
"""Infer the top-level entity type from checkpoint entities."""
def _detect_entity_type(
self, entry: dict[str, Any]
) -> Literal["crew", "flow", "agent"]:
for ent in entry.get("entities", []):
if ent.get("type") == "flow":
return "flow"
if ent.get("type") == "agent":
return "agent"
return "crew"
def _resolve_location(self, entry: dict[str, Any]) -> str:
"""Get the restore location string for a checkpoint entry."""
if "path" in entry:
return str(entry["path"])
if _is_sqlite(self._location):
return f"{self._location}#{entry['name']}"
return str(entry.get("name", ""))
# ── Events ─────────────────────────────────────────────────────
async def on_tree_node_highlighted(
self, event: Tree.NodeHighlighted[dict[str, Any]]
) -> None:
if event.node.data is not None:
await self._show_detail(event.node.data)
def on_button_pressed(self, event: Button.Pressed) -> None:
def _exit_with_action(self, action: str) -> None:
if self._selected_entry is None:
self.notify("No checkpoint selected", severity="warning")
return
inputs = self._collect_inputs()
overrides = self._collect_task_overrides()
loc = self._resolve_location(self._selected_entry)
etype = self._detect_entity_type(self._selected_entry)
if event.button.id == "btn-resume":
self.exit((loc, "resume", inputs, overrides, etype))
elif event.button.id == "btn-fork":
self.exit((loc, "fork", inputs, overrides, etype))
name = self._selected_entry.get("name", "")[:30]
self.notify(f"{action.title()}: {name}")
self.exit((loc, action, inputs, overrides, etype))
def action_resume(self) -> None:
self._exit_with_action("resume")
def action_fork(self) -> None:
self._exit_with_action("fork")
def action_refresh(self) -> None:
self._refresh_tree()
@@ -657,6 +833,21 @@ async def _run_checkpoint_tui_async(location: str) -> None:
click.echo(f"\nResult: {getattr(result, 'raw', result)}")
return
if entity_type == "agent":
from crewai.agent import Agent
if action == "fork":
click.echo(f"\nForking agent from: {selected}\n")
agent = Agent.fork(config)
else:
click.echo(f"\nResuming agent from: {selected}\n")
agent = Agent.from_checkpoint(config)
click.echo()
result = await agent.akickoff(messages="Resume execution.")
click.echo(f"\nResult: {getattr(result, 'raw', result)}")
return
from crewai.crew import Crew
if action == "fork":

2
lib/crewai/src/crewai/cli/templates/crew/pyproject.toml
View File

@@ -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.2rc1"
"crewai[tools]==1.14.3a3"
]
[project.scripts]

2
lib/crewai/src/crewai/cli/templates/flow/pyproject.toml
View File

@@ -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.2rc1"
"crewai[tools]==1.14.3a3"
]
[project.scripts]

2
lib/crewai/src/crewai/cli/templates/tool/pyproject.toml
View File

@@ -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.2rc1"
"crewai[tools]==1.14.3a3"
]
[tool.crewai]

24
lib/crewai/src/crewai/crew.py
View File

@@ -419,10 +419,32 @@ class Crew(FlowTrackable, BaseModel):
def _restore_runtime(self) -> None:
"""Re-create runtime objects after restoring from a checkpoint."""
from crewai.events.event_bus import crewai_event_bus
started_task_ids: set[str] = set()
state = crewai_event_bus._runtime_state
if state is not None:
for node in state.event_record.nodes.values():
if node.event.type == "task_started" and node.event.task_id:
started_task_ids.add(node.event.task_id)
resuming_task_agent_roles: set[str] = set()
for task in self.tasks:
if (
task.output is None
and task.agent is not None
and str(task.id) in started_task_ids
):
resuming_task_agent_roles.add(task.agent.role)
for agent in self.agents:
agent.crew = self
executor = agent.agent_executor
if executor and executor.messages:
if (
executor
and executor.messages
and agent.role in resuming_task_agent_roles
):
executor.crew = self
executor.agent = agent
executor._resuming = True

356
lib/crewai/src/crewai/events/__init__.py
View File

@@ -6,112 +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:
from crewai.events.types.agent_events import (
@@ -125,6 +33,223 @@ if TYPE_CHECKING:
LiteAgentExecutionErrorEvent,
LiteAgentExecutionStartedEvent,
)
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",
# 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__ = [
@@ -214,42 +339,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)

5
lib/crewai/src/crewai/events/listeners/tracing/trace_batch_manager.py
View File

@@ -81,8 +81,11 @@ class TraceBatchManager:
"""Initialize a new trace batch (thread-safe)"""
with self._batch_ready_cv:
if self.current_batch is not None:
# Lazy init (e.g. DefaultEnvEvent) may have created the batch without
# execution_type; merge metadata from a later flow/crew initializer.
self.current_batch.execution_metadata.update(execution_metadata)
logger.debug(
"Batch already initialized, skipping duplicate initialization"
"Batch already initialized, merged execution metadata and skipped duplicate initialization"
)
return self.current_batch

30
lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py
View File

@@ -60,12 +60,6 @@ from crewai.events.types.crew_events import (
CrewKickoffFailedEvent,
CrewKickoffStartedEvent,
)
from crewai.events.types.env_events import (
CCEnvEvent,
CodexEnvEvent,
CursorEnvEvent,
DefaultEnvEvent,
)
from crewai.events.types.flow_events import (
FlowCreatedEvent,
FlowFinishedEvent,
@@ -212,7 +206,6 @@ class TraceCollectionListener(BaseEventListener):
self._listeners_setup = True
return
self._register_env_event_handlers(crewai_event_bus)
self._register_flow_event_handlers(crewai_event_bus)
self._register_context_event_handlers(crewai_event_bus)
self._register_action_event_handlers(crewai_event_bus)
@@ -221,25 +214,6 @@ class TraceCollectionListener(BaseEventListener):
self._listeners_setup = True
def _register_env_event_handlers(self, event_bus: CrewAIEventsBus) -> None:
"""Register handlers for environment context events."""
@event_bus.on(CCEnvEvent)
def on_cc_env(source: Any, event: CCEnvEvent) -> None:
self._handle_action_event("cc_env", source, event)
@event_bus.on(CodexEnvEvent)
def on_codex_env(source: Any, event: CodexEnvEvent) -> None:
self._handle_action_event("codex_env", source, event)
@event_bus.on(CursorEnvEvent)
def on_cursor_env(source: Any, event: CursorEnvEvent) -> None:
self._handle_action_event("cursor_env", source, event)
@event_bus.on(DefaultEnvEvent)
def on_default_env(source: Any, event: DefaultEnvEvent) -> None:
self._handle_action_event("default_env", source, event)
def _register_flow_event_handlers(self, event_bus: CrewAIEventsBus) -> None:
"""Register handlers for flow events."""
@@ -286,8 +260,8 @@ class TraceCollectionListener(BaseEventListener):
if self.batch_manager.batch_owner_type != "flow":
# Always call _initialize_crew_batch to claim ownership.
# If batch was already initialized by a concurrent action event
# (race condition with DefaultEnvEvent), initialize_batch() returns
# early but batch_owner_type is still correctly set to "crew".
# (e.g. LLM/tool before crew_kickoff_started), initialize_batch()
# returns early but batch_owner_type is still correctly set to "crew".
# Skip only when a parent flow already owns the batch.
self._initialize_crew_batch(source, event)
self._handle_trace_event("crew_kickoff_started", source, event)

7
lib/crewai/src/crewai/flow/flow.py
View File

@@ -1503,6 +1503,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
except Exception:
logger.warning("FlowStartedEvent handler failed", exc_info=True)
get_env_context()
context = self._pending_feedback_context
emit = context.emit
default_outcome = context.default_outcome
@@ -2004,7 +2006,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
restored = apply_checkpoint(self, from_checkpoint)
if restored is not None:
return restored.kickoff(inputs=inputs, input_files=input_files)
get_env_context()
if self.stream:
result_holder: list[Any] = []
current_task_info: TaskInfo = {
@@ -2206,6 +2207,10 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
f"Flow started with ID: {self.flow_id}", color="bold magenta"
)
# After FlowStarted (when not suppressed): env events must not pre-empt
# trace batch init with implicit "crew" execution_type.
get_env_context()
if inputs is not None and "id" not in inputs:
self._initialize_state(inputs)

39
lib/crewai/src/crewai/llm.py
View File

@@ -175,6 +175,16 @@ LLM_CONTEXT_WINDOW_SIZES: Final[dict[str, int]] = {
"us.amazon.nova-pro-v1:0": 300000,
"us.amazon.nova-micro-v1:0": 128000,
"us.amazon.nova-lite-v1:0": 300000,
# Claude 4 models
"us.anthropic.claude-opus-4-7": 1000000,
"us.anthropic.claude-sonnet-4-6": 1000000,
"us.anthropic.claude-opus-4-6-v1": 1000000,
"us.anthropic.claude-opus-4-5-20251101-v1:0": 200000,
"us.anthropic.claude-haiku-4-5-20251001-v1:0": 200000,
"us.anthropic.claude-sonnet-4-5-20250929-v1:0": 200000,
"us.anthropic.claude-opus-4-1-20250805-v1:0": 200000,
"us.anthropic.claude-opus-4-20250514-v1:0": 200000,
"us.anthropic.claude-sonnet-4-20250514-v1:0": 200000,
"us.anthropic.claude-3-5-sonnet-20240620-v1:0": 200000,
"us.anthropic.claude-3-5-haiku-20241022-v1:0": 200000,
"us.anthropic.claude-3-5-sonnet-20241022-v2:0": 200000,
@@ -193,15 +203,44 @@ LLM_CONTEXT_WINDOW_SIZES: Final[dict[str, int]] = {
"eu.anthropic.claude-3-5-sonnet-20240620-v1:0": 200000,
"eu.anthropic.claude-3-sonnet-20240229-v1:0": 200000,
"eu.anthropic.claude-3-haiku-20240307-v1:0": 200000,
# Claude 4 EU
"eu.anthropic.claude-opus-4-7": 1000000,
"eu.anthropic.claude-sonnet-4-6": 1000000,
"eu.anthropic.claude-opus-4-6-v1": 1000000,
"eu.anthropic.claude-opus-4-5-20251101-v1:0": 200000,
"eu.anthropic.claude-haiku-4-5-20251001-v1:0": 200000,
"eu.anthropic.claude-sonnet-4-5-20250929-v1:0": 200000,
"eu.anthropic.claude-opus-4-1-20250805-v1:0": 200000,
"eu.anthropic.claude-opus-4-20250514-v1:0": 200000,
"eu.anthropic.claude-sonnet-4-20250514-v1:0": 200000,
"eu.meta.llama3-2-3b-instruct-v1:0": 131000,
"eu.meta.llama3-2-1b-instruct-v1:0": 131000,
"apac.anthropic.claude-3-5-sonnet-20240620-v1:0": 200000,
"apac.anthropic.claude-3-5-sonnet-20241022-v2:0": 200000,
"apac.anthropic.claude-3-sonnet-20240229-v1:0": 200000,
"apac.anthropic.claude-3-haiku-20240307-v1:0": 200000,
# Claude 4 APAC
"apac.anthropic.claude-opus-4-7": 1000000,
"apac.anthropic.claude-sonnet-4-6": 1000000,
"apac.anthropic.claude-opus-4-6-v1": 1000000,
"apac.anthropic.claude-opus-4-5-20251101-v1:0": 200000,
"apac.anthropic.claude-haiku-4-5-20251001-v1:0": 200000,
"apac.anthropic.claude-sonnet-4-5-20250929-v1:0": 200000,
"apac.anthropic.claude-opus-4-1-20250805-v1:0": 200000,
"apac.anthropic.claude-opus-4-20250514-v1:0": 200000,
"apac.anthropic.claude-sonnet-4-20250514-v1:0": 200000,
"amazon.nova-pro-v1:0": 300000,
"amazon.nova-micro-v1:0": 128000,
"amazon.nova-lite-v1:0": 300000,
"anthropic.claude-opus-4-7": 1000000,
"anthropic.claude-sonnet-4-6": 1000000,
"anthropic.claude-opus-4-6-v1": 1000000,
"anthropic.claude-opus-4-5-20251101-v1:0": 200000,
"anthropic.claude-haiku-4-5-20251001-v1:0": 200000,
"anthropic.claude-sonnet-4-5-20250929-v1:0": 200000,
"anthropic.claude-opus-4-1-20250805-v1:0": 200000,
"anthropic.claude-opus-4-20250514-v1:0": 200000,
"anthropic.claude-sonnet-4-20250514-v1:0": 200000,
"anthropic.claude-3-5-sonnet-20240620-v1:0": 200000,
"anthropic.claude-3-5-haiku-20241022-v1:0": 200000,
"anthropic.claude-3-5-sonnet-20241022-v2:0": 200000,

56
lib/crewai/src/crewai/llms/constants.py
View File

@@ -423,6 +423,34 @@ AZURE_MODELS: list[AzureModels] = [
BedrockModels: TypeAlias = Literal[
# Inference profiles (regional) - Claude 4
"us.anthropic.claude-sonnet-4-5-20250929-v1:0",
"us.anthropic.claude-sonnet-4-20250514-v1:0",
"us.anthropic.claude-opus-4-5-20251101-v1:0",
"us.anthropic.claude-opus-4-20250514-v1:0",
"us.anthropic.claude-opus-4-1-20250805-v1:0",
"us.anthropic.claude-haiku-4-5-20251001-v1:0",
"us.anthropic.claude-sonnet-4-6",
"us.anthropic.claude-opus-4-6-v1",
# Inference profiles - shorter versions
"us.anthropic.claude-sonnet-4-5-v1:0",
"us.anthropic.claude-opus-4-5-v1:0",
"us.anthropic.claude-opus-4-6-v1:0",
"us.anthropic.claude-haiku-4-5-v1:0",
"eu.anthropic.claude-sonnet-4-5-v1:0",
"eu.anthropic.claude-opus-4-5-v1:0",
"eu.anthropic.claude-haiku-4-5-v1:0",
"apac.anthropic.claude-sonnet-4-5-v1:0",
"apac.anthropic.claude-opus-4-5-v1:0",
"apac.anthropic.claude-haiku-4-5-v1:0",
# Global inference profiles
"global.anthropic.claude-sonnet-4-5-20250929-v1:0",
"global.anthropic.claude-sonnet-4-20250514-v1:0",
"global.anthropic.claude-opus-4-5-20251101-v1:0",
"global.anthropic.claude-opus-4-6-v1",
"global.anthropic.claude-haiku-4-5-20251001-v1:0",
"global.anthropic.claude-sonnet-4-6",
# Direct model IDs
"ai21.jamba-1-5-large-v1:0",
"ai21.jamba-1-5-mini-v1:0",
"amazon.nova-lite-v1:0",
@@ -496,6 +524,34 @@ BedrockModels: TypeAlias = Literal[
"twelvelabs.pegasus-1-2-v1:0",
]
BEDROCK_MODELS: list[BedrockModels] = [
# Inference profiles (regional) - Claude 4
"us.anthropic.claude-sonnet-4-5-20250929-v1:0",
"us.anthropic.claude-sonnet-4-20250514-v1:0",
"us.anthropic.claude-opus-4-5-20251101-v1:0",
"us.anthropic.claude-opus-4-20250514-v1:0",
"us.anthropic.claude-opus-4-1-20250805-v1:0",
"us.anthropic.claude-haiku-4-5-20251001-v1:0",
"us.anthropic.claude-sonnet-4-6",
"us.anthropic.claude-opus-4-6-v1",
# Inference profiles - shorter versions
"us.anthropic.claude-sonnet-4-5-v1:0",
"us.anthropic.claude-opus-4-5-v1:0",
"us.anthropic.claude-opus-4-6-v1:0",
"us.anthropic.claude-haiku-4-5-v1:0",
"eu.anthropic.claude-sonnet-4-5-v1:0",
"eu.anthropic.claude-opus-4-5-v1:0",
"eu.anthropic.claude-haiku-4-5-v1:0",
"apac.anthropic.claude-sonnet-4-5-v1:0",
"apac.anthropic.claude-opus-4-5-v1:0",
"apac.anthropic.claude-haiku-4-5-v1:0",
# Global inference profiles
"global.anthropic.claude-sonnet-4-5-20250929-v1:0",
"global.anthropic.claude-sonnet-4-20250514-v1:0",
"global.anthropic.claude-opus-4-5-20251101-v1:0",
"global.anthropic.claude-opus-4-6-v1",
"global.anthropic.claude-haiku-4-5-20251001-v1:0",
"global.anthropic.claude-sonnet-4-6",
# Direct model IDs
"ai21.jamba-1-5-large-v1:0",
"ai21.jamba-1-5-mini-v1:0",
"amazon.nova-lite-v1:0",

34
lib/crewai/src/crewai/llms/providers/azure/completion.py
View File

@@ -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()

3
lib/crewai/src/crewai/llms/providers/bedrock/completion.py
View File

@@ -2075,6 +2075,9 @@ class BedrockCompletion(BaseLLM):
# Context window sizes for common Bedrock models
context_windows = {
"anthropic.claude-sonnet-4": 200000,
"anthropic.claude-opus-4": 200000,
"anthropic.claude-haiku-4": 200000,
"anthropic.claude-3-5-sonnet": 200000,
"anthropic.claude-3-5-haiku": 200000,
"anthropic.claude-3-opus": 200000,

18
lib/crewai/src/crewai/llms/providers/gemini/completion.py
View File

@@ -976,6 +976,7 @@ class GeminiCompletion(BaseLLM):
"id": call_id,
"name": part.function_call.name,
"args": args_dict,
"raw_part": part,
}
self._emit_stream_chunk_event(
@@ -1060,29 +1061,20 @@ class GeminiCompletion(BaseLLM):
if call_data.get("name") != STRUCTURED_OUTPUT_TOOL_NAME
}
# If there are function calls but no available_functions,
# return them for the executor to handle
if non_structured_output_calls and not available_functions:
formatted_function_calls = [
{
"id": call_data["id"],
"function": {
"name": call_data["name"],
"arguments": json.dumps(call_data["args"]),
},
"type": "function",
}
raw_parts = [
call_data["raw_part"]
for call_data in non_structured_output_calls.values()
]
self._emit_call_completed_event(
response=formatted_function_calls,
response=raw_parts,
call_type=LLMCallType.TOOL_CALL,
from_task=from_task,
from_agent=from_agent,
messages=self._convert_contents_to_dict(contents),
usage=usage_data,
)
return formatted_function_calls
return raw_parts
# Handle completed function calls (excluding structured_output)
if non_structured_output_calls and available_functions:

34
lib/crewai/src/crewai/mcp/__init__.py
View File

@@ -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__ = [

2
lib/crewai/src/crewai/project/annotations.py
View File

@@ -237,6 +237,8 @@ def crew(
self.tasks = instantiated_tasks
crew_instance: Crew = _call_method(meth, self, *args, **kwargs)
if "name" not in crew_instance.model_fields_set:
crew_instance.name = getattr(self, "_crew_name", None) or crew_instance.name
def callback_wrapper(
hook: Callable[Concatenate[CrewInstance, P2], R2], instance: CrewInstance

3
lib/crewai/src/crewai/state/runtime.py
View File

@@ -44,9 +44,12 @@ def _sync_checkpoint_fields(entity: object) -> None:
entity: The entity whose private runtime attributes will be
copied into its public checkpoint fields.
"""
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.crew import Crew
from crewai.flow.flow import Flow
if isinstance(entity, BaseAgent):
entity.checkpoint_kickoff_event_id = entity._kickoff_event_id
if isinstance(entity, Flow):
entity.checkpoint_completed_methods = (
set(entity._completed_methods) if entity._completed_methods else None

85
lib/crewai/src/crewai/task.py
View File

@@ -32,6 +32,7 @@ from pydantic import (
field_validator,
model_validator,
)
from pydantic.functional_serializers import PlainSerializer
from pydantic_core import PydanticCustomError
from typing_extensions import Self
@@ -86,6 +87,22 @@ from crewai.utilities.printer import PRINTER
from crewai.utilities.string_utils import interpolate_only
def _serialize_model_class(v: type[BaseModel] | None) -> dict[str, Any] | None:
"""Serialize a Pydantic model class reference to its JSON schema."""
return v.model_json_schema() if v else None
def _deserialize_model_class(v: Any) -> type[BaseModel] | None:
"""Hydrate a model class reference from checkpoint data."""
if v is None or isinstance(v, type):
return v
if isinstance(v, dict):
from crewai.utilities.pydantic_schema_utils import create_model_from_schema
return create_model_from_schema(v)
return None
class Task(BaseModel):
"""Class that represents a task to be executed.
@@ -141,15 +158,33 @@ class Task(BaseModel):
description="Whether the task should be executed asynchronously or not.",
default=False,
)
output_json: type[BaseModel] | None = Field(
output_json: Annotated[
type[BaseModel] | None,
BeforeValidator(_deserialize_model_class),
PlainSerializer(
_serialize_model_class, return_type=dict | None, when_used="json"
),
] = Field(
description="A Pydantic model to be used to create a JSON output.",
default=None,
)
output_pydantic: type[BaseModel] | None = Field(
output_pydantic: Annotated[
type[BaseModel] | None,
BeforeValidator(_deserialize_model_class),
PlainSerializer(
_serialize_model_class, return_type=dict | None, when_used="json"
),
] = Field(
description="A Pydantic model to be used to create a Pydantic output.",
default=None,
)
response_model: type[BaseModel] | None = Field(
response_model: Annotated[
type[BaseModel] | None,
BeforeValidator(_deserialize_model_class),
PlainSerializer(
_serialize_model_class, return_type=dict | None, when_used="json"
),
] = Field(
description="A Pydantic model for structured LLM outputs using native provider features.",
default=None,
)
@@ -189,7 +224,13 @@ class Task(BaseModel):
description="Whether the task should instruct the agent to return the final answer formatted in Markdown",
default=False,
)
converter_cls: type[Converter] | None = Field(
converter_cls: Annotated[
type[Converter] | None,
BeforeValidator(lambda v: v if v is None or isinstance(v, type) else None),
PlainSerializer(
_serialize_model_class, return_type=dict | None, when_used="json"
),
] = Field(
description="A converter class used to export structured output",
default=None,
)
@@ -1241,12 +1282,26 @@ Follow these guidelines:
tools=tools,
)
pydantic_output, json_output = self._export_output(result)
if isinstance(result, BaseModel):
raw = result.model_dump_json()
if self.output_pydantic:
pydantic_output = result
json_output = None
elif self.output_json:
pydantic_output = None
json_output = result.model_dump()
else:
pydantic_output = None
json_output = None
else:
raw = result
pydantic_output, json_output = self._export_output(result)
task_output = TaskOutput(
name=self.name or self.description,
description=self.description,
expected_output=self.expected_output,
raw=result,
raw=raw,
pydantic=pydantic_output,
json_dict=json_output,
agent=agent.role,
@@ -1337,12 +1392,26 @@ Follow these guidelines:
tools=tools,
)
pydantic_output, json_output = self._export_output(result)
if isinstance(result, BaseModel):
raw = result.model_dump_json()
if self.output_pydantic:
pydantic_output = result
json_output = None
elif self.output_json:
pydantic_output = None
json_output = result.model_dump()
else:
pydantic_output = None
json_output = None
else:
raw = result
pydantic_output, json_output = self._export_output(result)
task_output = TaskOutput(
name=self.name or self.description,
description=self.description,
expected_output=self.expected_output,
raw=result,
raw=raw,
pydantic=pydantic_output,
json_dict=json_output,
agent=agent.role,

38
lib/crewai/tests/llms/azure/test_azure.py
View File

@@ -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

72
lib/crewai/tests/test_checkpoint.py
View File

@@ -562,3 +562,75 @@ class TestKickoffFromCheckpoint:
)
assert mock_restored.checkpoint.restore_from is None
assert result == "flow_result"
# ---------- Agent checkpoint/fork ----------
class TestAgentCheckpoint:
def _make_agent_state(self) -> RuntimeState:
agent = Agent(role="r", goal="g", backstory="b", llm="gpt-4o-mini")
return RuntimeState(root=[agent])
def test_agent_from_checkpoint_sets_runtime_state(self) -> None:
state = self._make_agent_state()
state._provider = JsonProvider()
with tempfile.TemporaryDirectory() as d:
loc = state.checkpoint(d)
cfg = CheckpointConfig(restore_from=loc)
from crewai.events.event_bus import crewai_event_bus
crewai_event_bus._runtime_state = None
Agent.from_checkpoint(cfg)
assert crewai_event_bus._runtime_state is not None
def test_agent_fork_sets_branch(self) -> None:
state = self._make_agent_state()
state._provider = JsonProvider()
with tempfile.TemporaryDirectory() as d:
loc = state.checkpoint(d)
cfg = CheckpointConfig(restore_from=loc)
from crewai.events.event_bus import crewai_event_bus
Agent.fork(cfg, branch="agent-experiment")
rt = crewai_event_bus._runtime_state
assert rt is not None
assert rt._branch == "agent-experiment"
def test_agent_fork_auto_branch(self) -> None:
state = self._make_agent_state()
state._provider = JsonProvider()
with tempfile.TemporaryDirectory() as d:
loc = state.checkpoint(d)
cfg = CheckpointConfig(restore_from=loc)
from crewai.events.event_bus import crewai_event_bus
Agent.fork(cfg)
rt = crewai_event_bus._runtime_state
assert rt is not None
assert rt._branch.startswith("fork/")
def test_sync_checkpoint_fields_agent(self) -> None:
from crewai.state.runtime import _sync_checkpoint_fields
agent = Agent(role="r", goal="g", backstory="b", llm="gpt-4o-mini")
agent._kickoff_event_id = "evt-123"
_sync_checkpoint_fields(agent)
assert agent.checkpoint_kickoff_event_id == "evt-123"
def test_agent_restore_kickoff_event_id(self) -> None:
agent = Agent(role="r", goal="g", backstory="b", llm="gpt-4o-mini")
agent._kickoff_event_id = "evt-456"
state = RuntimeState(root=[agent])
state._provider = JsonProvider()
with tempfile.TemporaryDirectory() as d:
from crewai.state.runtime import _prepare_entities
_prepare_entities(state.root)
loc = state.checkpoint(d)
cfg = CheckpointConfig(restore_from=loc)
restored = Agent.from_checkpoint(cfg)
assert restored._kickoff_event_id == "evt-456"

4
lib/crewai/tests/test_checkpoint_cli.py
View File

@@ -292,7 +292,7 @@ class TestPruneJson:
d, name="20250101T000000_old01111_p-none.json"
)
os.utime(old_path, (0, 0))
_write_json_checkpoint(d, name="20260417T000000_new01111_p-none.json")
_write_json_checkpoint(d, name="20990101T000000_new01111_p-none.json")
deleted = _prune_json(d, keep=None, older_than=timedelta(days=1))
assert deleted == 1
@@ -330,7 +330,7 @@ class TestPruneSqlite:
with tempfile.TemporaryDirectory() as d:
db_path = os.path.join(d, "test.db")
_create_sqlite_checkpoint(db_path, "20200101T000000_old01111")
_create_sqlite_checkpoint(db_path, "20260417T000000_new01111")
_create_sqlite_checkpoint(db_path, "20990101T000000_new01111")
deleted = _prune_sqlite(db_path, keep=None, older_than=timedelta(days=1))
assert deleted >= 1
with sqlite3.connect(db_path) as conn:

57
lib/crewai/tests/test_crew.py
View File

@@ -8,6 +8,7 @@ from concurrent.futures import Future
from hashlib import md5
import re
import sys
from typing import Any, cast
from unittest.mock import ANY, MagicMock, call, patch
from crewai.agent import Agent
@@ -17,6 +18,7 @@ from crewai.crew import Crew
from crewai.crews.crew_output import CrewOutput
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.crew_events import (
CrewKickoffStartedEvent,
CrewTestCompletedEvent,
CrewTestStartedEvent,
CrewTrainCompletedEvent,
@@ -4741,6 +4743,61 @@ def test_default_crew_name(researcher, writer):
assert crew.name == "crew"
@pytest.mark.parametrize(
"explicit_name,expected",
[
(None, "ResearchAutomation"),
("My Research Automation", "My Research Automation"),
],
ids=["class_name_from_decorator", "explicit_name_preserved"],
)
def test_crew_kickoff_started_emits_display_name(
researcher, writer, explicit_name, expected
):
"""Kickoff events should use the decorator-provided display name when implicit."""
from crewai.crews.utils import prepare_kickoff
from crewai.project import CrewBase, agent, crew, task
@CrewBase
class ResearchAutomation:
agents_config = None
tasks_config = None
@agent
def researcher(self):
return researcher
@task
def first_task(self):
return Task(
description="Task 1",
expected_output="output",
agent=self.researcher(),
)
@crew
def crew(self):
crew_kwargs: dict[str, Any] = {
"agents": self.agents,
"tasks": self.tasks,
}
if explicit_name is not None:
crew_kwargs["name"] = explicit_name
return Crew(**crew_kwargs)
captured: list[str | None] = []
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(CrewKickoffStartedEvent)
def _capture(_source: Any, event: CrewKickoffStartedEvent) -> None:
captured.append(event.crew_name)
automation_cls = cast(type[Any], ResearchAutomation)
prepare_kickoff(cast(Any, automation_cls()).crew(), inputs=None)
assert captured == [expected]
@pytest.mark.vcr()
def test_memory_remember_receives_task_content():
"""With memory=True, extract_memories receives raw content with task, agent, expected output, and result."""

51
lib/crewai/tests/test_project.py
View File

@@ -1,4 +1,4 @@
from typing import Any, ClassVar
from typing import Any, ClassVar, cast
from unittest.mock import Mock, create_autospec, patch
import pytest
@@ -261,6 +261,55 @@ def test_crew_name():
assert crew._crew_name == "InternalCrew"
def test_crew_decorator_propagates_class_name_to_instance():
"""@crew-decorated factory method should set Crew.name to the decorated class name."""
sample_agent = Agent(role="r", goal="g", backstory="b")
sample_task = Task(description="d", expected_output="o", agent=sample_agent)
@CrewBase
class ImplicitNameCrewFactory:
agents_config = None
tasks_config = None
agents: list[BaseAgent] = [sample_agent]
tasks: list[Task] = [sample_task]
@crew
def crew(self):
return Crew(
agents=[sample_agent],
tasks=[sample_task],
)
factory_cls = cast(type[Any], ImplicitNameCrewFactory)
crew_instance: Crew = cast(Any, factory_cls()).crew()
assert crew_instance.name == "ImplicitNameCrewFactory"
def test_crew_decorator_preserves_explicit_name():
"""Explicit Crew(name=...) inside @crew should win over the @CrewBase class name."""
sample_agent = Agent(role="r", goal="g", backstory="b")
sample_task = Task(description="d", expected_output="o", agent=sample_agent)
@CrewBase
class NamedCrewFactory:
agents_config = None
tasks_config = None
agents: list[BaseAgent] = [sample_agent]
tasks: list[Task] = [sample_task]
@crew
def crew(self):
return Crew(
name="My Explicit Name",
agents=[sample_agent],
tasks=[sample_task],
)
factory_cls = cast(type[Any], NamedCrewFactory)
crew_instance: Crew = cast(Any, factory_cls()).crew()
assert crew_instance.name == "My Explicit Name"
@tool
def simple_tool():
"""Return 'Hi!'"""

40
lib/crewai/tests/tracing/test_tracing.py
View File

@@ -1640,3 +1640,43 @@ class TestBackendInitializedGatedOnSuccess:
assert bm.backend_initialized is False
assert bm.trace_batch_id is None
class TestTraceBatchManagerDuplicateInitMerge:
"""Second initialize_batch call merges execution_metadata (flow after lazy action)."""
def test_duplicate_initialize_merges_execution_metadata(self):
with (
patch(
"crewai.events.listeners.tracing.trace_batch_manager.should_auto_collect_first_time_traces",
return_value=True,
),
patch(
"crewai.events.listeners.tracing.trace_batch_manager.is_tracing_enabled_in_context",
return_value=True,
),
):
bm = TraceBatchManager()
bm.initialize_batch(
user_context={"privacy_level": "standard"},
execution_metadata={
"crew_name": "Unknown Crew",
"crewai_version": "9.9.9",
},
)
first_batch_id = bm.current_batch.batch_id
bm.initialize_batch(
user_context={"privacy_level": "standard"},
execution_metadata={
"flow_name": "ResearchFlow",
"execution_type": "flow",
"crewai_version": "9.9.9",
"execution_start": "2026-01-01T00:00:00+00:00",
},
)
assert bm.current_batch.batch_id == first_batch_id
meta = bm.current_batch.execution_metadata
assert meta.get("execution_type") == "flow"
assert meta.get("flow_name") == "ResearchFlow"
assert meta.get("crew_name") == "Unknown Crew"

2
lib/devtools/pyproject.toml
View File

@@ -13,7 +13,7 @@ dependencies = [
"click~=8.1.7",
"tomlkit~=0.13.2",
"openai>=1.83.0,<3",
"python-dotenv~=1.1.1",
"python-dotenv>=1.2.2,<2",
"pygithub~=1.59.1",
"rich>=13.9.4",
]

2
lib/devtools/src/crewai_devtools/__init__.py
View File

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

134
lib/devtools/src/crewai_devtools/cli.py
View File

@@ -154,6 +154,117 @@ def check_git_clean() -> None:
sys.exit(1)
def _branch_exists_local(branch: str, cwd: Path | None = None) -> bool:
try:
subprocess.run( # noqa: S603
["git", "show-ref", "--verify", "--quiet", f"refs/heads/{branch}"], # noqa: S607
cwd=cwd,
check=True,
capture_output=True,
)
return True
except subprocess.CalledProcessError:
return False
def _branch_exists_remote(branch: str, cwd: Path | None = None) -> bool:
try:
output = run_command(["git", "ls-remote", "--heads", "origin", branch], cwd=cwd)
return bool(output.strip())
except subprocess.CalledProcessError:
return False
def _open_pr_url_for_branch(branch: str, cwd: Path | None = None) -> str | None:
"""Return URL of open PR for branch, or None if no open PR exists."""
try:
url = run_command(
[
"gh",
"pr",
"list",
"--head",
branch,
"--state",
"open",
"--json",
"url",
"--jq",
".[0].url // empty",
],
cwd=cwd,
)
return url or None
except subprocess.CalledProcessError:
return None
def create_or_reset_branch(branch: str, cwd: Path | None = None) -> None:
"""Create ``branch`` from current HEAD, resetting any stale copy.
If the branch exists locally or on origin, prompts the user to
choose between resetting it or aborting. If an open PR exists on
the branch, the prompt surfaces the PR URL and includes a
close-and-reset option so in-flight work isn't silently clobbered.
Raises:
SystemExit: If the user declines to reset.
"""
local_exists = _branch_exists_local(branch, cwd=cwd)
remote_exists = _branch_exists_remote(branch, cwd=cwd)
open_pr = _open_pr_url_for_branch(branch, cwd=cwd) if remote_exists else None
if local_exists or remote_exists:
if open_pr:
console.print(
f"\n[yellow]![/yellow] Branch [bold]{branch}[/bold] already has an open PR: {open_pr}"
)
prompt = "Close the PR, reset the branch, and continue?"
else:
where = []
if local_exists:
where.append("local")
if remote_exists:
where.append("remote")
console.print(
f"\n[yellow]![/yellow] Branch [bold]{branch}[/bold] already exists ({', '.join(where)}) with no open PR"
)
prompt = "Delete it and recreate?"
if not Confirm.ask(prompt, default=False):
console.print("[red]Aborted.[/red]")
sys.exit(1)
if open_pr:
console.print(f"Closing PR {open_pr}...")
run_command(
["gh", "pr", "close", branch, "--delete-branch"],
cwd=cwd,
)
# `gh pr close --delete-branch` removes the remote branch
# and, when checked out, the local branch too.
local_exists = _branch_exists_local(branch, cwd=cwd)
remote_exists = False
if local_exists:
current = run_command(
["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=cwd
).strip()
if current == branch:
console.print(
f"[yellow]![/yellow] Currently on {branch}, switching to main before delete"
)
run_command(["git", "checkout", "main"], cwd=cwd)
console.print(f"[yellow]![/yellow] Deleting local branch {branch}")
run_command(["git", "branch", "-D", branch], cwd=cwd)
if remote_exists:
console.print(f"[yellow]![/yellow] Deleting remote branch {branch}")
run_command(["git", "push", "origin", "--delete", branch], cwd=cwd)
run_command(["git", "checkout", "-b", branch], cwd=cwd)
def update_version_in_file(file_path: Path, new_version: str) -> bool:
"""Update __version__ attribute in a Python file.
@@ -980,7 +1091,7 @@ def _update_docs_and_create_pr(
if docs_files_staged:
docs_branch = f"docs/changelog-v{version}"
run_command(["git", "checkout", "-b", docs_branch])
create_or_reset_branch(docs_branch)
for f in docs_files_staged:
run_command(["git", "add", f])
run_command(
@@ -1418,7 +1529,7 @@ def _release_enterprise(version: str, is_prerelease: bool, dry_run: bool) -> Non
console.print("[green]✓[/green] Workspace synced")
branch_name = f"feat/bump-version-{version}"
run_command(["git", "checkout", "-b", branch_name], cwd=repo_dir)
create_or_reset_branch(branch_name, cwd=repo_dir)
run_command(["git", "add", "."], cwd=repo_dir)
run_command(
["git", "commit", "-m", f"feat: bump versions to {version}"],
@@ -1616,18 +1727,20 @@ def bump(version: str, dry_run: bool, no_push: bool, no_commit: bool) -> None:
for pkg in packages:
console.print(f" - {pkg.name}")
console.print(f"\nUpdating version to {version}...")
_update_all_versions(cwd, lib_dir, version, packages, dry_run)
if no_commit:
console.print(f"\nUpdating version to {version}...")
_update_all_versions(cwd, lib_dir, version, packages, dry_run)
console.print("\nSkipping git operations (--no-commit flag set)")
else:
branch_name = f"feat/bump-version-{version}"
if not dry_run:
console.print(f"\nCreating branch {branch_name}...")
run_command(["git", "checkout", "-b", branch_name])
create_or_reset_branch(branch_name)
console.print("[green]✓[/green] Branch created")
console.print(f"\nUpdating version to {version}...")
_update_all_versions(cwd, lib_dir, version, packages, dry_run)
console.print("\nCommitting changes...")
run_command(["git", "add", "."])
run_command(
@@ -1643,6 +1756,8 @@ def bump(version: str, dry_run: bool, no_push: bool, no_commit: bool) -> None:
console.print(
f"[dim][DRY RUN][/dim] Would create branch: {branch_name}"
)
console.print(f"\nUpdating version to {version}...")
_update_all_versions(cwd, lib_dir, version, packages, dry_run)
console.print(
f"[dim][DRY RUN][/dim] Would commit: feat: bump versions to {version}"
)
@@ -1906,14 +2021,14 @@ def release(
console.print(f"\n[bold cyan]Phase 1: Bumping versions to {version}[/bold cyan]")
try:
_update_all_versions(cwd, lib_dir, version, packages, dry_run)
branch_name = f"feat/bump-version-{version}"
if not dry_run:
console.print(f"\nCreating branch {branch_name}...")
run_command(["git", "checkout", "-b", branch_name])
create_or_reset_branch(branch_name)
console.print("[green]✓[/green] Branch created")
_update_all_versions(cwd, lib_dir, version, packages, dry_run)
console.print("\nCommitting changes...")
run_command(["git", "add", "."])
run_command(["git", "commit", "-m", f"feat: bump versions to {version}"])
@@ -1943,6 +2058,7 @@ def release(
_poll_pr_until_merged(branch_name, "bump PR")
else:
console.print(f"[dim][DRY RUN][/dim] Would create branch: {branch_name}")
_update_all_versions(cwd, lib_dir, version, packages, dry_run)
console.print(
f"[dim][DRY RUN][/dim] Would commit: feat: bump versions to {version}"
)