feat(crewai): START/CONTENT/END triples as the default wire shape (PNI-136, part 2)

integrations/crew-ai/python/README.md

@@ -52,6 +52,35 @@ add_crewai_flow_fastapi_endpoint(app, MyFlow(), "/flow")
 
 ## Protocol surface
 
+### Wire shape: START / CONTENT / END triples (default)
+
+Text and tool-call output is emitted as `TEXT_MESSAGE_START` / `TEXT_MESSAGE_CONTENT`
+/ `TEXT_MESSAGE_END` and `TOOL_CALL_START` / `TOOL_CALL_ARGS` / `TOOL_CALL_END` — the
+protocol's canonical discrete form. `emission_shape="chunks"` (or
+`AGUI_CREWAI_EMISSION_SHAPE=chunks`) opts back into the previous
+`TEXT_MESSAGE_CHUNK` / `TOOL_CALL_CHUNK` form.
+
+```python
+add_crewai_flow_fastapi_endpoint(app, MyFlow(), "/flow")                     # triples
+add_crewai_flow_fastapi_endpoint(app, MyFlow(), "/flow", emission_shape="chunks")
+```
+
+The two shapes are **not** equivalent. In `chunks` mode a `copilotkit_emit_state` /
+`copilotkit_predict_state` call that lands between two tool-call argument deltas makes
+`@ag-ui/client`'s chunk transform close the call and then throw on the next delta;
+triples keep the call open server-side, because the server knows more deltas are
+coming and the client does not. Triples are also the form any consumer can apply
+directly — `apply/default.ts` throws if a chunk reaches it untransformed — so a raw
+SSE reader (the Python SDK, conformance tooling, custom clients) needs no
+chunk-transform stage.
+
+Both transports (the crewai >= 1.6 `StreamFrame` path and the legacy
+event-bus-listener fallback) route through one `EmissionShaper`, so the event shape
+and payload never depend on the installed crewai version. A run never ends with an
+open sequence: any open message, tool call, or step is closed before `RUN_FINISHED`.
+MCP tool executions always use triples regardless of this setting — their name, args
+and result arrive together rather than streamed.
+
 ### RAW passthrough (opt-in, default OFF)
 
 `emit_raw_events=True` mirrors the crewai events this bridge does **not** map onto

integrations/crew-ai/python/ag_ui_crewai/_capabilities.py

@@ -696,6 +696,7 @@ def _reasoning_capability(llm: Any, *, raw_enabled: bool = False) -> dict:
 def get_capabilities(
     *,
     llm: Any = None,
+    emission_shape: str | None = None,
     emit_raw_events: bool | None = None,
 ) -> dict:
     """Return the CrewAI bridge's capability declaration.
@@ -739,9 +740,32 @@ def get_capabilities(
     # ``_config`` is a leaf (``_env`` + stdlib only), imported locally purely to
     # keep this module's "crewai / litellm / stdlib only" property for every path
     # that never calls ``get_capabilities``.
-    from ._config import DEFAULT_EMIT_RAW_EVENTS, resolve_emit_raw_events
+    from ._config import (
+        DEFAULT_EMIT_RAW_EVENTS,
+        resolve_emission_shape,
+        resolve_emit_raw_events,
+    )
 
     resolved_raw = resolve_emit_raw_events(emit_raw_events)
+    resolved_shape = resolve_emission_shape(emission_shape)
+    text_events = (
+        [EventType.TEXT_MESSAGE_CHUNK.value]
+        if resolved_shape == "chunks"
+        else [
+            EventType.TEXT_MESSAGE_START.value,
+            EventType.TEXT_MESSAGE_CONTENT.value,
+            EventType.TEXT_MESSAGE_END.value,
+        ]
+    )
+    tool_events = (
+        [EventType.TOOL_CALL_CHUNK.value]
+        if resolved_shape == "chunks"
+        else [
+            EventType.TOOL_CALL_START.value,
+            EventType.TOOL_CALL_ARGS.value,
+            EventType.TOOL_CALL_END.value,
+        ]
+    )
     return {
         "identity": {"type": "crewai", "crewaiVersion": CAPABILITIES.crewai_version},
         "humanInTheLoop": {
@@ -768,12 +792,12 @@ def get_capabilities(
             "streamFrames": CAPABILITIES.stream_frame_available,
         },
         "wireShape": {
-            # This build streams LLM text / tool calls as CHUNK events. MCP tool
-            # executions are the exception: their name, args and result arrive
-            # together, so they already emit canonical TOOL_CALL_* triples.
-            "emissionShape": "chunks",
-            "textMessages": [EventType.TEXT_MESSAGE_CHUNK.value],
-            "toolCalls": [EventType.TOOL_CALL_CHUNK.value],
+            # START/CONTENT/END triples by default; "chunks" is a compatibility
+            # opt-out. MCP tool executions always use triples (name, args and result
+            # arrive together, not streamed), independent of this setting.
+            "emissionShape": resolved_shape,
+            "textMessages": text_events,
+            "toolCalls": tool_events,
             "mcpToolCalls": [
                 EventType.TOOL_CALL_START.value,
                 EventType.TOOL_CALL_ARGS.value,

integrations/crew-ai/python/ag_ui_crewai/_config.py

@@ -26,6 +26,15 @@
 
 EMIT_RAW_EVENTS_ENV_VAR = "AGUI_CREWAI_EMIT_RAW_EVENTS"
 
+# Wire shape for streamed text / tool-call output. Triples (START/CONTENT/END) is
+# the canonical discrete form and the shipped default; "chunks" is a compatibility
+# opt-out. Kept here (a leaf module) so the capability declaration can report it
+# without importing the streaming stack.
+SUPPORTED_EMISSION_SHAPES = frozenset({"triples", "chunks"})
+DEFAULT_EMISSION_SHAPE = "triples"
+
+EMISSION_SHAPE_ENV_VAR = "AGUI_CREWAI_EMISSION_SHAPE"
+
 # Vocabulary ``_parse_env_bool`` accepts, so the "was this value used?" check stays
 # in step with the parser instead of duplicating its token list.
 _BOOL_TOKENS = _TRUE_VALUES | _FALSE_VALUES
@@ -76,3 +85,29 @@ def resolve_emit_raw_events(emit_raw_events: bool | None) -> bool:
     used = raw is not None and raw.strip().casefold() in _BOOL_TOKENS
     _warn_if_env_value_ignored(EMIT_RAW_EVENTS_ENV_VAR, raw, used)
     return resolved
+
+
+def resolve_emission_shape(emission_shape: str | None) -> str:
+    """Resolve the wire shape: explicit argument > env var > shipped default."""
+    if emission_shape is not None:
+        if not isinstance(emission_shape, str):
+            raise ValueError(
+                f"emission_shape must be a string, got "
+                f"{type(emission_shape).__name__} ({emission_shape!r})"
+            )
+        normalized = emission_shape.strip().casefold()
+        if normalized not in SUPPORTED_EMISSION_SHAPES:
+            raise ValueError(
+                f"Unknown emission_shape {emission_shape!r}; "
+                f"expected one of {sorted(SUPPORTED_EMISSION_SHAPES)}"
+            )
+        return normalized
+    raw = os.environ.get(EMISSION_SHAPE_ENV_VAR)
+    resolved = DEFAULT_EMISSION_SHAPE
+    used = False
+    if raw is not None:
+        token = raw.strip().casefold()
+        if token in SUPPORTED_EMISSION_SHAPES:
+            resolved, used = token, True
+    _warn_if_env_value_ignored(EMISSION_SHAPE_ENV_VAR, raw, used)
+    return resolved