andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-01 21:28:10 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

fix: keep json deploy archives python-free

Browse Source
This commit is contained in:
Joao Moura
2026-06-19 09:53:01 -07:00
parent c51e491b91
commit cddf658942
2 changed files with 18 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

266
lib/cli/src/crewai_cli/deploy/archive.py
View File

@@ -1,15 +1,11 @@
from __future__ import annotations
from pathlib import Path
import re
import shutil
import tempfile
from typing import Any
import zipfile
from crewai_cli import git
from crewai_cli.deploy.validate import normalize_package_name
from crewai_cli.utils import parse_toml
_EXCLUDED_DIRS = {
@@ -38,8 +34,6 @@ _EXCLUDED_SUFFIXES = {
".pyc",
".pyo",
}
_SCRIPT_KEY_PATTERN = re.compile(r"^\s*(?P<key>[A-Za-z0-9_.-]+|\"[^\"]+\"|'[^']+')\s*=")
_SECTION_PATTERN = re.compile(r"^\s*\[[^\]]+\]\s*(?:#.*)?$")
def create_project_zip(
@@ -143,267 +137,7 @@ def _stage_project(root: Path, files: list[Path]) -> Path:
destination = staging_root / relative_path
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, destination)
if _is_json_crew_project(staging_root):
_add_json_crew_deploy_wrapper(staging_root)
except Exception:
shutil.rmtree(staging_root, ignore_errors=True)
raise
return staging_root
def _is_json_crew_project(root: Path) -> bool:
"""Return True for JSON crew projects that need a Python deploy wrapper."""
if not ((root / "crew.jsonc").is_file() or (root / "crew.json").is_file()):
return False
project = _read_pyproject(root)
tool_config = project.get("tool") or {}
crewai_config = tool_config.get("crewai") if isinstance(tool_config, dict) else None
declared_type = (
crewai_config.get("type") if isinstance(crewai_config, dict) else None
)
if declared_type == "flow":
return False
package_name = _package_name(root)
if package_name is None:
raise ValueError(
"Could not derive a valid Python package name from [project].name."
)
return not (root / "src" / package_name / "crew.py").is_file()
def _read_pyproject(root: Path) -> dict[str, Any]:
"""Read pyproject.toml, returning an empty mapping on missing or invalid data."""
pyproject_path = root / "pyproject.toml"
if not pyproject_path.is_file():
return {}
try:
pyproject = parse_toml(pyproject_path.read_text())
except Exception:
return {}
return pyproject if isinstance(pyproject, dict) else {}
def _package_name(root: Path) -> str | None:
"""Return the normalized Python package name for the project."""
project = _read_pyproject(root).get("project")
if not isinstance(project, dict):
return None
name = project.get("name")
if not isinstance(name, str) or not name.strip():
return None
package_name = normalize_package_name(name)
return package_name or None
def _class_name(package_name: str) -> str:
"""Return the generated wrapper class name for a package."""
parts = [part for part in re.split(r"[^a-zA-Z0-9]+", package_name) if part]
class_name = "".join(part[:1].upper() + part[1:] for part in parts)
if not class_name:
return "JsonCrew"
if class_name[0].isdigit():
return f"Crew{class_name}"
return class_name
def _add_json_crew_deploy_wrapper(root: Path) -> None:
"""Add Python wrapper files required to deploy a JSON crew project."""
package_name = _package_name(root)
if package_name is None:
raise ValueError(
"Could not derive a valid Python package name from [project].name."
)
package_dir = root / "src" / package_name
config_dir = package_dir / "config"
config_dir.mkdir(parents=True, exist_ok=True)
class_name = _class_name(package_name)
crew_filename = "crew.jsonc" if (root / "crew.jsonc").is_file() else "crew.json"
(package_dir / "__init__.py").write_text("", encoding="utf-8")
(config_dir / "agents.yaml").write_text("{}\n", encoding="utf-8")
(config_dir / "tasks.yaml").write_text("{}\n", encoding="utf-8")
(package_dir / "crew.py").write_text(
_json_crew_py(class_name, crew_filename),
encoding="utf-8",
)
(package_dir / "main.py").write_text(
_json_main_py(package_name, class_name),
encoding="utf-8",
)
_ensure_project_scripts(root, package_name)
def _json_crew_py(class_name: str, crew_filename: str) -> str:
"""Render the generated crew.py module for a JSON crew."""
return f'''from pathlib import Path
from crewai import Crew
from crewai.project import CrewBase, crew
from crewai.project.crew_loader import load_crew
def _crew_path() -> Path:
return Path(__file__).resolve().parents[2] / "{crew_filename}"
@CrewBase
class {class_name}:
"""Compatibility wrapper for a JSON-defined CrewAI project."""
@crew
def crew(self) -> Crew:
crew_instance, default_inputs = load_crew(_crew_path())
self.default_inputs = default_inputs
return crew_instance
'''
def _json_main_py(package_name: str, class_name: str) -> str:
"""Render the generated main.py entrypoints for a JSON crew."""
return f"""#!/usr/bin/env python
import json
import sys
from {package_name}.crew import {class_name}
def _load():
wrapper = {class_name}()
crew = wrapper.crew()
return crew, getattr(wrapper, "default_inputs", {{}})
def run():
crew, inputs = _load()
return crew.kickoff(inputs=inputs)
def train():
crew, inputs = _load()
return crew.train(
n_iterations=int(sys.argv[1]),
filename=sys.argv[2],
inputs=inputs,
)
def replay():
crew, _ = _load()
return crew.replay(task_id=sys.argv[1])
def test():
crew, inputs = _load()
return crew.test(
n_iterations=int(sys.argv[1]),
eval_llm=sys.argv[2],
inputs=inputs,
)
def run_with_trigger():
if len(sys.argv) < 2:
raise ValueError("No trigger payload provided.")
crew, inputs = _load()
trigger_payload = json.loads(sys.argv[1])
return crew.kickoff(
inputs={{**inputs, "crewai_trigger_payload": trigger_payload}}
)
"""
def _ensure_project_scripts(root: Path, package_name: str) -> None:
"""Ensure generated wrappers have project script entrypoints."""
pyproject_path = root / "pyproject.toml"
if not pyproject_path.is_file():
return
content = pyproject_path.read_text(encoding="utf-8")
entries = _project_script_entries(package_name)
pyproject_path.write_text(
_update_project_scripts(content, entries),
encoding="utf-8",
)
def _project_script_entries(package_name: str) -> dict[str, str]:
"""Return script entrypoints required by the generated JSON wrapper."""
return {
package_name: f"{package_name}.main:run",
"run_crew": f"{package_name}.main:run",
"train": f"{package_name}.main:train",
"replay": f"{package_name}.main:replay",
"test": f"{package_name}.main:test",
"run_with_trigger": f"{package_name}.main:run_with_trigger",
}
def _update_project_scripts(content: str, entries: dict[str, str]) -> str:
"""Add or replace generated script entries in pyproject.toml content."""
lines = content.rstrip().splitlines()
header_index = _project_scripts_header_index(lines)
if header_index is None:
return content.rstrip() + _project_scripts_block(entries)
end_index = _section_end_index(lines, header_index + 1)
seen: set[str] = set()
for index in range(header_index + 1, end_index):
key = _script_key(lines[index])
if key in entries:
lines[index] = _script_line(key, entries[key])
seen.add(key)
missing_lines = [
_script_line(key, value) for key, value in entries.items() if key not in seen
]
lines[end_index:end_index] = missing_lines
return "\n".join(lines).rstrip() + "\n"
def _project_scripts_header_index(lines: list[str]) -> int | None:
"""Return the line index of the project scripts table, if present."""
for index, line in enumerate(lines):
if line.strip() == "[project.scripts]":
return index
return None
def _section_end_index(lines: list[str], start_index: int) -> int:
"""Return the exclusive end index for a TOML table section."""
for index in range(start_index, len(lines)):
if _SECTION_PATTERN.match(lines[index]):
return index
return len(lines)
def _script_key(line: str) -> str | None:
"""Return the script key for a pyproject script line."""
match = _SCRIPT_KEY_PATTERN.match(line)
if not match:
return None
key = match.group("key")
if key.startswith(("'", '"')) and key.endswith(("'", '"')):
return key[1:-1]
return key
def _script_line(key: str, value: str) -> str:
"""Render a project script TOML entry."""
return f'{key} = "{value}"'
def _project_scripts_block(entries: dict[str, str]) -> str:
"""Render a project scripts TOML table."""
lines = ["", "", "[project.scripts]"]
lines.extend(_script_line(key, value) for key, value in entries.items())
return "\n".join(lines) + "\n"

48
lib/cli/tests/deploy/test_archive.py
View File

@@ -132,7 +132,7 @@ def test_create_project_zip_excludes_symlinked_files(tmp_path: Path):
assert names == {"pyproject.toml"}
def test_create_project_zip_adds_json_project_wrapper(tmp_path: Path):
def test_create_project_zip_preserves_json_project_shape(tmp_path: Path):
(tmp_path / "pyproject.toml").write_text(
"""
[project]
@@ -157,8 +157,6 @@ type = "crew"
try:
with zipfile.ZipFile(archive_path) as archive:
names = set(archive.namelist())
crew_py = archive.read("src/json_crew/crew.py").decode()
main_py = archive.read("src/json_crew/main.py").decode()
pyproject = archive.read("pyproject.toml").decode()
finally:
archive_path.unlink(missing_ok=True)
@@ -166,18 +164,11 @@ type = "crew"
assert "uv.lock" not in names
assert "crew.jsonc" in names
assert "agents/researcher.jsonc" in names
assert "src/json_crew/__init__.py" in names
assert "src/json_crew/crew.py" in names
assert "src/json_crew/main.py" in names
assert "src/json_crew/config/agents.yaml" in names
assert "src/json_crew/config/tasks.yaml" in names
assert "load_crew(_crew_path())" in crew_py
assert "JsonCrew" in crew_py
assert "from json_crew.crew import JsonCrew" in main_py
assert "run_crew = \"json_crew.main:run\"" in pyproject
assert all(not name.startswith("src/") for name in names)
assert "run_crew" not in pyproject
def test_create_project_zip_updates_existing_json_project_scripts(tmp_path: Path):
def test_create_project_zip_does_not_rewrite_json_project_scripts(tmp_path: Path):
(tmp_path / "pyproject.toml").write_text(
"""
[project]
@@ -203,14 +194,9 @@ type = "crew"
finally:
archive_path.unlink(missing_ok=True)
assert 'json_crew = "json_crew.main:run"' in pyproject
assert 'run_crew = "json_crew.main:run"' in pyproject
assert 'train = "json_crew.main:train"' in pyproject
assert 'replay = "json_crew.main:replay"' in pyproject
assert 'test = "json_crew.main:test"' in pyproject
assert 'run_with_trigger = "json_crew.main:run_with_trigger"' in pyproject
assert 'json_crew = "old.module:run"' in pyproject
assert 'run_crew = "old.module:run"' in pyproject
assert 'custom = "custom.module:main"' in pyproject
assert "old.module:run" not in pyproject
assert "[tool.crewai]" in pyproject
@@ -221,7 +207,7 @@ type = "crew"
'[tool]\ncrewai = "invalid"\n',
],
)
def test_create_project_zip_adds_json_wrapper_for_malformed_tool_config(
def test_create_project_zip_preserves_json_project_with_malformed_tool_config(
tmp_path: Path, tool_config: str
):
(tmp_path / "pyproject.toml").write_text(
@@ -244,12 +230,11 @@ version = "0.1.0"
finally:
archive_path.unlink(missing_ok=True)
assert "src/json_crew/crew.py" in names
assert "src/json_crew/main.py" in names
assert "run_crew = \"json_crew.main:run\"" in pyproject
assert names == {"crew.jsonc", "pyproject.toml"}
assert "run_crew" not in pyproject
def test_create_project_zip_rejects_empty_normalized_package_name(tmp_path: Path):
def test_create_project_zip_accepts_json_project_without_package_name(tmp_path: Path):
(tmp_path / "pyproject.toml").write_text(
"""
[project]
@@ -263,8 +248,11 @@ type = "crew"
)
(tmp_path / "crew.jsonc").write_text("{}\n")
with pytest.raises(
ValueError,
match=r"Could not derive a valid Python package name",
):
create_project_zip("invalid", project_dir=tmp_path)
archive_path = create_project_zip("invalid", project_dir=tmp_path)
try:
with zipfile.ZipFile(archive_path) as archive:
names = set(archive.namelist())
finally:
archive_path.unlink(missing_ok=True)
assert names == {"crew.jsonc", "pyproject.toml"}