Warn locally when a persisted flow gets a non-UUID id input (AMP parity)

CrewAI AMP deployments reject non-UUID inputs.id values with HTTP 422,
while the OSS library accepts any string locally, so the mismatch only
surfaces on first deployed run. Emit a one-time UserWarning at kickoff
when persistence is active and the provided id is not a valid UUID.
Local behavior is unchanged — non-UUID ids keep working.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

lib/crewai/src/crewai/flow/runtime/__init__.py

@@ -30,7 +30,8 @@ from typing import (
     TypeVar,
     cast,
 )
-from uuid import uuid4
+from uuid import UUID, uuid4
+import warnings
 
 from opentelemetry import baggage
 from opentelemetry.context import attach, detach
@@ -161,6 +162,27 @@ ExecutionContext = Any  # type: ignore[assignment,misc]
 logger = logging.getLogger(__name__)
 
 
+def _warn_if_non_uuid_flow_state_id(value: Any) -> None:
+    """Warn once per kickoff when a persisted flow gets a non-UUID `id` input.
+
+    CrewAI AMP deployments reject non-UUID `inputs.id` values with HTTP 422,
+    while the OSS library accepts any string locally. Surfacing the mismatch
+    here keeps local development aligned with deployed behavior without
+    changing how the flow runs.
+    """
+    try:
+        UUID(str(value))
+    except (ValueError, TypeError):
+        warnings.warn(
+            f"Flow state id {value!r} is not a valid UUID. Non-UUID flow state "
+            "ids are rejected by CrewAI AMP deployments with HTTP 422; use a "
+            "UUID (e.g. str(uuid4())) for the 'id' input to keep local and "
+            "deployed behavior aligned.",
+            UserWarning,
+            stacklevel=2,
+        )
+
+
 def _condition_branches(
     condition: dict[str, Any],
 ) -> tuple[Literal["and", "or"], list[FlowDefinitionCondition]]:
@@ -2062,6 +2084,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             self._attach_usage_aggregation_listener()
 
         try:
+            # AMP parity: persisted flows deployed to CrewAI AMP reject
+            # non-UUID `id` inputs with HTTP 422, so surface that locally.
+            if inputs and "id" in inputs and self.persistence is not None:
+                _warn_if_non_uuid_flow_state_id(inputs["id"])
+
             # Reset flow state for fresh execution unless restoring from persistence
             is_restoring = (
                 inputs and "id" in inputs and self.persistence is not None

lib/crewai/tests/test_flow_id_format_warning.py

@@ -0,0 +1,101 @@
+"""Tests for the non-UUID flow state id warning (CrewAI AMP parity).
+
+CrewAI AMP deployments reject non-UUID ``inputs.id`` values with HTTP 422,
+while the OSS library accepts any string locally. A persisted flow kicked off
+with a non-UUID ``id`` should surface a one-time warning so the mismatch is
+caught before deployment — without changing local behavior.
+"""
+
+import os
+import warnings
+from uuid import uuid4
+
+import pytest
+from crewai.flow.flow import Flow, start
+from crewai.flow.persistence import persist
+from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
+
+
+AMP_WARNING_MATCH = "CrewAI AMP"
+
+
+def _build_persisted_flow_class(persistence):
+    class PersistedFlow(Flow[dict]):
+        initial_state = dict
+
+        @start()
+        @persist(persistence)
+        def init_step(self):
+            self.state["message"] = "ran"
+
+    return PersistedFlow
+
+
+def test_non_uuid_id_with_persistence_warns_once_and_still_runs(tmp_path):
+    """A non-UUID `id` input on a persisted flow warns once but runs fine."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+    flow = _build_persisted_flow_class(persistence)(persistence=persistence)
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        flow.kickoff(inputs={"id": "session-1"})
+
+    amp_warnings = [
+        w for w in caught if AMP_WARNING_MATCH in str(w.message)
+    ]
+    assert len(amp_warnings) == 1, (
+        f"expected exactly one AMP parity warning, got {len(amp_warnings)}"
+    )
+    assert issubclass(amp_warnings[0].category, UserWarning)
+    message = str(amp_warnings[0].message)
+    assert "422" in message
+    assert "UUID" in message
+
+    # Behavior is unchanged: the flow ran and kept the non-UUID id locally.
+    assert flow.state["id"] == "session-1"
+    assert flow.state["message"] == "ran"
+    assert persistence.load_state("session-1") is not None
+
+
+def test_valid_uuid_id_with_persistence_does_not_warn(tmp_path):
+    """A valid UUID `id` input on a persisted flow emits no AMP warning."""
+    db_path = os.path.join(tmp_path, "test_flows.db")
+    persistence = SQLiteFlowPersistence(db_path)
+    flow = _build_persisted_flow_class(persistence)(persistence=persistence)
+    valid_id = str(uuid4())
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        flow.kickoff(inputs={"id": valid_id})
+
+    amp_warnings = [
+        w for w in caught if AMP_WARNING_MATCH in str(w.message)
+    ]
+    assert amp_warnings == []
+    assert flow.state["id"] == valid_id
+    assert flow.state["message"] == "ran"
+
+
+def test_non_uuid_id_without_persistence_does_not_warn():
+    """Without persistence, a non-UUID `id` input emits no AMP warning."""
+
+    class PlainFlow(Flow[dict]):
+        initial_state = dict
+
+        @start()
+        def init_step(self):
+            self.state["message"] = "ran"
+
+    flow = PlainFlow()
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        flow.kickoff(inputs={"id": "session-1"})
+
+    amp_warnings = [
+        w for w in caught if AMP_WARNING_MATCH in str(w.message)
+    ]
+    assert amp_warnings == []
+    assert flow.state["id"] == "session-1"
+    assert flow.state["message"] == "ran"