Merge branch 'main' into fix/codeql-stagehand-url-assertion

AGENTS.md

@@ -1,26 +1,142 @@
-# Agent Instructions for CrewAI OSS
+# Docs contributor guide
 
-CrewAI is a Python based framework for building AI agents and agentic systems.
-Follow these guidelines when contributing:
+The `docs/` directory is published at [docs.crewai.com](https://docs.crewai.com)
+by [Mintlify](https://www.mintlify.com/). Mintlify watches `docs/docs.json`
+and the MDX files referenced from it.
 
-## Key Guidelines
+## TL;DR for editing docs
 
-1. Follow Python best practices and idiomatic patterns.
-2. Maintain existing code structure and organization.
-3. Write unit tests for new functionality focusing on behaivor and not
-   implementation.
-4. Document public APIs and complex logic.
-5. Suggest changes to the `docs/` folder when appropriate
-6. Follow software principles such as DRY and YAGNI.
-7. Keep diffs as minimal as possible.
+- Edit MDX under `docs/edge/<lang>/...` (e.g. `docs/edge/en/concepts/agents.mdx`).
+- Your change ships under the **Edge** version selector the moment it merges
+  to `main`. Edge follows `main` and is the channel for unreleased work.
+- On release cut, the current Edge state is frozen into `docs/v<X.Y.Z>/` and
+  that snapshot becomes the new default version in the selector (tag:
+  `Latest`). Canonical URLs (`/<lang>/...`) auto-redirect to the new default.
+- Never modify files under `docs/v*/`. Those are frozen release snapshots
+  and the `docs-snapshots` CI guard rejects writes. The only exception is a
+  release-cut PR (auto-generated by `devtools release` or the manual
+  `scripts/docs/freeze_current_edge.py` wrapper), which uses a
+  `[docs-freeze]` title prefix to opt out.
+- Never delete or rename files under `docs/images/`. Images are append-only.
+  See [Images](#images) below.
 
-## Changing Docs
+## The version model
 
-1. Edit MDX under `docs/edge/en/*` and reference it from `docs/docs.json` if
-   needed.
-2. Do not modify files under `docs/v*/`. Those are frozen release snapshots
-   managed by devtools.
-3. Do not delete or rename files under `docs/images/` as frozen snapshots
-   may reference them.
-4. If you want to preview your changes locally, use `cd docs && mintlify dev`.
-   To check for broken links, run `cd docs && mintlify broken-links`.
+The site has one rolling channel (Edge) plus one frozen snapshot per
+release.
+
+```
+docs/
+  edge/                  <-- Edge sources (you edit here)
+    en/...
+    pt-BR/  ko/  ar/
+    enterprise-api.*.yaml
+
+  v1.14.7/               <-- frozen snapshot of v1.14.7
+    en/...
+    pt-BR/  ko/  ar/
+    enterprise-api.*.yaml
+  v1.14.6/...
+  ...
+
+  images/                <-- shared, append-only
+  docs.json              <-- Mintlify config: navigation + redirects
+```
+
+`docs/docs.json` lists one navigation block per version per language. Edge
+points at `docs/edge/<lang>/...`; every other version points at its own
+`docs/v<X.Y.Z>/<lang>/...` subtree. Mintlify scopes both the sidebar and the
+in-site search to whichever version the reader selects, so picking
+`v1.10.0` genuinely shows the v1.10.0 docs (and only those).
+
+### URLs and canonical redirects
+
+Each Mintlify version corresponds to its own URL prefix:
+
+- Edge: `/edge/<lang>/<page>` (e.g. `/edge/en/concepts/agents`)
+- Frozen: `/v<X.Y.Z>/<lang>/<page>` (e.g. `/v1.14.7/en/concepts/agents`)
+
+External links to the old, unversioned `/<lang>/<page>` URLs would 404 under
+this layout. To keep them working, `docs.json` ships wildcard redirects:
+
+```jsonc
+{ "source": "/en/:slug*", "destination": "/v1.14.7/en/:slug*", "permanent": false }
+```
+
+The release-cut step rewrites the destination on every release so canonical
+`/<lang>/...` URLs always resolve to the latest stable docs.
+
+## Lifecycle
+
+1. **During development.** You add or edit pages under
+   `docs/edge/<lang>/...` in normal PRs. They land in Edge as soon as the PR
+   merges. Both `/edge/<lang>/<page>` and the version selector's `Edge` entry
+   reflect the change immediately.
+2. **Release cut.** The release engineer runs `devtools release X.Y.Z`. As
+   part of that flow the CLI opens a `[docs-freeze]` PR that copies Edge into
+   `docs/v<X.Y.Z>/`, rewrites internal OpenAPI references, updates
+   `docs/docs.json` to make `v<X.Y.Z>` the new default + `Latest`, and rewires
+   the canonical-URL redirects to the new default. The PR must merge before
+   the tag and PyPI publish run.
+3. **After release.** Edge keeps rolling. Patch fixes to the just-released
+   docs go into Edge and ship with the next release. We do not back-edit
+   frozen snapshots.
+
+See [`RELEASING.md`](RELEASING.md) for the full release runbook.
+
+## Images
+
+Snapshots share a single `docs/images/` directory. If an image is deleted
+or renamed, every frozen snapshot that referenced it breaks. So the rule
+is:
+
+- Adding new images is always fine.
+- Deleting or renaming an existing image fails CI unless the PR is a
+  `[docs-freeze]` release-cut PR.
+- If an asset is wrong, add a new file with a new name and reference the
+  new name in the Edge MDX (`docs/edge/<lang>/...`). Leave the old file
+  alone.
+
+## Local preview
+
+Install the Mintlify CLI and run from `docs/`:
+
+```bash
+npm i -g mintlify
+mintlify dev
+```
+
+Use the version selector at the top of the rendered page to switch between
+Edge and frozen versions.
+
+To check links across every version:
+
+```bash
+mintlify broken-links
+```
+
+CI runs the broken-links check on every PR that touches `docs/**` via
+[`.github/workflows/docs-broken-links.yml`](.github/workflows/docs-broken-links.yml).
+
+## Scripts
+
+- `scripts/docs/freeze_historical_versions.py` — one-time migration that
+  reconstructed `docs/v1.10.0/` through `docs/v1.14.7/` from git tags. You
+  should not need to run this again.
+- `scripts/docs/prefix_version_paths.py` — one-time migration that switched
+  `docs/docs.json` to directory-based versioning, inserted Edge, and added
+  the canonical-URL redirects. You should not need to run this again.
+- `scripts/docs/freeze_current_edge.py` — thin CLI wrapper around
+  `crewai_devtools.docs_versioning.freeze`. `devtools release` calls the
+  same module during its docs PR step; this script is the manual escape
+  hatch (e.g. retroactively freezing a forgotten release).
+
+## CI guards
+
+- [`.github/workflows/docs-snapshots.yml`](.github/workflows/docs-snapshots.yml)
+  enforces the two rules above (frozen snapshots immutable, images
+  append-only). Both checks accept the `[docs-freeze]` PR-title escape
+  hatch.
+- [`.github/workflows/docs-broken-links.yml`](.github/workflows/docs-broken-links.yml)
+  runs `mintlify broken-links` against the whole site, so adding a new
+  page or moving a snapshot file that breaks a link will fail CI.

docs/docs.json

@@ -159,7 +159,6 @@
                       "edge/en/concepts/tasks",
                       "edge/en/concepts/crews",
                       "edge/en/concepts/flows",
-                      "edge/en/concepts/streaming",
                       "edge/en/concepts/production-architecture",
                       "edge/en/concepts/knowledge",
                       "edge/en/concepts/skills",
@@ -365,8 +364,6 @@
                       "edge/en/learn/human-feedback-in-flows",
                       "edge/en/learn/kickoff-async",
                       "edge/en/learn/kickoff-for-each",
-                      "edge/en/learn/streaming-runtime-contract",
-                      "edge/en/learn/consuming-streams",
                       "edge/en/learn/llm-connections",
                       "edge/en/learn/litellm-removal-guide",
                       "edge/en/learn/multimodal-agents",
@@ -9590,7 +9587,6 @@
                       "edge/pt-BR/learn/human-feedback-in-flows",
                       "edge/pt-BR/learn/kickoff-async",
                       "edge/pt-BR/learn/kickoff-for-each",
-                      "edge/pt-BR/learn/streaming-runtime-contract",
                       "edge/pt-BR/learn/llm-connections",
                       "edge/pt-BR/learn/multimodal-agents",
                       "edge/pt-BR/learn/replay-tasks-from-latest-crew-kickoff",
@@ -18481,7 +18477,6 @@
                       "edge/ko/learn/human-feedback-in-flows",
                       "edge/ko/learn/kickoff-async",
                       "edge/ko/learn/kickoff-for-each",
-                      "edge/ko/learn/streaming-runtime-contract",
                       "edge/ko/learn/llm-connections",
                       "edge/ko/learn/multimodal-agents",
                       "edge/ko/learn/replay-tasks-from-latest-crew-kickoff",
@@ -27588,7 +27583,6 @@
                       "edge/ar/learn/human-feedback-in-flows",
                       "edge/ar/learn/kickoff-async",
                       "edge/ar/learn/kickoff-for-each",
-                      "edge/ar/learn/streaming-runtime-contract",
                       "edge/ar/learn/llm-connections",
                       "edge/ar/learn/multimodal-agents",
                       "edge/ar/learn/replay-tasks-from-latest-crew-kickoff",

docs/edge/ar/changelog.mdx

@@ -4,60 +4,6 @@ description: "تحديثات المنتج والتحسينات وإصلاحات
 icon: "clock"
 mode: "wide"
 ---
-<Update label="1 يوليو 2026">
-  ## v1.15.2a2
-
-  [عرض الإصدار على GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.15.2a2)
-
-  ## ما الذي تغير
-
-  ### الميزات
-  - إضافة aiobotocore إلى الإضافات الأساسية
-  - توثيق خيارات وكيل التدفق
-  - إضافة مساعد نصي إلى مثال مهارة التدفق
-  - إضافة مساعد نصي لمطالب CEL الخاصة بالتدفق
-  - إضافة وثائق البث إلى التنقل
-
-  ### إصلاحات الأخطاء
-  - رفض طرق التدفق ذات الاستماع الذاتي
-
-  ### الوثائق
-  - تحديث اللقطة وسجل التغييرات للإصدار v1.15.2a1
-  - ضغط ملف AGENTS.md
-
-  ## المساهمون
-
-  @akaKuruma, @github-code-quality[bot], @lorenzejay, @vinibrsl
-
-</Update>
-
-<Update label="30 يونيو 2026">
-  ## v1.15.2a1
-
-  [عرض الإصدار على GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.15.2a1)
-
-  ## ما الذي تغير
-
-  ### الميزات
-  - إعادة توجيه أوامر القالب إلى منظمة crewAIInc-fde
-  - دعم تعريفات المهارات المضمنة
-  - تعريف بروتوكول إطار التدفق للتدفقات
-  - إضافة أداة النوع والتطبيق في CrewDefinition
-  - إضافة مهارة تأليف تعريف التدفق المُنشأ
-
-  ### إصلاحات الأخطاء
-  - قطع تنقل إصدار الوثائق من Edge لمنع فقدان الصفحات الجديدة
-
-  ### الوثائق
-  - توثيق نوع قاعدة حد التكلفة في وحدة التحكم الخاصة بالوكيل
-  - إزالة مراجع CREWAI_LOG_FORMAT من دليل Datadog
-
-  ## المساهمون
-
-  @danielfsbarreto, @joaomdmoura, @lorenzejay, @lucasgomide, @vinibrsl
-
-</Update>
-
 <Update label="26 يونيو 2026">
   ## v1.15.1
 

docs/edge/ar/learn/streaming-flow-execution.mdx

@@ -11,7 +11,7 @@ mode: "wide"
 
 ## كيف يعمل بث التدفق
 
-عند تفعيل البث في تدفق، يلتقط CrewAI ويبث المخرجات من أي أطقم أو استدعاءات LLM أو أدوات أو أحداث دورة حياة داخل التدفق. يقدم البث عناصر `StreamFrame` مرتبة تحتوي على محتوى قابل للطباعة وبيانات حدث مهيكلة مع تقدم التنفيذ.
+عند تفعيل البث في تدفق، يلتقط CrewAI ويبث المخرجات من أي أطقم أو استدعاءات LLM داخل التدفق. يقدم البث أجزاء منظمة تحتوي على المحتوى وسياق المهمة ومعلومات الوكيل مع تقدم التنفيذ.
 
 ## تفعيل البث
 
@@ -52,7 +52,7 @@ class ResearchFlow(Flow):
 
 ## البث المتزامن
 
-عند استدعاء `kickoff()` على تدفق مع تفعيل البث، يُرجع جلسة stream تنتج عناصر `StreamFrame` مرتبة:
+عند استدعاء `kickoff()` على تدفق مع تفعيل البث، يُرجع كائن `FlowStreamingOutput` يمكنك التكرار عليه:
 
 ```python Code
 flow = ResearchFlow()
@@ -60,43 +60,44 @@ flow = ResearchFlow()
 # Start streaming execution
 streaming = flow.kickoff()
 
-# Iterate over stream items as they arrive
-for item in streaming:
-    print(item.content, end="", flush=True)
+# Iterate over chunks as they arrive
+for chunk in streaming:
+    print(chunk.content, end="", flush=True)
 
 # Access the final result after streaming completes
 result = streaming.result
 print(f"\n\nFinal output: {result}")
 ```
 
-### معلومات عنصر البث
+### معلومات جزء البث
 
-يوفر كل عنصر محتوى قابلاً للطباعة وبيانات حدث مهيكلة:
+يوفر كل جزء سياقاً حول مصدره في التدفق:
 
 ```python Code
 streaming = flow.kickoff()
 
-for item in streaming:
-    print(f"Channel: {item.channel}")
-    print(f"Type: {item.type}")
-    print(f"Content: {item.content}")
-    print(f"Event payload: {item.event}")
+for chunk in streaming:
+    print(f"Agent: {chunk.agent_role}")
+    print(f"Task: {chunk.task_name}")
+    print(f"Content: {chunk.content}")
+    print(f"Type: {chunk.chunk_type}")  # TEXT or TOOL_CALL
 ```
 
 ### الوصول إلى خصائص البث
 
-توفر جلسة stream خصائص وطرق مفيدة:
+يوفر كائن `FlowStreamingOutput` خصائص وطرق مفيدة:
 
 ```python Code
 streaming = flow.kickoff()
 
-# Iterate and collect items
-for item in streaming:
-    print(item.content, end="", flush=True)
+# Iterate and collect chunks
+for chunk in streaming:
+    print(chunk.content, end="", flush=True)
 
 # After iteration completes
 print(f"\nCompleted: {streaming.is_completed}")
-print(f"Total frames: {len(streaming.frames)}")
+print(f"Full text: {streaming.get_full_text()}")
+print(f"Total chunks: {len(streaming.chunks)}")
 print(f"Final result: {streaming.result}")
 ```
 
@@ -113,9 +114,9 @@ async def stream_flow():
     # Start async streaming
     streaming = await flow.kickoff_async()
 
-    # Async iteration over stream items
-    async for item in streaming:
-        print(item.content, end="", flush=True)
+    # Async iteration over chunks
+    async for chunk in streaming:
+        print(chunk.content, end="", flush=True)
 
     # Access final result
     result = streaming.result
@@ -180,14 +181,13 @@ flow = MultiStepFlow()
 streaming = flow.kickoff()
 
 current_step = ""
-for item in streaming:
+for chunk in streaming:
     # Track which flow step is executing
-    step_name = item.event.get("method_name") or item.event.get("task_name")
-    if step_name and step_name != current_step:
-        current_step = step_name
-        print(f"\n\n=== {step_name} ===\n")
+    if chunk.task_name != current_step:
+        current_step = chunk.task_name
+        print(f"\n\n=== {chunk.task_name} ===\n")
 
-    print(item.content, end="", flush=True)
+    print(chunk.content, end="", flush=True)
 
 result = streaming.result
 print(f"\n\nFinal analysis: {result}")
@@ -201,6 +201,7 @@ print(f"\n\nFinal analysis: {result}")
 import asyncio
 from crewai.flow.flow import Flow, listen, start
 from crewai import Agent, Crew, Task
+from crewai.types.streaming import StreamChunkType
 
 class ResearchPipeline(Flow):
     stream = True
@@ -253,35 +254,33 @@ async def run_with_dashboard():
 
     current_agent = ""
     current_task = ""
-    frame_count = 0
+    chunk_count = 0
 
-    async for item in streaming:
-        frame_count += 1
+    async for chunk in streaming:
+        chunk_count += 1
 
         # Display phase transitions
-        task_name = item.event.get("task_name", "")
-        agent_role = item.event.get("agent_role", "")
-        if task_name and task_name != current_task:
-            current_task = task_name
-            current_agent = agent_role
+        if chunk.task_name != current_task:
+            current_task = chunk.task_name
+            current_agent = chunk.agent_role
             print(f"\n\n📋 Phase: {current_task}")
             print(f"👤 Agent: {current_agent}")
             print("-" * 60)
 
         # Display text output
-        if item.content:
-            print(item.content, end="", flush=True)
+        if chunk.chunk_type == StreamChunkType.TEXT:
+            print(chunk.content, end="", flush=True)
 
         # Display tool usage
-        elif item.channel == "tools":
-            print(f"\n🔧 Tool event: {item.type}")
+        elif chunk.chunk_type == StreamChunkType.TOOL_CALL and chunk.tool_call:
+            print(f"\n🔧 Tool: {chunk.tool_call.tool_name}")
 
     # Show completion summary
     result = streaming.result
     print(f"\n\n{'='*60}")
     print("PIPELINE COMPLETE")
     print(f"{'='*60}")
-    print(f"Total frames: {frame_count}")
+    print(f"Total chunks: {chunk_count}")
     print(f"Final output length: {len(str(result))} characters")
 
 asyncio.run(run_with_dashboard())
@@ -354,8 +353,8 @@ class StatefulStreamingFlow(Flow[AnalysisState]):
 flow = StatefulStreamingFlow()
 streaming = flow.kickoff(inputs={"topic": "quantum computing"})
 
-for item in streaming:
-    print(item.content, end="", flush=True)
+for chunk in streaming:
+    print(chunk.content, end="", flush=True)
 
 result = streaming.result
 print(f"\n\nFinal state:")
@@ -375,29 +374,29 @@ print(f"Insights length: {len(flow.state.insights)}")
 - **تتبع التقدم**: إظهار المرحلة الحالية من سير العمل للمستخدمين
 - **لوحات المعلومات الحية**: إنشاء واجهات مراقبة لتدفقات الإنتاج
 
-## قنوات إطارات البث
+## أنواع أجزاء البث
 
-ينتج بث التدفق عناصر `StreamFrame` عبر عدة قنوات:
+مثل بث الطاقم، يمكن أن تكون أجزاء التدفق من أنواع مختلفة:
 
-### إطارات LLM
+### أجزاء TEXT
 
 محتوى نصي قياسي من استجابات LLM:
 
 ```python Code
-for item in streaming:
-    if item.channel == "llm" and item.content:
-        print(item.content, end="", flush=True)
+for chunk in streaming:
+    if chunk.chunk_type == StreamChunkType.TEXT:
+        print(chunk.content, end="", flush=True)
 ```
 
-### إطارات الأدوات
+### أجزاء TOOL_CALL
 
 معلومات حول استدعاءات الأدوات داخل التدفق:
 
 ```python Code
-for item in streaming:
-    if item.channel == "tools":
-        print(f"\nTool event: {item.type}")
-        print(f"Payload: {item.event}")
+for chunk in streaming:
+    if chunk.chunk_type == StreamChunkType.TOOL_CALL and chunk.tool_call:
+        print(f"\nTool: {chunk.tool_call.tool_name}")
+        print(f"Args: {chunk.tool_call.arguments}")
 ```
 
 ## معالجة الأخطاء
@@ -409,8 +408,8 @@ flow = ResearchFlow()
 streaming = flow.kickoff()
 
 try:
-    for item in streaming:
-        print(item.content, end="", flush=True)
+    for chunk in streaming:
+        print(chunk.content, end="", flush=True)
 
     result = streaming.result
     print(f"\nSuccess! Result: {result}")
@@ -423,7 +422,7 @@ except Exception as e:
 
 ## الإلغاء وتنظيف الموارد
 
-تدعم جلسة stream الإلغاء السلس بحيث يتوقف العمل الجاري فوراً عند انقطاع اتصال المستهلك.
+يدعم `FlowStreamingOutput` الإلغاء السلس بحيث يتوقف العمل الجاري فوراً عند انقطاع اتصال المستهلك.
 
 ### مدير السياق غير المتزامن
 
@@ -431,8 +430,8 @@ except Exception as e:
 streaming = await flow.kickoff_async()
 
 async with streaming:
-    async for item in streaming:
-        print(item.content, end="", flush=True)
+    async for chunk in streaming:
+        print(chunk.content, end="", flush=True)
 ```
 
 ### الإلغاء الصريح
@@ -440,8 +439,8 @@ async with streaming:
 ```python Code
 streaming = await flow.kickoff_async()
 try:
-    async for item in streaming:
-        print(item.content, end="", flush=True)
+    async for chunk in streaming:
+        print(chunk.content, end="", flush=True)
 finally:
     await streaming.aclose()  # غير متزامن
     # streaming.close()       # المكافئ المتزامن
@@ -452,10 +451,10 @@ finally:
 ## ملاحظات مهمة
 
 - يفعّل البث تلقائياً بث LLM لأي أطقم مستخدمة داخل التدفق
-- يجب التكرار عبر جميع عناصر stream قبل الوصول إلى خاصية `.result`
+- يجب التكرار عبر جميع الأجزاء قبل الوصول إلى خاصية `.result`
 - يعمل البث مع كل من حالة التدفق المنظمة وغير المنظمة
 - يلتقط بث التدفق المخرجات من جميع الأطقم واستدعاءات LLM في التدفق
-- يتضمن كل إطار سياق حدث مهيكلاً مثل القناة والنوع والنطاق والحمولة
+- يتضمن كل جزء سياقاً حول الوكيل والمهمة التي ولدته
 - يضيف البث حملاً ضئيلاً لتنفيذ التدفق
 
 ## الدمج مع تصور التدفق
@@ -469,8 +468,8 @@ flow.plot("research_flow")  # Creates HTML visualization
 
 # Run with streaming
 streaming = flow.kickoff()
-for item in streaming:
-    print(item.content, end="", flush=True)
+for chunk in streaming:
+    print(chunk.content, end="", flush=True)
 
 result = streaming.result
 print(f"\nFlow complete! View structure at: research_flow.html")

docs/edge/ar/learn/streaming-runtime-contract.mdx

@@ -1,194 +0,0 @@
----
-title: عقد بث وقت التشغيل
-description: بث إطارات وقت تشغيل مرتبة من التدفقات واستدعاءات LLM المباشرة ودورات المحادثة.
-icon: tower-broadcast
-mode: "wide"
----
-
-## نظرة عامة
-
-يوفر CrewAI عقد بث قائمًا على الإطارات للأنظمة التي تحتاج إلى أكثر من أجزاء نصية بسيطة. يصدر العقد كائنات `StreamFrame` مرتبة لأحداث دورة حياة Flow، وتوكنات LLM المباشرة، ونشاط الأدوات، ورسائل المحادثة، والأحداث المخصصة.
-
-استخدم هذه الواجهة عندما تبني واجهة مستخدم، أو جسر خدمة، أو تطبيق طرفية، أو وقت تشغيل نشر يحتاج إلى تدفق ثابت من الأحداث المهيكلة أثناء تشغيل Flow أو دورة محادثة أو استدعاء LLM مباشر.
-
-## StreamFrame
-
-لكل إطار نفس الغلاف:
-
-```python
-from crewai.types.streaming import StreamFrame
-
-frame.id           # معرف إطار فريد
-frame.seq          # ترتيب محلي للتنفيذ، عند توفره
-frame.type         # نوع الحدث المصدر، مثل "flow_started"
-frame.channel      # "llm", "flow", "tools", "messages", "lifecycle", or "custom"
-frame.namespace    # نطاق المصدر/وقت التشغيل
-frame.timestamp    # طابع وقت الحدث
-frame.parent_id    # معرف الحدث الأب، عند توفره
-frame.previous_id  # معرف الحدث السابق، عند توفره
-frame.data         # حمولة الحدث
-frame.event        # اسم بديل لـ frame.data
-frame.content      # نص قابل للطباعة لإطارات التوكن، وإلا ""
-```
-
-حقل `channel` هو أسرع طريقة لتوجيه الإطارات في المستهلكين:
-
-| القناة | تحتوي على |
-|--------|-----------|
-| `llm` | توكنات وأجزاء التفكير من أحداث بث LLM |
-| `flow` | دورة حياة Flow، وتنفيذ الدوال، والتوجيه، وأحداث الإيقاف/الاستئناف |
-| `tools` | أحداث استخدام الأدوات |
-| `messages` | أحداث سجل المحادثة |
-| `lifecycle` | أحداث دورة حياة وقت التشغيل التي لا تخص قناة أخرى |
-| `custom` | أحداث لا تُطابق قناة مدمجة |
-
-يحافظ `frame.type` على نوع الحدث المصدر، حتى يتمكن المستهلكون من التعامل مع أحداث محددة داخل القناة.
-
-## بث Flow
-
-عيّن `stream=True` على Flow لجعل `kickoff()` يعيد جلسة stream:
-
-```python
-from crewai.flow import Flow, start
-
-
-class ReportFlow(Flow):
-    @start()
-    def generate(self):
-        return "done"
-
-
-flow = ReportFlow(stream=True)
-stream = flow.kickoff()
-
-with stream:
-    for chunk in stream:
-        print(chunk.content, end="", flush=True)
-        if chunk.type == "tool_usage_started":
-            print(chunk.event["tool_name"])
-
-result = stream.result
-```
-
-يجب استهلاك stream قبل قراءة `stream.result`. يؤدي الوصول إلى النتيجة مبكرًا إلى رفع `RuntimeError` حتى لا يتعامل المستهلكون بالخطأ مع تشغيل جزئي على أنه مكتمل.
-
-يمكنك أيضًا استدعاء `flow.stream_events(...)` مباشرة عندما تريد البث لاستدعاء واحد بدون تعيين `stream=True` على مثيل Flow.
-
-## التصفية حسب القناة
-
-يوفر `StreamSession` إسقاطات حسب القناة تحافظ على ترتيب الإطارات العالمي داخل القناة المحددة:
-
-```python
-stream = flow.stream_events()
-
-with stream:
-    for frame in stream.llm:
-        print(frame.content, end="", flush=True)
-
-result = stream.result
-```
-
-الإسقاطات المتاحة هي:
-
-| الإسقاط | الإطارات |
-|---------|----------|
-| `stream.events` | كل الإطارات |
-| `stream.llm` | إطارات LLM |
-| `stream.messages` | إطارات رسائل المحادثة |
-| `stream.flow` | إطارات Flow |
-| `stream.tools` | إطارات الأدوات |
-| `stream.interleave([...])` | مجموعة مختارة من القنوات |
-
-استخدم `stream.interleave(["flow", "llm", "messages"])` عندما يريد المستهلك بعض القنوات فقط لكنه ما زال يحتاج إلى ترتيبها النسبي.
-
-## البث غير المتزامن
-
-استخدم `astream()` للمستهلكين غير المتزامنين:
-
-```python
-flow = ReportFlow()
-stream = flow.astream()
-
-async with stream:
-    async for chunk in stream.events:
-        print(chunk.channel, chunk.type, chunk.content)
-
-result = stream.result
-```
-
-تملك الجلسة غير المتزامنة نفس إسقاطات الجلسة المتزامنة.
-
-## بث استدعاء LLM مباشر
-
-ما زال `llm.call(...)` يعيد النتيجة النهائية المجمعة. استخدم `llm.stream_events(...)` عندما تريد التكرار على الأجزاء فور وصولها مع الحفاظ على حمولة الحدث المهيكلة:
-
-```python
-from crewai import LLM
-
-
-llm = LLM(model="gpt-4o-mini")
-stream = llm.stream_events(
-    messages=[
-        {
-            "role": "user",
-            "content": "Explain CrewAI streaming in two short sentences.",
-        }
-    ]
-)
-
-with stream:
-    for chunk in stream:
-        print(chunk.content, end="", flush=True)
-
-result = stream.result
-```
-
-يفعل `llm.stream_events(...)` البث مؤقتًا للاستدعاء المغلف ثم يستعيد إعداد `stream` السابق في LLM بعد ذلك. تستمر تكاملات المزودين في إصدار أحداث بث LLM الأساسية؛ يوفر هذا المساعد واجهة مكرر مشتركة فوق تلك الأحداث لكل مزودي LLM.
-
-## دورات المحادثة
-
-يمكن للتدفقات المحادثية بث دورة مستخدم واحدة باستخدام `stream_turn()`:
-
-```python
-from crewai import Flow
-from crewai.experimental.conversational import ConversationConfig, ConversationState
-
-
-@ConversationConfig(llm="gpt-4o-mini", defer_trace_finalization=True)
-class ChatFlow(Flow[ConversationState]):
-    conversational = True
-
-
-flow = ChatFlow()
-stream = flow.stream_turn("What can you help me with?", session_id="session-1")
-
-with stream:
-    for frame in stream.events:
-        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
-            print(frame.content, end="", flush=True)
-
-reply = stream.result
-```
-
-أثناء `stream_turn()`، يفعّل مسار الإجابة المحادثية المدمج بث توكنات LLM لذلك الدور ثم يستعيد إعداد `stream` السابق في LLM بعد ذلك. يجب على معالجات المسارات المخصصة التي تنشئ Agents أو مثيلات LLM خاصة بها تهيئة تلك النماذج للبث إذا احتاجت إلى إخراج على مستوى التوكن.
-
-## التنظيف
-
-استخدم الجلسة كمدير سياق عندما يكون ذلك ممكنًا. إذا انقطع اتصال العميل قبل استهلاك stream بالكامل، فأغلق الجلسة صراحة:
-
-```python
-stream = flow.stream_events()
-
-try:
-    for frame in stream.events:
-        print(frame.type)
-finally:
-    if not stream.is_exhausted:
-        stream.close()
-```
-
-للتدفقات غير المتزامنة، استخدم `await stream.aclose()`.
-
-## بث الأجزاء القديم
-
-ما زال بث Crew مع `stream=True` يعيد واجهة `CrewStreamingOutput` المعتمدة على الأجزاء والموضحة في [بث تنفيذ Crew](/ar/learn/streaming-crew-execution). وما زالت استدعاءات `llm.call(...)` المباشرة تعيد نتيجة LLM النهائية. عقد الإطارات مخصص لأوقات التشغيل التي تحتاج إلى غلاف حدث ثابت عبر Flows، واستدعاءات LLM المباشرة، ودورات المحادثة، والأدوات، والرسائل.

docs/edge/en/changelog.mdx

@@ -4,60 +4,6 @@ description: "Product updates, improvements, and bug fixes for CrewAI"
 icon: "clock"
 mode: "wide"
 ---
-<Update label="Jul 01, 2026">
-  ## v1.15.2a2
-
-  [View release on GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.15.2a2)
-
-  ## What's Changed
-
-  ### Features
-  - Add aiobotocore to the bedrock extra
-  - Document flow agent options
-  - Add text helper to flow skill example
-  - Add text helper for flow CEL prompts
-  - Add streaming docs to the navigation
-
-  ### Bug Fixes
-  - Reject self-listening flow methods
-
-  ### Documentation
-  - Update snapshot and changelog for v1.15.2a1
-  - Squeeze AGENTS.md file
-
-  ## Contributors
-
-  @akaKuruma, @github-code-quality[bot], @lorenzejay, @vinibrsl
-
-</Update>
-
-<Update label="Jun 30, 2026">
-  ## v1.15.2a1
-
-  [View release on GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.15.2a1)
-
-  ## What's Changed
-
-  ### Features
-  - Repoint template commands to crewAIInc-fde org
-  - Support inline skill definitions
-  - Define stream frame protocol for flows
-  - Add type tool and app in CrewDefinition
-  - Add generated Flow Definition authoring skill
-
-  ### Bug Fixes
-  - Cut docs version navigation from Edge to prevent new pages from being dropped
-
-  ### Documentation
-  - Document Cost Limit rule type in Agent Control Plane
-  - Drop CREWAI_LOG_FORMAT references from Datadog guide
-
-  ## Contributors
-
-  @danielfsbarreto, @joaomdmoura, @lorenzejay, @lucasgomide, @vinibrsl
-
-</Update>
-
 <Update label="Jun 26, 2026">
   ## v1.15.1
 

docs/edge/en/concepts/streaming.mdx

@@ -1,137 +0,0 @@
----
-title: Streaming
-description: Understand CrewAI's streaming model for Flows, direct LLM calls, tools, and conversational turns.
-icon: radio
-mode: "wide"
----
-
-## Overview
-
-Streaming lets your application receive execution updates while work is still running. Instead of waiting for the final result, you can render LLM tokens, tool activity, Flow lifecycle events, and conversation messages as they happen.
-
-CrewAI has two streaming surfaces:
-
-| Surface | Used by | Output |
-|---------|---------|--------|
-| Frame streaming | Flows, direct LLM calls, conversational turns | Ordered `StreamFrame` objects |
-| Crew chunk streaming | Crews with `stream=True` | `CrewStreamingOutput` chunks |
-
-For new runtime integrations, UIs, terminal apps, service bridges, and conversational surfaces, use frame streaming. It provides one stable event envelope across the runtime.
-
-## StreamFrame
-
-A `StreamFrame` is the common object emitted by streamable runtimes:
-
-```python
-frame.id           # unique frame id
-frame.seq          # execution-local order, when available
-frame.type         # source event type, such as "llm_stream_chunk"
-frame.channel      # "llm", "flow", "tools", "messages", "lifecycle", or "custom"
-frame.namespace    # source/runtime namespace
-frame.timestamp    # event timestamp
-frame.parent_id    # parent event id, when available
-frame.previous_id  # previous event id, when available
-frame.data         # structured event payload
-frame.event        # alias for frame.data
-frame.content      # printable text for token-like frames, otherwise ""
-```
-
-The important fields for most consumers are:
-
-| Field | Use it for |
-|-------|------------|
-| `channel` | Routing frames to the right UI region |
-| `type` | Handling a specific event inside a channel |
-| `content` | Printing token-like text |
-| `event` | Reading structured metadata, such as tool names or message roles |
-| `seq` | Preserving execution order |
-
-## Channels
-
-Frames are grouped into high-level channels:
-
-| Channel | Contains |
-|---------|----------|
-| `llm` | LLM call lifecycle, text chunks, and thinking chunks |
-| `flow` | Flow lifecycle, method execution, routing, pause, and resume events |
-| `tools` | Tool usage start, finish, and error events |
-| `messages` | Conversation transcript events |
-| `lifecycle` | Runtime lifecycle events that do not belong to another channel |
-| `custom` | Events that do not map to a built-in channel |
-
-The stream itself remains one ordered timeline. Channel projections let consumers focus on only part of that timeline.
-
-```mermaid
-flowchart LR
-    A["flow<br/>flow_started"] --> B["llm<br/>llm_call_started"]
-    B --> C["llm<br/>llm_stream_chunk"]
-    C --> D["tools<br/>tool_usage_started"]
-    D --> E["tools<br/>tool_usage_finished"]
-    E --> F["llm<br/>llm_stream_chunk"]
-    F --> G["flow<br/>flow_finished"]
-```
-
-## Stream Sessions
-
-Frame streaming returns a stream session:
-
-```python
-stream = flow.stream_events(inputs={"topic": "AI agents"})
-```
-
-The session is both an iterator and the holder for the final result:
-
-```python
-with stream:
-    for frame in stream:
-        print(frame.content, end="", flush=True)
-
-result = stream.result
-```
-
-Consume the stream before reading `stream.result`. Reading the result too early raises an error because the runtime may still be producing frames.
-
-## Channel Projections
-
-Use channel projections when you only need one kind of frame:
-
-```python
-with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
-    for frame in stream.llm:
-        print(frame.content, end="", flush=True)
-
-result = stream.result
-```
-
-Available projections:
-
-| Projection | Frames |
-|------------|--------|
-| `stream.events` | All frames |
-| `stream.llm` | LLM frames |
-| `stream.flow` | Flow frames |
-| `stream.tools` | Tool frames |
-| `stream.messages` | Conversation message frames |
-| `stream.interleave([...])` | Selected channels in relative order |
-
-## Entrypoints
-
-Use the entrypoint that matches the runtime you are streaming:
-
-| Runtime | Streaming entrypoint |
-|---------|----------------------|
-| Flow | `flow.stream_events(...)` |
-| Flow with `stream=True` | `flow.kickoff(...)` returns a stream session |
-| Async Flow | `flow.astream(...)` or `await flow.kickoff_async(...)` when `stream=True` |
-| Direct LLM call | `llm.stream_events(...)` |
-| Conversational Flow turn | `flow.stream_turn(...)` |
-| Crew | `Crew(..., stream=True).kickoff(...)` returns `CrewStreamingOutput` |
-
-Direct `llm.call(...)` still returns the final assembled LLM result. Use `llm.stream_events(...)` when you want to iterate over LLM chunks as they arrive.
-
-## Related Guides
-
-- [Consuming Streams](/edge/en/learn/consuming-streams)
-- [Streaming Runtime Contract](/edge/en/learn/streaming-runtime-contract)
-- [Streaming Flow Execution](/edge/en/learn/streaming-flow-execution)
-- [Streaming Crew Execution](/edge/en/learn/streaming-crew-execution)

docs/edge/en/guides/flows/conversational-flows.mdx

@@ -25,7 +25,6 @@ Use **`flow.handle_turn(message, session_id=...)`** for every user message from
 | API | Use for |
 |-----|---------|
 | `handle_turn(message, session_id=...)` | Ergonomic one-turn wrapper for conversational `Flow` |
-| `stream_turn(message, session_id=...)` | Stream one conversational turn as ordered runtime frames |
 | `chat()` | Local terminal REPL for conversational `Flow` |
 | `kickoff(inputs={...})` | Advanced flow execution without conversational turn handling |
 | `ask()` | Blocking prompt **inside** one step (wizard, clarification) |
@@ -86,23 +85,6 @@ finally:
     flow.finalize_session_traces()  # one trace link for the whole chat
 ```
 
-## Streaming a turn
-
-Use `stream_turn()` when a UI or runtime needs structured events for one chat turn. It returns a stream session with ordered frames for Flow routing, LLM chunks, tool activity, and conversation messages.
-
-```python
-stream = flow.stream_turn("Where is my order?", session_id=session_id)
-
-with stream:
-    for frame in stream.events:
-        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
-            print(frame.data.get("chunk", ""), end="", flush=True)
-
-result = stream.result
-```
-
-For the full frame contract, channel list, and async API, see [Streaming Runtime Contract](/edge/en/learn/streaming-runtime-contract).
-
 ## Turn lifecycle
 
 Each `handle_turn` runs this pipeline:

docs/edge/en/learn/consuming-streams.mdx

@@ -1,177 +0,0 @@
----
-title: Consuming Streams
-description: Print LLM chunks, observe tool events, and read final results from CrewAI streams.
-icon: square-terminal
-mode: "wide"
----
-
-## Overview
-
-Use this guide when you want to subscribe to a CrewAI stream and print or route frames as they arrive.
-
-The basic pattern is:
-
-```python
-stream = flow.stream_events(inputs={"topic": "AI agents"})
-
-with stream:
-    for frame in stream:
-        ...
-
-result = stream.result
-```
-
-Always consume the stream before reading `stream.result`.
-
-## Print LLM Output
-
-If you only care about text generated by LLM calls, subscribe to the `llm` projection and print `frame.content`:
-
-```python
-stream = flow.stream_events(inputs={"topic": "AI agents"})
-
-with stream:
-    for frame in stream.llm:
-        print(frame.content, end="", flush=True)
-
-print()
-result = stream.result
-```
-
-`frame.content` is an empty string for frames that do not carry printable text, so this is also safe:
-
-```python
-with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
-    for frame in stream.events:
-        if frame.channel == "llm" and frame.content:
-            print(frame.content, end="", flush=True)
-
-result = stream.result
-```
-
-## Print Tool Activity
-
-Tool events arrive on the `tools` channel. Use `frame.type` to distinguish starts, finishes, and errors.
-
-```python
-with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
-    for frame in stream.events:
-        if frame.channel == "llm" and frame.content:
-            print(frame.content, end="", flush=True)
-
-        if frame.channel == "tools" and frame.type == "tool_usage_started":
-            print(f"\nTool started: {frame.event.get('tool_name')}")
-
-        if frame.channel == "tools" and frame.type == "tool_usage_finished":
-            print(f"\nTool finished: {frame.event.get('tool_name')}")
-
-result = stream.result
-```
-
-`frame.event` is the structured payload for the source event. Use it for metadata such as tool names, arguments, message roles, and runtime identifiers.
-
-## Watch Flow Progress
-
-Flow lifecycle and method execution frames arrive on the `flow` channel:
-
-```python
-with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
-    for frame in stream.flow:
-        print(frame.type, frame.namespace)
-
-result = stream.result
-```
-
-Use this when you want a progress log instead of token-level output.
-
-## Interleave Selected Channels
-
-Use `interleave()` when you want a subset of channels while preserving their relative order:
-
-```python
-with flow.stream_events(inputs={"topic": "AI agents"}) as stream:
-    for frame in stream.interleave(["llm", "tools"]):
-        if frame.channel == "llm":
-            print(frame.content, end="", flush=True)
-        elif frame.type == "tool_usage_started":
-            print(f"\nTool: {frame.event.get('tool_name')}")
-
-result = stream.result
-```
-
-## Stream a Direct LLM Call
-
-Direct `llm.call(...)` returns the final assembled result. To stream a direct LLM call, use `llm.stream_events(...)`:
-
-```python
-from crewai import LLM
-
-
-llm = LLM(model="gpt-4o-mini")
-stream = llm.stream_events("Explain streaming in one sentence.")
-
-with stream:
-    for frame in stream.llm:
-        print(frame.content, end="", flush=True)
-
-print()
-result = stream.result
-```
-
-## Stream a Conversational Turn
-
-Conversational Flows expose `stream_turn()` for one user message:
-
-```python
-stream = flow.stream_turn(
-    "What can you help me with?",
-    session_id="session-1",
-)
-
-with stream:
-    for frame in stream.interleave(["llm", "messages"]):
-        if frame.channel == "llm":
-            print(frame.content, end="", flush=True)
-        elif frame.channel == "messages":
-            print(f"\n{frame.event.get('role')}: {frame.event.get('content')}")
-
-reply = stream.result
-```
-
-## Async Consumers
-
-Async streams use the same channel projections:
-
-```python
-stream = flow.astream(inputs={"topic": "AI agents"})
-
-async with stream:
-    async for frame in stream.llm:
-        print(frame.content, end="", flush=True)
-
-result = stream.result
-```
-
-## Cleanup
-
-Use the stream as a context manager when possible. If a client disconnects or you stop consuming early, close the stream:
-
-```python
-stream = flow.stream_events(inputs={"topic": "AI agents"})
-
-try:
-    for frame in stream.events:
-        print(frame.content, end="", flush=True)
-finally:
-    if not stream.is_exhausted:
-        stream.close()
-```
-
-For async streams, call `await stream.aclose()`.
-
-## See Also
-
-- [Streaming](/edge/en/concepts/streaming)
-- [Streaming Runtime Contract](/edge/en/learn/streaming-runtime-contract)
-- [Streaming Flow Execution](/edge/en/learn/streaming-flow-execution)
-- [Streaming Crew Execution](/edge/en/learn/streaming-crew-execution)

docs/edge/en/learn/streaming-flow-execution.mdx

@@ -11,7 +11,7 @@ CrewAI Flows support streaming output, allowing you to receive real-time updates
 
 ## How Flow Streaming Works
 
-When streaming is enabled on a Flow, CrewAI captures and streams output from any crews, LLM calls, tools, and lifecycle events within the flow. The stream delivers ordered `StreamFrame` items with printable content plus structured event data as execution progresses.
+When streaming is enabled on a Flow, CrewAI captures and streams output from any crews or LLM calls within the flow. The stream delivers structured chunks containing the content, task context, and agent information as execution progresses.
 
 ## Enabling Streaming
 
@@ -52,7 +52,7 @@ class ResearchFlow(Flow):
 
 ## Synchronous Streaming
 
-When you call `kickoff()` on a flow with streaming enabled, it returns a stream session that yields ordered `StreamFrame` items:
+When you call `kickoff()` on a flow with streaming enabled, it returns a `FlowStreamingOutput` object that you can iterate over:
 
 ```python Code
 flow = ResearchFlow()
@@ -60,43 +60,44 @@ flow = ResearchFlow()
 # Start streaming execution
 streaming = flow.kickoff()
 
-# Iterate over stream items as they arrive
-for item in streaming:
-    print(item.content, end="", flush=True)
+# Iterate over chunks as they arrive
+for chunk in streaming:
+    print(chunk.content, end="", flush=True)
 
 # Access the final result after streaming completes
 result = streaming.result
 print(f"\n\nFinal output: {result}")
 ```
 
-### Stream Item Information
+### Stream Chunk Information
 
-Each item provides both printable content and structured event data:
+Each chunk provides context about where it originated in the flow:
 
 ```python Code
 streaming = flow.kickoff()
 
-for item in streaming:
-    print(f"Channel: {item.channel}")
-    print(f"Type: {item.type}")
-    print(f"Content: {item.content}")
-    print(f"Event payload: {item.event}")
+for chunk in streaming:
+    print(f"Agent: {chunk.agent_role}")
+    print(f"Task: {chunk.task_name}")
+    print(f"Content: {chunk.content}")
+    print(f"Type: {chunk.chunk_type}")  # TEXT or TOOL_CALL
 ```
 
 ### Accessing Streaming Properties
 
-The stream session provides useful properties and methods:
+The `FlowStreamingOutput` object provides useful properties and methods:
 
 ```python Code
 streaming = flow.kickoff()
 
-# Iterate and collect items
-for item in streaming:
-    print(item.content, end="", flush=True)
+# Iterate and collect chunks
+for chunk in streaming:
+    print(chunk.content, end="", flush=True)
 
 # After iteration completes
 print(f"\nCompleted: {streaming.is_completed}")
-print(f"Total frames: {len(streaming.frames)}")
+print(f"Full text: {streaming.get_full_text()}")
+print(f"Total chunks: {len(streaming.chunks)}")
 print(f"Final result: {streaming.result}")
 ```
 
@@ -113,9 +114,9 @@ async def stream_flow():
     # Start async streaming
     streaming = await flow.kickoff_async()
 
-    # Async iteration over stream items
-    async for item in streaming:
-        print(item.content, end="", flush=True)
+    # Async iteration over chunks
+    async for chunk in streaming:
+        print(chunk.content, end="", flush=True)
 
     # Access final result
     result = streaming.result
@@ -180,14 +181,13 @@ flow = MultiStepFlow()
 streaming = flow.kickoff()
 
 current_step = ""
-for item in streaming:
+for chunk in streaming:
     # Track which flow step is executing
-    step_name = item.event.get("method_name") or item.event.get("task_name")
-    if step_name and step_name != current_step:
-        current_step = step_name
-        print(f"\n\n=== {step_name} ===\n")
+    if chunk.task_name != current_step:
+        current_step = chunk.task_name
+        print(f"\n\n=== {chunk.task_name} ===\n")
 
-    print(item.content, end="", flush=True)
+    print(chunk.content, end="", flush=True)
 
 result = streaming.result
 print(f"\n\nFinal analysis: {result}")
@@ -201,6 +201,7 @@ Here's a complete example showing how to build a progress dashboard with streami
 import asyncio
 from crewai.flow.flow import Flow, listen, start
 from crewai import Agent, Crew, Task
+from crewai.types.streaming import StreamChunkType
 
 class ResearchPipeline(Flow):
     stream = True
@@ -253,35 +254,33 @@ async def run_with_dashboard():
 
     current_agent = ""
     current_task = ""
-    frame_count = 0
+    chunk_count = 0
 
-    async for item in streaming:
-        frame_count += 1
+    async for chunk in streaming:
+        chunk_count += 1
 
         # Display phase transitions
-        task_name = item.event.get("task_name", "")
-        agent_role = item.event.get("agent_role", "")
-        if task_name and task_name != current_task:
-            current_task = task_name
-            current_agent = agent_role
+        if chunk.task_name != current_task:
+            current_task = chunk.task_name
+            current_agent = chunk.agent_role
             print(f"\n\n📋 Phase: {current_task}")
             print(f"👤 Agent: {current_agent}")
             print("-" * 60)
 
         # Display text output
-        if item.content:
-            print(item.content, end="", flush=True)
+        if chunk.chunk_type == StreamChunkType.TEXT:
+            print(chunk.content, end="", flush=True)
 
         # Display tool usage
-        elif item.channel == "tools":
-            print(f"\n🔧 Tool event: {item.type}")
+        elif chunk.chunk_type == StreamChunkType.TOOL_CALL and chunk.tool_call:
+            print(f"\n🔧 Tool: {chunk.tool_call.tool_name}")
 
     # Show completion summary
     result = streaming.result
     print(f"\n\n{'='*60}")
     print("PIPELINE COMPLETE")
     print(f"{'='*60}")
-    print(f"Total frames: {frame_count}")
+    print(f"Total chunks: {chunk_count}")
     print(f"Final output length: {len(str(result))} characters")
 
 asyncio.run(run_with_dashboard())
@@ -354,8 +353,8 @@ class StatefulStreamingFlow(Flow[AnalysisState]):
 flow = StatefulStreamingFlow()
 streaming = flow.kickoff(inputs={"topic": "quantum computing"})
 
-for item in streaming:
-    print(item.content, end="", flush=True)
+for chunk in streaming:
+    print(chunk.content, end="", flush=True)
 
 result = streaming.result
 print(f"\n\nFinal state:")
@@ -375,29 +374,29 @@ Flow streaming is particularly valuable for:
 - **Progress Tracking**: Show users which stage of the workflow is currently executing
 - **Live Dashboards**: Create monitoring interfaces for production flows
 
-## Stream Frame Channels
+## Stream Chunk Types
 
-Flow streaming yields `StreamFrame` items across several channels:
+Like crew streaming, flow chunks can be of different types:
 
-### LLM Frames
+### TEXT Chunks
 
 Standard text content from LLM responses:
 
 ```python Code
-for item in streaming:
-    if item.channel == "llm" and item.content:
-        print(item.content, end="", flush=True)
+for chunk in streaming:
+    if chunk.chunk_type == StreamChunkType.TEXT:
+        print(chunk.content, end="", flush=True)
 ```
 
-### Tool Frames
+### TOOL_CALL Chunks
 
 Information about tool calls within the flow:
 
 ```python Code
-for item in streaming:
-    if item.channel == "tools":
-        print(f"\nTool event: {item.type}")
-        print(f"Payload: {item.event}")
+for chunk in streaming:
+    if chunk.chunk_type == StreamChunkType.TOOL_CALL and chunk.tool_call:
+        print(f"\nTool: {chunk.tool_call.tool_name}")
+        print(f"Args: {chunk.tool_call.arguments}")
 ```
 
 ## Error Handling
@@ -409,8 +408,8 @@ flow = ResearchFlow()
 streaming = flow.kickoff()
 
 try:
-    for item in streaming:
-        print(item.content, end="", flush=True)
+    for chunk in streaming:
+        print(chunk.content, end="", flush=True)
 
     result = streaming.result
     print(f"\nSuccess! Result: {result}")
@@ -423,7 +422,7 @@ except Exception as e:
 
 ## Cancellation and Resource Cleanup
 
-The stream session supports graceful cancellation so that in-flight work stops promptly when the consumer disconnects.
+`FlowStreamingOutput` supports graceful cancellation so that in-flight work stops promptly when the consumer disconnects.
 
 ### Async Context Manager
 
@@ -431,8 +430,8 @@ The stream session supports graceful cancellation so that in-flight work stops p
 streaming = await flow.kickoff_async()
 
 async with streaming:
-    async for item in streaming:
-        print(item.content, end="", flush=True)
+    async for chunk in streaming:
+        print(chunk.content, end="", flush=True)
 ```
 
 ### Explicit Cancellation
@@ -440,8 +439,8 @@ async with streaming:
 ```python Code
 streaming = await flow.kickoff_async()
 try:
-    async for item in streaming:
-        print(item.content, end="", flush=True)
+    async for chunk in streaming:
+        print(chunk.content, end="", flush=True)
 finally:
     await streaming.aclose()  # async
     # streaming.close()       # sync equivalent
@@ -452,10 +451,10 @@ After cancellation, `streaming.is_cancelled` and `streaming.is_completed` are bo
 ## Important Notes
 
 - Streaming automatically enables LLM streaming for any crews used within the flow
-- You must iterate through all stream items before accessing the `.result` property
+- You must iterate through all chunks before accessing the `.result` property
 - Streaming works with both structured and unstructured flow state
 - Flow streaming captures output from all crews and LLM calls in the flow
-- Each frame includes structured event context such as channel, type, namespace, and payload
+- Each chunk includes context about which agent and task generated it
 - Streaming adds minimal overhead to flow execution
 
 ## Combining with Flow Visualization
@@ -469,11 +468,11 @@ flow.plot("research_flow")  # Creates HTML visualization
 
 # Run with streaming
 streaming = flow.kickoff()
-for item in streaming:
-    print(item.content, end="", flush=True)
+for chunk in streaming:
+    print(chunk.content, end="", flush=True)
 
 result = streaming.result
 print(f"\nFlow complete! View structure at: research_flow.html")
 ```
 
-By leveraging flow streaming, you can build sophisticated, responsive applications that provide users with real-time visibility into complex multi-stage workflows, making your AI automations more transparent and engaging.
+By leveraging flow streaming, you can build sophisticated, responsive applications that provide users with real-time visibility into complex multi-stage workflows, making your AI automations more transparent and engaging.

docs/edge/en/learn/streaming-runtime-contract.mdx

@@ -1,194 +0,0 @@
----
-title: Streaming Runtime Contract
-description: Stream ordered runtime frames from Flows, direct LLM calls, and conversational turns.
-icon: tower-broadcast
-mode: "wide"
----
-
-## Overview
-
-CrewAI exposes a frame-based streaming contract for runtimes that need more than plain text chunks. The contract emits ordered `StreamFrame` objects for Flow lifecycle events, direct LLM tokens, tool activity, conversation messages, and custom events.
-
-Use this API when you are building a UI, service bridge, terminal app, or deployment runtime that needs a stable stream of structured events while a Flow, chat turn, or direct LLM call is running.
-
-## StreamFrame
-
-Every frame has the same envelope:
-
-```python
-from crewai.types.streaming import StreamFrame
-
-frame.id           # unique frame id
-frame.seq          # execution-local order, when available
-frame.type         # source event type, such as "flow_started"
-frame.channel      # "llm", "flow", "tools", "messages", "lifecycle", or "custom"
-frame.namespace    # source/runtime namespace
-frame.timestamp    # event timestamp
-frame.parent_id    # parent event id, when available
-frame.previous_id  # previous event id, when available
-frame.data         # event payload
-frame.event        # alias for frame.data
-frame.content      # printable text for token-like frames, otherwise ""
-```
-
-The `channel` field is the fastest way to route frames in consumers:
-
-| Channel | Contains |
-|---------|----------|
-| `llm` | Token and thinking chunks from LLM streaming events |
-| `flow` | Flow lifecycle, method execution, routing, and pause/resume events |
-| `tools` | Tool usage events |
-| `messages` | Conversation transcript events |
-| `lifecycle` | Runtime lifecycle events that are not specific to another channel |
-| `custom` | Events that do not map to a built-in channel |
-
-`frame.type` preserves the source event type, so consumers can handle specific events inside a channel.
-
-## Stream a Flow
-
-Set `stream=True` on a Flow to make `kickoff()` return a stream session:
-
-```python
-from crewai.flow import Flow, start
-
-
-class ReportFlow(Flow):
-    @start()
-    def generate(self):
-        return "done"
-
-
-flow = ReportFlow(stream=True)
-stream = flow.kickoff()
-
-with stream:
-    for chunk in stream:
-        print(chunk.content, end="", flush=True)
-        if chunk.type == "tool_usage_started":
-            print(chunk.event["tool_name"])
-
-result = stream.result
-```
-
-You must consume the stream before reading `stream.result`. Accessing the result early raises a `RuntimeError` so consumers do not accidentally treat a partial run as complete.
-
-You can also call `flow.stream_events(...)` directly when you want streaming for a single invocation without setting `stream=True` on the Flow instance.
-
-## Filter by Channel
-
-`StreamSession` exposes channel projections that preserve global frame order within the selected channel:
-
-```python
-stream = flow.stream_events()
-
-with stream:
-    for frame in stream.llm:
-        print(frame.content, end="", flush=True)
-
-result = stream.result
-```
-
-Available projections are:
-
-| Projection | Frames |
-|------------|--------|
-| `stream.events` | All frames |
-| `stream.llm` | LLM frames |
-| `stream.messages` | Conversation message frames |
-| `stream.flow` | Flow frames |
-| `stream.tools` | Tool frames |
-| `stream.interleave([...])` | A selected set of channels |
-
-Use `stream.interleave(["flow", "llm", "messages"])` when a consumer wants only some channels but still needs their relative order.
-
-## Async Streaming
-
-Use `astream()` for async consumers:
-
-```python
-flow = ReportFlow()
-stream = flow.astream()
-
-async with stream:
-    async for chunk in stream.events:
-        print(chunk.channel, chunk.type, chunk.content)
-
-result = stream.result
-```
-
-The async session has the same projections as the sync session.
-
-## Stream a Direct LLM Call
-
-`llm.call(...)` still returns the final assembled result. Use `llm.stream_events(...)` when you want to iterate over chunks as they arrive while keeping the structured event payload:
-
-```python
-from crewai import LLM
-
-
-llm = LLM(model="gpt-4o-mini")
-stream = llm.stream_events(
-    messages=[
-        {
-            "role": "user",
-            "content": "Explain CrewAI streaming in two short sentences.",
-        }
-    ]
-)
-
-with stream:
-    for chunk in stream:
-        print(chunk.content, end="", flush=True)
-
-result = stream.result
-```
-
-`llm.stream_events(...)` temporarily enables streaming for the wrapped call and restores the LLM's previous `stream` setting afterward. Provider integrations continue to emit the underlying LLM stream events; this helper provides a common iterator API over those events for every LLM provider.
-
-## Conversational Turns
-
-Conversational Flows can stream one user turn with `stream_turn()`:
-
-```python
-from crewai import Flow
-from crewai.experimental.conversational import ConversationConfig, ConversationState
-
-
-@ConversationConfig(llm="gpt-4o-mini", defer_trace_finalization=True)
-class ChatFlow(Flow[ConversationState]):
-    conversational = True
-
-
-flow = ChatFlow()
-stream = flow.stream_turn("What can you help me with?", session_id="session-1")
-
-with stream:
-    for frame in stream.events:
-        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
-            print(frame.content, end="", flush=True)
-
-reply = stream.result
-```
-
-During `stream_turn()`, the built-in conversational answer path enables LLM token streaming for that turn and restores the LLM's previous `stream` setting afterward. Custom route handlers that create their own agents or LLM instances should configure those LLMs for streaming if they need token-level output.
-
-## Cleanup
-
-Use the session as a context manager when possible. If a client disconnects before the stream is exhausted, close the session explicitly:
-
-```python
-stream = flow.stream_events()
-
-try:
-    for frame in stream.events:
-        print(frame.type)
-finally:
-    if not stream.is_exhausted:
-        stream.close()
-```
-
-For async streams, use `await stream.aclose()`.
-
-## Legacy Chunk Streaming
-
-Crew streaming with `stream=True` still returns the chunk-oriented `CrewStreamingOutput` API described in [Streaming Crew Execution](/en/learn/streaming-crew-execution). Direct `llm.call(...)` still returns the final LLM result. The frame contract is intended for runtimes that need a stable event envelope across Flows, direct LLM calls, conversational turns, tools, and messages.

docs/edge/ko/changelog.mdx

@@ -4,60 +4,6 @@ description: "CrewAI의 제품 업데이트, 개선 사항 및 버그 수정"
 icon: "clock"
 mode: "wide"
 ---
-<Update label="2026년 7월 1일">
-  ## v1.15.2a2
-
-  [GitHub 릴리스 보기](https://github.com/crewAIInc/crewAI/releases/tag/1.15.2a2)
-
-  ## 변경 사항
-
-  ### 기능
-  - bedrock 추가에 aiobotocore 추가
-  - 흐름 에이전트 옵션 문서화
-  - 흐름 기술 예제에 텍스트 도우미 추가
-  - 흐름 CEL 프롬프트를 위한 텍스트 도우미 추가
-  - 탐색에 스트리밍 문서 추가
-
-  ### 버그 수정
-  - 자기 청취 흐름 메서드 거부
-
-  ### 문서
-  - v1.15.2a1에 대한 스냅샷 및 변경 로그 업데이트
-  - AGENTS.md 파일 압축
-
-  ## 기여자
-
-  @akaKuruma, @github-code-quality[bot], @lorenzejay, @vinibrsl
-
-</Update>
-
-<Update label="2026년 6월 30일">
-  ## v1.15.2a1
-
-  [GitHub 릴리스 보기](https://github.com/crewAIInc/crewAI/releases/tag/1.15.2a1)
-
-  ## 변경 사항
-
-  ### 기능
-  - 템플릿 명령을 crewAIInc-fde 조직으로 재지정
-  - 인라인 기술 정의 지원
-  - 흐름을 위한 스트림 프레임 프로토콜 정의
-  - CrewDefinition에 타입 도구 및 앱 추가
-  - 생성된 흐름 정의 저작 기술 추가
-
-  ### 버그 수정
-  - 새로운 페이지가 삭제되는 것을 방지하기 위해 Edge에서 문서 버전 탐색 제거
-
-  ### 문서
-  - 에이전트 제어 평면에서 비용 한도 규칙 유형 문서화
-  - Datadog 가이드에서 CREWAI_LOG_FORMAT 참조 제거
-
-  ## 기여자
-
-  @danielfsbarreto, @joaomdmoura, @lorenzejay, @lucasgomide, @vinibrsl
-
-</Update>
-
 <Update label="2026년 6월 26일">
   ## v1.15.1
 

docs/edge/ko/learn/streaming-runtime-contract.mdx

@@ -1,194 +0,0 @@
----
-title: 스트리밍 런타임 계약
-description: Flow, 직접 LLM 호출, 대화 턴에서 정렬된 런타임 프레임을 스트리밍합니다.
-icon: tower-broadcast
-mode: "wide"
----
-
-## 개요
-
-CrewAI는 단순한 텍스트 청크보다 더 많은 정보가 필요한 런타임을 위해 프레임 기반 스트리밍 계약을 제공합니다. 이 계약은 Flow 생명주기 이벤트, 직접 LLM 토큰, 도구 활동, 대화 메시지, 사용자 지정 이벤트에 대해 정렬된 `StreamFrame` 객체를 방출합니다.
-
-UI, 서비스 브리지, 터미널 앱, 배포 런타임을 만들 때 Flow, 채팅 턴, 직접 LLM 호출이 실행되는 동안 안정적인 구조화 이벤트 스트림이 필요하다면 이 API를 사용하세요.
-
-## StreamFrame
-
-모든 프레임은 같은 envelope를 가집니다:
-
-```python
-from crewai.types.streaming import StreamFrame
-
-frame.id           # 고유 프레임 id
-frame.seq          # 사용 가능한 경우 실행 로컬 순서
-frame.type         # "flow_started" 같은 원본 이벤트 타입
-frame.channel      # "llm", "flow", "tools", "messages", "lifecycle", "custom"
-frame.namespace    # 소스/런타임 namespace
-frame.timestamp    # 이벤트 timestamp
-frame.parent_id    # 사용 가능한 경우 부모 이벤트 id
-frame.previous_id  # 사용 가능한 경우 이전 이벤트 id
-frame.data         # 이벤트 payload
-frame.event        # frame.data의 alias
-frame.content      # 토큰류 프레임의 출력 가능한 텍스트, 그 외에는 ""
-```
-
-`channel` 필드는 소비자에서 프레임을 라우팅하는 가장 빠른 방법입니다:
-
-| 채널 | 포함 내용 |
-|------|-----------|
-| `llm` | LLM 스트리밍 이벤트의 토큰 및 thinking 청크 |
-| `flow` | Flow 생명주기, 메서드 실행, 라우팅, pause/resume 이벤트 |
-| `tools` | 도구 사용 이벤트 |
-| `messages` | 대화 transcript 이벤트 |
-| `lifecycle` | 다른 채널에 속하지 않는 런타임 생명주기 이벤트 |
-| `custom` | 내장 채널에 매핑되지 않는 이벤트 |
-
-`frame.type`은 원본 이벤트 타입을 보존하므로, 소비자는 채널 안에서 특정 이벤트를 처리할 수 있습니다.
-
-## Flow 스트리밍
-
-Flow에 `stream=True`를 설정하면 `kickoff()`가 stream session을 반환합니다:
-
-```python
-from crewai.flow import Flow, start
-
-
-class ReportFlow(Flow):
-    @start()
-    def generate(self):
-        return "done"
-
-
-flow = ReportFlow(stream=True)
-stream = flow.kickoff()
-
-with stream:
-    for chunk in stream:
-        print(chunk.content, end="", flush=True)
-        if chunk.type == "tool_usage_started":
-            print(chunk.event["tool_name"])
-
-result = stream.result
-```
-
-`stream.result`를 읽기 전에 stream을 소비해야 합니다. 결과를 너무 일찍 접근하면 `RuntimeError`가 발생하여, 소비자가 부분 실행을 완료된 실행으로 잘못 처리하지 않도록 합니다.
-
-Flow 인스턴스에 `stream=True`를 설정하지 않고 단일 호출만 스트리밍하려면 `flow.stream_events(...)`를 직접 호출할 수도 있습니다.
-
-## 채널별 필터링
-
-`StreamSession`은 선택한 채널 안에서 전역 프레임 순서를 보존하는 채널 projection을 제공합니다:
-
-```python
-stream = flow.stream_events()
-
-with stream:
-    for frame in stream.llm:
-        print(frame.content, end="", flush=True)
-
-result = stream.result
-```
-
-사용 가능한 projection은 다음과 같습니다:
-
-| Projection | 프레임 |
-|------------|--------|
-| `stream.events` | 모든 프레임 |
-| `stream.llm` | LLM 프레임 |
-| `stream.messages` | 대화 메시지 프레임 |
-| `stream.flow` | Flow 프레임 |
-| `stream.tools` | 도구 프레임 |
-| `stream.interleave([...])` | 선택한 채널 집합 |
-
-소비자가 일부 채널만 원하지만 상대 순서도 필요하다면 `stream.interleave(["flow", "llm", "messages"])`를 사용하세요.
-
-## 비동기 스트리밍
-
-비동기 소비자는 `astream()`을 사용하세요:
-
-```python
-flow = ReportFlow()
-stream = flow.astream()
-
-async with stream:
-    async for chunk in stream.events:
-        print(chunk.channel, chunk.type, chunk.content)
-
-result = stream.result
-```
-
-비동기 세션은 동기 세션과 같은 projection을 제공합니다.
-
-## 직접 LLM 호출 스트리밍
-
-`llm.call(...)`은 계속 최종 조립 결과를 반환합니다. 구조화된 이벤트 payload를 유지하면서 청크가 도착하는 대로 반복 처리하려면 `llm.stream_events(...)`를 사용하세요:
-
-```python
-from crewai import LLM
-
-
-llm = LLM(model="gpt-4o-mini")
-stream = llm.stream_events(
-    messages=[
-        {
-            "role": "user",
-            "content": "Explain CrewAI streaming in two short sentences.",
-        }
-    ]
-)
-
-with stream:
-    for chunk in stream:
-        print(chunk.content, end="", flush=True)
-
-result = stream.result
-```
-
-`llm.stream_events(...)`는 감싼 호출 동안 일시적으로 streaming을 활성화하고, 이후 LLM의 이전 `stream` 설정을 복원합니다. provider 통합은 계속 기본 LLM stream 이벤트를 방출하며, 이 helper는 모든 LLM provider에서 그 이벤트 위에 공통 iterator API를 제공합니다.
-
-## 대화 턴
-
-대화형 Flow는 `stream_turn()`으로 사용자 턴 하나를 스트리밍할 수 있습니다:
-
-```python
-from crewai import Flow
-from crewai.experimental.conversational import ConversationConfig, ConversationState
-
-
-@ConversationConfig(llm="gpt-4o-mini", defer_trace_finalization=True)
-class ChatFlow(Flow[ConversationState]):
-    conversational = True
-
-
-flow = ChatFlow()
-stream = flow.stream_turn("What can you help me with?", session_id="session-1")
-
-with stream:
-    for frame in stream.events:
-        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
-            print(frame.content, end="", flush=True)
-
-reply = stream.result
-```
-
-`stream_turn()` 중에는 내장 대화 응답 경로가 해당 턴에 대해 LLM 토큰 스트리밍을 활성화하고 이후 LLM의 이전 `stream` 설정을 복원합니다. 자체 agent 또는 LLM 인스턴스를 만드는 사용자 지정 route handler는 토큰 단위 출력이 필요하다면 해당 LLM을 streaming으로 구성해야 합니다.
-
-## 정리
-
-가능하면 세션을 context manager로 사용하세요. stream이 끝나기 전에 클라이언트 연결이 끊기면 세션을 명시적으로 닫으세요:
-
-```python
-stream = flow.stream_events()
-
-try:
-    for frame in stream.events:
-        print(frame.type)
-finally:
-    if not stream.is_exhausted:
-        stream.close()
-```
-
-비동기 stream에서는 `await stream.aclose()`를 사용하세요.
-
-## 레거시 청크 스트리밍
-
-`stream=True`를 사용하는 Crew 스트리밍은 계속 [스트리밍 Crew 실행](/ko/learn/streaming-crew-execution)에 설명된 청크 중심 `CrewStreamingOutput` API를 반환합니다. 직접 `llm.call(...)` 호출도 계속 최종 LLM 결과를 반환합니다. 프레임 계약은 Flow, 직접 LLM 호출, 대화 턴, 도구, 메시지 전반에서 안정적인 이벤트 envelope가 필요한 런타임을 위한 것입니다.

docs/edge/pt-BR/changelog.mdx

@@ -4,60 +4,6 @@ description: "Atualizações de produto, melhorias e correções do CrewAI"
 icon: "clock"
 mode: "wide"
 ---
-<Update label="01 jul 2026">
-  ## v1.15.2a2
-
-  [Ver release no GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.15.2a2)
-
-  ## O que Mudou
-
-  ### Recursos
-  - Adicionar aiobotocore ao extra bedrock
-  - Documentar opções do agente de fluxo
-  - Adicionar helper de texto ao exemplo de habilidade de fluxo
-  - Adicionar helper de texto para prompts CEL de fluxo
-  - Adicionar documentação de streaming à navegação
-
-  ### Correções de Bugs
-  - Rejeitar métodos de fluxo de autoescuta
-
-  ### Documentação
-  - Atualizar snapshot e changelog para v1.15.2a1
-  - Compactar arquivo AGENTS.md
-
-  ## Contribuidores
-
-  @akaKuruma, @github-code-quality[bot], @lorenzejay, @vinibrsl
-
-</Update>
-
-<Update label="30 jun 2026">
-  ## v1.15.2a1
-
-  [Ver release no GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.15.2a1)
-
-  ## O que Mudou
-
-  ### Recursos
-  - Reapontar comandos de template para a organização crewAIInc-fde
-  - Suporte a definições de habilidades inline
-  - Definir protocolo de quadro de fluxo para fluxos
-  - Adicionar ferramenta de tipo e aplicativo em CrewDefinition
-  - Adicionar habilidade de autoria de Definição de Fluxo gerada
-
-  ### Correções de Bugs
-  - Remover a navegação de versão de docs do Edge para evitar que novas páginas sejam descartadas
-
-  ### Documentação
-  - Documentar o tipo de regra Limite de Custo no Painel de Controle do Agente
-  - Remover referências a CREWAI_LOG_FORMAT do guia Datadog
-
-  ## Contribuidores
-
-  @danielfsbarreto, @joaomdmoura, @lorenzejay, @lucasgomide, @vinibrsl
-
-</Update>
-
 <Update label="26 jun 2026">
   ## v1.15.1
 

docs/edge/pt-BR/learn/streaming-runtime-contract.mdx

@@ -1,194 +0,0 @@
----
-title: Contrato de Streaming do Runtime
-description: Transmita frames ordenados do runtime a partir de Flows, chamadas diretas de LLM e turnos conversacionais.
-icon: tower-broadcast
-mode: "wide"
----
-
-## Visão geral
-
-O CrewAI expõe um contrato de streaming baseado em frames para runtimes que precisam de mais do que chunks de texto simples. O contrato emite objetos `StreamFrame` ordenados para eventos de ciclo de vida de Flow, tokens de LLM diretos, atividade de ferramentas, mensagens de conversa e eventos personalizados.
-
-Use esta API ao criar uma UI, ponte de serviço, aplicativo de terminal ou runtime de implantação que precise de um fluxo estável de eventos estruturados enquanto um Flow, turno de chat ou chamada direta de LLM está em execução.
-
-## StreamFrame
-
-Todo frame tem o mesmo envelope:
-
-```python
-from crewai.types.streaming import StreamFrame
-
-frame.id           # id único do frame
-frame.seq          # ordem local da execução, quando disponível
-frame.type         # tipo do evento de origem, como "flow_started"
-frame.channel      # "llm", "flow", "tools", "messages", "lifecycle" ou "custom"
-frame.namespace    # namespace de origem/runtime
-frame.timestamp    # timestamp do evento
-frame.parent_id    # id do evento pai, quando disponível
-frame.previous_id  # id do evento anterior, quando disponível
-frame.data         # payload do evento
-frame.event        # alias para frame.data
-frame.content      # texto imprimível para frames de token, caso contrário ""
-```
-
-O campo `channel` é a forma mais rápida de rotear frames em consumidores:
-
-| Canal | Contém |
-|-------|--------|
-| `llm` | Tokens e chunks de raciocínio de eventos de streaming de LLM |
-| `flow` | Ciclo de vida do Flow, execução de métodos, roteamento e eventos de pausa/retomada |
-| `tools` | Eventos de uso de ferramentas |
-| `messages` | Eventos do transcript da conversa |
-| `lifecycle` | Eventos de ciclo de vida do runtime que não pertencem a outro canal |
-| `custom` | Eventos que não mapeiam para um canal integrado |
-
-`frame.type` preserva o tipo do evento de origem, para que consumidores possam tratar eventos específicos dentro de um canal.
-
-## Transmitir um Flow
-
-Defina `stream=True` em um Flow para fazer `kickoff()` retornar uma sessão de stream:
-
-```python
-from crewai.flow import Flow, start
-
-
-class ReportFlow(Flow):
-    @start()
-    def generate(self):
-        return "done"
-
-
-flow = ReportFlow(stream=True)
-stream = flow.kickoff()
-
-with stream:
-    for chunk in stream:
-        print(chunk.content, end="", flush=True)
-        if chunk.type == "tool_usage_started":
-            print(chunk.event["tool_name"])
-
-result = stream.result
-```
-
-Você deve consumir o stream antes de ler `stream.result`. Acessar o resultado cedo demais gera um `RuntimeError`, para que consumidores não tratem uma execução parcial como concluída.
-
-Você também pode chamar `flow.stream_events(...)` diretamente quando quiser streaming para uma única invocação sem definir `stream=True` na instância do Flow.
-
-## Filtrar por canal
-
-`StreamSession` expõe projeções por canal que preservam a ordem global dos frames dentro do canal selecionado:
-
-```python
-stream = flow.stream_events()
-
-with stream:
-    for frame in stream.llm:
-        print(frame.content, end="", flush=True)
-
-result = stream.result
-```
-
-As projeções disponíveis são:
-
-| Projeção | Frames |
-|----------|--------|
-| `stream.events` | Todos os frames |
-| `stream.llm` | Frames de LLM |
-| `stream.messages` | Frames de mensagens de conversa |
-| `stream.flow` | Frames de Flow |
-| `stream.tools` | Frames de ferramentas |
-| `stream.interleave([...])` | Um conjunto selecionado de canais |
-
-Use `stream.interleave(["flow", "llm", "messages"])` quando um consumidor quiser apenas alguns canais, mas ainda precisar da ordem relativa entre eles.
-
-## Streaming assíncrono
-
-Use `astream()` para consumidores assíncronos:
-
-```python
-flow = ReportFlow()
-stream = flow.astream()
-
-async with stream:
-    async for chunk in stream.events:
-        print(chunk.channel, chunk.type, chunk.content)
-
-result = stream.result
-```
-
-A sessão assíncrona tem as mesmas projeções da sessão síncrona.
-
-## Transmitir uma chamada direta de LLM
-
-`llm.call(...)` ainda retorna o resultado final montado. Use `llm.stream_events(...)` quando quiser iterar pelos chunks conforme eles chegam, mantendo o payload estruturado do evento:
-
-```python
-from crewai import LLM
-
-
-llm = LLM(model="gpt-4o-mini")
-stream = llm.stream_events(
-    messages=[
-        {
-            "role": "user",
-            "content": "Explain CrewAI streaming in two short sentences.",
-        }
-    ]
-)
-
-with stream:
-    for chunk in stream:
-        print(chunk.content, end="", flush=True)
-
-result = stream.result
-```
-
-`llm.stream_events(...)` ativa temporariamente o streaming para a chamada encapsulada e restaura a configuração anterior de `stream` do LLM depois. As integrações de provedores continuam emitindo os eventos de stream de LLM subjacentes; esse helper fornece uma API de iterador comum sobre esses eventos para todos os provedores de LLM.
-
-## Turnos conversacionais
-
-Flows conversacionais podem transmitir um turno de usuário com `stream_turn()`:
-
-```python
-from crewai import Flow
-from crewai.experimental.conversational import ConversationConfig, ConversationState
-
-
-@ConversationConfig(llm="gpt-4o-mini", defer_trace_finalization=True)
-class ChatFlow(Flow[ConversationState]):
-    conversational = True
-
-
-flow = ChatFlow()
-stream = flow.stream_turn("What can you help me with?", session_id="session-1")
-
-with stream:
-    for frame in stream.events:
-        if frame.channel == "llm" and frame.type == "llm_stream_chunk":
-            print(frame.content, end="", flush=True)
-
-reply = stream.result
-```
-
-Durante `stream_turn()`, o caminho de resposta conversacional integrado ativa o streaming de tokens de LLM para esse turno e restaura a configuração anterior de `stream` do LLM depois. Handlers de rota personalizados que criam seus próprios agentes ou instâncias de LLM devem configurar esses LLMs para streaming se precisarem de saída em nível de token.
-
-## Limpeza
-
-Use a sessão como gerenciador de contexto quando possível. Se um cliente se desconectar antes de o stream ser esgotado, feche a sessão explicitamente:
-
-```python
-stream = flow.stream_events()
-
-try:
-    for frame in stream.events:
-        print(frame.type)
-finally:
-    if not stream.is_exhausted:
-        stream.close()
-```
-
-Para streams assíncronos, use `await stream.aclose()`.
-
-## Streaming de chunks legado
-
-O streaming de Crew com `stream=True` ainda retorna a API orientada a chunks `CrewStreamingOutput` descrita em [Streaming da Execução de Crew](/pt-BR/learn/streaming-crew-execution). Chamadas diretas `llm.call(...)` ainda retornam o resultado final do LLM. O contrato de frames é destinado a runtimes que precisam de um envelope de evento estável em Flows, chamadas diretas de LLM, turnos conversacionais, ferramentas e mensagens.

lib/cli/pyproject.toml

@@ -8,7 +8,7 @@ authors = [
 ]
 requires-python = ">=3.10, <3.14"
 dependencies = [
-    "crewai-core==1.15.2a2",
+    "crewai-core==1.15.1",
     "click>=8.1.7,<9",
     "pydantic>=2.11.9,<2.13",
     "pydantic-settings~=2.10.1",

lib/cli/src/crewai_cli/__init__.py

@@ -1 +1 @@
-__version__ = "1.15.2a2"
+__version__ = "1.15.1"

lib/cli/src/crewai_cli/create_flow.py

@@ -126,7 +126,10 @@ def _create_declarative_flow(
 
     package_dir = Path(__file__).parent
     templates_dir = package_dir / "templates" / "declarative_flow"
-    root_template_files = {".gitignore", "AGENTS.md", "README.md", "pyproject.toml"}
+
+    agents_md_src = package_dir / "templates" / "AGENTS.md"
+    if agents_md_src.exists():
+        shutil.copy2(agents_md_src, project_root / "AGENTS.md")
 
     for src_file in templates_dir.rglob("*"):
         if not src_file.is_file():
@@ -135,7 +138,7 @@ def _create_declarative_flow(
         relative_path = src_file.relative_to(templates_dir)
         dst_file = (
             project_root / relative_path
-            if relative_path.name in root_template_files
+            if relative_path.name in {".gitignore", "README.md", "pyproject.toml"}
             else package_root / relative_path
         )
         dst_file.parent.mkdir(parents=True, exist_ok=True)

lib/cli/src/crewai_cli/remote_template/main.py

@@ -17,7 +17,7 @@ from crewai_cli.command import BaseCommand
 logger = logging.getLogger(__name__)
 console = Console()
 
-GITHUB_ORG = "crewAIInc-fde"
+GITHUB_ORG = "crewAIInc"
 TEMPLATE_PREFIX = "template_"
 GITHUB_API_BASE = "https://api.github.com"
 
@@ -218,7 +218,7 @@ class TemplateCommand(BaseCommand):
     def _extract_zip(self, zip_bytes: bytes, dest: str) -> None:
         """Extract a GitHub zipball into dest, stripping the top-level directory."""
         with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
-            # GitHub zipballs have a single top-level dir like 'crewAIInc-fde-template_xxx-<sha>/'
+            # GitHub zipballs have a single top-level dir like 'crewAIInc-template_xxx-<sha>/'
             members = zf.namelist()
             if not members:
                 click.secho("Downloaded archive is empty.", fg="red")

lib/cli/src/crewai_cli/templates/declarative_flow/AGENTS.md

@@ -1,352 +0,0 @@
-# Flow Definition
-
-You are writing a CrewAI Flow declaration for the user.
-Use these instructions when the user asks you to create or edit a Flow.
-Return one valid `crewai.flow/v1` YAML or JSON document.
-
-Treat this document as instructions for you, not as text to show the user.
-Follow the examples for shape and formatting, then use the API reference to check exact fields.
-
-## Output Format
-
-Return one valid `crewai.flow/v1` Flow declaration.
-Do not include explanatory prose unless the user asks for it.
-
-## Build It In This Order
-
-1. Define `state` first. Use `type: json_schema` and put the JSON Schema inline.
-2. Put required input fields in `state.json_schema.required`. Do not rely on `state.default` to make fields required.
-3. Add exactly one method with `start: true`.
-4. Add later methods with `listen`.
-5. Give each method exactly one `do` action object. Never make `do` a list.
-6. Pass data with `${...}` mappings from `state` and completed `outputs`.
-7. Before final output, check every `listen`, `emit`, and `outputs.some_method` reference.
-
-Set optional fields only when you are confident they are needed. Otherwise, trust CrewAI defaults and omit them.
-
-Method names must match `^[A-Za-z_][A-Za-z0-9_]*$`.
-
-## Choose One Action Per Method
-
-Pick the simplest action that does the job.
-
-- Use `call: expression` for simple reads, filters, computed values, and deterministic routing.
-- Use `call: agent` for one AI worker that classifies, decides, summarizes, writes, or drafts. Put `role`, `goal`, `backstory`, and `input` under `with`. Do not add an action-level `inputs` map to an agent.
-- Use `call: crew` for coordinated AI work with multiple agents or tasks. Define the crew under `with`. Pass runtime values with the action-level `inputs` map.
-
-## Wire Methods Explicitly
-
-- `state` is the initial shared data shape. Action results do not automatically merge into `state`.
-- Read method results with `outputs.method_name` after that method can run.
-- `listen` targets a method name or a router-emitted event name.
-- Methods must not listen to their own method name.
-- Method names and emitted event names share one namespace. Avoid reusing the same string for both unless the user explicitly wants that.
-- Use `router: true` plus `emit` when one method chooses between named branches.
-- A router action must return exactly one emitted event string. It must not return JSON, a list, or an explanation.
-- Use `start: true` for the single entrypoint.
-
-If an agent is a router, make its goal say exactly what to return, for example:
-`Return exactly one bare value: approved, rejected, or needs_review. Do not include explanation.`
-Prefer `call: expression` when routing can be computed without an agent.
-
-## CEL And Dynamic Values
-
-CEL is the expression language for reading Flow data and making small decisions.
-Use agents and crews for larger work or side effects.
-
-Use these expression forms correctly:
-
-- Raw CEL: use in `expr`. Do not wrap raw CEL in `${...}`.
-- Use `${...}` inside action mapping strings to read Flow data with CEL. Example value: `Ticket: ${state.ticket_id}`.
-- Use `state` for input data. Use `outputs.step_name` for a completed method result.
-- If a value is only one `${...}` expression, the result keeps its type. Use this for numbers, booleans, objects, and lists.
-- If the string has other text, the final value is text. Non-text values become JSON. `null` becomes empty text.
-- Use `text(root, "path", "default")` for values that may be missing or null. The default is optional and is `""`.
-
-Expression examples:
-
-Mix text and Flow data:
-
-```yaml
-query: "News about ${state.topic}"
-```
-
-Keep a list or number type:
-
-```yaml
-domains: "${state.domains}"
-limit: "${state.limit}"
-```
-
-Use a default for missing text:
-
-```yaml
-input: "Ticket ${text(state, \"ticket.id\", \"unknown\")}"
-```
-
-- Crew text: use `{name}` placeholders from crew inputs. Example: `Research {topic}`.
-- Crew inputs become prompt text only when agent or task text references matching `{name}` placeholders.
-- Passing an input that is not referenced by any `{name}` placeholder does not ground the crew. If the crew needs a field, put that placeholder in an agent `goal`, task `description`, or task `expected_output`.
-
-Available CEL variables:
-
-- `state`: initial input data, for example `state.ticket.subject`.
-- `outputs`: completed method outputs, for example `outputs.classify_ticket`.
-
-Dynamic value rules:
-
-- When an agent needs multiple fields, write one text value with labels and separators. Example value: `Ticket ID: ${state.ticket_id}; Message: ${state.message}`.
-- Crew action-level `inputs` are the actual Crew kickoff inputs. Use `${...}` values there for runtime data from `state` or `outputs`.
-- Crew action-level `inputs` alone are not grounding. Include placeholders for the facts the model must use.
-- Crew outputs are objects. Use `${outputs.research_brief.raw}` for text.
-- For structured crew output, use fields like `${outputs.research_brief.json_dict.field}` or `${outputs.research_brief.pydantic.field}`.
-- Do not pass a whole crew output to an agent input, like `${outputs.research_brief}`.
-- Agent outputs may also be objects. Use fields like `${outputs.classify_ticket.raw}` or `${outputs.classify_ticket.pydantic.category}`.
-- Use `with.inputs` only for static Crew input defaults.
-- Agent action `with.input` is the agent's single input value.
-
-## Do Not
-
-- Do not invent top-level keys outside the Flow declaration shape.
-- Do not use fields outside the declaration schema.
-- Do not put more than one action under a method's `do`.
-- Do not make `do` a list.
-- Do not reference `outputs.some_method` before `some_method` can run.
-- Do not set a method's `listen` to its own method name.
-- Do not use the same string for an emitted event and a method name unless the user asks for it.
-- Do not use `emit` without `router: true`.
-- Do not rely on crew action-level `inputs` alone to ground agent behavior. Inputs that do not match placeholders are effectively unused by the prompt.
-- Do not ask agents to infer missing facts when accuracy matters. Tell them to mark missing dates, amounts, offers, logs, or constraints as unknown.
-- Do not set `config.stream: true` unless the caller is expected to consume a streaming result. For normal generated flows and CLI smoke tests, omit it.
-
-## Examples
-
-### Crew review with routed follow-up
-
-```yaml
-schema: crewai.flow/v1
-name: ResearchReviewFlow
-state:
-  type: json_schema
-  json_schema:
-    type: object
-    properties:
-      topic:
-        type: string
-      audience:
-        type: string
-    required:
-      - topic
-      - audience
-  default:
-    topic: AI agent orchestration
-    audience: platform engineering leaders
-methods:
-  research_brief:
-    start: true
-    do:
-      call: crew
-      with:
-        agents:
-          researcher:
-            role: Research analyst
-            goal: Research {topic} for {audience}
-            backstory: Expert at concise technical research.
-          reviewer:
-            role: Strategy reviewer
-            goal: Decide whether the research needs an executive follow-up
-            backstory: Experienced at reviewing technical briefs for leaders.
-        tasks:
-          - name: research_task
-            description: Research {topic} for {audience}.
-            expected_output: Key findings and tradeoffs.
-            agent: researcher
-          - name: review_task
-            description: Review the research and decide if an executive follow-up is needed.
-            expected_output: 'A brief review ending with `needs_followup: true` or `needs_followup: false`.'
-            agent: reviewer
-        inputs:
-          topic: Default topic
-          audience: Default audience
-      inputs:
-        topic: "${state.topic}"
-        audience: "${state.audience}"
-  route_followup:
-    listen: research_brief
-    router: true
-    emit:
-      - followup
-      - done
-    do:
-      call: agent
-      with:
-        role: Follow-up router
-        goal: 'Return exactly one bare value: followup or done. Do not include explanation.'
-        backstory: Skilled at routing reviewed research briefs.
-        input: "Reviewed research: ${outputs.research_brief.raw}"
-  write_followup:
-    listen: followup
-    do:
-      call: agent
-      with:
-        role: Executive communications specialist
-        goal: Draft a concise executive follow-up from the reviewed research
-        backstory: Writes crisp follow-ups for technical leaders.
-        input: "${outputs.research_brief.raw}"
-```
-
-## API Reference
-
-Use this appendix to check exact field names, required fields, linked object types, and allowed action/state shapes. Linked type names point to another section in this reference.
-
-### Flow Definition
-
-Fields:
-- `schema` (optional): must be `crewai.flow/v1`; default `crewai.flow/v1`. Declarative Flow schema identifier and version. Include it explicitly in authored declarations.
-- `name` (required): string. Unique flow name used in logs, events, and traces.
-- `description` (optional): string | null; default `null`. Human-readable summary of the flow.
-- `state` (required): [State](#json-schema-state-statetypejson_schema). State contract for the initial state and updates during execution.
-- `config` (optional): [Config (`config`)](#config-config); default generated default. Serializable flow-level execution configuration.
-- `methods` (required): map of string to [Method](#method-methods). Mapping of method names to method definitions.
-
-### JSON Schema State (`state[type=json_schema]`)
-
-Shape:
-- `type: json_schema`
-
-Fields:
-- `type` (optional): must be `json_schema`; default `json_schema`. Inline JSON Schema used as the Flow state contract.
-- `json_schema` (required): map of string to any. JSON Schema used to validate and document flow state. Declare required fields with JSON Schema's `required` array.
-- `default` (optional): map of string to any | null; default `null`. Default values used to initialize Flow state. Defaults are not the same as schema-required fields.
-
-### Method (`methods.<name>`)
-
-Fields:
-- `description` (optional): string | null; default `null`. Human-readable summary of what this method does.
-- `do` (required): [Action](#action). Single action object executed when this method runs.
-- `start` (optional): boolean | string | map of string to any | null; default `null`. Marks the single normal entrypoint. Use `true`.
-- `listen` (optional): string | map of string to any | null; default `null`. Runs this method after one upstream method or router-emitted event.
-- `router` (optional): boolean; default `false`. Whether the method output should be treated as the next event name. Router actions must return one event name string, with no surrounding explanation.
-- `emit` (optional): list[string] | null; default `null`. Declared router events this method may emit. Each emitted event name should be unique and should not collide with method names.
-
-### Action
-
-Discriminated union by `call`.
-
-Allowed shapes:
-- [`call: crew`](#crew-action-methodsdocallcrew)
-- [`call: agent`](#agent-action-methodsdocallagent)
-- [`call: expression`](#expression-action-methodsdocallexpression)
-
-### Crew Action (`methods.<name>.do[call=crew]`)
-
-Shape:
-- `call: crew`
-
-Fields:
-- `call` (required): must be `crew`. Action discriminator. Use crew to run an inline Crew definition. Example: `crew`
-- `with` (required): inline crew definition. Inline Crew definition to load and execute for this action. Example: `{"agents": {"researcher": {"backstory": "Knows the domain.", "goal": "Research {topic}", "role": "Researcher"}}, "name": "inline_research", "tasks": [{"agent": "researcher", "description": "Research {topic}", "expected_output": "Findings about {topic}", "name": "research_task"}]}`
-- `inputs` (optional): map of string to expression data | null; default `null`. Actual kickoff inputs passed to the Crew. Use `${...}` inside action mapping strings to read Flow data with CEL. Example value: `Ticket: ${state.ticket_id}`. Use `state` for input data. Use `outputs.step_name` for a completed method result. If a value is only one `${...}` expression, the result keeps its type. Use this for numbers, booleans, objects, and lists. If the string has other text, the final value is text. Non-text values become JSON. `null` becomes empty text. Use `text(root, "path", "default")` for values that may be missing or null. The default is optional and is `""`. The evaluated values are available to crew agent and task interpolation as `{name}` placeholders; reference each input the crew needs in agent or task text. Example: `{"topic": "${state.topic}"}`
-
-#### Crew Definition (`methods.<name>.do[call=crew].with`)
-
-Fields:
-- `agents` (required): map of string to any | list[map of string to any]. Inline crew agents keyed by agent name. Example: `{"researcher": {"backstory": "Expert at concise technical research.", "goal": "Research {topic}", "role": "Research analyst"}}`
-- `tasks` (required): list[any]. Ordered crew tasks. Example: `[{"agent": "researcher", "description": "Research {topic}.", "expected_output": "Key findings about {topic}.", "name": "research_task"}]`
-- `inputs` (optional): map of string to any. Static default crew inputs. Values are available to crew agent and task interpolation as `{name}` placeholders, for example `{topic}`. Prefer action-level crew `inputs` for runtime values from `state` or `outputs`, and include placeholders for any inputs the crew must reason over. Example: `{"topic": "AI agents"}`
-
-#### Crew Agent Definition (`methods.<name>.do[call=crew].with.agents.<name>`)
-
-Fields:
-- `role` (required): string. Crew agent role. Crew inputs are interpolated with `{name}` placeholders such as `{topic}`; this is not CEL. Example: `Research analyst`
-- `goal` (required): string. Crew agent goal. Crew inputs are interpolated with `{name}` placeholders such as `{topic}`; this is not CEL. Example: `Research {topic}`
-- `backstory` (required): string. Crew agent backstory. Crew inputs are interpolated with `{name}` placeholders such as `{topic}`; this is not CEL. Example: `Expert at concise technical research.`
-- `settings` (optional): map of string to any. Additional agent settings passed to the loader. Example: `{"llm": "openai/gpt-4o-mini"}`
-- `llm` (optional): string or inline LLM config; default `null`. Language model that runs this crew agent. Use an object when setting LLM options such as `max_tokens`. Example: `{"max_tokens": 4096, "model": "openai/gpt-4o-mini"}`
-- `planning_config` (optional): object | null; default `null`. Agent planning configuration. Set `max_attempts` to limit planning refinement attempts before task execution. Example: `{"max_attempts": 3}`
-- `allow_delegation` (optional): boolean | null; default `null`. Enable agent to delegate and ask questions among each other. Example: `false`
-- `max_iter` (optional): integer | null; default `null`. Maximum iterations for an agent to execute a task Example: `25`
-- `max_rpm` (optional): integer | null; default `null`. Maximum number of requests per minute for the agent execution to be respected. Example: `10`
-- `max_execution_time` (optional): integer | null; default `null`. Maximum execution time in seconds for an agent to execute a task Example: `300`
-- `tools` (optional): list[string | map of string to any] | null; default `null`. Tool refs or serialized tool definitions available to this agent. String refs can use CrewAI tool names, `custom:<name>`, or fully qualified `module:Class` references. Example: `["crewai_tools:SerperDevTool", "custom:file_read"]`
-- `apps` (optional): list[string] | null; default `null`. Platform apps available to this agent. Can contain app names such as `gmail` or app/action refs such as `gmail/send_email`. Example: `["gmail", "slack/send_message"]`
-- `mcps` (optional): list[string | map of string to any] | null; default `null`. MCP server refs or serialized MCP server configs available to this agent. String refs can use HTTPS URLs, connected MCP integration slugs, or refs with a `#tool_name` suffix for specific tools. Example: `["https://api.weather.com/mcp#get_current_weather", "snowflake", "stripe#list_invoices", {"cache_tools_list": true, "headers": {"Authorization": "Bearer your_token"}, "streamable": true, "url": "https://api.example.com/mcp"}]`
-
-#### LLM Definition
-
-Fields:
-- `model` (required): string. Model identifier used to instantiate the LLM. Example: `openai/gpt-4o-mini`
-- `max_tokens` (optional): integer | null; default `null`. Maximum number of tokens the LLM can generate. If null, CrewAI does not set an explicit output token cap and the provider's default applies. Example: `4096`
-
-#### Crew Task Definition (`methods.<name>.do[call=crew].with.tasks[]`)
-
-Fields:
-- `description` (required): string. Task instructions. Crew inputs are interpolated with `{name}` placeholders such as `{topic}`; this is not CEL. Example: `Research {topic}.`
-- `expected_output` (required): string. Expected task output. Crew inputs are interpolated with `{name}` placeholders such as `{topic}`; this is not CEL. Example: `Key findings about {topic}.`
-- `name` (optional): string | null; default `null`. Optional task name. Example: `research_task`
-- `agent` (optional): string | null; default `null`. Name of the crew agent assigned to this task. Example: `researcher`
-
-### Agent Action (`methods.<name>.do[call=agent]`)
-
-Shape:
-- `call: agent`
-
-Fields:
-- `call` (required): must be `agent`. Action discriminator. Use agent to run an individual inline Agent definition outside of a crew. Example: `agent`
-- `with` (required): any. Individual Agent definition to load and execute outside of a crew for this action. Put the agent input in `with.input`; agent actions do not support action-level `inputs`. Example: `{"backstory": "Precise and concise.", "goal": "Answer user questions", "input": "${state.question}", "role": "Analyst", "settings": {"llm": "openai/gpt-4o-mini"}}`
-
-#### Agent Definition (`methods.<name>.do[call=agent].with`)
-
-Fields:
-- `role` (required): string. Individual agent role used by a Flow agent action outside of a crew. Example: `Support specialist`
-- `goal` (required): string. Individual agent goal for the Flow agent action outside of a crew. Example: `Draft a concise customer reply`
-- `backstory` (required): string. Individual agent backstory used to shape behavior outside of a crew. Example: `Expert at resolving SaaS support questions.`
-- `settings` (optional): map of string to any. Additional agent settings passed to the loader. Example: `{"llm": "openai/gpt-4o-mini"}`
-- `llm` (optional): string or inline LLM config; default `null`. Language model that runs this agent. Use an object when setting LLM options such as `max_tokens`. Example: `{"max_tokens": 4096, "model": "openai/gpt-4o-mini"}`
-- `planning_config` (optional): object | null; default `null`. Agent planning configuration. Set `max_attempts` to limit planning refinement attempts before task execution. Example: `{"max_attempts": 3}`
-- `allow_delegation` (optional): boolean | null; default `null`. Enable agent to delegate and ask questions among each other. Example: `false`
-- `max_iter` (optional): integer | null; default `null`. Maximum iterations for an agent to execute a task Example: `25`
-- `max_rpm` (optional): integer | null; default `null`. Maximum number of requests per minute for the agent execution to be respected. Example: `10`
-- `max_execution_time` (optional): integer | null; default `null`. Maximum execution time in seconds for an agent to execute a task Example: `300`
-- `tools` (optional): list[string | map of string to any] | null; default `null`. Tool refs or serialized tool definitions available to this agent. String refs can use CrewAI tool names, `custom:<name>`, or fully qualified `module:Class` references. Example: `["crewai_tools:SerperDevTool", "custom:file_read"]`
-- `apps` (optional): list[string] | null; default `null`. Platform apps available to this agent. Can contain app names such as `gmail` or app/action refs such as `gmail/send_email`. Example: `["gmail", "slack/send_message"]`
-- `mcps` (optional): list[string | map of string to any] | null; default `null`. MCP server refs or serialized MCP server configs available to this agent. String refs can use HTTPS URLs, connected MCP integration slugs, or refs with a `#tool_name` suffix for specific tools. Example: `["https://api.weather.com/mcp#get_current_weather", "snowflake", "stripe#list_invoices", {"cache_tools_list": true, "headers": {"Authorization": "Bearer your_token"}, "streamable": true, "url": "https://api.example.com/mcp"}]`
-- `input` (required): string. Input passed to the individual agent kickoff outside of a crew. Use one string. Use `${...}` inside action mapping strings to read Flow data with CEL. Example value: `Ticket: ${state.ticket_id}`. Use `state` for input data. Use `outputs.step_name` for a completed method result. If a value is only one `${...}` expression, the result keeps its type. Use this for numbers, booleans, objects, and lists. If the string has other text, the final value is text. Non-text values become JSON. `null` becomes empty text. Use `text(root, "path", "default")` for values that may be missing or null. The default is optional and is `""`. When an agent needs multiple fields, write one string with labels and separators, for example `Ticket ID: ${state.ticket_id}; Message: ${state.message}`. Example: `${state.ticket.body}`
-
-#### LLM Definition
-
-Fields:
-- `model` (required): string. Model identifier used to instantiate the LLM. Example: `openai/gpt-4o-mini`
-- `max_tokens` (optional): integer | null; default `null`. Maximum number of tokens the LLM can generate. If null, CrewAI does not set an explicit output token cap and the provider's default applies. Example: `4096`
-
-### Expression Action (`methods.<name>.do[call=expression]`)
-
-Shape:
-- `call: expression`
-
-Fields:
-- `call` (required): must be `expression`. Action discriminator. Use expression to evaluate a CEL expression.
-- `expr` (required): string. CEL expression evaluated against state, outputs, and local context.
-
-### Config (`config`)
-
-Fields:
-- `tracing` (optional): boolean | null; default `null`. Override for flow tracing; when omitted, execution defaults apply.
-- `stream` (optional): boolean; default `false`. Whether the flow should emit streaming events when supported.
-- `memory` (optional): map of string to any | null; default `null`. Serializable memory configuration passed to flow execution.
-- `input_provider` (optional): string | null; default `null`. Provider key used to supply initial state.
-- `suppress_flow_events` (optional): boolean; default `false`. Disable flow event emission for this definition.
-- `max_method_calls` (optional): integer; default `100`. Maximum number of method executions allowed during one kickoff.
-- `defer_trace_finalization` (optional): boolean; default `false`. Defer trace finalization so callers can complete tracing later.
-- `checkpoint` (optional): boolean | map of string to any | null; default `null`. Checkpointing configuration, or true to use default checkpointing.
-
-### Cross-Field Rules
-
-- A method has exactly one `do` action object with one `call` discriminator.
-- `listen` targets method names and router-emitted event names in one shared namespace.
-- Methods cannot listen to their own method name.
-- A router method result must match one declared `emit` value.
-- Crew action-level `inputs` are the Crew kickoff inputs; use CEL-wrapped strings there for runtime values.
-- Crew agent/task interpolation uses `{name}` placeholders from evaluated crew inputs.
-- Agent `with.input` must be text. Use `${outputs.method_name.raw}` or a text field like `${outputs.method_name.json_dict.summary}`.
-

lib/cli/tests/test_create_flow.py

@@ -26,15 +26,6 @@ def test_create_flow_declarative_project_can_run(
     assert pyproject["project"]["requires-python"]
     assert pyproject["project"]["dependencies"]
     assert (project_root / pyproject["tool"]["crewai"]["definition"]).is_file()
-    agents_md = (project_root / "AGENTS.md").read_text(encoding="utf-8")
-    assert "CrewAI Flow declaration" in agents_md
-    assert "schema: crewai.flow/v1" in agents_md
-    assert 'text(root, "path", "default")' in agents_md
-    assert "call: expression" in agents_md
-    assert "call: tool" not in agents_md
-    assert "call: script" not in agents_md
-    assert "call: each" not in agents_md
-    assert "human_feedback" not in agents_md
 
     monkeypatch.chdir(project_root)
     result = CliRunner().invoke(crewai, ["run"], env={"UV_RUN_RECURSION_DEPTH": "1"})

lib/cli/tests/test_plus_api.py

@@ -1,6 +1,8 @@
 import os
 import unittest
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import ANY, AsyncMock, MagicMock, patch
+
+import pytest
 
 from crewai_cli.plus_api import PlusAPI
 
@@ -341,23 +343,28 @@ class TestPlusAPI(unittest.TestCase):
         )
 
 
-@patch("crewai_core.plus_api.PlusAPI._make_request")
-def test_get_agent(mock_make_request):
+@pytest.mark.asyncio
+@patch("httpx.AsyncClient")
+async def test_get_agent(mock_async_client_class):
     api = PlusAPI("test_api_key")
     mock_response = MagicMock()
-    mock_make_request.return_value = mock_response
+    mock_client_instance = AsyncMock()
+    mock_client_instance.get.return_value = mock_response
+    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
 
-    response = api.get_agent("test_agent_handle")
+    response = await api.get_agent("test_agent_handle")
 
-    mock_make_request.assert_called_once_with(
-        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
+    mock_client_instance.get.assert_called_once_with(
+        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
+        headers=api.headers,
     )
     assert response == mock_response
 
 
-@patch("crewai_core.plus_api.PlusAPI._make_request")
+@pytest.mark.asyncio
+@patch("httpx.AsyncClient")
 @patch("crewai_core.plus_api.Settings")
-def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
+async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_class):
     org_uuid = "test-org-uuid"
     mock_settings = MagicMock()
     mock_settings.org_uuid = org_uuid
@@ -367,12 +374,15 @@ def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
     api = PlusAPI("test_api_key")
 
     mock_response = MagicMock()
-    mock_make_request.return_value = mock_response
+    mock_client_instance = AsyncMock()
+    mock_client_instance.get.return_value = mock_response
+    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
 
-    response = api.get_agent("test_agent_handle")
+    response = await api.get_agent("test_agent_handle")
 
-    mock_make_request.assert_called_once_with(
-        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
+    mock_client_instance.get.assert_called_once_with(
+        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
+        headers=api.headers,
     )
     assert "X-Crewai-Organization-Id" in api.headers
     assert api.headers["X-Crewai-Organization-Id"] == org_uuid

lib/crewai-core/src/crewai_core/__init__.py

@@ -1 +1 @@
-__version__ = "1.15.2a2"
+__version__ = "1.15.1"

lib/crewai-core/src/crewai_core/plus_api.py

@@ -232,8 +232,10 @@ class PlusAPI:
     def get_tool(self, handle: str) -> httpx.Response:
         return self._make_request("GET", f"{self.TOOLS_RESOURCE}/{handle}")
 
-    def get_agent(self, handle: str) -> httpx.Response:
-        return self._make_request("GET", f"{self.AGENTS_RESOURCE}/{handle}")
+    async def get_agent(self, handle: str) -> httpx.Response:
+        url = urljoin(self.base_url, f"{self.AGENTS_RESOURCE}/{handle}")
+        async with httpx.AsyncClient() as client:
+            return await client.get(url, headers=cast(dict[str, str], self.headers))
 
     def publish_tool(
         self,

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

@@ -152,4 +152,4 @@ __all__ = [
     "wrap_file_source",
 ]
 
-__version__ = "1.15.2a2"
+__version__ = "1.15.1"

lib/crewai-tools/pyproject.toml

@@ -10,7 +10,7 @@ requires-python = ">=3.10, <3.14"
 dependencies = [
     "pytube~=15.0.0",
     "requests>=2.33.0,<3",
-    "crewai==1.15.2a2",
+    "crewai==1.15.1",
     "tiktoken>=0.8.0,<0.13",
     "beautifulsoup4~=4.13.4",
     "python-docx~=1.2.0",

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

@@ -330,4 +330,4 @@ __all__ = [
     "ZapierActionTools",
 ]
 
-__version__ = "1.15.2a2"
+__version__ = "1.15.1"

lib/crewai-tools/tests/tools/stagehand_tool_test.py

@@ -164,7 +164,13 @@ def test_navigate_command(mock_run, stagehand_tool):
     )
 
     # Assertions
-    assert "https://example.com" in result
+    assert result == "Successfully navigated to https://example.com"
+    mock_run.assert_called_once_with(
+        stagehand_tool,
+        instruction="Go to example.com",
+        url="https://example.com",
+        command_type="navigate",
+    )
 
 
 @patch(

lib/crewai/pyproject.toml

@@ -8,8 +8,8 @@ authors = [
 ]
 requires-python = ">=3.10, <3.14"
 dependencies = [
-    "crewai-core==1.15.2a2",
-    "crewai-cli==1.15.2a2",
+    "crewai-core==1.15.1",
+    "crewai-cli==1.15.1",
     # Core Dependencies
     "pydantic>=2.11.9,<2.13",
     "openai>=2.30.0,<3",
@@ -55,7 +55,7 @@ Repository = "https://github.com/crewAIInc/crewAI"
 
 [project.optional-dependencies]
 tools = [
-    "crewai-tools==1.15.2a2",
+    "crewai-tools==1.15.1",
 ]
 embeddings = [
     "tiktoken>=0.8.0,<0.13"
@@ -92,7 +92,6 @@ litellm = [
 ]
 bedrock = [
     "boto3~=1.42.90",
-    "aiobotocore~=3.5.0",
 ]
 google-genai = [
     "google-genai~=1.65.0",

lib/crewai/src/crewai/__init__.py

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

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

@@ -73,6 +73,7 @@ from crewai.events.types.memory_events import (
     MemoryRetrievalFailedEvent,
     MemoryRetrievalStartedEvent,
 )
+from crewai.events.types.skill_events import SkillActivatedEvent
 from crewai.experimental.agent_executor import AgentExecutor
 from crewai.knowledge.knowledge import Knowledge
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
@@ -81,8 +82,8 @@ from crewai.llms.base_llm import BaseLLM
 from crewai.mcp.config import MCPServerConfig
 from crewai.rag.embeddings.types import EmbedderConfig
 from crewai.security.fingerprint import Fingerprint
-from crewai.skills.loader import load_skills
-from crewai.skills.models import Skill as SkillModel
+from crewai.skills.loader import activate_skill, discover_skills
+from crewai.skills.models import INSTRUCTIONS, Skill as SkillModel
 from crewai.state.checkpoint_config import CheckpointConfig, apply_checkpoint
 from crewai.tools.agent_tools.agent_tools import AgentTools
 from crewai.types.callback import SerializableCallable
@@ -201,7 +202,7 @@ class Agent(BaseAgent):
     _last_messages: list[LLMMessage] = PrivateAttr(default_factory=list)
     max_execution_time: int | None = Field(
         default=None,
-        description="Maximum execution time in seconds for an agent to execute a task",
+        description="Maximum execution time for an agent to execute a task",
     )
     step_callback: SerializableCallable | None = Field(
         default=None,
@@ -428,12 +429,13 @@ class Agent(BaseAgent):
         self,
         resolved_crew_skills: list[SkillModel] | None = None,
     ) -> None:
-        """Load configured skills while preserving explicit disclosure levels.
+        """Resolve skill paths while preserving explicit disclosure levels.
 
-        Path strings, Path objects, inline SKILL.md strings, and registry refs
-        are loaded through the shared skill loader. Pre-loaded Skill objects
-        keep their current disclosure level so callers can attach METADATA-only
-        skills and progressively activate them later.
+        Path entries trigger discovery and activation because directory-based
+        skills opt into eager loading. Pre-loaded Skill objects keep their
+        current disclosure level so callers can attach METADATA-only skills and
+        progressively activate them later. Crew-level skills are merged in with
+        event emission so observability is consistent regardless of origin.
 
         Args:
             resolved_crew_skills: Pre-resolved crew skills. When provided,
@@ -442,7 +444,7 @@ class Agent(BaseAgent):
         from crewai.crew import Crew
 
         if resolved_crew_skills is None:
-            crew_skills = (
+            crew_skills: list[Path | SkillModel | str] | None = (
                 self.crew.skills
                 if isinstance(self.crew, Crew) and isinstance(self.crew.skills, list)
                 else None
@@ -453,14 +455,58 @@ class Agent(BaseAgent):
         if not self.skills and not crew_skills:
             return
 
-        items = list(self.skills) if self.skills else []
+        needs_work = self.skills and any(
+            isinstance(s, (Path, str))
+            or (isinstance(s, SkillModel) and s.disclosure_level < INSTRUCTIONS)
+            for s in self.skills
+        )
+        if not needs_work and not crew_skills:
+            return
+
+        seen: set[str] = set()
+        resolved: list[Path | SkillModel | str] = []
+        items: list[Path | SkillModel | str] = list(self.skills) if self.skills else []
+
         if crew_skills:
             items.extend(crew_skills)
 
-        self.skills = cast(
-            list[Path | SkillModel | str] | None,
-            load_skills(items, source=self) or None,
-        )
+        for item in items:
+            if isinstance(item, str):
+                from crewai.experimental.skills.registry import (
+                    is_registry_ref,
+                    parse_registry_ref,
+                    resolve_registry_ref,
+                )
+
+                if is_registry_ref(item):
+                    skill = resolve_registry_ref(item, source=self)
+                    org, _ = parse_registry_ref(item)
+                    dedup_key = f"{org}/{skill.name}"
+                    if dedup_key not in seen:
+                        seen.add(dedup_key)
+                        resolved.append(skill)
+            elif isinstance(item, Path):
+                discovered = discover_skills(item, source=self)
+                for skill in discovered:
+                    if skill.name not in seen:
+                        seen.add(skill.name)
+                        resolved.append(activate_skill(skill, source=self))
+            elif isinstance(item, SkillModel):
+                if item.name not in seen:
+                    seen.add(item.name)
+                    if item.disclosure_level >= INSTRUCTIONS:
+                        crewai_event_bus.emit(
+                            self,
+                            event=SkillActivatedEvent(
+                                from_agent=self,
+                                skill_name=item.name,
+                                skill_path=item.path,
+                                disclosure_level=item.disclosure_level,
+                            ),
+                        )
+                    resolved.append(item)
+
+        self.skills = resolved if resolved else None
 
     def _is_any_available_memory(self) -> bool:
         """Check if unified memory is available (agent or crew)."""

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -388,7 +388,7 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     )
     skills: list[Path | Skill | str] | None = Field(
         default=None,
-        description="Agent Skills. Accepts paths for discovery, inline SKILL.md strings, pre-loaded Skill objects, or '@org/name' registry refs.",
+        description="Agent Skills. Accepts paths for discovery, pre-loaded Skill objects, or '@org/name' registry refs.",
         min_length=1,
     )
     execution_context: ExecutionContext | None = Field(default=None)
@@ -494,6 +494,20 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     def process_model_config(cls, values: Any) -> dict[str, Any]:
         return process_config(values, cls)
 
+    @field_validator("skills", mode="before")
+    @classmethod
+    def coerce_skill_strings(cls, skills: Any) -> Any:
+        """Coerce plain path strings to Path objects; keep @-prefixed refs as str."""
+        if not isinstance(skills, list):
+            return skills
+        result = []
+        for item in skills:
+            if isinstance(item, str) and not item.startswith("@"):
+                result.append(Path(item))
+            else:
+                result.append(item)
+        return result
+
     @field_validator("tools")
     @classmethod
     def validate_tools(cls, tools: list[Any]) -> list[BaseTool]:

lib/crewai/src/crewai/crew.py

@@ -361,7 +361,7 @@ class Crew(FlowTrackable, BaseModel):
     )
     skills: list[Path | Skill | str] | None = Field(
         default=None,
-        description="Skill search paths, inline SKILL.md strings, pre-loaded Skill objects, or '@org/name' registry refs applied to all agents in the crew.",
+        description="Skill search paths, pre-loaded Skill objects, or '@org/name' registry refs applied to all agents in the crew.",
     )
 
     security_config: SecurityConfig = Field(
@@ -574,6 +574,20 @@ class Crew(FlowTrackable, BaseModel):
         if max_seq > 0:
             set_emission_counter(max_seq)
 
+    @field_validator("skills", mode="before")
+    @classmethod
+    def coerce_skill_strings(cls, skills: Any) -> Any:
+        """Coerce plain path strings to Path objects; keep @-prefixed refs as str."""
+        if not isinstance(skills, list):
+            return skills
+        result = []
+        for item in skills:
+            if isinstance(item, str) and not item.startswith("@"):
+                result.append(Path(item))
+            else:
+                result.append(item)
+        return result
+
     @field_validator("id", mode="before")
     @classmethod
     def _deny_user_set_id(cls, v: UUID4 | None, info: Any) -> UUID4 | None:

lib/crewai/src/crewai/crews/utils.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import Callable, Coroutine, Iterable, Mapping
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from opentelemetry import baggage
@@ -12,7 +13,7 @@ from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.crews.crew_output import CrewOutput
 from crewai.llms.base_llm import BaseLLM
 from crewai.rag.embeddings.types import EmbedderConfig
-from crewai.skills.loader import activate_skill, load_skills
+from crewai.skills.loader import activate_skill, discover_skills
 from crewai.skills.models import INSTRUCTIONS, Skill as SkillModel
 from crewai.types.streaming import CrewStreamingOutput, FlowStreamingOutput
 from crewai.utilities.file_store import store_files
@@ -59,13 +60,23 @@ def _resolve_crew_skills(crew: Crew) -> list[SkillModel] | None:
     if not isinstance(crew.skills, list) or not crew.skills:
         return None
 
-    resolved = load_skills(crew.skills)
-    if not resolved:
-        return None
-    return [
-        activate_skill(skill) if skill.disclosure_level < INSTRUCTIONS else skill
-        for skill in resolved
-    ]
+    resolved: list[SkillModel] = []
+    seen: set[str] = set()
+    for item in crew.skills:
+        if isinstance(item, Path):
+            for skill in discover_skills(item):
+                if skill.name not in seen:
+                    seen.add(skill.name)
+                    resolved.append(activate_skill(skill))
+        elif isinstance(item, SkillModel):
+            if item.name not in seen:
+                seen.add(item.name)
+                resolved.append(
+                    activate_skill(item)
+                    if item.disclosure_level < INSTRUCTIONS
+                    else item
+                )
+    return resolved
 
 
 def setup_agents(

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

@@ -40,7 +40,6 @@ from crewai.events.event_context import (
     set_last_event_id,
 )
 from crewai.events.handler_graph import build_execution_plan
-from crewai.events.stream_context import publish_stream_event
 from crewai.events.types.event_bus_types import (
     AsyncHandler,
     AsyncHandlerSet,
@@ -566,7 +565,6 @@ class CrewAIEventsBus:
 
         set_last_event_id(event.event_id)
 
-        publish_stream_event(source, event)
         self._record_event(event)
 
     def emit(self, source: Any, event: BaseEvent) -> Future[None] | None:
@@ -777,7 +775,9 @@ class CrewAIEventsBus:
             source: The object emitting the event
             event: The event instance to emit
         """
-        self._prepare_event(source, event)
+        self._register_source(source)
+        event.emission_sequence = get_next_emission_sequence()
+        self._record_event(event)
 
         event_type = type(event)
 

lib/crewai/src/crewai/events/stream_context.py

@@ -1,30 +0,0 @@
-"""Scoped stream sinks for converting emitted events into public frames."""
-
-from __future__ import annotations
-
-from collections.abc import Callable
-import contextvars
-from typing import Any
-
-
-StreamSink = Callable[[Any, Any], None]
-
-_stream_sinks: contextvars.ContextVar[tuple[StreamSink, ...]] = contextvars.ContextVar(
-    "crewai_stream_sinks", default=()
-)
-
-
-def add_stream_sink(sink: StreamSink) -> contextvars.Token[tuple[StreamSink, ...]]:
-    """Register a sink in the current context."""
-    return _stream_sinks.set((*_stream_sinks.get(), sink))
-
-
-def reset_stream_sinks(token: contextvars.Token[tuple[StreamSink, ...]]) -> None:
-    """Restore the stream sink context."""
-    _stream_sinks.reset(token)
-
-
-def publish_stream_event(source: Any, event: Any) -> None:
-    """Publish a prepared event to sinks scoped to the current execution."""
-    for sink in _stream_sinks.get():
-        sink(source, event)

lib/crewai/src/crewai/experimental/conversational_mixin.py

@@ -18,8 +18,7 @@ Import surface:
 
 from __future__ import annotations
 
-from collections.abc import Callable, Iterator, Mapping, Sequence
-from contextlib import contextmanager
+from collections.abc import Callable, Mapping, Sequence
 from enum import Enum
 import json
 import logging
@@ -45,7 +44,6 @@ from crewai.experimental.conversational import (
     _conversational_only,
     message_to_llm_dict,
 )
-from crewai.flow.async_feedback import HumanFeedbackPending
 from crewai.flow.conversation import (
     append_message as _append_conversation_message,
     get_conversation_messages,
@@ -223,9 +221,7 @@ class _ConversationalMixin:
             messages.append({"role": "system", "content": system_prompt})
         messages.extend(self.conversation_messages)
 
-        llm_instance = self._coerce_llm(llm)
-        with self._conversation_streaming_enabled(llm_instance):
-            response = llm_instance.call(messages=messages)
+        response = self._coerce_llm(llm).call(messages=messages)
         content = self._stringify_result(response)
         self.append_assistant_message(content)
         return content
@@ -258,8 +254,7 @@ class _ConversationalMixin:
             },
             *self.build_agent_context("answer_from_history"),
         ]
-        with self._conversation_streaming_enabled(llm_instance):
-            response = llm_instance.call(messages=messages)
+        response = llm_instance.call(messages=messages)
         content = self._stringify_result(response)
         self.append_assistant_message(content)
         return content
@@ -342,102 +337,6 @@ class _ConversationalMixin:
         )
         return result
 
-    def stream_turn(
-        self,
-        message: str,
-        *,
-        session_id: str | None = None,
-        intents: Sequence[str] | None = None,
-        intent_llm: str | BaseLLM | None = None,
-        **kickoff_kwargs: Any,
-    ) -> Any:
-        """Append a user message and stream one conversational turn as frames."""
-        if not self._is_conversational_enabled():
-            raise ValueError(
-                "Flow.stream_turn() is only available on conversational flows"
-            )
-
-        from crewai.types.streaming import StreamSession
-        from crewai.utilities.streaming import (
-            create_frame_generator,
-            create_frame_streaming_state,
-        )
-
-        state = cast(ConversationState, self.state)
-        sid = session_id or state.id
-        result_holder: list[Any] = []
-        frame_state = create_frame_streaming_state(result_holder, use_async=False)
-        output_holder: list[StreamSession[Any]] = []
-
-        def run_turn() -> Any:
-            crewai_event_bus.emit(
-                self,
-                ConversationTurnStartedEvent(
-                    type="conversation_turn_started",
-                    flow_name=self.name or self.__class__.__name__,
-                    session_id=sid,
-                ),
-            )
-
-            self._pending_user_message = message
-            self._pending_intents = list(intents) if intents else None
-            self._pending_intent_llm = intent_llm
-
-            try:
-                if "from_checkpoint" not in kickoff_kwargs:
-                    self._reset_turn_execution_state()
-
-                assistant_count = self._assistant_message_count()
-                original_stream = bool(getattr(self, "stream", False))
-                original_streaming_turn = getattr(
-                    self, "_streaming_conversation_turn", False
-                )
-                try:
-                    object.__setattr__(self, "stream", False)
-                    object.__setattr__(self, "_streaming_conversation_turn", True)
-                    result = self.kickoff(inputs={"id": sid}, **kickoff_kwargs)
-                finally:
-                    object.__setattr__(self, "stream", original_stream)
-                    object.__setattr__(
-                        self, "_streaming_conversation_turn", original_streaming_turn
-                    )
-                if (
-                    result is not None
-                    and self._assistant_message_count() == assistant_count
-                    and self._is_public_turn_result(result)
-                ):
-                    self.append_assistant_message(self._stringify_result(result))
-            except HumanFeedbackPending as exc:
-                return exc
-            except Exception as exc:
-                failed_event = ConversationTurnFailedEvent(
-                    type="conversation_turn_failed",
-                    flow_name=self.name or self.__class__.__name__,
-                    session_id=sid,
-                    error=exc,
-                )
-                self._emit_terminal_conversation_turn_event(failed_event)
-                raise
-            finally:
-                self._pending_user_message = None
-                self._pending_intents = None
-                self._pending_intent_llm = None
-
-            self._emit_terminal_conversation_turn_event(
-                ConversationTurnCompletedEvent(
-                    type="conversation_turn_completed",
-                    flow_name=self.name or self.__class__.__name__,
-                    session_id=sid,
-                ),
-            )
-            return result
-
-        stream_session: StreamSession[Any] = StreamSession(
-            sync_iterator=create_frame_generator(frame_state, run_turn, output_holder)
-        )
-        output_holder.append(stream_session)
-        return stream_session
-
     def _emit_terminal_conversation_turn_event(
         self,
         event: ConversationTurnCompletedEvent | ConversationTurnFailedEvent,
@@ -786,8 +685,6 @@ class _ConversationalMixin:
             object.__setattr__(self, "_pending_intents", None)
         if not hasattr(self, "_pending_intent_llm"):
             object.__setattr__(self, "_pending_intent_llm", None)
-        if not hasattr(self, "_streaming_conversation_turn"):
-            object.__setattr__(self, "_streaming_conversation_turn", False)
 
     def _create_default_extension_state(self) -> ConversationState | None:
         initial_state_t = getattr(self, "_initial_state_t", None)
@@ -1158,19 +1055,6 @@ class _ConversationalMixin:
             return llm
         raise ValueError(f"Invalid llm type: {type(llm)}. Expected str or BaseLLM.")
 
-    @contextmanager
-    def _conversation_streaming_enabled(self, llm: Any) -> Iterator[None]:
-        if not getattr(self, "_streaming_conversation_turn", False) or not hasattr(
-            llm, "stream"
-        ):
-            yield
-            return
-
-        from crewai.llms.base_llm import call_stream_override
-
-        with call_stream_override(llm, True):
-            yield
-
     def finalize_session_traces(self) -> None:
         """Emit a final ``FlowFinishedEvent`` and finalize the trace batch.
 

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

@@ -3,17 +3,13 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
-from functools import lru_cache
 import json
-from typing import TYPE_CHECKING, Any, NamedTuple, TypeAlias, cast
+from typing import TYPE_CHECKING, Any, TypeAlias, cast
 
 from crewai.utilities.serialization import to_serializable
 
 
 if TYPE_CHECKING:
-    from celpy.celtypes import StringType
-    from celpy.evaluation import CELFunction
-
     from crewai.flow.runtime import Flow
 else:
     from typing_extensions import TypeAliasType
@@ -22,141 +18,6 @@ else:
 _CEL_MACROS_WITH_LOCAL_BINDINGS = frozenset(
     {"all", "exists", "exists_one", "filter", "map"}
 )
-
-
-def _handle_text_custom_expression(
-    root: Any, path: Any, default: Any = ""
-) -> StringType:
-    from celpy.celtypes import StringType
-
-    fallback = StringType("" if default is None else str(default))
-    current = root
-    for part in str(path).split("."):
-        if current is None:
-            return fallback
-        try:
-            if isinstance(current, list):
-                current = current[int(part)]
-            else:
-                current = current[StringType(part)]
-        except (KeyError, IndexError, TypeError, ValueError):
-            return fallback
-
-    if current is None:
-        return fallback
-    return StringType(_stringify_cel_value(current))
-
-
-def _stringify_cel_value(value: Any) -> str:
-    from celpy.adapter import CELJSONEncoder
-
-    if isinstance(value, str):
-        return value
-    return json.dumps(value, cls=CELJSONEncoder, ensure_ascii=False)
-
-
-class _ExpressionSegment(NamedTuple):
-    source: str
-
-
-def _marker_end(value: str, start: int) -> int:
-    from celpy.celparser import CELParser
-
-    CELParser()
-    parser: Any = CELParser.CEL_PARSER
-    depth = 1
-    try:
-        for token in parser.lex(value[start:]):
-            if token.type == "LBRACE":
-                depth += 1
-            elif token.type == "RBRACE":
-                depth -= 1
-                if depth == 0:
-                    return start + int(token.start_pos)
-    except Exception as e:
-        raise ExpressionError(
-            f"unterminated or invalid ${{...}} expression in {value!r}: {e}"
-        ) from e
-    raise ExpressionError(f"unterminated ${{...}} expression in {value!r}")
-
-
-@lru_cache(maxsize=256)
-def _parse_template_segments(value: str) -> tuple[str | _ExpressionSegment, ...]:
-    segments: list[str | _ExpressionSegment] = []
-    index = 0
-    while (start := value.find("${", index)) != -1:
-        if start > index:
-            segments.append(value[index:start])
-        end = _marker_end(value, start + 2)
-        source = value[start + 2 : end].strip()
-        if not source:
-            raise ExpressionError(f"empty CEL expression in {value!r}")
-        segments.append(_ExpressionSegment(source))
-        index = end + 1
-    if index < len(value) or not segments:
-        segments.append(value[index:])
-    return tuple(segments)
-
-
-_EXPRESSION_FUNCTIONS: dict[str, CELFunction] = {
-    "text": _handle_text_custom_expression,
-}
-
-FLOW_TEMPLATE_EXPRESSION_RULES: tuple[str, ...] = (
-    "Use `${...}` inside action mapping strings to read Flow data with CEL. "
-    "Example value: `Ticket: ${state.ticket_id}`.",
-    "Use `state` for input data. Use `outputs.step_name` for a completed "
-    "method result.",
-    "If a value is only one `${...}` expression, the result keeps its type. "
-    "Use this for numbers, booleans, objects, and lists.",
-    "If the string has other text, the final value is text. Non-text values "
-    "become JSON. `null` becomes empty text.",
-    'Use `text(root, "path", "default")` for values that may be missing '
-    'or null. The default is optional and is `""`.',
-)
-FLOW_TEMPLATE_EXPRESSION_CONTRACT = " ".join(FLOW_TEMPLATE_EXPRESSION_RULES)
-FLOW_TEMPLATE_EXPRESSION_EXAMPLES: dict[str, tuple[dict[str, str], ...]] = {
-    "yaml": (
-        {
-            "title": "Mix text and Flow data",
-            "code": 'query: "News about ${state.topic}"',
-        },
-        {
-            "title": "Keep a list or number type",
-            "code": 'domains: "${state.domains}"\nlimit: "${state.limit}"',
-        },
-        {
-            "title": "Use a default for missing text",
-            "code": 'input: "Ticket ${text(state, \\"ticket.id\\", \\"unknown\\")}"',
-        },
-    ),
-    "json": (
-        {
-            "title": "Mix text and Flow data",
-            "code": '{\n  "query": "News about ${state.topic}"\n}',
-        },
-        {
-            "title": "Keep a list or number type",
-            "code": (
-                '{\n  "domains": "${state.domains}",\n  "limit": "${state.limit}"\n}'
-            ),
-        },
-        {
-            "title": "Use a default for missing text",
-            "code": (
-                "{\n"
-                '  "input": "Ticket ${text(state, \\"ticket.id\\", \\"unknown\\")}"\n'
-                "}"
-            ),
-        },
-    ),
-}
-
-
-def flow_template_expression_description(prefix: str) -> str:
-    return f"{prefix} {FLOW_TEMPLATE_EXPRESSION_CONTRACT}"
-
-
 if TYPE_CHECKING:
     ExpressionData: TypeAlias = (
         str
@@ -180,13 +41,9 @@ else:
     )
 
 __all__ = [
-    "FLOW_TEMPLATE_EXPRESSION_CONTRACT",
-    "FLOW_TEMPLATE_EXPRESSION_EXAMPLES",
-    "FLOW_TEMPLATE_EXPRESSION_RULES",
     "Expression",
     "ExpressionData",
     "ExpressionError",
-    "flow_template_expression_description",
 ]
 
 
@@ -242,7 +99,7 @@ class Expression:
         allowed_roots: Iterable[str],
         source: str = "with block",
     ) -> None:
-        """Validate ``${...}`` expressions inside nested strings as CEL."""
+        """Validate nested strings fully wrapped in ``${...}`` as CEL."""
         self._validate_template_value(
             self.value, allowed_roots=allowed_roots, source=source
         )
@@ -256,11 +113,7 @@ class Expression:
         )
 
     def render_template(self, context: dict[str, Any] | None = None) -> Any:
-        """Interpolate ``${...}`` expressions inside nested strings as CEL.
-
-        A string that is exactly one ``${...}`` keeps the evaluated value's
-        type; strings mixing literals and expressions render as text.
-        """
+        """Evaluate nested strings fully wrapped in ``${...}`` as CEL."""
         resolved_context = self.context if context is None else context
         return self._render_template_value(self.value, resolved_context or {})
 
@@ -272,23 +125,10 @@ class Expression:
         source: str,
     ) -> None:
         if isinstance(value, str):
-            try:
-                segments = _parse_template_segments(value)
-            except ExpressionError as e:
-                raise ExpressionError(f"{e} at {source}") from None
-            expressions = [
-                segment
-                for segment in segments
-                if isinstance(segment, _ExpressionSegment)
-            ]
-            for index, segment in enumerate(expressions):
-                segment_source = (
-                    source
-                    if len(expressions) == 1
-                    else f"{source} (expression {index + 1})"
-                )
-                Expression(segment.source).validate_expression(
-                    allowed_roots=allowed_roots, source=segment_source
+            expression = Expression._expression_marker_source(value, source=source)
+            if expression is not None:
+                Expression(expression).validate_expression(
+                    allowed_roots=allowed_roots, source=source
                 )
             return
         if isinstance(value, dict):
@@ -347,23 +187,28 @@ class Expression:
 
     @staticmethod
     def _render_template_string(value: str, context: dict[str, Any]) -> Any:
-        segments = _parse_template_segments(value)
-        expressions = [
-            segment for segment in segments if isinstance(segment, _ExpressionSegment)
-        ]
-        if not expressions:
+        expression = Expression._expression_marker_source(value)
+        if expression is None:
             return value
-        literals = [segment for segment in segments if isinstance(segment, str)]
-        if len(expressions) == 1 and all(not literal.strip() for literal in literals):
-            return Expression._evaluate_cel(expressions[0].source, context)
-        rendered: list[str] = []
-        for segment in segments:
-            if isinstance(segment, str):
-                rendered.append(segment)
-                continue
-            result = Expression._evaluate_cel(segment.source, context)
-            rendered.append("" if result is None else _stringify_cel_value(result))
-        return "".join(rendered)
+        return Expression._evaluate_cel(expression, context)
+
+    @staticmethod
+    def _expression_marker_source(
+        value: str, *, source: str | None = None
+    ) -> str | None:
+        """Return CEL source when the trimmed string starts with ``${`` and ends with ``}``."""
+        stripped = value.strip()
+        if not stripped.startswith("${"):
+            return None
+        if not stripped.endswith("}"):
+            return None
+
+        expression = stripped[2:-1].strip()
+        if not expression:
+            if source is None:
+                raise ExpressionError("empty CEL expression in with block")
+            raise ExpressionError(f"empty CEL expression at {source}")
+        return expression
 
     @staticmethod
     def _evaluate_cel(expression: str, context: dict[str, Any]) -> Any:
@@ -374,8 +219,7 @@ class Expression:
 
             environment = Environment()
             program = environment.program(
-                Expression._compile_cel(expression, environment=environment),
-                functions=_EXPRESSION_FUNCTIONS,
+                Expression._compile_cel(expression, environment=environment)
             )
             result = program.evaluate(cast(Context, json_to_cel(context)))
             return json.loads(json.dumps(result, cls=CELJSONEncoder))

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

@@ -9,7 +9,6 @@ layer that may have produced it and of the engine that runs it (see
 
 from __future__ import annotations
 
-from collections.abc import Sequence
 import logging
 from pathlib import Path
 import re
@@ -29,10 +28,7 @@ from crewai.flow.conversational_definition import (
     FlowConversationalDefinition,
     FlowConversationalRouterDefinition,
 )
-from crewai.flow.expressions import (
-    ExpressionData,
-    flow_template_expression_description,
-)
+from crewai.flow.expressions import ExpressionData
 from crewai.project.crew_definition import AgentDefinition, CrewDefinition
 
 
@@ -90,7 +86,7 @@ class FlowDictStateDefinition(BaseModel):
     )
     default: dict[str, Any] | None = Field(
         default=None,
-        description="Default values used to initialize Flow state.",
+        description="Default state values applied before kickoff inputs.",
         examples=[{"topic": "AI agents", "limit": 3}],
     )
 
@@ -125,7 +121,7 @@ class FlowPydanticStateDefinition(BaseModel):
     )
     default: dict[str, Any] | None = Field(
         default=None,
-        description="Default values used to initialize Flow state.",
+        description="Default state values applied before kickoff inputs.",
         examples=[{"topic": "AI agents", "limit": 3}],
     )
 
@@ -152,7 +148,7 @@ class FlowJsonSchemaStateDefinition(BaseModel):
     )
     default: dict[str, Any] | None = Field(
         default=None,
-        description="Default values used to initialize Flow state.",
+        description="Default state values applied before kickoff inputs.",
         examples=[{"topic": "AI agents", "limit": 3}],
     )
 
@@ -164,7 +160,7 @@ class FlowUnknownStateDefinition(BaseModel):
 
     type: Literal["unknown"] = Field(
         default="unknown",
-        description="Unknown state representation; execution uses dictionary state.",
+        description="Unknown state representation; runtime falls back to dictionary state.",
         examples=["unknown"],
     )
     ref: str | None = Field(
@@ -174,7 +170,7 @@ class FlowUnknownStateDefinition(BaseModel):
     )
     default: dict[str, Any] | None = Field(
         default=None,
-        description="Default values used to initialize Flow state.",
+        description="Default state values applied before kickoff inputs.",
         examples=[{"topic": "AI agents", "limit": 3}],
     )
 
@@ -193,7 +189,7 @@ class FlowConfigDefinition(BaseModel):
 
     tracing: bool | None = Field(
         default=None,
-        description="Override for flow tracing; when omitted, execution defaults apply.",
+        description="Override for flow tracing; when omitted, runtime defaults apply.",
         examples=[True],
     )
     stream: bool = Field(
@@ -208,7 +204,7 @@ class FlowConfigDefinition(BaseModel):
     )
     input_provider: str | None = Field(
         default=None,
-        description="Provider key used to supply initial state.",
+        description="Import reference or provider key used to supply flow inputs.",
         examples=["my_project.inputs:load_inputs"],
     )
     suppress_flow_events: bool = Field(
@@ -365,10 +361,12 @@ class FlowCodeActionDefinition(BaseModel):
     with_: dict[str, ExpressionData] | None = Field(
         default=None,
         alias="with",
-        description=flow_template_expression_description(
-            "Keyword arguments passed to the callable."
+        description=(
+            "Keyword arguments passed to the callable. String values are evaluated "
+            "as CEL only when the trimmed value starts with ${ and ends with }; "
+            "all other values are literal."
         ),
-        examples=[{"topic": "${state.topic}", "query": "News about ${state.topic}"}],
+        examples=[{"topic": "${state.topic}"}],
     )
 
 
@@ -385,13 +383,17 @@ class FlowToolActionDefinition(BaseModel):
         examples=["tool"],
     )
     ref: str = Field(
-        description="Reference to the CrewAI tool to run.",
+        description="Import reference for a BaseTool class, formatted as module:qualname.",
         examples=["my_project.tools:SearchTool"],
     )
     with_: dict[str, ExpressionData] | None = Field(
         default=None,
         alias="with",
-        description=flow_template_expression_description("Tool input arguments."),
+        description=(
+            "Tool input arguments. String values are evaluated as CEL only when "
+            "the trimmed value starts with ${ and ends with }; all other values "
+            "are literal."
+        ),
         examples=[{"query": "${outputs.normalize_topic}", "limit": 5}],
     )
 
@@ -443,12 +445,10 @@ class FlowCrewActionDefinition(BaseModel):
     )
     inputs: dict[str, ExpressionData] | None = Field(
         default=None,
-        description=flow_template_expression_description(
-            "Input overrides passed to the Crew."
-        )
-        + (
-            " The resulting values are available to crew agent and task "
-            "interpolation as `{name}` placeholders."
+        description=(
+            "Input overrides passed to the Crew. String values are evaluated as CEL "
+            "only when the trimmed value starts with ${ and ends with }; all other "
+            "values are literal."
         ),
         examples=[{"topic": "${state.topic}"}],
     )
@@ -463,7 +463,7 @@ class FlowCrewActionDefinition(BaseModel):
 
 
 class FlowAgentActionDefinition(BaseModel):
-    """A Flow method action that builds and kicks off one agent outside a crew."""
+    """A Flow method action that builds and kicks off a CrewAI agent."""
 
     model_config = ConfigDict(
         populate_by_name=True,
@@ -471,18 +471,12 @@ class FlowAgentActionDefinition(BaseModel):
     )
 
     call: Literal["agent"] = Field(
-        description=(
-            "Action discriminator. Use agent to run an individual inline Agent "
-            "definition outside of a crew."
-        ),
+        description="Action discriminator. Use agent to run an inline Agent definition.",
         examples=["agent"],
     )
     with_: AgentDefinition = Field(
         alias="with",
-        description=(
-            "Individual Agent definition to load and execute outside of a crew "
-            "for this action."
-        ),
+        description="Inline Agent definition to load and execute for this action.",
         examples=[
             {
                 "role": "Analyst",
@@ -521,11 +515,12 @@ class FlowScriptActionDefinition(BaseModel):
     )
     code: str = Field(
         description=(
-            "Trusted inline Python source. Values are available as state and outputs; "
-            "they are not interpolated into the source. This is not sandboxed."
+            "Trusted Python source executed as a generated function. Runtime values are "
+            "passed as state, outputs, input, and item; they are not interpolated into "
+            "the source. This is not sandboxed."
         ),
         examples=[
-            "state['normalized_topic'] = state['topic'].strip()\n"
+            "state['normalized_topic'] = input.strip()\n"
             "return state['normalized_topic']"
         ],
     )
@@ -650,13 +645,13 @@ class FlowMethodDefinition(BaseModel):
     )
     do: FlowActionDefinition = Field(
         description="Action executed when this method runs.",
-        examples=[{"call": "expression", "expr": "state.topic"}],
+        examples=[{"call": "script", "code": "return input.strip()"}],
     )
     start: bool | FlowDefinitionCondition | None = Field(
         default=None,
         description=(
             "Marks a start method. True starts unconditionally; a condition starts "
-            "when the initial state or events satisfy it."
+            "when the kickoff inputs or events satisfy it."
         ),
         examples=[True],
     )
@@ -734,12 +729,12 @@ class FlowDefinition(BaseModel):
     )
     state: FlowStateDefinition | None = Field(
         default=None,
-        description="State contract for the initial state and updates during execution.",
+        description="State contract for kickoff inputs and runtime state.",
         examples=[{"type": "dict", "default": {"topic": "AI agents"}}],
     )
     config: FlowConfigDefinition = Field(
         default_factory=FlowConfigDefinition,
-        description="Serializable flow-level execution configuration.",
+        description="Serializable flow-level runtime configuration.",
         examples=[{"stream": True, "max_method_calls": 20}],
     )
     persist: FlowPersistenceDefinition | None = Field(
@@ -770,15 +765,6 @@ class FlowDefinition(BaseModel):
             _validate_step_name(method_name, field="Flow method names")
         return self
 
-    @model_validator(mode="after")
-    def _validate_trigger_namespace(self) -> FlowDefinition:
-        for method_name, method in self.methods.items():
-            if _condition_references(method.listen, method_name):
-                raise ValueError(
-                    f"methods.{method_name}.listen must not reference itself"
-                )
-        return self
-
     @model_validator(mode="after")
     def _validate_cel_expressions(self) -> FlowDefinition:
         for method_name, method in self.methods.items():
@@ -849,18 +835,6 @@ class FlowDefinition(BaseModel):
         log_flow_definition_issues(definition)
         return definition
 
-    @classmethod
-    def skill(
-        cls,
-        *,
-        skips: Sequence[str] = (),
-        examples_format: Literal["yaml", "json"] = "yaml",
-    ) -> str:
-        """Return a portable Markdown skill for authoring Flow declarations."""
-        from crewai.flow.skill import render_skill_markdown
-
-        return render_skill_markdown(skips=skips, examples_format=examples_format)
-
 
 def _validate_step_name(name: str, *, field: str) -> None:
     if not isinstance(name, str) or not _STEP_NAME_PATTERN.fullmatch(name):
@@ -876,18 +850,6 @@ def _validate_step_list(steps: list[FlowEachStepDefinition], *, field: str) -> N
         seen.add(name)
 
 
-def _condition_references(condition: FlowDefinitionCondition | None, name: str) -> bool:
-    if condition is None:
-        return False
-    if isinstance(condition, str):
-        return condition == name
-    return any(
-        _condition_references(child, name)
-        for key in ("and", "or")
-        for child in condition.get(key, [])
-    )
-
-
 def _validate_action_cel(
     action: FlowActionDefinition,
     *,

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

@@ -9,7 +9,11 @@ Structure (see ``flow_definition``) and executed here.
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Callable, Iterator, Sequence
+from collections.abc import (
+    Callable,
+    Iterator,
+    Sequence,
+)
 from concurrent.futures import Future, ThreadPoolExecutor
 import contextvars
 import copy
@@ -136,16 +140,17 @@ if TYPE_CHECKING:
     from crewai.llms.base_llm import BaseLLM
 
 from crewai.flow.visualization import build_flow_structure, render_interactive
-from crewai.types.streaming import (
-    AsyncStreamSession,
-    StreamSession,
-)
+from crewai.types.streaming import CrewStreamingOutput, FlowStreamingOutput
 from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities.env import get_env_context
 from crewai.utilities.streaming import (
-    create_async_frame_generator,
-    create_frame_generator,
-    create_frame_streaming_state,
+    TaskInfo,
+    create_async_chunk_generator,
+    create_chunk_generator,
+    create_streaming_state,
+    register_cleanup,
+    signal_end,
+    signal_error,
 )
 
 
@@ -1827,79 +1832,13 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 if hasattr(self._state, key):
                     object.__setattr__(self._state, key, value)
 
-    def stream_events(
-        self,
-        inputs: dict[str, Any] | None = None,
-        input_files: dict[str, FileInput] | None = None,
-        from_checkpoint: CheckpointConfig | None = None,
-        restore_from_state_id: str | None = None,
-    ) -> StreamSession[Any]:
-        """Run the flow and stream all scoped public ``StreamFrame`` events."""
-        result_holder: list[Any] = []
-        state = create_frame_streaming_state(result_holder, use_async=False)
-        output_holder: list[StreamSession[Any]] = []
-
-        def run_flow() -> Any:
-            original_stream = self.stream
-            try:
-                self.stream = False
-                return self.kickoff(
-                    inputs=inputs,
-                    input_files=input_files,
-                    from_checkpoint=from_checkpoint,
-                    restore_from_state_id=restore_from_state_id,
-                )
-            except HumanFeedbackPending as e:
-                return e
-            finally:
-                self.stream = original_stream
-
-        stream_session: StreamSession[Any] = StreamSession(
-            sync_iterator=create_frame_generator(state, run_flow, output_holder)
-        )
-        output_holder.append(stream_session)
-        return stream_session
-
-    def astream(
-        self,
-        inputs: dict[str, Any] | None = None,
-        input_files: dict[str, FileInput] | None = None,
-        from_checkpoint: CheckpointConfig | None = None,
-        restore_from_state_id: str | None = None,
-    ) -> AsyncStreamSession[Any]:
-        """Run the flow asynchronously and stream scoped public frames."""
-        result_holder: list[Any] = []
-        state = create_frame_streaming_state(result_holder, use_async=True)
-        output_holder: list[AsyncStreamSession[Any]] = []
-
-        async def run_flow() -> Any:
-            original_stream = self.stream
-            try:
-                self.stream = False
-                return await self.kickoff_async(
-                    inputs=inputs,
-                    input_files=input_files,
-                    from_checkpoint=from_checkpoint,
-                    restore_from_state_id=restore_from_state_id,
-                )
-            except HumanFeedbackPending as e:
-                return e
-            finally:
-                self.stream = original_stream
-
-        stream_session: AsyncStreamSession[Any] = AsyncStreamSession(
-            async_iterator=create_async_frame_generator(state, run_flow, output_holder)
-        )
-        output_holder.append(stream_session)
-        return stream_session
-
     def kickoff(
         self,
         inputs: dict[str, Any] | None = None,
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
         restore_from_state_id: str | None = None,
-    ) -> Any | StreamSession[Any]:
+    ) -> Any | FlowStreamingOutput:
         """Start the flow execution in a synchronous context.
 
         This method wraps kickoff_async so that all state initialization and event
@@ -1920,7 +1859,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 ``from_checkpoint``; passing both raises ``ValueError``.
 
         Returns:
-            The final output from the flow or StreamSession if streaming.
+            The final output from the flow or FlowStreamingOutput if streaming.
         """
         if from_checkpoint is not None and restore_from_state_id is not None:
             raise ValueError(
@@ -1932,12 +1871,46 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         if restored is not None:
             return restored.kickoff(inputs=inputs, input_files=input_files)
         if self.stream:
-            return self.stream_events(
-                inputs=inputs,
-                input_files=input_files,
-                from_checkpoint=from_checkpoint,
-                restore_from_state_id=restore_from_state_id,
+            result_holder: list[Any] = []
+            current_task_info: TaskInfo = {
+                "index": 0,
+                "name": "",
+                "id": "",
+                "agent_role": "",
+                "agent_id": "",
+            }
+
+            state = create_streaming_state(
+                current_task_info, result_holder, use_async=False
             )
+            output_holder: list[CrewStreamingOutput | FlowStreamingOutput] = []
+
+            def run_flow() -> None:
+                try:
+                    self.stream = False
+                    result = self.kickoff(
+                        inputs=inputs,
+                        input_files=input_files,
+                        restore_from_state_id=restore_from_state_id,
+                    )
+                    result_holder.append(result)
+                except Exception as e:
+                    # HumanFeedbackPending is expected control flow, not an error
+                    if isinstance(e, HumanFeedbackPending):
+                        result_holder.append(e)
+                    else:
+                        signal_error(state, e)
+                finally:
+                    self.stream = True
+                    signal_end(state)
+
+            streaming_output = FlowStreamingOutput(
+                sync_iterator=create_chunk_generator(state, run_flow, output_holder)
+            )
+            register_cleanup(streaming_output, state)
+            output_holder.append(streaming_output)
+
+            return streaming_output
 
         async def _run_flow() -> Any:
             return await self.kickoff_async(
@@ -1964,7 +1937,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
         restore_from_state_id: str | None = None,
-    ) -> Any | AsyncStreamSession[Any]:
+    ) -> Any | FlowStreamingOutput:
         """Start the flow execution asynchronously.
 
         This method performs state restoration (if an 'id' is provided and persistence is available)
@@ -1998,12 +1971,48 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         if restored is not None:
             return await restored.kickoff_async(inputs=inputs, input_files=input_files)
         if self.stream:
-            return self.astream(
-                inputs=inputs,
-                input_files=input_files,
-                from_checkpoint=from_checkpoint,
-                restore_from_state_id=restore_from_state_id,
+            result_holder: list[Any] = []
+            current_task_info: TaskInfo = {
+                "index": 0,
+                "name": "",
+                "id": "",
+                "agent_role": "",
+                "agent_id": "",
+            }
+
+            state = create_streaming_state(
+                current_task_info, result_holder, use_async=True
             )
+            output_holder: list[CrewStreamingOutput | FlowStreamingOutput] = []
+
+            async def run_flow() -> None:
+                try:
+                    self.stream = False
+                    result = await self.kickoff_async(
+                        inputs=inputs,
+                        input_files=input_files,
+                        restore_from_state_id=restore_from_state_id,
+                    )
+                    result_holder.append(result)
+                except Exception as e:
+                    # HumanFeedbackPending is expected control flow, not an error
+                    if isinstance(e, HumanFeedbackPending):
+                        result_holder.append(e)
+                    else:
+                        signal_error(state, e, is_async=True)
+                finally:
+                    self.stream = True
+                    signal_end(state, is_async=True)
+
+            streaming_output = FlowStreamingOutput(
+                async_iterator=create_async_chunk_generator(
+                    state, run_flow, output_holder
+                )
+            )
+            register_cleanup(streaming_output, state)
+            output_holder.append(streaming_output)
+
+            return streaming_output
 
         ctx = baggage.set_baggage("flow_inputs", inputs or {})
         ctx = baggage.set_baggage("flow_input_files", input_files or {}, context=ctx)
@@ -2347,7 +2356,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         input_files: dict[str, FileInput] | None = None,
         from_checkpoint: CheckpointConfig | None = None,
         restore_from_state_id: str | None = None,
-    ) -> Any | AsyncStreamSession[Any]:
+    ) -> Any | FlowStreamingOutput:
         """Native async method to start the flow execution. Alias for kickoff_async.
 
         Args:

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

@@ -138,12 +138,11 @@ class CrewAction:
 
         local_context = _pop_local_context(kwargs)
         if self.definition.from_declaration is not None:
-            crew, default_inputs = await asyncio.to_thread(
-                load_crew,
+            crew, default_inputs = load_crew(
                 _resolve_crew_declaration(
                     self.definition.from_declaration,
                     base_dir=self.flow._definition.source_dir,
-                ),
+                )
             )
             input_template = {**default_inputs, **(self.definition.inputs or {})}
         else:
@@ -156,9 +155,7 @@ class CrewAction:
                 **crew_definition.inputs,
                 **(self.definition.inputs or {}),
             }
-            crew, _ = await asyncio.to_thread(
-                load_crew_from_definition, crew_definition, source="crew action"
-            )
+            crew, _ = load_crew_from_definition(crew_definition, source="crew action")
 
         inputs = Expression.from_flow(
             cast(ExpressionData, input_template),
@@ -187,8 +184,7 @@ class AgentAction:
         if not isinstance(rendered_input, str):
             raise ValueError("agent input must render to a string")
 
-        agent, response_format = await asyncio.to_thread(
-            load_agent_from_definition,
+        agent, response_format = load_agent_from_definition(
             self.definition.with_,
             source="agent action",
         )

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

@@ -1,577 +0,0 @@
-"""Markdown skill rendering for Flow Definition authoring."""
-
-from collections.abc import Sequence
-from dataclasses import dataclass, field
-import json
-from pathlib import Path
-import re
-from typing import Any, Literal
-
-from jinja2 import Environment, FileSystemLoader
-import yaml
-
-from crewai.flow.expressions import (
-    FLOW_TEMPLATE_EXPRESSION_CONTRACT,
-    FLOW_TEMPLATE_EXPRESSION_EXAMPLES,
-    FLOW_TEMPLATE_EXPRESSION_RULES,
-)
-from crewai.flow.flow_definition import FlowDefinition
-
-
-SKIP_BY_MODEL: dict[str, str] = {
-    "FlowScriptActionDefinition": "script_action",
-    "FlowToolActionDefinition": "tool_action",
-    "FlowExpressionActionDefinition": "expression_action",
-    "FlowEachActionDefinition": "each",
-    "FlowEachStepDefinition": "each",
-    "FlowConfigDefinition": "config",
-    "FlowHumanFeedbackDefinition": "hitl",
-    "FlowPersistenceDefinition": "persistence",
-}
-
-FIELD_TYPE_OVERRIDES: dict[tuple[str, str], str] = {
-    ("FlowDefinition", "state"): "[State](#json-schema-state-statetypejson_schema)",
-    ("FlowDefinition", "methods"): "map of string to [Method](#method-methods)",
-    ("FlowMethodDefinition", "do"): "[Action](#action)",
-    ("FlowCrewActionDefinition", "with"): "inline crew definition",
-    ("CrewAgentDefinition", "llm"): "string or inline LLM config",
-    ("AgentDefinition", "llm"): "string or inline LLM config",
-}
-
-_TEMPLATES_DIR = Path(__file__).parent / "templates"
-_ENVIRONMENT = Environment(  # noqa: S701 - renders trusted Markdown, not HTML.
-    loader=FileSystemLoader(_TEMPLATES_DIR),
-    trim_blocks=True,
-    lstrip_blocks=True,
-    keep_trailing_newline=False,
-)
-
-
-def render_skill_markdown(
-    *,
-    skips: Sequence[str] = (),
-    examples_format: Literal["yaml", "json"] = "yaml",
-) -> str:
-    if examples_format not in ("yaml", "json"):
-        raise ValueError("Flow skill examples_format must be 'yaml' or 'json'")
-
-    skips_set = frozenset(skips)
-    rendered = _ENVIRONMENT.get_template("flow_definition_skill.md.j2").render(
-        template_context(skips_set, examples_format)
-    )
-    rendered = re.sub(r"\n{3,}", "\n\n", rendered)
-    return rendered.strip() + "\n"
-
-
-def template_context(
-    skips: frozenset[str], examples_format: Literal["yaml", "json"] = "yaml"
-) -> dict[str, Any]:
-    return {
-        "examples_format": examples_format,
-        "example": render_flow_example(examples_format),
-        "example_language": examples_format,
-        "include_each_action": "each" not in skips,
-        "include_conversational": "conversational" not in skips,
-        "include_hitl": "hitl" not in skips,
-        "include_non_linear_flows": "non_linear_flows" not in skips,
-        "include_persistence": "persistence" not in skips,
-        "include_expression_action": "expression_action" not in skips,
-        "include_script_action": "script_action" not in skips,
-        "include_tool_action": "tool_action" not in skips,
-        "expression_contract_examples": FLOW_TEMPLATE_EXPRESSION_EXAMPLES[
-            examples_format
-        ],
-        "expression_contract_rules": FLOW_TEMPLATE_EXPRESSION_RULES,
-        "sections": FlowSkillReferenceExtractor(skips=skips).extract(),
-    }
-
-
-def render_flow_example(examples_format: Literal["yaml", "json"]) -> str:
-    example_yaml = (_TEMPLATES_DIR / "flow_definition_example.yaml").read_text(
-        encoding="utf-8"
-    )
-    if examples_format == "json":
-        return json.dumps(yaml.safe_load(example_yaml), indent=2)
-    return example_yaml.rstrip()
-
-
-@dataclass(frozen=True)
-class ModelSpec:
-    name: str
-    section: str
-    address: str = ""
-    label: str = ""
-    hidden: bool = False
-    examples: bool = False
-    descriptions: dict[str, str] = field(default_factory=dict)
-
-    @property
-    def display_title(self) -> str:
-        return self.label or MODEL_TITLES.get(self.name, self.section)
-
-    @property
-    def display_label(self) -> str:
-        if not self.address:
-            return self.display_title
-        return f"{self.display_title} (`{self.address}`)"
-
-
-MODEL_TITLES = {
-    "FlowDefinition": "Flow Definition",
-    "FlowDictStateDefinition": "Dict State",
-    "FlowPydanticStateDefinition": "Pydantic State",
-    "FlowJsonSchemaStateDefinition": "JSON Schema State",
-    "FlowUnknownStateDefinition": "Unknown State",
-    "FlowMethodDefinition": "Method",
-    "FlowCodeActionDefinition": "Code Action",
-    "FlowScriptActionDefinition": "Script Action",
-    "FlowToolActionDefinition": "Tool Action",
-    "FlowCrewActionDefinition": "Crew Action",
-    "FlowAgentActionDefinition": "Agent Action",
-    "FlowExpressionActionDefinition": "Expression Action",
-    "FlowEachActionDefinition": "Each Action",
-    "FlowEachStepDefinition": "Each Step",
-    "CrewDefinition": "Crew Definition",
-    "CrewAgentDefinition": "Crew Agent Definition",
-    "CrewTaskDefinition": "Crew Task Definition",
-    "AgentDefinition": "Agent Definition",
-    "LLMDefinition": "LLM Definition",
-    "FlowConfigDefinition": "Config",
-    "FlowPersistenceDefinition": "Persistence",
-    "FlowHumanFeedbackDefinition": "Human Feedback",
-}
-
-
-MODEL_SPECS: tuple[ModelSpec, ...] = (
-    ModelSpec(
-        "FlowDefinition",
-        "Flow Definition",
-        descriptions={
-            "schema": "Declarative Flow schema identifier and version. Include it explicitly in authored declarations.",
-            "conversational": "Top-level conversational flow configuration, only when the flow supports chat.",
-        },
-    ),
-    ModelSpec("FlowDictStateDefinition", "State", "state[type=dict]", hidden=True),
-    ModelSpec(
-        "FlowPydanticStateDefinition", "State", "state[type=pydantic]", hidden=True
-    ),
-    ModelSpec(
-        "FlowJsonSchemaStateDefinition",
-        "State",
-        "state[type=json_schema]",
-        descriptions={
-            "json_schema": "JSON Schema used to validate and document flow state. Declare required fields with JSON Schema's `required` array.",
-            "default": "Default values used to initialize Flow state. Defaults are not the same as schema-required fields.",
-        },
-    ),
-    ModelSpec(
-        "FlowUnknownStateDefinition", "State", "state[type=unknown]", hidden=True
-    ),
-    ModelSpec(
-        "FlowMethodDefinition",
-        "Method",
-        "methods.<name>",
-        descriptions={
-            "do": "Single action object executed when this method runs.",
-            "start": "Marks a start method. Use `true` for the normal entrypoint. String or map conditions are advanced trigger conditions; use them only when the user asks for event/condition-based starts.",
-            "listen": 'Trigger condition that runs this method after upstream events. A string target can be a method name or a router-emitted event name, and both live in the same trigger namespace. Methods must not listen to their own method name. Map conditions are for `and`/`or` trigger composition, for example `{"and": ["validated", "processed"]}`.',
-            "router": "Whether the method output should be treated as the next event name. Router actions must return one event name string, with no surrounding explanation.",
-            "emit": "Declared router events this method may emit. Each emitted event name should be unique and should not collide with method names.",
-        },
-    ),
-    ModelSpec(
-        "FlowCodeActionDefinition",
-        "Action",
-        "methods.<name>.do[call=code]",
-        hidden=True,
-    ),
-    ModelSpec("FlowScriptActionDefinition", "Action", "methods.<name>.do[call=script]"),
-    ModelSpec("FlowToolActionDefinition", "Action", "methods.<name>.do[call=tool]"),
-    ModelSpec(
-        "FlowCrewActionDefinition",
-        "Action",
-        "methods.<name>.do[call=crew]",
-        examples=True,
-        descriptions={
-            "call": "Action discriminator. Use crew to run an inline Crew definition.",
-            "inputs": f"Actual kickoff inputs passed to the Crew. {FLOW_TEMPLATE_EXPRESSION_CONTRACT} The evaluated values are available to crew agent and task interpolation as `{{name}}` placeholders; reference each input the crew needs in agent or task text.",
-        },
-    ),
-    ModelSpec(
-        "FlowAgentActionDefinition",
-        "Action",
-        "methods.<name>.do[call=agent]",
-        examples=True,
-        descriptions={
-            "with": "Individual Agent definition to load and execute outside of a crew for this action. Put the agent input in `with.input`; agent actions do not support action-level `inputs`.",
-        },
-    ),
-    ModelSpec(
-        "FlowExpressionActionDefinition",
-        "Action",
-        "methods.<name>.do[call=expression]",
-    ),
-    ModelSpec("FlowEachActionDefinition", "Action", "methods.<name>.do[call=each]"),
-    ModelSpec(
-        "FlowEachStepDefinition",
-        "Each Step",
-        "methods.<name>.do[call=each].do[]",
-    ),
-    ModelSpec(
-        "CrewDefinition",
-        "Crew Definition",
-        "methods.<name>.do[call=crew].with",
-        hidden=True,
-        examples=True,
-        descriptions={
-            "inputs": "Static default crew inputs. Values are available to crew agent and task interpolation as `{name}` placeholders, for example `{topic}`. Prefer action-level crew `inputs` for runtime values from `state` or `outputs`, and include placeholders for any inputs the crew must reason over.",
-            "manager_agent": "Optional manager agent name.",
-        },
-    ),
-    ModelSpec(
-        "CrewAgentDefinition",
-        "Crew Agent Definition",
-        "methods.<name>.do[call=crew].with.agents.<name>",
-        hidden=True,
-        examples=True,
-        descriptions={
-            "llm": "Language model that runs this crew agent. Use an object when setting LLM options such as `max_tokens`.",
-            "planning_config": "Agent planning configuration. Set `max_attempts` to limit planning refinement attempts before task execution.",
-        },
-    ),
-    ModelSpec(
-        "LLMDefinition",
-        "LLM Definition",
-        hidden=True,
-        examples=True,
-    ),
-    ModelSpec(
-        "CrewTaskDefinition",
-        "Crew Task Definition",
-        "methods.<name>.do[call=crew].with.tasks[]",
-        hidden=True,
-        examples=True,
-        descriptions={
-            "name": "Optional task name.",
-        },
-    ),
-    ModelSpec(
-        "AgentDefinition",
-        "Agent Definition",
-        "methods.<name>.do[call=agent].with",
-        hidden=True,
-        examples=True,
-        descriptions={
-            "input": f"Input passed to the individual agent kickoff outside of a crew. Use one string. {FLOW_TEMPLATE_EXPRESSION_CONTRACT} When an agent needs multiple fields, write one string with labels and separators, for example `Ticket ID: ${{state.ticket_id}}; Message: ${{state.message}}`.",
-            "llm": "Language model that runs this agent. Use an object when setting LLM options such as `max_tokens`.",
-            "planning_config": "Agent planning configuration. Set `max_attempts` to limit planning refinement attempts before task execution.",
-        },
-    ),
-    ModelSpec("FlowConfigDefinition", "Config", "config"),
-    ModelSpec("FlowPersistenceDefinition", "Persistence", "persist"),
-    ModelSpec(
-        "FlowHumanFeedbackDefinition",
-        "Human Feedback",
-        "methods.<name>.human_feedback",
-    ),
-)
-
-_SPECS_BY_NAME: dict[str, ModelSpec] = {spec.name: spec for spec in MODEL_SPECS}
-
-
-@dataclass(frozen=True)
-class FlowSkillReferenceExtractor:
-    skips: frozenset[str]
-    schema: dict[str, Any] = field(
-        default_factory=lambda: FlowDefinition.model_json_schema(by_alias=True)
-    )
-
-    def extract(self) -> list[dict[str, Any]]:
-        sections: list[dict[str, Any]] = []
-
-        for spec in MODEL_SPECS:
-            if spec.hidden or self.model_is_skipped(spec.name):
-                continue
-
-            if not sections or sections[-1]["label"] != spec.section:
-                sections.append(
-                    {
-                        "label": spec.section,
-                        "models": [],
-                        "kind": "union" if spec.section == "Action" else "object",
-                    }
-                )
-            sections[-1]["models"].append(self.extract_model(spec))
-
-        for section in sections:
-            if section["kind"] != "union" and section["models"]:
-                section["label"] = section["models"][0]["label"]
-
-        return sections
-
-    def extract_model(self, spec: ModelSpec) -> dict[str, Any]:
-        model_name = spec.name
-        model_schema = (
-            self.schema
-            if model_name == "FlowDefinition"
-            else self.schema["$defs"][model_name]
-        )
-        required_from_schema = set(model_schema.get("required", ()))
-        fields = []
-
-        for field_name, field_schema in model_schema.get("properties", {}).items():
-            if self.field_is_hidden(model_name, field_name):
-                continue
-
-            required = (
-                field_name in required_from_schema
-                or (
-                    model_name == "FlowDefinition"
-                    and field_name in ("state", "methods")
-                )
-                or (model_name == "FlowCrewActionDefinition" and field_name == "with")
-            )
-            fields.append(
-                {
-                    "name": field_name,
-                    "type": self.render_field_type(
-                        model_name, field_name, field_schema
-                    ),
-                    "required": required,
-                    "default": render_field_default(
-                        model_name, field_name, field_schema, required
-                    ),
-                    "description": self.render_field_description(
-                        spec, model_name, field_name, field_schema
-                    ),
-                    "examples": render_field_examples(spec, field_name, field_schema),
-                }
-            )
-
-        return {
-            "label": spec.display_label,
-            "anchor": f"#{markdown_heading_anchor(spec.display_label)}",
-            "link": (
-                f"[{spec.display_label}](#{markdown_heading_anchor(spec.display_label)})"
-            ),
-            "discriminator": extract_discriminator(model_schema),
-            "fields": fields,
-            "inline_models": self.inline_models_for(model_name),
-        }
-
-    def inline_models_for(self, model_name: str) -> list[dict[str, Any]]:
-        names_by_model = {
-            "FlowCrewActionDefinition": (
-                "CrewDefinition",
-                "CrewAgentDefinition",
-                "CrewTaskDefinition",
-            ),
-            "CrewAgentDefinition": ("LLMDefinition",),
-            "FlowAgentActionDefinition": ("AgentDefinition",),
-            "AgentDefinition": ("LLMDefinition",),
-        }
-        return [
-            self.extract_model(_SPECS_BY_NAME[name])
-            for name in names_by_model.get(model_name, ())
-        ]
-
-    def model_is_skipped(self, model_name: str) -> bool:
-        skip = SKIP_BY_MODEL.get(model_name)
-        return skip in self.skips if skip is not None else False
-
-    def field_is_hidden(
-        self,
-        model_name: str,
-        field_name: str,
-    ) -> bool:
-        return (
-            ("hitl" in self.skips and field_name == "human_feedback")
-            or ("persistence" in self.skips and field_name == "persist")
-            or ("config" in self.skips and field_name == "config")
-            or ("conversational" in self.skips and field_name == "conversational")
-            or (model_name == "AgentDefinition" and field_name == "response_format")
-            or (model_name == "CrewDefinition" and field_name == "manager_agent")
-            or (model_name == "CrewTaskDefinition" and field_name == "context")
-            or (
-                field_name == "type"
-                and model_name
-                in {"AgentDefinition", "CrewAgentDefinition", "CrewTaskDefinition"}
-            )
-            or (field_name == "ref" and model_name != "FlowToolActionDefinition")
-            or (
-                model_name == "FlowCrewActionDefinition"
-                and field_name == "from_declaration"
-            )
-        )
-
-    def render_field_type(
-        self,
-        model_name: str,
-        field_name: str,
-        field_schema: dict[str, Any],
-    ) -> str:
-        if override := FIELD_TYPE_OVERRIDES.get((model_name, field_name)):
-            return override
-        return self.render_schema_type(field_schema) or "any"
-
-    def render_schema_type(self, field_schema: dict[str, Any]) -> str | None:
-        if "$ref" in field_schema:
-            return self.render_schema_ref(field_schema["$ref"])
-        if "const" in field_schema:
-            return f"must be {format_inline_value(field_schema['const'])}"
-        if "enum" in field_schema:
-            values = ", ".join(
-                format_inline_value(value) for value in field_schema["enum"]
-            )
-            return f"one of {values}"
-
-        for union_key in ("anyOf", "oneOf", "allOf"):
-            if union_key in field_schema:
-                return join_unique(
-                    self.render_schema_type(option)
-                    for option in field_schema[union_key]
-                )
-
-        json_type = field_schema.get("type")
-        if isinstance(json_type, list):
-            return join_unique(
-                self.render_schema_type({"type": item}) for item in json_type
-            )
-        if json_type == "array":
-            item_type = self.render_schema_type(field_schema.get("items", {})) or "any"
-            return (
-                f"list of {item_type}"
-                if item_type.startswith("[")
-                else f"list[{item_type}]"
-            )
-        if json_type == "object":
-            additional_properties = field_schema.get("additionalProperties")
-            if isinstance(additional_properties, dict):
-                value_type = self.render_schema_type(additional_properties) or "any"
-                return f"map of string to {value_type}"
-            return "map of string to any" if additional_properties is True else "object"
-        if isinstance(json_type, str):
-            return json_type
-        return "object" if "properties" in field_schema else "any"
-
-    def render_schema_ref(self, ref: str) -> str | None:
-        schema_name = ref.rsplit("/", 1)[-1]
-        if schema_name == "ExpressionData":
-            return (
-                "expression data"
-                if "expression_action" not in self.skips
-                else "dynamic value"
-            )
-        if schema_name == "PythonReferenceDefinition":
-            return None
-        spec = _SPECS_BY_NAME.get(schema_name)
-        if (spec and spec.hidden) or self.model_is_skipped(schema_name):
-            return None
-        if spec is None:
-            return "object"
-        return f"[{spec.display_label}](#{markdown_heading_anchor(spec.display_label)})"
-
-    def render_field_description(
-        self,
-        spec: ModelSpec,
-        model_name: str,
-        field_name: str,
-        field_schema: dict[str, Any],
-    ) -> str | None:
-        if "non_linear_flows" in self.skips and model_name == "FlowMethodDefinition":
-            if field_name == "start":
-                return "Marks the single normal entrypoint. Use `true`."
-            if field_name == "listen":
-                return "Runs this method after one upstream method or router-emitted event."
-        return render_field_description(spec, field_name, field_schema)
-
-
-def render_field_default(
-    model_name: str,
-    field_name: str,
-    field_schema: dict[str, Any],
-    required: bool,
-) -> str | None:
-    if required:
-        return None
-    if model_name == "FlowDefinition" and field_name == "config":
-        return "generated default"
-    if "default" in field_schema:
-        return format_inline_value(field_schema["default"])
-    return None
-
-
-def extract_discriminator(model_schema: dict[str, Any]) -> dict[str, str] | None:
-    properties = model_schema.get("properties", {})
-    for field_name in ("call", "type"):
-        if field_name not in properties:
-            continue
-        value = properties[field_name].get(
-            "const", properties[field_name].get("default")
-        )
-        if value is not None:
-            return {"name": field_name, "value": str(value)}
-    return None
-
-
-def join_unique(values: Any) -> str | None:
-    rendered_values = list(
-        dict.fromkeys(value for value in values if value is not None)
-    )
-    return " | ".join(rendered_values) or None
-
-
-def markdown_heading_anchor(text: str) -> str:
-    heading = re.sub(r"<[^>]+>", "", text)
-    heading = re.sub(r"`([^`]*)`", r"\1", heading)
-    heading = heading.lower()
-    heading = re.sub(r"[^\w\s-]", "", heading)
-    return re.sub(r"\s+", "-", heading.strip())
-
-
-def format_inline_value(value: Any) -> str:
-    if value is None:
-        return "`null`"
-    if isinstance(value, bool):
-        return f"`{str(value).lower()}`"
-    return f"`{value}`"
-
-
-def render_field_description(
-    spec: ModelSpec, field_name: str, field_schema: dict[str, Any]
-) -> str | None:
-    if field_name in spec.descriptions:
-        return spec.descriptions[field_name]
-    return field_schema.get("description")
-
-
-def render_field_examples(
-    spec: ModelSpec, field_name: str, field_schema: dict[str, Any]
-) -> list[str]:
-    if not spec.examples:
-        return []
-
-    examples = (
-        example
-        for example in field_schema.get("examples", ())
-        if not contains_python_reference(example)
-    )
-    return [format_inline_example(example) for example in examples]
-
-
-def contains_python_reference(value: Any) -> bool:
-    if isinstance(value, dict):
-        return "python" in value or any(
-            contains_python_reference(item) for item in value.values()
-        )
-    if isinstance(value, list):
-        return any(contains_python_reference(item) for item in value)
-    return False
-
-
-def format_inline_example(value: Any) -> str:
-    if isinstance(value, str):
-        return format_inline_value(value.replace("\n", "\\n"))
-    if value is None or isinstance(value, (bool, int, float)):
-        return format_inline_value(value)
-    return f"`{json.dumps(value, ensure_ascii=True)}`"

lib/crewai/src/crewai/flow/templates/flow_definition_example.yaml

@@ -1,69 +0,0 @@
-schema: crewai.flow/v1
-name: ResearchReviewFlow
-state:
-  type: json_schema
-  json_schema:
-    type: object
-    properties:
-      topic:
-        type: string
-      audience:
-        type: string
-    required:
-      - topic
-      - audience
-  default:
-    topic: AI agent orchestration
-    audience: platform engineering leaders
-methods:
-  research_brief:
-    start: true
-    do:
-      call: crew
-      with:
-        agents:
-          researcher:
-            role: Research analyst
-            goal: Research {topic} for {audience}
-            backstory: Expert at concise technical research.
-          reviewer:
-            role: Strategy reviewer
-            goal: Decide whether the research needs an executive follow-up
-            backstory: Experienced at reviewing technical briefs for leaders.
-        tasks:
-          - name: research_task
-            description: Research {topic} for {audience}.
-            expected_output: Key findings and tradeoffs.
-            agent: researcher
-          - name: review_task
-            description: Review the research and decide if an executive follow-up is needed.
-            expected_output: 'A brief review ending with `needs_followup: true` or `needs_followup: false`.'
-            agent: reviewer
-        inputs:
-          topic: Default topic
-          audience: Default audience
-      inputs:
-        topic: "${state.topic}"
-        audience: "${state.audience}"
-  route_followup:
-    listen: research_brief
-    router: true
-    emit:
-      - followup
-      - done
-    do:
-      call: agent
-      with:
-        role: Follow-up router
-        goal: 'Return exactly one bare value: followup or done. Do not include explanation.'
-        backstory: Skilled at routing reviewed research briefs.
-        input: "Reviewed research: ${outputs.research_brief.raw}"
-  write_followup:
-    listen: followup
-    do:
-      call: agent
-      with:
-        role: Executive communications specialist
-        goal: Draft a concise executive follow-up from the reviewed research
-        backstory: Writes crisp follow-ups for technical leaders.
-        input: "${outputs.research_brief.raw}"

lib/crewai/src/crewai/flow/templates/flow_definition_skill.md.j2

@@ -1,217 +0,0 @@
----
-name: flow-definition
-description: Create or edit CrewAI Flow declarations. Use when the user needs a YAML or JSON flow with methods, state, agents, crews, tools, outputs, or conditional branches.
----
-
-# Flow Definition
-
-You are writing a CrewAI Flow declaration for the user.
-Use these instructions when the user asks you to create or edit a Flow.
-Return one valid `crewai.flow/v1` YAML or JSON document.
-
-Treat this document as instructions for you, not as text to show the user.
-Follow the examples for shape and formatting, then use the API reference to check exact fields.
-
-## Output Format
-
-Return one valid `crewai.flow/v1` Flow declaration.
-Do not include explanatory prose unless the user asks for it.
-
-## Build It In This Order
-
-1. Define `state` first. Use `type: json_schema` and put the JSON Schema inline.
-2. Put required input fields in `state.json_schema.required`. Do not rely on `state.default` to make fields required.
-{% if include_non_linear_flows %}
-3. Add at least one method with `start: true`.
-{% else %}
-3. Add exactly one method with `start: true`.
-{% endif %}
-4. Add later methods with `listen`.
-5. Give each method exactly one `do` action object. Never make `do` a list.
-6. Pass data with `${...}` mappings from `state` and completed `outputs`.
-7. Before final output, check every `listen`, `emit`, and `outputs.some_method` reference.
-
-Set optional fields only when you are confident they are needed. Otherwise, trust CrewAI defaults and omit them.
-
-Method names must match `^[A-Za-z_][A-Za-z0-9_]*$`.
-
-## Choose One Action Per Method
-
-Pick the simplest action that does the job.
-
-{% if include_expression_action %}
-- Use `call: expression` for simple reads, filters, computed values, and deterministic routing.
-{% endif %}
-{% if include_tool_action %}
-- Use `call: tool` for packaged deterministic work: API calls, searches, lookups, scoring, file work, or custom CrewAI tools.
-{% endif %}
-- Use `call: agent` for one AI worker that classifies, decides, summarizes, writes, or drafts. Put `role`, `goal`, `backstory`, and `input` under `with`. Do not add an action-level `inputs` map to an agent.
-- Use `call: crew` for coordinated AI work with multiple agents or tasks. Define the crew under `with`. Pass runtime values with the action-level `inputs` map.
-{% if include_each_action %}
-- Use `call: each` when the same ordered mini-pipeline must run once per item. Give every step a `name`.
-{% endif %}
-{% if include_hitl %}
-- Use `human_feedback` when a method needs a human checkpoint.
-{% endif %}
-{% if include_script_action %}
-- Use `call: script` only for trusted inline Python. Scripts are not sandboxed.
-{% endif %}
-
-## Wire Methods Explicitly
-
-- `state` is the initial shared data shape. Action results do not automatically merge into `state`.
-- Read method results with `outputs.method_name` after that method can run.
-- `listen` targets a method name or a router-emitted event name.
-- Methods must not listen to their own method name.
-- Method names and emitted event names share one namespace. Avoid reusing the same string for both unless the user explicitly wants that.
-- Use `router: true` plus `emit` when one method chooses between named branches.
-- A router action must return exactly one emitted event string. It must not return JSON, a list, or an explanation.
-{% if include_non_linear_flows %}
-- Conditional `listen` values use `and` / `or`, for example `listen: {and: [validated, enriched]}`. They are not CEL.
-- Use `start: true` for normal entrypoints. Use conditional `start` only for advanced event-driven starts.
-{% else %}
-- Use `start: true` for the single entrypoint.
-{% endif %}
-
-If an agent is a router, make its goal say exactly what to return, for example:
-`Return exactly one bare value: approved, rejected, or needs_review. Do not include explanation.`
-{% if include_expression_action %}
-Prefer `call: expression` when routing can be computed without an agent.
-{% endif %}
-
-## CEL And Dynamic Values
-
-CEL is the expression language for reading Flow data and making small decisions.
-Use {% if include_tool_action %}tools, {% endif %}agents and crews{% if include_script_action %}, or trusted scripts{% endif %} for larger work or side effects.
-
-Use these expression forms correctly:
-
-{% if include_expression_action or include_each_action %}
-- Raw CEL: use in {% if include_expression_action and include_each_action %}`expr`, `in`, and `if`{% elif include_expression_action %}`expr`{% else %}`in` and `if`{% endif %}. Do not wrap raw CEL in `${...}`.
-{% endif %}
-{% for rule in expression_contract_rules %}
-- {{ rule }}
-{% endfor %}
-
-Expression examples:
-{% for example in expression_contract_examples %}
-
-{{ example.title }}:
-
-```{{ example_language }}
-{{ example.code }}
-```
-{% endfor %}
-
-- Crew text: use `{name}` placeholders from crew inputs. Example: `Research {topic}`.
-- Crew inputs become prompt text only when agent or task text references matching `{name}` placeholders.
-- Passing an input that is not referenced by any `{name}` placeholder does not ground the crew. If the crew needs a field, put that placeholder in an agent `goal`, task `description`, or task `expected_output`.
-
-Available CEL variables:
-
-- `state`: initial input data, for example `state.ticket.subject`.
-- `outputs`: completed method outputs, for example `outputs.classify_ticket`.
-{% if include_each_action %}
-- `item`: the current item inside `each`, for example `item.company_domain`.
-- `outputs`: inside `each`, prior step outputs for the current item, for example `outputs.enrich`.
-{% endif %}
-
-Dynamic value rules:
-
-- When an agent needs multiple fields, write one text value with labels and separators. Example value: `Ticket ID: ${state.ticket_id}; Message: ${state.message}`.
-- Crew action-level `inputs` are the actual Crew kickoff inputs. Use `${...}` values there for runtime data from `state` or `outputs`.
-- Crew action-level `inputs` alone are not grounding. Include placeholders for the facts the model must use.
-- Crew outputs are objects. Use `${outputs.research_brief.raw}` for text.
-- For structured crew output, use fields like `${outputs.research_brief.json_dict.field}` or `${outputs.research_brief.pydantic.field}`.
-- Do not pass a whole crew output to an agent input, like `${outputs.research_brief}`.
-- Agent outputs may also be objects. Use fields like `${outputs.classify_ticket.raw}` or `${outputs.classify_ticket.pydantic.category}`.
-- Use `with.inputs` only for static Crew input defaults.
-- Agent action `with.input` is the agent's single input value.
-{% if include_script_action %}
-- Trusted `script` actions can mutate state explicitly. Use them only when the user asks for that behavior.
-{% endif %}
-
-## Do Not
-
-- Do not invent top-level keys outside the Flow declaration shape.
-- Do not use fields outside the declaration schema{% if include_tool_action %} or tool refs shown here{% endif %}.
-- Do not put more than one action under a method's `do`.
-- Do not make `do` a list.
-- Do not reference `outputs.some_method` before `some_method` can run.
-- Do not set a method's `listen` to its own method name.
-- Do not use the same string for an emitted event and a method name unless the user asks for it.
-- Do not use `emit` without `router: true`{% if include_hitl %} or `human_feedback.emit`{% endif %}.
-- Do not rely on crew action-level `inputs` alone to ground agent behavior. Inputs that do not match placeholders are effectively unused by the prompt.
-- Do not ask agents to infer missing facts when accuracy matters. Tell them to mark missing dates, amounts, offers, logs, or constraints as unknown.
-- Do not set `config.stream: true` unless the caller is expected to consume a streaming result. For normal generated flows and CLI smoke tests, omit it.
-{% if include_conversational %}
-- Do not put conversational settings under `state`, `config`, or a method. Use top-level `conversational` only when the user asks for a chat or conversation flow.
-{% endif %}
-{% if include_each_action %}
-- Do not use `each` without at least one named step.
-{% endif %}
-{% if include_script_action %}
-- Do not use `script` for untrusted input or user-authored code.
-{% endif %}
-
-## Examples
-
-### Crew review with routed follow-up
-
-```{{ example_language }}
-{{ example }}
-```
-
-## API Reference
-
-Use this appendix to check exact field names, required fields, linked object types, and allowed action/state shapes. Linked type names point to another section in this reference.
-{% macro render_model(model) -%}
-{% if model.discriminator %}
-
-Shape:
-- `{{ model.discriminator.name }}: {{ model.discriminator.value }}`
-{% endif %}
-
-Fields:
-{% for field in model.fields %}
-- `{{ field.name }}` ({% if field.required %}required{% else %}optional{% endif %}): {{ field.type }}{% if field.default %}; default {{ field.default }}{% endif %}{% if field.description %}. {{ field.description }}{% endif %}{% if field.examples %}{% if not field.description %}.{% endif %} Example{% if field.examples|length > 1 %}s{% endif %}: {{ field.examples | join(", ") }}{% endif +%}
-{% endfor %}
-{% for inline_model in model.inline_models %}
-
-#### {{ inline_model.label }}
-{{ render_model(inline_model) }}
-{% endfor %}
-{%- endmacro %}
-{% for section in sections %}
-
-### {{ section.label }}
-{% if section.kind == "union" %}
-
-Discriminated union by `{% if section.label == "State" %}type{% else %}call{% endif %}`.
-
-Allowed shapes:
-{% for model in section.models %}
-- [`{{ model.discriminator.name }}: {{ model.discriminator.value }}`]({{ model.anchor }})
-{% endfor %}
-{% for model in section.models %}
-
-### {{ model.label }}
-{{ render_model(model) }}
-{% endfor %}
-{% else %}
-{{ render_model(section.models[0]) }}
-{% endif %}
-
-{% endfor %}
-### Cross-Field Rules
-
-- A method has exactly one `do` action object with one `call` discriminator.
-- `listen` targets method names and router-emitted event names in one shared namespace.
-- Methods cannot listen to their own method name.
-- A router method result must match one declared `emit` value.
-- Crew action-level `inputs` are the Crew kickoff inputs; use CEL-wrapped strings there for runtime values.
-- Crew agent/task interpolation uses `{name}` placeholders from evaluated crew inputs.
-- Agent `with.input` must be text. Use `${outputs.method_name.raw}` or a text field like `${outputs.method_name.json_dict.summary}`.
-{% if include_each_action %}
-- `each.do` must contain at least one named step.
-{% endif %}

lib/crewai/src/crewai/llm.py

@@ -749,7 +749,7 @@ class LLM(BaseLLM):
             "base_url": self.base_url,
             "api_version": self.api_version,
             "api_key": self.api_key,
-            "stream": self._effective_stream(),
+            "stream": self.stream,
             "tools": tools,
             "reasoning_effort": self.reasoning_effort,
             **self.additional_params,
@@ -1841,7 +1841,7 @@ class LLM(BaseLLM):
                     self.set_callbacks(callbacks)
                 try:
                     params = self._prepare_completion_params(messages, tools)
-                    if self._effective_stream():
+                    if self.stream:
                         result = self._handle_streaming_response(
                             params=params,
                             callbacks=callbacks,
@@ -1983,7 +1983,7 @@ class LLM(BaseLLM):
                         messages, tools, skip_file_processing=True
                     )
 
-                    if self._effective_stream():
+                    if self.stream:
                         return await self._ahandle_streaming_response(
                             params=params,
                             callbacks=callbacks,

lib/crewai/src/crewai/llms/base_llm.py

@@ -42,13 +42,8 @@ from crewai.events.types.tool_usage_events import (
     ToolUsageFinishedEvent,
     ToolUsageStartedEvent,
 )
-from crewai.types.streaming import StreamSession
 from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities.pydantic_schema_utils import serialize_model_class
-from crewai.utilities.streaming import (
-    create_frame_generator,
-    create_frame_streaming_state,
-)
 
 
 try:
@@ -82,9 +77,6 @@ _current_call_id: contextvars.ContextVar[str | None] = contextvars.ContextVar(
 _call_stop_override_var: contextvars.ContextVar[dict[int, list[str]] | None] = (
     contextvars.ContextVar("_call_stop_override_var", default=None)
 )
-_call_stream_override_var: contextvars.ContextVar[dict[int, bool] | None] = (
-    contextvars.ContextVar("_call_stream_override_var", default=None)
-)
 
 
 @contextmanager
@@ -123,19 +115,6 @@ def call_stop_override(
         _call_stop_override_var.reset(token)
 
 
-@contextmanager
-def call_stream_override(llm: BaseLLM, stream: bool) -> Generator[None, None, None]:
-    """Override streaming for ``llm`` within the current call scope."""
-    current = _call_stream_override_var.get()
-    new_overrides: dict[int, bool] = dict(current) if current else {}
-    new_overrides[id(llm)] = stream
-    token = _call_stream_override_var.set(new_overrides)
-    try:
-        yield
-    finally:
-        _call_stream_override_var.reset(token)
-
-
 def get_current_call_id() -> str:
     """Get current call_id from context"""
     call_id = _current_call_id.get()
@@ -234,13 +213,6 @@ class BaseLLM(BaseModel, ABC):
                 return override
         return self.stop
 
-    def _effective_stream(self) -> bool | None:
-        """Return the call-scoped streaming mode for this instance."""
-        overrides = _call_stream_override_var.get()
-        if overrides is not None and id(self) in overrides:
-            return overrides[id(self)]
-        return self.stream
-
     _token_usage: dict[str, int] = PrivateAttr(
         default_factory=lambda: {
             "total_tokens": 0,
@@ -346,39 +318,6 @@ class BaseLLM(BaseModel, ABC):
             RuntimeError: If the LLM request fails for other reasons.
         """
 
-    def stream_events(
-        self,
-        messages: str | list[LLMMessage],
-        tools: list[dict[str, BaseTool]] | None = None,
-        callbacks: list[Any] | None = None,
-        available_functions: dict[str, Any] | None = None,
-        from_task: Task | None = None,
-        from_agent: BaseAgent | None = None,
-        response_model: type[BaseModel] | None = None,
-    ) -> StreamSession[Any]:
-        """Run the LLM call and stream scoped public ``StreamFrame`` events."""
-        result_holder: list[Any] = []
-        state = create_frame_streaming_state(result_holder, use_async=False)
-        output_holder: list[StreamSession[Any]] = []
-
-        def run_llm_call() -> Any:
-            with call_stream_override(self, True):
-                return self.call(
-                    messages=messages,
-                    tools=tools,
-                    callbacks=callbacks,
-                    available_functions=available_functions,
-                    from_task=from_task,
-                    from_agent=from_agent,
-                    response_model=response_model,
-                )
-
-        stream_session: StreamSession[Any] = StreamSession(
-            sync_iterator=create_frame_generator(state, run_llm_call, output_holder)
-        )
-        output_holder.append(stream_session)
-        return stream_session
-
     async def acall(
         self,
         messages: str | list[LLMMessage],
@@ -570,7 +509,7 @@ class BaseLLM(BaseModel, ABC):
         if max_tokens is None:
             max_tokens = self._effective_max_tokens()
         if stream is None:
-            stream = self._effective_stream()
+            stream = self.stream
         if seed is None:
             seed = self.seed
         if stop_sequences is None:

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

@@ -323,7 +323,7 @@ class AnthropicCompletion(BaseLLM):
 
                 effective_response_model = response_model or self.response_format
 
-                if self._effective_stream():
+                if self.stream:
                     return self._handle_streaming_completion(
                         completion_params,
                         available_functions,
@@ -393,7 +393,7 @@ class AnthropicCompletion(BaseLLM):
 
                 effective_response_model = response_model or self.response_format
 
-                if self._effective_stream():
+                if self.stream:
                     return await self._ahandle_streaming_completion(
                         completion_params,
                         available_functions,
@@ -441,7 +441,7 @@ class AnthropicCompletion(BaseLLM):
             "model": self.model,
             "messages": messages,
             "max_tokens": self.max_tokens,
-            "stream": self._effective_stream(),
+            "stream": self.stream,
         }
 
         if system_message:

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

@@ -42,7 +42,7 @@ try:
     )
 
     from crewai.events.types.llm_events import LLMCallType
-    from crewai.llms.base_llm import BaseLLM, call_stream_override, llm_call_context
+    from crewai.llms.base_llm import BaseLLM, llm_call_context
 
 except ImportError:
     raise ImportError(
@@ -493,18 +493,15 @@ class AzureCompletion(BaseLLM):
             Completion response or tool call result
         """
         if self.api == "responses":
-            with call_stream_override(
-                self._responses_delegate, bool(self._effective_stream())
-            ):
-                return self._responses_delegate.call(
-                    messages=messages,
-                    tools=tools,
-                    callbacks=callbacks,
-                    available_functions=available_functions,
-                    from_task=from_task,
-                    from_agent=from_agent,
-                    response_model=response_model,
-                )
+            return self._responses_delegate.call(
+                messages=messages,
+                tools=tools,
+                callbacks=callbacks,
+                available_functions=available_functions,
+                from_task=from_task,
+                from_agent=from_agent,
+                response_model=response_model,
+            )
 
         with llm_call_context():
             try:
@@ -530,7 +527,7 @@ class AzureCompletion(BaseLLM):
                     formatted_messages, tools, effective_response_model
                 )
 
-                if self._effective_stream():
+                if self.stream:
                     return self._handle_streaming_completion(
                         completion_params,
                         available_functions,
@@ -575,18 +572,15 @@ class AzureCompletion(BaseLLM):
             Completion response or tool call result
         """
         if self.api == "responses":
-            with call_stream_override(
-                self._responses_delegate, bool(self._effective_stream())
-            ):
-                return await self._responses_delegate.acall(
-                    messages=messages,
-                    tools=tools,
-                    callbacks=callbacks,
-                    available_functions=available_functions,
-                    from_task=from_task,
-                    from_agent=from_agent,
-                    response_model=response_model,
-                )
+            return await self._responses_delegate.acall(
+                messages=messages,
+                tools=tools,
+                callbacks=callbacks,
+                available_functions=available_functions,
+                from_task=from_task,
+                from_agent=from_agent,
+                response_model=response_model,
+            )
 
         with llm_call_context():
             try:
@@ -607,7 +601,7 @@ class AzureCompletion(BaseLLM):
                     formatted_messages, tools, effective_response_model
                 )
 
-                if self._effective_stream():
+                if self.stream:
                     return await self._ahandle_streaming_completion(
                         completion_params,
                         available_functions,
@@ -645,11 +639,11 @@ class AzureCompletion(BaseLLM):
         """
         params: AzureCompletionParams = {
             "messages": messages,
-            "stream": bool(self._effective_stream()),
+            "stream": self.stream,
         }
 
         model_extras: dict[str, Any] = {}
-        if self._effective_stream():
+        if self.stream:
             model_extras["stream_options"] = {"include_usage": True}
 
         if response_model and self.is_openai_model:

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

@@ -428,7 +428,7 @@ class BedrockCompletion(BaseLLM):
                         self.additional_model_response_field_paths
                     )
 
-                if self._effective_stream():
+                if self.stream:
                     return self._handle_streaming_converse(
                         formatted_messages,
                         body,
@@ -492,7 +492,7 @@ class BedrockCompletion(BaseLLM):
         if not AIOBOTOCORE_AVAILABLE:
             raise NotImplementedError(
                 "Async support for AWS Bedrock requires aiobotocore. "
-                'Install with: uv add "crewai[bedrock]"'
+                'Install with: uv add "crewai[bedrock-async]"'
             )
 
         with llm_call_context():
@@ -556,7 +556,7 @@ class BedrockCompletion(BaseLLM):
                         self.additional_model_response_field_paths
                     )
 
-                if self._effective_stream():
+                if self.stream:
                     return await self._ahandle_streaming_converse(
                         formatted_messages,
                         body,

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

@@ -322,7 +322,7 @@ class GeminiCompletion(BaseLLM):
                     system_instruction, tools, effective_response_model
                 )
 
-                if self._effective_stream():
+                if self.stream:
                     return self._handle_streaming_completion(
                         formatted_content,
                         config,
@@ -401,7 +401,7 @@ class GeminiCompletion(BaseLLM):
                     system_instruction, tools, effective_response_model
                 )
 
-                if self._effective_stream():
+                if self.stream:
                     return await self._ahandle_streaming_completion(
                         formatted_content,
                         config,

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

@@ -469,7 +469,7 @@ class OpenAICompletion(BaseLLM):
             messages=messages, tools=tools
         )
 
-        if self._effective_stream():
+        if self.stream:
             return self._handle_streaming_completion(
                 params=completion_params,
                 available_functions=available_functions,
@@ -564,7 +564,7 @@ class OpenAICompletion(BaseLLM):
             messages=messages, tools=tools
         )
 
-        if self._effective_stream():
+        if self.stream:
             return await self._ahandle_streaming_completion(
                 params=completion_params,
                 available_functions=available_functions,
@@ -595,7 +595,7 @@ class OpenAICompletion(BaseLLM):
             messages=messages, tools=tools, response_model=response_model
         )
 
-        if self._effective_stream():
+        if self.stream:
             return self._handle_streaming_responses(
                 params=params,
                 available_functions=available_functions,
@@ -626,7 +626,7 @@ class OpenAICompletion(BaseLLM):
             messages=messages, tools=tools, response_model=response_model
         )
 
-        if self._effective_stream():
+        if self.stream:
             return await self._ahandle_streaming_responses(
                 params=params,
                 available_functions=available_functions,
@@ -685,7 +685,7 @@ class OpenAICompletion(BaseLLM):
         if instructions:
             params["instructions"] = instructions
 
-        if self._effective_stream():
+        if self.stream:
             params["stream"] = True
 
         if self.store is not None:
@@ -1540,8 +1540,8 @@ class OpenAICompletion(BaseLLM):
             "model": self.model,
             "messages": messages,
         }
-        if self._effective_stream():
-            params["stream"] = self._effective_stream()
+        if self.stream:
+            params["stream"] = self.stream
             params["stream_options"] = {"include_usage": True}
 
         params.update(self.additional_params)

lib/crewai/src/crewai/project/crew_definition.py

@@ -6,15 +6,12 @@ from typing import Any, TypeAlias
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
-from crewai.agent.planning_config import PlanningConfig
-
 
 __all__ = [
     "AgentDefinition",
     "CrewAgentDefinition",
     "CrewDefinition",
     "CrewTaskDefinition",
-    "LLMDefinition",
     "PythonReferenceDefinition",
 ]
 
@@ -22,10 +19,7 @@ __all__ = [
 class PythonReferenceDefinition(BaseModel):
     """Dotted Python reference used by crew definitions."""
 
-    python: str = Field(
-        description="Dotted Python import path to load.",
-        examples=["my_project.schemas.SupportReply"],
-    )
+    python: str
 
     @field_validator("python")
     @classmethod
@@ -41,148 +35,16 @@ class PythonReferenceDefinition(BaseModel):
         return path
 
 
-class LLMDefinition(BaseModel):
-    """LLM configuration used by inline agent definitions."""
-
-    model_config = ConfigDict(extra="allow")
-
-    model: str = Field(
-        description="Model identifier used to instantiate the LLM.",
-        examples=["openai/gpt-4o-mini"],
-    )
-    max_tokens: int | None = Field(
-        default=None,
-        description=(
-            "Maximum number of tokens the LLM can generate. If null, CrewAI "
-            "does not set an explicit output token cap and the provider's "
-            "default applies."
-        ),
-        examples=[4096],
-    )
-
-
 class CrewAgentDefinition(BaseModel):
     """Inline agent definition used by a crew definition."""
 
     model_config = ConfigDict(extra="allow")
 
-    role: str | None = Field(
-        default=None,
-        description=(
-            "Crew agent role. Crew inputs are interpolated with `{name}` "
-            "placeholders such as `{topic}`; this is not CEL."
-        ),
-        examples=["Research analyst"],
-    )
-    goal: str | None = Field(
-        default=None,
-        description=(
-            "Crew agent goal. Crew inputs are interpolated with `{name}` "
-            "placeholders such as `{topic}`; this is not CEL."
-        ),
-        examples=["Research {topic}"],
-    )
-    backstory: str | None = Field(
-        default=None,
-        description=(
-            "Crew agent backstory. Crew inputs are interpolated with `{name}` "
-            "placeholders such as `{topic}`; this is not CEL."
-        ),
-        examples=["Expert at concise technical research."],
-    )
-    type: str | PythonReferenceDefinition | None = Field(
-        default=None,
-        description="Optional built-in type or Python reference used to load the agent.",
-        examples=["agent", {"python": "my_project.agents.ResearchAgent"}],
-    )
-    from_repository: str | None = Field(
-        default=None,
-        description=(
-            "Agent repository name to load. Repository values supply missing "
-            "agent configuration; explicitly provided local fields override the "
-            "repository values."
-        ),
-        examples=["researcher"],
-    )
-    settings: dict[str, Any] = Field(
-        default_factory=dict,
-        description="Additional agent settings passed to the loader.",
-        examples=[{"llm": "openai/gpt-4o-mini"}],
-    )
-    llm: str | LLMDefinition | None = Field(
-        default=None,
-        description=(
-            "Language model that runs the agent. Use a string model name or an "
-            "object with model settings such as max_tokens."
-        ),
-        examples=[{"model": "openai/gpt-4o-mini", "max_tokens": 4096}],
-    )
-    planning_config: PlanningConfig | None = Field(
-        default=None,
-        description="Configuration for agent planning before task execution.",
-        examples=[{"max_attempts": 3}],
-    )
-    allow_delegation: bool | None = Field(
-        default=None,
-        description="Enable agent to delegate and ask questions among each other.",
-        examples=[False],
-    )
-    max_iter: int | None = Field(
-        default=None,
-        description="Maximum iterations for an agent to execute a task",
-        examples=[25],
-    )
-    max_rpm: int | None = Field(
-        default=None,
-        description=(
-            "Maximum number of requests per minute for the agent execution to be "
-            "respected."
-        ),
-        examples=[10],
-    )
-    max_execution_time: int | None = Field(
-        default=None,
-        description="Maximum execution time in seconds for an agent to execute a task",
-        examples=[300],
-    )
-    tools: list[str | dict[str, Any]] | None = Field(
-        default=None,
-        description=(
-            "Tool refs or serialized tool definitions available to this agent. "
-            "String refs can use CrewAI tool names, `custom:<name>`, or fully "
-            "qualified `module:Class` references."
-        ),
-        examples=[["crewai_tools:SerperDevTool", "custom:file_read"]],
-    )
-    apps: list[str] | None = Field(
-        default=None,
-        description=(
-            "Platform apps available to this agent. Can contain app names such as "
-            "`gmail` or app/action refs such as `gmail/send_email`."
-        ),
-        examples=[["gmail", "slack/send_message"]],
-    )
-    mcps: list[str | dict[str, Any]] | None = Field(
-        default=None,
-        description=(
-            "MCP server refs or serialized MCP server configs available to this "
-            "agent. String refs can use HTTPS URLs, connected MCP integration "
-            "slugs, or refs with a `#tool_name` suffix for specific tools."
-        ),
-        examples=[
-            [
-                "https://api.weather.com/mcp#get_current_weather",
-                "snowflake",
-                "stripe#list_invoices",
-                {
-                    "url": "https://api.example.com/mcp",
-                    "headers": {"Authorization": "Bearer your_token"},
-                    "streamable": True,
-                    "cache_tools_list": True,
-                },
-            ]
-        ],
-    )
+    role: str
+    goal: str
+    backstory: str
+    type: str | PythonReferenceDefinition | None = None
+    settings: dict[str, Any] = Field(default_factory=dict)
 
     @field_validator("settings", mode="before")
     @classmethod
@@ -193,44 +55,10 @@ class CrewAgentDefinition(BaseModel):
 
 
 class AgentDefinition(CrewAgentDefinition):
-    """Inline individual agent definition used outside of a crew."""
+    """Inline agent definition used by a Flow agent action."""
 
-    role: str | None = Field(
-        default=None,
-        description="Individual agent role used by a Flow agent action outside of a crew.",
-        examples=["Support specialist"],
-    )
-    goal: str | None = Field(
-        default=None,
-        description="Individual agent goal for the Flow agent action outside of a crew.",
-        examples=["Draft a concise customer reply"],
-    )
-    backstory: str | None = Field(
-        default=None,
-        description=(
-            "Individual agent backstory used to shape behavior outside of a crew."
-        ),
-        examples=["Expert at resolving SaaS support questions."],
-    )
-    type: str | PythonReferenceDefinition | None = Field(
-        default=None,
-        description="Optional built-in type or Python reference used to load the agent.",
-        examples=["agent", {"python": "my_project.agents.SupportAgent"}],
-    )
-    settings: dict[str, Any] = Field(
-        default_factory=dict,
-        description="Additional agent settings passed to the loader.",
-        examples=[{"llm": "openai/gpt-4o-mini"}],
-    )
-    input: str = Field(
-        description="Input passed to the individual agent kickoff outside of a crew.",
-        examples=["${state.ticket.body}"],
-    )
-    response_format: PythonReferenceDefinition | None = Field(
-        default=None,
-        description="Optional Python reference to a Pydantic response format.",
-        examples=[{"python": "my_project.schemas.SupportReply"}],
-    )
+    input: str
+    response_format: PythonReferenceDefinition | None = None
 
     @field_validator("input", mode="before")
     @classmethod
@@ -245,40 +73,12 @@ class CrewTaskDefinition(BaseModel):
 
     model_config = ConfigDict(extra="allow")
 
-    description: str = Field(
-        description=(
-            "Task instructions. Crew inputs are interpolated with `{name}` "
-            "placeholders such as `{topic}`; this is not CEL."
-        ),
-        examples=["Research {topic}."],
-    )
-    expected_output: str = Field(
-        description=(
-            "Expected task output. Crew inputs are interpolated with `{name}` "
-            "placeholders such as `{topic}`; this is not CEL."
-        ),
-        examples=["Key findings about {topic}."],
-    )
-    name: str | None = Field(
-        default=None,
-        description="Optional task name used by context references.",
-        examples=["research_task"],
-    )
-    agent: str | None = Field(
-        default=None,
-        description="Name of the crew agent assigned to this task.",
-        examples=["researcher"],
-    )
-    context: list[str] | None = Field(
-        default=None,
-        description="Names of previous tasks whose outputs should be used as context.",
-        examples=[["research_task"]],
-    )
-    type: str | PythonReferenceDefinition | None = Field(
-        default=None,
-        description="Optional built-in type or Python reference used to load the task.",
-        examples=["task", {"python": "my_project.tasks.ResearchTask"}],
-    )
+    description: str
+    expected_output: str
+    name: str | None = None
+    agent: str | None = None
+    context: list[str] | None = None
+    type: str | PythonReferenceDefinition | None = None
 
 
 _CrewAgentsInput: TypeAlias = dict[str, CrewAgentDefinition] | list[dict[str, Any]]
@@ -289,44 +89,10 @@ class CrewDefinition(BaseModel):
 
     model_config = ConfigDict(extra="allow")
 
-    agents: dict[str, CrewAgentDefinition] = Field(
-        description="Inline crew agents keyed by agent name.",
-        examples=[
-            {
-                "researcher": {
-                    "role": "Research analyst",
-                    "goal": "Research {topic}",
-                    "backstory": "Expert at concise technical research.",
-                }
-            }
-        ],
-    )
-    tasks: list[CrewTaskDefinition] = Field(
-        description="Ordered crew tasks.",
-        examples=[
-            [
-                {
-                    "name": "research_task",
-                    "description": "Research {topic}.",
-                    "expected_output": "Key findings about {topic}.",
-                    "agent": "researcher",
-                }
-            ]
-        ],
-    )
-    inputs: dict[str, Any] = Field(
-        default_factory=dict,
-        description=(
-            "Default crew inputs. Values are available to crew agent and task "
-            "interpolation as `{name}` placeholders, for example `{topic}`."
-        ),
-        examples=[{"topic": "AI agents"}],
-    )
-    manager_agent: str | PythonReferenceDefinition | None = Field(
-        default=None,
-        description="Optional manager agent name or Python reference.",
-        examples=["manager", {"python": "my_project.agents.ManagerAgent"}],
-    )
+    agents: dict[str, CrewAgentDefinition]
+    tasks: list[CrewTaskDefinition]
+    inputs: dict[str, Any] = Field(default_factory=dict)
+    manager_agent: str | PythonReferenceDefinition | None = None
 
     @field_validator("inputs", mode="before")
     @classmethod

lib/crewai/src/crewai/project/json_loader.py

@@ -978,10 +978,9 @@ def _agent_kwargs_from_definition(
         extra_allowed,
         skip_unknown=skip_unknown,
     )
-    if not defn.get("from_repository"):
-        for required in ("role", "goal", "backstory"):
-            if defn.get(required) is None:
-                errors.append(f"{path}: missing required field '{required}'")
+    for required in ("role", "goal", "backstory"):
+        if required not in defn:
+            errors.append(f"{path}: missing required field '{required}'")
 
     settings = defn.get("settings", {})
     if settings is None:

lib/crewai/src/crewai/skills/__init__.py

@@ -3,12 +3,7 @@
 Provides filesystem-based skill packaging with progressive disclosure.
 """
 
-from crewai.skills.loader import (
-    activate_skill,
-    discover_skills,
-    load_skill,
-    load_skills,
-)
+from crewai.skills.loader import activate_skill, discover_skills
 from crewai.skills.models import Skill, SkillFrontmatter
 from crewai.skills.parser import SkillParseError
 
@@ -19,6 +14,4 @@ __all__ = [
     "SkillParseError",
     "activate_skill",
     "discover_skills",
-    "load_skill",
-    "load_skills",
 ]

lib/crewai/src/crewai/skills/loader.py

@@ -6,7 +6,6 @@ for agent use, and format skill context for prompt injection.
 
 from __future__ import annotations
 
-from collections.abc import Iterable
 import logging
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -19,13 +18,12 @@ from crewai.events.types.skill_events import (
     SkillLoadFailedEvent,
     SkillLoadedEvent,
 )
-from crewai.skills.models import INSTRUCTIONS, RESOURCES, Skill, SkillFrontmatter
+from crewai.skills.models import INSTRUCTIONS, RESOURCES, Skill
 from crewai.skills.parser import (
     SKILL_FILENAME,
     load_skill_instructions,
     load_skill_metadata,
     load_skill_resources,
-    parse_frontmatter,
 )
 
 
@@ -145,72 +143,6 @@ def activate_skill(
     return activated
 
 
-def load_skill(
-    skill: Path | Skill | str,
-    source: BaseAgent | None = None,
-) -> list[Skill]:
-    """Load one skill input into Skill objects.
-
-    Accepts a pre-loaded Skill object, skill search path, inline SKILL.md
-    string, or '@org/name' registry reference. Path inputs can expand to many
-    skills. Path and inline inputs are activated immediately; pre-loaded Skill
-    objects keep their disclosure level.
-    """
-    if isinstance(skill, Skill):
-        return [skill]
-    if isinstance(skill, Path):
-        return [
-            activate_skill(s, source=source)
-            for s in discover_skills(skill, source=source)
-        ]
-    if isinstance(skill, str) and skill.startswith("@"):
-        from crewai.experimental.skills.registry import resolve_registry_ref
-
-        return [resolve_registry_ref(skill, source=source)]
-    if isinstance(skill, str) and skill.lstrip().startswith("---\n"):
-        frontmatter_dict, body = parse_frontmatter(skill.strip())
-        return [
-            Skill(
-                frontmatter=SkillFrontmatter(**frontmatter_dict),
-                instructions=body,
-                path=Path("."),
-                disclosure_level=INSTRUCTIONS,
-            )
-        ]
-    if isinstance(skill, str):
-        return [
-            activate_skill(s, source=source)
-            for s in discover_skills(Path(skill), source=source)
-        ]
-
-    msg = f"Unsupported skill input: {skill!r}"
-    raise TypeError(msg)
-
-
-def load_skills(
-    skills: Iterable[Path | Skill | str],
-    source: BaseAgent | None = None,
-) -> list[Skill]:
-    """Load skill inputs into de-duplicated Skill objects.
-
-    Preserves first-seen order when multiple inputs resolve to the same skill
-    name. Registry refs are scoped by org so different orgs can publish skills
-    that share a frontmatter name.
-    """
-    loaded: dict[str, Skill] = {}
-    for skill_input in skills:
-        for skill in load_skill(skill_input, source=source):
-            dedup_key = skill.name
-            if isinstance(skill_input, str) and skill_input.startswith("@"):
-                from crewai.experimental.skills.registry import parse_registry_ref
-
-                org, _ = parse_registry_ref(skill_input)
-                dedup_key = f"{org}/{skill.name}"
-            if dedup_key not in loaded:
-                loaded[dedup_key] = skill
-    return list(loaded.values())
-
-
 def load_resources(skill: Skill) -> Skill:
     """Promote a skill to RESOURCES disclosure level.
 

lib/crewai/src/crewai/types/streaming.py

@@ -2,10 +2,9 @@
 
 from __future__ import annotations
 
-from collections.abc import AsyncIterator, Callable, Iterator, Sequence
-from datetime import datetime
+from collections.abc import AsyncIterator, Callable, Iterator
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar
+from typing import TYPE_CHECKING, Any, Generic, TypeVar
 
 from pydantic import BaseModel, Field
 from typing_extensions import Self
@@ -16,262 +15,6 @@ if TYPE_CHECKING:
 
 
 T = TypeVar("T")
-_MISSING = object()
-
-StreamChannel = Literal[
-    "llm",
-    "flow",
-    "tools",
-    "messages",
-    "lifecycle",
-    "custom",
-]
-
-
-class StreamFrame(BaseModel):
-    """Stable public stream frame emitted by streamable runtimes."""
-
-    id: str = Field(description="Unique frame/event identifier")
-    seq: int | None = Field(default=None, description="Execution-local order")
-    type: str = Field(description="Source event type")
-    channel: StreamChannel = Field(description="High-level stream channel")
-    namespace: list[str] = Field(default_factory=list)
-    timestamp: datetime
-    parent_id: str | None = None
-    previous_id: str | None = None
-    data: dict[str, Any] = Field(default_factory=dict)
-
-    @property
-    def content(self) -> str:
-        """Printable text content for chunk-like consumers."""
-        chunk = self.data.get("chunk")
-        if isinstance(chunk, str):
-            return chunk
-        return ""
-
-    @property
-    def event(self) -> dict[str, Any]:
-        """Structured source event payload."""
-        return self.data
-
-
-class StreamSessionBase(Generic[T]):
-    """Base stream session with ordered frame iteration and result access."""
-
-    def __init__(
-        self,
-        sync_iterator: Iterator[StreamFrame] | None = None,
-        async_iterator: AsyncIterator[StreamFrame] | None = None,
-    ) -> None:
-        self._result: T | object = _MISSING
-        self._completed = False
-        self._frames: list[StreamFrame] = []
-        self._error: Exception | None = None
-        self._cancelled = False
-        self._exhausted = False
-        self._on_cleanup: Callable[[], None] | None = None
-        self._sync_iterator = sync_iterator
-        self._async_iterator = async_iterator
-
-    @property
-    def result(self) -> T:
-        """Return the final result after stream exhaustion or completion."""
-        if not self._completed:
-            raise RuntimeError(
-                "Streaming has not completed yet. "
-                "Iterate over all frames before accessing result."
-            )
-        if self._error is not None:
-            raise self._error
-        if self._result is _MISSING:
-            raise RuntimeError("No result available")
-        return self._result  # type: ignore[return-value]
-
-    @property
-    def is_completed(self) -> bool:
-        """Check if the stream has completed."""
-        return self._completed
-
-    @property
-    def is_cancelled(self) -> bool:
-        """Check if the stream was cancelled."""
-        return self._cancelled
-
-    @property
-    def is_exhausted(self) -> bool:
-        """Check if the stream iterator was fully consumed."""
-        return self._exhausted
-
-    @property
-    def frames(self) -> list[StreamFrame]:
-        """Return collected frames."""
-        return self._frames.copy()
-
-    def _set_result(self, result: T) -> None:
-        self._result = result
-        self._completed = True
-
-
-class StreamSession(StreamSessionBase[T]):
-    """Synchronous stream session for ordered public frames."""
-
-    def __enter__(self) -> Self:
-        return self
-
-    def __exit__(self, *exc_info: Any) -> None:
-        if not self._exhausted:
-            self.close()
-
-    @property
-    def events(self) -> Iterator[StreamFrame]:
-        """Iterate over all ordered frames."""
-        return self.subscribe()
-
-    def __iter__(self) -> Iterator[StreamFrame]:
-        """Iterate over all ordered frames."""
-        return self.events
-
-    @property
-    def llm(self) -> Iterator[StreamFrame]:
-        """Iterate over LLM token and thinking frames."""
-        return self.subscribe(channels=["llm"])
-
-    @property
-    def messages(self) -> Iterator[StreamFrame]:
-        """Iterate over conversation message frames."""
-        return self.subscribe(channels=["messages"])
-
-    @property
-    def flow(self) -> Iterator[StreamFrame]:
-        """Iterate over Flow lifecycle and method frames."""
-        return self.subscribe(channels=["flow"])
-
-    @property
-    def tools(self) -> Iterator[StreamFrame]:
-        """Iterate over tool execution frames."""
-        return self.subscribe(channels=["tools"])
-
-    def interleave(self, channels: Sequence[StreamChannel]) -> Iterator[StreamFrame]:
-        """Iterate over selected channels while preserving global order."""
-        return self.subscribe(channels=channels)
-
-    def subscribe(
-        self, channels: Sequence[StreamChannel] | None = None
-    ) -> Iterator[StreamFrame]:
-        """Iterate over frames, optionally filtered by channel."""
-        selected = set(channels) if channels is not None else None
-        if self._exhausted:
-            for frame in self._frames:
-                if selected is None or frame.channel in selected:
-                    yield frame
-            return
-        if self._sync_iterator is None:
-            raise RuntimeError("Sync iterator not available")
-        try:
-            for frame in self._sync_iterator:
-                self._frames.append(frame)
-                if selected is None or frame.channel in selected:
-                    yield frame
-            self._exhausted = True
-        except Exception as e:
-            self._error = e
-            raise
-        finally:
-            self._completed = True
-
-    def close(self) -> None:
-        """Cancel streaming and clean up resources."""
-        if self._cancelled or self._exhausted or self._error is not None:
-            return
-        self._cancelled = True
-        self._completed = True
-        if self._sync_iterator is not None and hasattr(self._sync_iterator, "close"):
-            self._sync_iterator.close()
-        if self._on_cleanup is not None:
-            self._on_cleanup()
-            self._on_cleanup = None
-
-
-class AsyncStreamSession(StreamSessionBase[T]):
-    """Asynchronous stream session for ordered public frames."""
-
-    async def __aenter__(self) -> Self:
-        return self
-
-    async def __aexit__(self, *exc_info: Any) -> None:
-        if not self._exhausted:
-            await self.aclose()
-
-    @property
-    def events(self) -> AsyncIterator[StreamFrame]:
-        """Iterate over all ordered frames."""
-        return self.subscribe()
-
-    def __aiter__(self) -> AsyncIterator[StreamFrame]:
-        """Iterate over all ordered frames."""
-        return self.events
-
-    @property
-    def llm(self) -> AsyncIterator[StreamFrame]:
-        """Iterate over LLM token and thinking frames."""
-        return self.subscribe(channels=["llm"])
-
-    @property
-    def messages(self) -> AsyncIterator[StreamFrame]:
-        """Iterate over conversation message frames."""
-        return self.subscribe(channels=["messages"])
-
-    @property
-    def flow(self) -> AsyncIterator[StreamFrame]:
-        """Iterate over Flow lifecycle and method frames."""
-        return self.subscribe(channels=["flow"])
-
-    @property
-    def tools(self) -> AsyncIterator[StreamFrame]:
-        """Iterate over tool execution frames."""
-        return self.subscribe(channels=["tools"])
-
-    def interleave(
-        self, channels: Sequence[StreamChannel]
-    ) -> AsyncIterator[StreamFrame]:
-        """Iterate over selected channels while preserving global order."""
-        return self.subscribe(channels=channels)
-
-    async def subscribe(
-        self, channels: Sequence[StreamChannel] | None = None
-    ) -> AsyncIterator[StreamFrame]:
-        """Iterate over frames, optionally filtered by channel."""
-        selected = set(channels) if channels is not None else None
-        if self._exhausted:
-            for frame in self._frames:
-                if selected is None or frame.channel in selected:
-                    yield frame
-            return
-        if self._async_iterator is None:
-            raise RuntimeError("Async iterator not available")
-        try:
-            async for frame in self._async_iterator:
-                self._frames.append(frame)
-                if selected is None or frame.channel in selected:
-                    yield frame
-            self._exhausted = True
-        except Exception as e:
-            self._error = e
-            raise
-        finally:
-            self._completed = True
-
-    async def aclose(self) -> None:
-        """Cancel streaming and clean up resources."""
-        if self._cancelled or self._exhausted or self._error is not None:
-            return
-        self._cancelled = True
-        self._completed = True
-        if self._async_iterator is not None and hasattr(self._async_iterator, "aclose"):
-            await self._async_iterator.aclose()
-        if self._on_cleanup is not None:
-            self._on_cleanup()
-            self._on_cleanup = None
 
 
 class StreamChunkType(Enum):

lib/crewai/src/crewai/utilities/agent_utils.py

@@ -1125,7 +1125,7 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
 
             client = PlusAPI(api_key=get_auth_token())
         _print_current_organization()
-        response = client.get_agent(from_repository)
+        response = asyncio.run(client.get_agent(from_repository))
         if response.status_code == 404:
             raise AgentRepositoryError(
                 f"Agent {from_repository} does not exist, make sure the name is correct or the agent is available on your organization."
@@ -1158,8 +1158,6 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
                         raise AgentRepositoryError(
                             f"Tool {tool['name']} could not be loaded: {e}"
                         ) from e
-            elif key == "skills" and value == []:
-                continue
             else:
                 attributes[key] = value
     return attributes

lib/crewai/src/crewai/utilities/streaming.py

@@ -6,32 +6,19 @@ import contextvars
 import logging
 import queue
 import threading
-from typing import Any, NamedTuple, cast
+from typing import Any, NamedTuple
 import uuid
 
 from typing_extensions import TypedDict
 
 from crewai.events.base_events import BaseEvent
 from crewai.events.event_bus import crewai_event_bus
-from crewai.events.stream_context import add_stream_sink, reset_stream_sinks
-from crewai.events.types.flow_events import ConversationMessageAddedEvent, FlowEvent
-from crewai.events.types.llm_events import (
-    LLMEventBase,
-    LLMStreamChunkEvent,
-)
-from crewai.events.types.tool_usage_events import (
-    ToolExecutionErrorEvent,
-    ToolUsageEvent,
-)
+from crewai.events.types.llm_events import LLMStreamChunkEvent
 from crewai.types.streaming import (
-    AsyncStreamSession,
     CrewStreamingOutput,
     FlowStreamingOutput,
-    StreamChannel,
     StreamChunk,
     StreamChunkType,
-    StreamFrame,
-    StreamSession,
     ToolCallChunk,
 )
 from crewai.utilities.string_utils import sanitize_tool_name
@@ -66,16 +53,6 @@ class StreamingState(NamedTuple):
     stream_id: str | None = None
 
 
-class FrameStreamingState(NamedTuple):
-    """Immutable state for public frame streaming execution."""
-
-    result_holder: list[Any]
-    sync_queue: queue.Queue[StreamFrame | None | Exception]
-    async_queue: asyncio.Queue[StreamFrame | None | Exception] | None
-    loop: asyncio.AbstractEventLoop | None
-    sink: Callable[[Any, BaseEvent], None]
-
-
 def _extract_tool_call_info(
     event: LLMStreamChunkEvent,
 ) -> tuple[StreamChunkType, ToolCallChunk | None]:
@@ -130,207 +107,6 @@ def _create_stream_chunk(
     )
 
 
-_FRAME_DATA_EXCLUDE = {
-    "timestamp",
-    "type",
-    "event_id",
-    "parent_event_id",
-    "previous_event_id",
-    "emission_sequence",
-}
-
-
-def _stream_channel(event: BaseEvent) -> StreamChannel:
-    if isinstance(event, LLMEventBase):
-        return "llm"
-    if isinstance(event, ConversationMessageAddedEvent):
-        return "messages"
-    if isinstance(event, FlowEvent):
-        return "flow"
-    if isinstance(event, ToolUsageEvent | ToolExecutionErrorEvent):
-        return "tools"
-    if "error" in event.type or "failed" in event.type:
-        return "lifecycle"
-    return "custom"
-
-
-def _stream_namespace(event: BaseEvent, channel: StreamChannel) -> list[str]:
-    namespace: list[str] = [channel]
-    for attr in (
-        "flow_name",
-        "method_name",
-        "session_id",
-        "call_id",
-        "tool_name",
-        "agent_role",
-        "task_name",
-    ):
-        value = getattr(event, attr, None)
-        if value is not None:
-            namespace.append(str(value))
-    return namespace
-
-
-def stream_frame_from_event(event: BaseEvent) -> StreamFrame:
-    """Convert an internal CrewAI event into the public stream frame contract."""
-    channel = _stream_channel(event)
-    data = event.to_json(exclude=_FRAME_DATA_EXCLUDE)
-    if not isinstance(data, dict):
-        data = {"value": data}
-    return StreamFrame(
-        id=event.event_id,
-        seq=event.emission_sequence,
-        type=event.type,
-        channel=channel,
-        namespace=_stream_namespace(event, channel),
-        timestamp=event.timestamp,
-        parent_id=event.parent_event_id,
-        previous_id=event.previous_event_id,
-        data=cast(dict[str, Any], data),
-    )
-
-
-def _create_frame_sink(
-    sync_queue: queue.Queue[StreamFrame | None | Exception],
-    async_queue: asyncio.Queue[StreamFrame | None | Exception] | None = None,
-    loop: asyncio.AbstractEventLoop | None = None,
-) -> Callable[[Any, BaseEvent], None]:
-    def frame_sink(_: Any, event: BaseEvent) -> None:
-        frame = stream_frame_from_event(event)
-        if async_queue is not None and loop is not None:
-            loop.call_soon_threadsafe(async_queue.put_nowait, frame)
-        else:
-            sync_queue.put(frame)
-
-    return frame_sink
-
-
-def create_frame_streaming_state(
-    result_holder: list[Any],
-    use_async: bool = False,
-) -> FrameStreamingState:
-    """Create state for a scoped public frame stream."""
-    sync_queue: queue.Queue[StreamFrame | None | Exception] = queue.Queue()
-    async_queue: asyncio.Queue[StreamFrame | None | Exception] | None = None
-    loop: asyncio.AbstractEventLoop | None = None
-    if use_async:
-        async_queue = asyncio.Queue()
-        loop = asyncio.get_running_loop()
-    sink = _create_frame_sink(sync_queue, async_queue, loop)
-    return FrameStreamingState(
-        result_holder=result_holder,
-        sync_queue=sync_queue,
-        async_queue=async_queue,
-        loop=loop,
-        sink=sink,
-    )
-
-
-def _signal_frame_end(state: FrameStreamingState, is_async: bool = False) -> None:
-    if is_async and state.async_queue is not None and state.loop is not None:
-        state.loop.call_soon_threadsafe(state.async_queue.put_nowait, None)
-    else:
-        state.sync_queue.put(None)
-
-
-def _signal_frame_error(
-    state: FrameStreamingState, error: Exception, is_async: bool = False
-) -> None:
-    if is_async and state.async_queue is not None and state.loop is not None:
-        state.loop.call_soon_threadsafe(state.async_queue.put_nowait, error)
-    else:
-        state.sync_queue.put(error)
-
-
-def _finalize_frame_streaming(
-    state: FrameStreamingState,
-    stream_session: StreamSession[Any] | AsyncStreamSession[Any],
-) -> None:
-    stream_session._on_cleanup = None
-    if state.result_holder:
-        stream_session._set_result(state.result_holder[0])
-
-
-def create_frame_generator(
-    state: FrameStreamingState,
-    run_func: Callable[[], Any],
-    output_holder: list[StreamSession[Any]],
-) -> Iterator[StreamFrame]:
-    """Create a scoped synchronous public frame generator."""
-
-    def run_with_sink() -> None:
-        token = add_stream_sink(state.sink)
-        try:
-            result = run_func()
-            state.result_holder.append(result)
-        except Exception as e:
-            _signal_frame_error(state, e)
-        finally:
-            reset_stream_sinks(token)
-            _signal_frame_end(state)
-
-    ctx = contextvars.copy_context()
-    thread = threading.Thread(target=ctx.run, args=(run_with_sink,), daemon=True)
-    thread.start()
-
-    try:
-        while True:
-            item = state.sync_queue.get()
-            if item is None:
-                break
-            if isinstance(item, Exception):
-                raise item
-            yield item
-    finally:
-        thread.join()
-        if output_holder:
-            _finalize_frame_streaming(state, output_holder[0])
-
-
-async def create_async_frame_generator(
-    state: FrameStreamingState,
-    run_coro: Callable[[], Any],
-    output_holder: list[AsyncStreamSession[Any]],
-) -> AsyncIterator[StreamFrame]:
-    """Create a scoped asynchronous public frame generator."""
-    if state.async_queue is None:
-        raise RuntimeError(
-            "Async queue not initialized. Use create_frame_streaming_state(use_async=True)."
-        )
-
-    async def run_with_sink() -> None:
-        token = add_stream_sink(state.sink)
-        try:
-            result = await run_coro()
-            state.result_holder.append(result)
-        except Exception as e:
-            _signal_frame_error(state, e, is_async=True)
-        finally:
-            reset_stream_sinks(token)
-            _signal_frame_end(state, is_async=True)
-
-    task = asyncio.create_task(run_with_sink())
-    try:
-        while True:
-            item = await state.async_queue.get()
-            if item is None:
-                break
-            if isinstance(item, Exception):
-                raise item
-            yield item
-    finally:
-        if not task.done():
-            task.cancel()
-        try:
-            await task
-        except asyncio.CancelledError:
-            pass
-        except Exception:
-            logger.debug("Background frame streaming task failed", exc_info=True)
-        if output_holder:
-            _finalize_frame_streaming(state, output_holder[0])
-
-
 def _create_stream_handler(
     current_task_info: TaskInfo,
     sync_queue: queue.Queue[StreamChunk | None | Exception],

lib/crewai/tests/agents/test_agent.py

@@ -2243,27 +2243,6 @@ def test_agent_from_repository_override_attributes(mock_get_agent, mock_get_auth
     assert isinstance(agent.tools[0], SerperDevTool)
 
 
-@patch("crewai.plus_api.PlusAPI.get_agent")
-def test_agent_from_repository_ignores_empty_skills(
-    mock_get_agent, mock_get_auth_token
-):
-    mock_get_response = MagicMock()
-    mock_get_response.status_code = 200
-    mock_get_response.json.return_value = {
-        "role": "test role",
-        "goal": "test goal",
-        "backstory": "test backstory",
-        "tools": [],
-        "skills": [],
-    }
-    mock_get_agent.return_value = mock_get_response
-
-    agent = Agent(from_repository="test_agent")
-
-    assert agent.role == "test role"
-    assert agent.skills is None
-
-
 @patch("crewai.plus_api.PlusAPI.get_agent")
 def test_agent_from_repository_with_invalid_tools(mock_get_agent, mock_get_auth_token):
     mock_get_response = MagicMock()

lib/crewai/tests/cli/remote_template/test_main.py

@@ -24,7 +24,7 @@ SAMPLE_REPOS = [
 ]
 
 
-def _make_zipball(files: dict[str, str], top_dir: str = "crewAIInc-fde-template_test-abc123") -> bytes:
+def _make_zipball(files: dict[str, str], top_dir: str = "crewAIInc-template_test-abc123") -> bytes:
     """Create an in-memory zipball mimicking GitHub's format."""
     buf = io.BytesIO()
     with zipfile.ZipFile(buf, "w") as zf:

lib/crewai/tests/cli/test_plus_api.py

@@ -1,6 +1,8 @@
 import os
 import unittest
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import ANY, AsyncMock, MagicMock, patch
+
+import pytest
 
 from crewai.plus_api import PlusAPI
 
@@ -394,23 +396,28 @@ class TestPlusAPI(unittest.TestCase):
         )
 
 
-@patch("crewai_core.plus_api.PlusAPI._make_request")
-def test_get_agent(mock_make_request):
+@pytest.mark.asyncio
+@patch("httpx.AsyncClient")
+async def test_get_agent(mock_async_client_class):
     api = PlusAPI("test_api_key")
     mock_response = MagicMock()
-    mock_make_request.return_value = mock_response
+    mock_client_instance = AsyncMock()
+    mock_client_instance.get.return_value = mock_response
+    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
 
-    response = api.get_agent("test_agent_handle")
+    response = await api.get_agent("test_agent_handle")
 
-    mock_make_request.assert_called_once_with(
-        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
+    mock_client_instance.get.assert_called_once_with(
+        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
+        headers=api.headers,
     )
     assert response == mock_response
 
 
-@patch("crewai_core.plus_api.PlusAPI._make_request")
+@pytest.mark.asyncio
+@patch("httpx.AsyncClient")
 @patch("crewai_core.plus_api.Settings")
-def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
+async def test_get_agent_with_org_uuid(mock_settings_class, mock_async_client_class):
     org_uuid = "test-org-uuid"
     mock_settings = MagicMock()
     mock_settings.org_uuid = org_uuid
@@ -420,12 +427,15 @@ def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
     api = PlusAPI("test_api_key")
 
     mock_response = MagicMock()
-    mock_make_request.return_value = mock_response
+    mock_client_instance = AsyncMock()
+    mock_client_instance.get.return_value = mock_response
+    mock_async_client_class.return_value.__aenter__.return_value = mock_client_instance
 
-    response = api.get_agent("test_agent_handle")
+    response = await api.get_agent("test_agent_handle")
 
-    mock_make_request.assert_called_once_with(
-        "GET", "/crewai_plus/api/v1/agents/test_agent_handle"
+    mock_client_instance.get.assert_called_once_with(
+        f"{api.base_url}/crewai_plus/api/v1/agents/test_agent_handle",
+        headers=api.headers,
     )
     assert "X-Crewai-Organization-Id" in api.headers
     assert api.headers["X-Crewai-Organization-Id"] == org_uuid

lib/crewai/tests/project/test_json_loader.py

@@ -355,24 +355,6 @@ class TestLoadAgent:
         with pytest.raises(Exception):
             load_agent(agent_file)
 
-    @pytest.mark.parametrize("field", ["role", "goal", "backstory"])
-    def test_load_agent_rejects_null_required_fields(
-        self, tmp_path: Path, field: str
-    ):
-        agent_def = {
-            "role": "Researcher",
-            "goal": "Find information",
-            "backstory": "Expert researcher.",
-        }
-        agent_def[field] = None
-        agent_file = tmp_path / "agent.json"
-        agent_file.write_text(json.dumps(agent_def))
-
-        with pytest.raises(
-            JSONProjectValidationError, match=f"missing required field '{field}'"
-        ):
-            load_agent(agent_file)
-
     def test_load_agent_file_not_found(self):
         with pytest.raises(FileNotFoundError):
             load_agent(Path("/nonexistent/agent.json"))

lib/crewai/tests/skills/test_integration.py

@@ -4,13 +4,8 @@ from pathlib import Path
 
 import pytest
 
-from crewai import Agent, Crew, Task
-from crewai.crews.utils import _resolve_crew_skills
-from crewai.skills.loader import (
-    activate_skill,
-    discover_skills,
-    format_skill_context,
-)
+from crewai import Agent
+from crewai.skills.loader import activate_skill, discover_skills, format_skill_context
 from crewai.skills.models import INSTRUCTIONS, METADATA
 from crewai.utilities.prompts import Prompts
 
@@ -105,104 +100,3 @@ class TestSkillDiscoveryAndActivation:
         assert "Skill travel" in system
         # METADATA-level skills must not leak full instructions into the prompt
         assert "Use this skill for travel planning." not in system
-
-    def test_agent_accepts_inline_skill_string(self) -> None:
-        agent = Agent(
-            role="Reviewer",
-            goal="Review changes.",
-            backstory="An experienced reviewer.",
-            skills=[
-                "---\n"
-                "name: inline-review\n"
-                "description: Inline review guidance\n"
-                "---\n"
-                "Focus on behavior and missing tests."
-            ],
-        )
-
-        assert agent.skills is not None
-        assert [skill.name for skill in agent.skills] == ["inline-review"]
-        assert [skill.disclosure_level for skill in agent.skills] == [INSTRUCTIONS]
-        assert [skill.instructions for skill in agent.skills] == [
-            "Focus on behavior and missing tests."
-        ]
-
-        result = Prompts(agent=agent, has_tools=False, use_system_prompt=True).task_execution()
-        system = getattr(result, "system", "") or result.prompt
-        assert '<skill name="inline-review">' in system
-        assert "Focus on behavior and missing tests." in system
-
-    def test_agent_treats_plain_skill_string_as_path(self, tmp_path: Path) -> None:
-        _create_skill_dir(tmp_path, "path-skill", body="Use the path skill.")
-
-        agent = Agent(
-            role="Reviewer",
-            goal="Review changes.",
-            backstory="An experienced reviewer.",
-            skills=[str(tmp_path)],
-        )
-
-        assert agent.skills is not None
-        assert [skill.name for skill in agent.skills] == ["path-skill"]
-        assert [skill.instructions for skill in agent.skills] == ["Use the path skill."]
-
-    def test_crew_resolves_inline_skill_string(self) -> None:
-        agent = Agent(
-            role="Reviewer",
-            goal="Review changes.",
-            backstory="An experienced reviewer.",
-        )
-        task = Task(
-            description="Review the diff.",
-            expected_output="Findings.",
-            agent=agent,
-        )
-        crew = Crew(
-            agents=[agent],
-            tasks=[task],
-            skills=[
-                "---\n"
-                "name: crew-inline-review\n"
-                "description: Crew-level inline review guidance\n"
-                "---\n"
-                "Apply this to every agent."
-            ],
-        )
-
-        skills = _resolve_crew_skills(crew)
-
-        assert skills is not None
-        assert [skill.name for skill in skills] == ["crew-inline-review"]
-        assert [skill.instructions for skill in skills] == ["Apply this to every agent."]
-
-    def test_crew_activates_preloaded_metadata_skill(self, tmp_path: Path) -> None:
-        _create_skill_dir(
-            tmp_path,
-            "crew-preloaded",
-            body="Apply this crew-level guidance to every agent.",
-        )
-        metadata_skill = discover_skills(tmp_path)[0]
-        agent = Agent(
-            role="Reviewer",
-            goal="Review changes.",
-            backstory="An experienced reviewer.",
-        )
-        task = Task(
-            description="Review the diff.",
-            expected_output="Findings.",
-            agent=agent,
-        )
-        crew = Crew(
-            agents=[agent],
-            tasks=[task],
-            skills=[metadata_skill],
-        )
-
-        skills = _resolve_crew_skills(crew)
-
-        assert skills is not None
-        assert [skill.name for skill in skills] == ["crew-preloaded"]
-        assert [skill.disclosure_level for skill in skills] == [INSTRUCTIONS]
-        assert [skill.instructions for skill in skills] == [
-            "Apply this crew-level guidance to every agent."
-        ]

lib/crewai/tests/skills/test_loader.py

@@ -9,11 +9,9 @@ from crewai.skills.loader import (
     discover_skills,
     format_skill_context,
     load_resources,
-    load_skill,
-    load_skills,
 )
 from crewai.skills.models import INSTRUCTIONS, METADATA, RESOURCES, Skill, SkillFrontmatter
-from crewai.skills.parser import SkillParseError, load_skill_metadata
+from crewai.skills.parser import load_skill_metadata
 
 
 def _create_skill_dir(parent: Path, name: str, body: str = "Body.") -> Path:
@@ -86,126 +84,6 @@ class TestActivateSkill:
         assert again is activated
 
 
-class TestLoadSkill:
-    """Tests for load_skill."""
-
-    @pytest.mark.parametrize("as_string", [False, True])
-    def test_loads_path_input(self, tmp_path: Path, as_string: bool) -> None:
-        _create_skill_dir(tmp_path, "first-skill", body="First.")
-        _create_skill_dir(tmp_path, "second-skill", body="Second.")
-        path = str(tmp_path) if as_string else tmp_path
-
-        skills = load_skill(path)
-
-        assert [skill.name for skill in skills] == ["first-skill", "second-skill"]
-        assert [skill.disclosure_level for skill in skills] == [
-            INSTRUCTIONS,
-            INSTRUCTIONS,
-        ]
-        assert [skill.instructions for skill in skills] == ["First.", "Second."]
-
-    def test_loads_preloaded_skill(self, tmp_path: Path) -> None:
-        preloaded = Skill(
-            frontmatter=SkillFrontmatter(
-                name="preloaded-skill",
-                description="Preloaded skill",
-            ),
-            path=tmp_path / "preloaded-skill",
-        )
-
-        skills = load_skill(preloaded)
-
-        assert skills == [preloaded]
-
-    def test_loads_inline_skill(self) -> None:
-        inline_skill = (
-            "---\n"
-            "name: inline-skill\n"
-            "description: Inline guidance\n"
-            "---\n"
-            "Follow these instructions."
-        )
-
-        skills = load_skill(inline_skill)
-
-        assert [skill.name for skill in skills] == ["inline-skill"]
-        assert [skill.disclosure_level for skill in skills] == [INSTRUCTIONS]
-        assert [skill.instructions for skill in skills] == [
-            "Follow these instructions."
-        ]
-
-    def test_invalid_inline_skill_raises_parse_error(self) -> None:
-        with pytest.raises(SkillParseError, match="missing closing"):
-            load_skill("---\nname: inline-skill\n")
-
-    def test_missing_path_raises_file_not_found(self, tmp_path: Path) -> None:
-        with pytest.raises(FileNotFoundError):
-            load_skill(tmp_path / "missing")
-
-    def test_unsupported_input_raises_type_error(self) -> None:
-        with pytest.raises(TypeError, match="Unsupported skill input"):
-            load_skill(object())  # type: ignore[arg-type]
-
-    def test_load_skills_deduplicates_by_name(self, tmp_path: Path) -> None:
-        first = Skill(
-            frontmatter=SkillFrontmatter(
-                name="duplicate-skill",
-                description="First skill",
-            ),
-            path=tmp_path / "first",
-        )
-        second = Skill(
-            frontmatter=SkillFrontmatter(
-                name="duplicate-skill",
-                description="Second skill",
-            ),
-            path=tmp_path / "second",
-        )
-
-        skills = load_skills([first, second])
-
-        assert skills == [first]
-
-    def test_load_skills_keeps_registry_refs_from_different_orgs(
-        self,
-        monkeypatch: pytest.MonkeyPatch,
-        tmp_path: Path,
-    ) -> None:
-        first = Skill(
-            frontmatter=SkillFrontmatter(
-                name="shared-skill",
-                description="First registry skill",
-            ),
-            path=tmp_path / "first",
-            disclosure_level=INSTRUCTIONS,
-            instructions="First instructions.",
-        )
-        second = Skill(
-            frontmatter=SkillFrontmatter(
-                name="shared-skill",
-                description="Second registry skill",
-            ),
-            path=tmp_path / "second",
-            disclosure_level=INSTRUCTIONS,
-            instructions="Second instructions.",
-        )
-
-        def resolve_registry_ref(ref: str, source: object = None) -> Skill:
-            return {
-                "@first/shared-skill": first,
-                "@second/shared-skill": second,
-            }[ref]
-
-        monkeypatch.setattr(
-            "crewai.experimental.skills.registry.resolve_registry_ref",
-            resolve_registry_ref,
-        )
-
-        skills = load_skills(["@first/shared-skill", "@second/shared-skill"])
-
-        assert skills == [first, second]
-
-
 class TestLoadResources:
     """Tests for load_resources."""
 

lib/crewai/tests/test_flow.py

@@ -2144,7 +2144,14 @@ def test_cyclic_flow_works_with_persist_and_id_input():
 
 
 @pytest.mark.timeout(5)
-def test_self_listening_method_is_rejected():
+def test_self_listening_method_does_not_loop():
+    """A method whose @listen label matches its own name must not loop forever.
+
+    Without the guard, 'process' re-triggers itself on every completion,
+    running indefinitely (timeout → FAIL).  The fix caps method calls
+    and raises RecursionError (PASS).
+    """
+
     class SelfListenFlow(Flow):
         @start()
         def begin(self):
@@ -2158,11 +2165,15 @@ def test_self_listening_method_is_rejected():
         def process(self):
             pass
 
-    with pytest.raises(ValueError, match="methods.process.listen"):
-        SelfListenFlow.flow_definition()
+    flow = SelfListenFlow()
+    with pytest.raises(RecursionError, match="infinite loop"):
+        flow.kickoff()
 
 
-def test_or_condition_self_listen_is_rejected():
+def test_or_condition_self_listen_fires_once():
+    """or_() with a self-referencing label only fires once due to or_() guard."""
+    call_count = 0
+
     class OrSelfListenFlow(Flow):
         @start()
         def begin(self):
@@ -2174,25 +2185,12 @@ def test_or_condition_self_listen_is_rejected():
 
         @listen(or_("other_trigger", "process"))
         def process(self):
-            pass
-
-    with pytest.raises(ValueError, match="methods.process.listen"):
-        OrSelfListenFlow.flow_definition()
-
-
-def test_router_self_listening_method_is_rejected():
-    class RouterSelfListenFlow(Flow):
-        @start()
-        def begin(self):
-            return "route"
-
-        @router("route")
-        def route(self):
-            return "done"
-
-    with pytest.raises(ValueError, match="methods.route.listen"):
-        RouterSelfListenFlow.flow_definition()
+            nonlocal call_count
+            call_count += 1
 
+    flow = OrSelfListenFlow()
+    flow.kickoff()
+    assert call_count == 1
 
 class ListState(BaseModel):
     items: list = []

lib/crewai/tests/test_flow_conversation.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any, ClassVar, Literal
+from typing import Any, Literal
 from unittest.mock import MagicMock, patch
 from uuid import uuid4
 
@@ -21,7 +21,7 @@ from crewai.events.types.flow_events import (
     MethodExecutionFinishedEvent,
     MethodExecutionStartedEvent,
 )
-from crewai.events.types.llm_events import LLMCallStartedEvent, LLMStreamChunkEvent
+from crewai.events.types.llm_events import LLMCallStartedEvent
 from crewai.experimental import (
     ConversationConfig,
     ConversationMessage,
@@ -29,13 +29,11 @@ from crewai.experimental import (
     RouterConfig,
 )
 from crewai.flow import Flow, ChatState, listen, start
-from crewai.flow.async_feedback import HumanFeedbackPending, PendingFeedbackContext
 from crewai.flow.flow_context import (
     current_flow_defer_trace_finalization,
     current_flow_id,
     current_flow_name,
 )
-from crewai.llms.base_llm import BaseLLM
 from crewai.flow.conversation import (
     append_message,
     get_conversation_messages,
@@ -139,109 +137,6 @@ class TestClassifyIntent:
 
 
 class TestConversationalFlow:
-    def test_stream_turn_emits_ordered_conversation_frames(self) -> None:
-        flow = ConversationalFlow()
-        flow.stream = True
-        stream_values_seen_by_kickoff: list[bool] = []
-
-        def kickoff_side_effect(*_: Any, **__: Any) -> str:
-            stream_values_seen_by_kickoff.append(flow.stream)
-            crewai_event_bus.emit(
-                flow,
-                LLMStreamChunkEvent(
-                    type="llm_stream_chunk",
-                    chunk="pong",
-                    call_id="call-1",
-                ),
-            )
-            return "pong"
-
-        with patch.object(flow, "kickoff", side_effect=kickoff_side_effect):
-            stream = flow.stream_turn("ping", session_id="session-1")
-
-            with pytest.raises(RuntimeError, match="Streaming has not completed yet"):
-                _ = stream.result
-
-            frames = list(stream.events)
-
-        assert stream.result == "pong"
-        assert stream_values_seen_by_kickoff == [False]
-        assert flow.stream is True
-        assert [frame.seq for frame in frames] == sorted(frame.seq for frame in frames)
-        assert [frame.type for frame in frames] == [
-            "conversation_turn_started",
-            "llm_stream_chunk",
-            "conversation_message_added",
-            "conversation_turn_completed",
-        ]
-        assert [frame.channel for frame in frames] == [
-            "flow",
-            "llm",
-            "messages",
-            "flow",
-        ]
-        assert frames[1].data["chunk"] == "pong"
-        assert flow.state.messages[-1].content == "pong"
-
-    def test_stream_turn_enables_streaming_on_conversation_llm(self) -> None:
-        class FakeLLM(BaseLLM):
-            stream_values: ClassVar[list[bool | None]] = []
-
-            def call(self, messages: Any, *args: Any, **kwargs: Any) -> str:
-                self.stream_values.append(self._effective_stream())
-                for chunk in ("po", "ng"):
-                    crewai_event_bus.emit(
-                        flow,
-                        LLMStreamChunkEvent(
-                            type="llm_stream_chunk",
-                            chunk=chunk,
-                            call_id="call-1",
-                        ),
-                    )
-                return "pong"
-
-        FakeLLM.stream_values = []
-        llm = FakeLLM(model="gpt-4o-mini", stream=False)
-
-        @ConversationConfig(llm=llm)
-        class StreamingChatFlow(ConversationalFlow):
-            pass
-
-        flow = StreamingChatFlow()
-        stream = flow.stream_turn("ping", session_id="session-1")
-        frames = list(stream.events)
-
-        assert stream.result == "pong"
-        assert llm.stream_values == [True]
-        assert llm.stream is False
-        assert [
-            frame.data["chunk"]
-            for frame in frames
-            if frame.type == "llm_stream_chunk"
-        ] == ["po", "ng"]
-
-    def test_stream_turn_returns_pending_feedback_without_failure_event(self) -> None:
-        flow = ConversationalFlow()
-        pending = HumanFeedbackPending(
-            context=PendingFeedbackContext(
-                flow_id="session-1",
-                flow_class="tests.PendingFeedbackFlow",
-                method_name="review",
-                method_output="draft",
-                message="Please review",
-            )
-        )
-
-        def kickoff_side_effect(*_: Any, **__: Any) -> None:
-            raise pending
-
-        with patch.object(flow, "kickoff", side_effect=kickoff_side_effect):
-            stream = flow.stream_turn("review this", session_id="session-1")
-            frames = list(stream.events)
-
-        assert stream.result is pending
-        assert [frame.type for frame in frames] == ["conversation_turn_started"]
-
     def test_deferred_multi_turn_emits_single_flow_finished(self) -> None:
         """A deferred multi-turn session lands as one trace: exactly one
         ``FlowFinishedEvent`` is emitted at ``finalize_session_traces()``, not

lib/crewai/tests/test_flow_definition.py

@@ -14,10 +14,6 @@ import crewai.flow.dsl as flow_dsl
 import crewai.flow.flow_definition as flow_definition
 import crewai.flow.visualization.builder as visualization_builder
 from crewai.experimental import ConversationConfig, RouterConfig
-from crewai.flow.expressions import (
-    FLOW_TEMPLATE_EXPRESSION_EXAMPLES,
-    FLOW_TEMPLATE_EXPRESSION_RULES,
-)
 from crewai.flow import Flow, and_, human_feedback, listen, or_, persist, router, start
 
 
@@ -86,17 +82,8 @@ def test_flow_definition_json_schema_carries_reference_descriptions():
     assert "not sandboxed" in script_properties["code"]["description"]
 
     agent_properties = defs["FlowAgentActionDefinition"]["properties"]
-    assert "Individual Agent definition" in agent_properties["with"]["description"]
-    assert "outside of a crew" in agent_properties["with"]["description"]
-    assert "individual inline Agent" in agent_properties["call"]["description"]
-
-    expression_rule = FLOW_TEMPLATE_EXPRESSION_RULES[0]
-    code_properties = defs["FlowCodeActionDefinition"]["properties"]
-    tool_properties = defs["FlowToolActionDefinition"]["properties"]
-    crew_properties = defs["FlowCrewActionDefinition"]["properties"]
-    assert expression_rule in code_properties["with"]["description"]
-    assert expression_rule in tool_properties["with"]["description"]
-    assert expression_rule in crew_properties["inputs"]["description"]
+    assert "Inline Agent definition" in agent_properties["with"]["description"]
+    assert "run an inline Agent" in agent_properties["call"]["description"]
 
     state_schema = next(
         branch
@@ -167,16 +154,14 @@ def test_flow_definition_json_schema_carries_field_examples_only():
 
     script_properties = defs["FlowScriptActionDefinition"]["properties"]
     assert script_properties["call"]["examples"] == ["script"]
-    assert "state['topic'].strip()" in script_properties["code"]["examples"][0]
+    assert "input.strip()" in script_properties["code"]["examples"][0]
     assert script_properties["language"]["examples"] == ["python"]
 
     action_properties = defs["FlowCodeActionDefinition"]["properties"]
     assert action_properties["ref"]["examples"] == [
         "my_project.flows:normalize_topic"
     ]
-    assert action_properties["with"]["examples"] == [
-        {"topic": "${state.topic}", "query": "News about ${state.topic}"}
-    ]
+    assert action_properties["with"]["examples"] == [{"topic": "${state.topic}"}]
 
     agent_properties = defs["FlowAgentActionDefinition"]["properties"]
     assert agent_properties["call"]["examples"] == ["agent"]
@@ -639,7 +624,7 @@ def test_flow_definition_from_declaration_accepts_json_and_yaml_strings():
             return "left"
 
         @listen("left")
-        def handle_left(self):
+        def left(self):
             return "left"
 
     expected = RoundTripFlow.flow_definition()
@@ -664,11 +649,11 @@ def test_flow_definition_from_declaration_accepts_json_and_yaml_strings():
                 "ref": "test_flow_definition:RoundTripFlow.decide"
               }
             },
-            "handle_left": {
+            "left": {
               "listen": "left",
               "do": {
                 "call": "code",
-                "ref": "test_flow_definition:RoundTripFlow.handle_left"
+                "ref": "test_flow_definition:RoundTripFlow.left"
               }
             }
           }
@@ -689,11 +674,11 @@ def test_flow_definition_from_declaration_accepts_json_and_yaml_strings():
             do:
               call: code
               ref: test_flow_definition:RoundTripFlow.decide
-          handle_left:
+          left:
             listen: left
             do:
               call: code
-              ref: test_flow_definition:RoundTripFlow.handle_left
+              ref: test_flow_definition:RoundTripFlow.left
         """,
     ]
 
@@ -960,11 +945,11 @@ def test_flow_definition_infers_literal_router_emit():
             return "left"
 
         @listen("left")
-        def handle_left(self):
+        def left(self):
             return "left"
 
         @listen("right")
-        def handle_right(self):
+        def right(self):
             return "right"
 
     definition = LiteralRouterFlow.flow_definition()
@@ -987,11 +972,11 @@ def test_flow_definition_infers_enum_router_emit():
             return Decision.APPROVE
 
         @listen("approve")
-        def handle_approve(self):
+        def approve(self):
             return "approve"
 
         @listen("reject")
-        def handle_reject(self):
+        def reject(self):
             return "reject"
 
     definition = EnumRouterFlow.flow_definition()
@@ -1010,11 +995,11 @@ def test_flow_definition_infers_literal_union_router_emit():
             return "left"
 
         @listen("left")
-        def handle_left(self):
+        def left(self):
             return "left"
 
         @listen("right")
-        def handle_right(self):
+        def right(self):
             return "right"
 
     definition = LiteralUnionRouterFlow.flow_definition()
@@ -1068,7 +1053,7 @@ def test_flow_definition_does_not_infer_unannotated_router_body_emit():
             return "left"
 
         @listen("left")
-        def handle_left(self):
+        def left(self):
             return "left"
 
     definition = UnannotatedRouterFlow.flow_definition()
@@ -1087,11 +1072,11 @@ def test_flow_definition_accepts_explicit_router_events():
             return self.state["dynamic_event"]
 
         @listen("left")
-        def handle_left(self):
+        def left(self):
             return "left"
 
         @listen("right")
-        def handle_right(self):
+        def right(self):
             return "right"
 
     definition = ExplicitRouterFlow.flow_definition()
@@ -1163,7 +1148,7 @@ def test_router_human_feedback_preserves_existing_router_metadata():
             return "approved"
 
         @listen("approved")
-        def handle_approved(self):
+        def approved(self):
             return "approved"
 
     definition = RouterHumanFeedbackFlow.flow_definition()
@@ -1228,30 +1213,6 @@ def test_static_string_listener_is_allowed_by_contract():
     assert definition.methods["handle"].listen == "begni"
 
 
-@pytest.mark.parametrize("listen", ["publish", {"or": ["publish", "revise"]}])
-@pytest.mark.parametrize("router_enabled", [False, True])
-def test_flow_definition_rejects_method_self_listen(listen, router_enabled):
-    with pytest.raises(ValueError, match="methods.publish.listen"):
-        flow_definition.FlowDefinition.from_declaration(contents=
-            {
-                "schema": "crewai.flow/v1",
-                "name": "SelfListenFlow",
-                "methods": {
-                    "begin": {
-                        "do": {"ref": "loaded_flows:SelfListenFlow.begin"},
-                        "start": True,
-                    },
-                    "publish": {
-                        "do": {"ref": "loaded_flows:SelfListenFlow.publish"},
-                        "listen": listen,
-                        "router": router_enabled,
-                        "emit": ["done"] if router_enabled else None,
-                    },
-                },
-            }
-        )
-
-
 def test_start_false_not_classified_as_start_method():
     definition = flow_definition.FlowDefinition.from_declaration(contents=
         {
@@ -1339,68 +1300,3 @@ def test_flow_definition_allows_router_without_trigger(caplog):
     StandaloneRouterFlow.flow_definition()
 
     assert not caplog.records
-
-
-def test_skill_documents_flow_wiring():
-    skill = flow_definition.FlowDefinition.skill()
-
-    assert isinstance(skill, str)
-    assert "```yaml" in skill
-    assert "[Method](#method-methods)" in skill
-    assert 'input: "Reviewed research: ${outputs.research_brief.raw}"' in skill
-    assert 'text(root, "path", "default")' in skill
-    assert "trust CrewAI defaults and omit them" in skill
-    assert "#### LLM Definition" in skill
-    assert "`max_tokens` (optional): integer | null; default `null`" in skill
-    assert "CrewAI does not set an explicit output token cap" in skill
-    assert "`planning_config` (optional): object | null; default `null`" in skill
-    assert "Set `max_attempts` to limit planning refinement attempts" in skill
-    assert "`allow_delegation` (optional): boolean | null; default `null`" in skill
-    assert "`max_iter` (optional): integer | null; default `null`" in skill
-    assert "`max_rpm` (optional): integer | null; default `null`" in skill
-    assert "`max_execution_time` (optional): integer | null; default `null`" in skill
-    assert "Maximum execution time in seconds for an agent" in skill
-    for rule in FLOW_TEMPLATE_EXPRESSION_RULES:
-        assert rule in skill
-    for example in FLOW_TEMPLATE_EXPRESSION_EXAMPLES["yaml"]:
-        assert example["title"] in skill
-        assert example["code"] in skill
-
-
-def test_skill_can_render_json_examples():
-    skill = flow_definition.FlowDefinition.skill(examples_format="json")
-
-    assert "```json" in skill
-    assert '"schema": "crewai.flow/v1"' in skill
-    for example in FLOW_TEMPLATE_EXPRESSION_EXAMPLES["json"]:
-        assert example["title"] in skill
-        assert example["code"] in skill
-    assert FLOW_TEMPLATE_EXPRESSION_EXAMPLES["yaml"][0]["code"] not in skill
-    assert "```yaml" not in skill
-
-
-def test_skill_ignores_unknown_skips():
-    skill = flow_definition.FlowDefinition.skill(skips=["unknown"])
-
-    assert "[Method](#method-methods)" in skill
-
-
-def test_skill_with_skips_is_shorter():
-    full = flow_definition.FlowDefinition.skill()
-    trimmed = flow_definition.FlowDefinition.skill(
-        skips=[
-            "each",
-            "hitl",
-            "persistence",
-            "expression_action",
-            "script_action",
-            "tool_action",
-        ]
-    )
-
-    assert "[Method](#method-methods)" in trimmed
-    assert "call: expression" not in trimmed
-    assert "Prefer `call: expression`" not in trimmed
-    assert "call: script" not in trimmed
-    assert "call: tool" not in trimmed
-    assert len(trimmed) < len(full)

lib/crewai/tests/test_stream_frames.py

@@ -1,300 +0,0 @@
-"""Tests for the public stream frame contract."""
-
-from __future__ import annotations
-
-import asyncio
-from collections.abc import AsyncIterator
-from typing import Any, ClassVar
-
-import pytest
-
-from crewai.events.event_bus import crewai_event_bus
-from crewai.events.types.flow_events import ConversationMessageAddedEvent
-from crewai.events.types.llm_events import LLMStreamChunkEvent, LLMThinkingChunkEvent
-from crewai.events.types.tool_usage_events import ToolUsageStartedEvent
-from crewai.flow.flow import Flow, start
-from crewai.llms.base_llm import BaseLLM, call_stop_override
-from crewai.types.streaming import StreamFrame
-
-
-class FrameFlow(Flow):
-    @start()
-    def run(self) -> str:
-        crewai_event_bus.emit(
-            self,
-            LLMStreamChunkEvent(
-                type="llm_stream_chunk",
-                chunk="hello",
-                call_id="call-1",
-            ),
-        )
-        crewai_event_bus.emit(
-            self,
-            LLMThinkingChunkEvent(
-                type="llm_thinking_chunk",
-                chunk="thinking",
-                call_id="call-1",
-            ),
-        )
-        crewai_event_bus.emit(
-            self,
-            ConversationMessageAddedEvent(
-                type="conversation_message_added",
-                flow_name=self._definition.name,
-                session_id="session-1",
-                role="assistant",
-                content="hello",
-                message_index=0,
-            ),
-        )
-        crewai_event_bus.emit(
-            self,
-            ToolUsageStartedEvent(
-                type="tool_usage_started",
-                tool_name="search",
-                tool_args={"query": "crew"},
-            ),
-        )
-        return "done"
-
-
-class DirectStreamingLLM(BaseLLM):
-    call_stream_values: ClassVar[list[bool | None]] = []
-    raw_stream_values: ClassVar[list[bool | None]] = []
-    call_instance_ids: ClassVar[list[int]] = []
-    call_stop_values: ClassVar[list[list[str]]] = []
-
-    def call(self, messages: Any, *args: Any, **kwargs: Any) -> str:
-        self.call_stream_values.append(self._effective_stream())
-        self.raw_stream_values.append(self.stream)
-        self.call_instance_ids.append(id(self))
-        self.call_stop_values.append(list(self.stop_sequences))
-        self._track_token_usage_internal(
-            {
-                "prompt_tokens": 1,
-                "completion_tokens": 2,
-                "total_tokens": 3,
-            }
-        )
-        crewai_event_bus.emit(
-            self,
-            LLMStreamChunkEvent(
-                type="llm_stream_chunk",
-                chunk="hel",
-                call_id="call-1",
-            ),
-        )
-        crewai_event_bus.emit(
-            self,
-            LLMStreamChunkEvent(
-                type="llm_stream_chunk",
-                chunk="lo",
-                call_id="call-1",
-            ),
-        )
-        return "hello"
-
-
-def test_stream_frame_contract_and_ordering() -> None:
-    stream = FrameFlow().stream_events()
-
-    with pytest.raises(RuntimeError, match="Streaming has not completed yet"):
-        _ = stream.result
-
-    with stream:
-        frames = list(stream.events)
-
-    assert stream.result == "done"
-    assert all(isinstance(frame, StreamFrame) for frame in frames)
-    assert [frame.seq for frame in frames] == sorted(frame.seq for frame in frames)
-
-    by_type = {frame.type: frame for frame in frames}
-    assert by_type["flow_started"].channel == "flow"
-    assert by_type["method_execution_started"].parent_id == by_type["flow_started"].id
-    assert by_type["llm_stream_chunk"].channel == "llm"
-    assert by_type["llm_thinking_chunk"].channel == "llm"
-    assert by_type["conversation_message_added"].channel == "messages"
-    assert by_type["tool_usage_started"].channel == "tools"
-    assert "FrameFlow" in by_type["method_execution_started"].namespace
-    assert "run" in by_type["method_execution_started"].namespace
-
-
-def test_stream_subscribe_filters_channels_without_losing_order() -> None:
-    with FrameFlow().stream_events() as stream:
-        frames = list(stream.interleave(["messages", "tools"]))
-
-    assert [frame.channel for frame in frames] == ["messages", "tools"]
-    assert [frame.seq for frame in frames] == sorted(frame.seq for frame in frames)
-    assert stream.result == "done"
-
-
-def test_stream_projections_replay_cached_frames_after_exhaustion() -> None:
-    with FrameFlow().stream_events() as stream:
-        all_frames = list(stream.events)
-
-    assert [frame.content for frame in stream.llm if frame.content] == [
-        "hello",
-        "thinking",
-    ]
-    assert [frame.type for frame in stream.tools] == ["tool_usage_started"]
-    assert list(stream.events) == all_frames
-
-
-def test_stream_channel_projection_can_be_followed_by_cached_projection() -> None:
-    with FrameFlow().stream_events() as stream:
-        llm_frames = list(stream.llm)
-
-    assert [frame.content for frame in llm_frames if frame.content] == [
-        "hello",
-        "thinking",
-    ]
-    assert [frame.type for frame in stream.flow] == [
-        "flow_started",
-        "method_execution_started",
-        "method_execution_finished",
-        "flow_finished",
-    ]
-
-
-def test_stream_errors_surface_after_failed_frame() -> None:
-    class ErrorFlow(Flow):
-        @start()
-        def run(self) -> str:
-            raise ValueError("boom")
-
-    stream = ErrorFlow().stream_events()
-
-    with pytest.raises(ValueError, match="boom"):
-        list(stream.events)
-
-    assert any(frame.type == "method_execution_failed" for frame in stream.frames)
-    with pytest.raises(ValueError, match="boom"):
-        _ = stream.result
-
-
-def test_flow_streaming_returns_iterable_frame_session() -> None:
-    flow = FrameFlow()
-    flow.stream = True
-
-    stream = flow.kickoff()
-
-    with stream:
-        frames = list(stream)
-
-    assert all(isinstance(frame, StreamFrame) for frame in frames)
-    assert [frame.content for frame in frames if frame.content] == [
-        "hello",
-        "thinking",
-    ]
-    first_content_frame = next(frame for frame in frames if frame.content)
-    assert first_content_frame.event["chunk"] == "hello"
-    assert stream.result == "done"
-
-
-def test_direct_llm_stream_events_scope_and_restore_stream_flag() -> None:
-    DirectStreamingLLM.call_stream_values = []
-    DirectStreamingLLM.raw_stream_values = []
-    DirectStreamingLLM.call_instance_ids = []
-    DirectStreamingLLM.call_stop_values = []
-    llm = DirectStreamingLLM(model="gpt-4o-mini", stream=False)
-
-    with call_stop_override(llm, ["STOP"]):
-        with llm.stream_events("hello") as stream:
-            frames = list(stream)
-
-    assert [frame.content for frame in frames] == ["hel", "lo"]
-    assert frames[0].event["chunk"] == "hel"
-    assert stream.result == "hello"
-    assert llm.stream is False
-    assert DirectStreamingLLM.call_stream_values == [True]
-    assert DirectStreamingLLM.raw_stream_values == [False]
-    assert DirectStreamingLLM.call_instance_ids == [id(llm)]
-    assert DirectStreamingLLM.call_stop_values == [["STOP"]]
-    usage = llm.get_token_usage_summary()
-    assert usage.total_tokens == 3
-    assert usage.prompt_tokens == 1
-    assert usage.completion_tokens == 2
-    assert usage.successful_requests == 1
-
-
-@pytest.mark.asyncio
-async def test_astream_scopes_concurrent_executions() -> None:
-    class ConcurrentFlow(Flow):
-        @start()
-        async def run(self) -> str:
-            label = str(self.state["label"])
-            await asyncio.sleep(0)
-            crewai_event_bus.emit(
-                self,
-                LLMStreamChunkEvent(
-                    type="llm_stream_chunk",
-                    chunk=label,
-                    call_id=label,
-                ),
-            )
-            return label
-
-    async def collect(label: str) -> tuple[str, list[str]]:
-        async with ConcurrentFlow().astream(inputs={"label": label}) as stream:
-            frames = [frame async for frame in stream.llm]
-        return stream.result, [frame.data["chunk"] for frame in frames]
-
-    first, second = await asyncio.gather(collect("first"), collect("second"))
-
-    assert first == ("first", ["first"])
-    assert second == ("second", ["second"])
-
-
-@pytest.mark.asyncio
-async def test_async_stream_projections_replay_cached_frames_after_exhaustion() -> None:
-    async with FrameFlow().astream() as stream:
-        all_frames = [frame async for frame in stream.events]
-
-    llm_frames = [frame async for frame in stream.llm]
-    tool_frames = [frame async for frame in stream.tools]
-    replayed_frames = [frame async for frame in stream.events]
-
-    assert [frame.content for frame in llm_frames if frame.content] == [
-        "hello",
-        "thinking",
-    ]
-    assert [frame.type for frame in tool_frames] == ["tool_usage_started"]
-    assert replayed_frames == all_frames
-
-
-@pytest.mark.asyncio
-async def test_async_stream_channel_projection_can_be_followed_by_cached_projection() -> None:
-    async with FrameFlow().astream() as stream:
-        llm_frames = [frame async for frame in stream.llm]
-
-    flow_frames = [frame async for frame in stream.flow]
-
-    assert [frame.content for frame in llm_frames if frame.content] == [
-        "hello",
-        "thinking",
-    ]
-    assert [frame.type for frame in flow_frames] == [
-        "flow_started",
-        "method_execution_started",
-        "method_execution_finished",
-        "flow_finished",
-    ]
-
-
-@pytest.mark.asyncio
-async def test_astream_cancellation_cleans_up_task() -> None:
-    class SlowFlow(Flow):
-        @start()
-        async def run(self) -> str:
-            await asyncio.sleep(10)
-            return "too late"
-
-    stream = SlowFlow().astream()
-    events: AsyncIterator[StreamFrame] = stream.events
-    first_frame = await anext(events)
-
-    assert first_frame.type == "flow_started"
-    await stream.aclose()
-
-    assert stream.is_cancelled is True
-    assert stream.is_completed is True

lib/crewai/tests/test_streaming.py

@@ -11,15 +11,11 @@ from crewai import Agent, Crew, Task
 from crewai.events.event_bus import crewai_event_bus
 from crewai.events.types.llm_events import LLMStreamChunkEvent, ToolCall, FunctionCall
 from crewai.flow.flow import Flow, start
-from crewai.state.checkpoint_config import CheckpointConfig
 from crewai.types.streaming import (
-    AsyncStreamSession,
     CrewStreamingOutput,
     FlowStreamingOutput,
     StreamChunk,
     StreamChunkType,
-    StreamFrame,
-    StreamSession,
     ToolCallChunk,
 )
 
@@ -421,8 +417,8 @@ class TestCrewKickoffStreamingAsync:
 class TestFlowKickoffStreaming:
     """Tests for Flow(stream=True).kickoff() method."""
 
-    def test_kickoff_streaming_returns_stream_session(self) -> None:
-        """Test that flow kickoff with stream=True returns StreamSession."""
+    def test_kickoff_streaming_returns_streaming_output(self) -> None:
+        """Test that flow kickoff with stream=True returns FlowStreamingOutput."""
 
         class SimpleFlow(Flow[dict[str, Any]]):
             @start()
@@ -432,7 +428,7 @@ class TestFlowKickoffStreaming:
         flow = SimpleFlow()
         flow.stream = True
         streaming = flow.kickoff()
-        assert isinstance(streaming, StreamSession)
+        assert isinstance(streaming, FlowStreamingOutput)
 
     def test_flow_kickoff_streaming_captures_chunks(self) -> None:
         """Test that flow streaming captures LLM chunks from crew execution."""
@@ -473,7 +469,7 @@ class TestFlowKickoffStreaming:
 
         with patch.object(Flow, "kickoff", mock_kickoff_fn):
             streaming = flow.kickoff()
-            assert isinstance(streaming, StreamSession)
+            assert isinstance(streaming, FlowStreamingOutput)
             chunks = list(streaming)
 
         assert len(chunks) >= 2
@@ -504,38 +500,19 @@ class TestFlowKickoffStreaming:
 
         with patch.object(Flow, "kickoff", mock_kickoff_fn):
             streaming = flow.kickoff()
-            assert isinstance(streaming, StreamSession)
+            assert isinstance(streaming, FlowStreamingOutput)
             _ = list(streaming)
 
         result = streaming.result
         assert result == "flow result"
 
-    def test_streaming_kickoff_passes_checkpoint_config_to_stream_events(self) -> None:
-        """stream=True preserves checkpoint config when routing to stream_events."""
-
-        class TestFlow(Flow[dict[str, Any]]):
-            @start()
-            def generate(self) -> str:
-                return "flow result"
-
-        flow = TestFlow()
-        flow.stream = True
-        checkpoint = CheckpointConfig()
-
-        with patch.object(flow, "stream_events", wraps=flow.stream_events) as spy:
-            streaming = flow.kickoff(from_checkpoint=checkpoint)
-            list(streaming)
-
-        assert spy.call_args.kwargs["from_checkpoint"] is checkpoint
-        assert streaming.result == "flow result"
-
 
 class TestFlowKickoffStreamingAsync:
     """Tests for Flow(stream=True).kickoff_async() method."""
 
     @pytest.mark.asyncio
-    async def test_kickoff_streaming_async_returns_stream_session(self) -> None:
-        """Test that flow kickoff_async with stream=True returns AsyncStreamSession."""
+    async def test_kickoff_streaming_async_returns_streaming_output(self) -> None:
+        """Test that flow kickoff_async with stream=True returns FlowStreamingOutput."""
 
         class SimpleFlow(Flow[dict[str, Any]]):
             @start()
@@ -545,7 +522,7 @@ class TestFlowKickoffStreamingAsync:
         flow = SimpleFlow()
         flow.stream = True
         streaming = await flow.kickoff_async()
-        assert isinstance(streaming, AsyncStreamSession)
+        assert isinstance(streaming, FlowStreamingOutput)
 
     @pytest.mark.asyncio
     async def test_flow_kickoff_streaming_async_captures_chunks(self) -> None:
@@ -590,8 +567,8 @@ class TestFlowKickoffStreamingAsync:
 
         with patch.object(Flow, "kickoff_async", mock_kickoff_fn):
             streaming = await flow.kickoff_async()
-            assert isinstance(streaming, AsyncStreamSession)
-            chunks: list[StreamFrame] = []
+            assert isinstance(streaming, FlowStreamingOutput)
+            chunks: list[StreamChunk] = []
             async for chunk in streaming:
                 chunks.append(chunk)
 
@@ -624,36 +601,13 @@ class TestFlowKickoffStreamingAsync:
 
         with patch.object(Flow, "kickoff_async", mock_kickoff_fn):
             streaming = await flow.kickoff_async()
-            assert isinstance(streaming, AsyncStreamSession)
+            assert isinstance(streaming, FlowStreamingOutput)
             async for _ in streaming:
                 pass
 
         result = streaming.result
         assert result == "async flow result"
 
-    @pytest.mark.asyncio
-    async def test_streaming_kickoff_async_passes_checkpoint_config_to_astream(
-        self,
-    ) -> None:
-        """stream=True preserves checkpoint config when routing to astream."""
-
-        class TestFlow(Flow[dict[str, Any]]):
-            @start()
-            async def generate(self) -> str:
-                return "async flow result"
-
-        flow = TestFlow()
-        flow.stream = True
-        checkpoint = CheckpointConfig()
-
-        with patch.object(flow, "astream", wraps=flow.astream) as spy:
-            streaming = await flow.kickoff_async(from_checkpoint=checkpoint)
-            async for _ in streaming:
-                pass
-
-        assert spy.call_args.kwargs["from_checkpoint"] is checkpoint
-        assert streaming.result == "async flow result"
-
 
 class TestStreamingEdgeCases:
     """Tests for edge cases in streaming functionality."""

lib/crewai/tests/test_streaming_integration.py

@@ -3,10 +3,8 @@
 import pytest
 
 from crewai import Agent, Crew, Task
-from crewai.events.event_bus import crewai_event_bus
-from crewai.events.types.llm_events import LLMStreamChunkEvent
 from crewai.flow.flow import Flow, start
-from crewai.types.streaming import AsyncStreamSession, CrewStreamingOutput, StreamSession
+from crewai.types.streaming import CrewStreamingOutput, FlowStreamingOutput
 
 
 @pytest.fixture
@@ -214,7 +212,7 @@ class TestStreamingFlowIntegration:
 
         streaming = flow.kickoff()
 
-        assert isinstance(streaming, StreamSession)
+        assert isinstance(streaming, FlowStreamingOutput)
 
         chunks = []
         for chunk in streaming:
@@ -234,14 +232,6 @@ class TestStreamingFlowIntegration:
 
             @start()
             def execute(self) -> str:
-                crewai_event_bus.emit(
-                    self,
-                    LLMStreamChunkEvent(
-                        type="llm_stream_chunk",
-                        chunk="Flow result",
-                        call_id="call-1",
-                    ),
-                )
                 return "Flow result"
 
         flow = SimpleFlow()
@@ -251,11 +241,8 @@ class TestStreamingFlowIntegration:
             pass
 
         assert streaming.is_completed is True
-        content_frames = [frame for frame in streaming.frames if frame.content]
-        full_text = "".join(frame.content for frame in content_frames)
-        assert full_text == "Flow result"
-        assert len(content_frames) == 1
-        assert len(streaming.frames) > 0
+        streaming.get_full_text()
+        assert len(streaming.chunks) >= 0
 
         result = streaming.result
         assert result is not None
@@ -294,7 +281,7 @@ class TestStreamingFlowIntegration:
 
         streaming = await flow.kickoff_async()
 
-        assert isinstance(streaming, AsyncStreamSession)
+        assert isinstance(streaming, FlowStreamingOutput)
 
         chunks = []
         async for chunk in streaming:

lib/crewai/tests/utilities/events/test_async_event_bus.py

@@ -9,7 +9,6 @@ import pytest
 
 from crewai.events.base_events import BaseEvent
 from crewai.events.event_bus import crewai_event_bus
-from crewai.events.stream_context import add_stream_sink, reset_stream_sinks
 
 
 class AsyncTestEvent(BaseEvent):
@@ -54,24 +53,6 @@ async def test_aemit_with_async_handlers():
         assert received_events[0] == event
 
 
-@pytest.mark.asyncio
-async def test_aemit_publishes_to_active_stream_sinks():
-    published_events = []
-
-    def sink(source: object, event: BaseEvent) -> None:
-        published_events.append((source, event))
-
-    event = AsyncTestEvent(type="async_test")
-    token = add_stream_sink(sink)
-    try:
-        await crewai_event_bus.aemit("test_source", event)
-    finally:
-        reset_stream_sinks(token)
-
-    assert published_events == [("test_source", event)]
-    assert event.emission_sequence is not None
-
-
 @pytest.mark.asyncio
 async def test_multiple_async_handlers():
     received_events_1 = []

lib/devtools/src/crewai_devtools/__init__.py

@@ -1,3 +1,3 @@
 """CrewAI development tools."""
 
-__version__ = "1.15.2a2"
+__version__ = "1.15.1"

scripts/age90_file_input_runner.py

@@ -0,0 +1,226 @@
+# ruff: noqa: T201
+"""Manual runner for AGE-90 PDF input handling.
+
+Usage examples:
+    uv run python scripts/age90_file_input_runner.py
+    uv run python scripts/age90_file_input_runner.py --mode fallback
+    uv run python scripts/age90_file_input_runner.py --mode payload --pdf ./sample_story.pdf
+    uv run python scripts/age90_file_input_runner.py --mode kickoff --pdf ./sample_story.pdf
+"""
+
+from __future__ import annotations
+
+import argparse
+from collections.abc import Mapping, Sequence
+from contextlib import nullcontext
+import os
+from pathlib import Path
+from typing import Any
+from unittest.mock import patch
+
+from crewai_files import PDFFile, format_multimodal_content, get_supported_content_types
+
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_PDF = ROOT / "lib" / "crewai-files" / "tests" / "fixtures" / "agents.pdf"
+
+
+def _content_summary(block: dict[str, Any]) -> dict[str, str]:
+    """Return a compact, non-base64 summary of a content block."""
+    summary: dict[str, str] = {"type": str(block.get("type"))}
+    for key in ("file_id", "file_url", "filename", "image_url"):
+        if key in block:
+            value = str(block[key])
+            summary[key] = value[:100] + ("..." if len(value) > 100 else "")
+    if "file_data" in block:
+        value = str(block["file_data"])
+        summary["file_data"] = value[:80] + f"... ({len(value)} chars)"
+    return summary
+
+
+def _sanitize_payload(value: Any) -> Any:
+    """Shorten large fields before printing API payloads."""
+    if isinstance(value, Mapping):
+        sanitized: dict[str, Any] = {}
+        for key, item in value.items():
+            if key == "file_data" and isinstance(item, str):
+                sanitized[key] = item[:100] + f"... ({len(item)} chars)"
+            else:
+                sanitized[str(key)] = _sanitize_payload(item)
+        return sanitized
+
+    if isinstance(value, Sequence) and not isinstance(value, str | bytes):
+        return [_sanitize_payload(item) for item in value]
+
+    return value
+
+
+def inspect_native_path(pdf_path: Path, provider: str, api: str | None) -> None:
+    """Show whether the PDF is treated as a native multimodal input."""
+    pdf = PDFFile(source=str(pdf_path))
+    supported_types = get_supported_content_types(provider, api=api)
+    blocks = format_multimodal_content(
+        {"document": pdf},
+        provider=provider,
+        api=api,
+        text="Summarize this PDF.",
+    )
+
+    print("\n== Native File Formatting ==")
+    print(f"PDF: {pdf_path}")
+    print(f"Provider/API: {provider} / {api or 'default'}")
+    print(f"Supported content types: {supported_types}")
+    print(f"Content block count: {len(blocks)}")
+    for index, block in enumerate(blocks, start=1):
+        print(f"  {index}. {_content_summary(block)}")
+
+    has_pdf_block = any(block.get("type") == "input_file" for block in blocks)
+    print(f"PDF native input_file block: {'YES' if has_pdf_block else 'NO'}")
+
+
+def inspect_fallback_tool(pdf_path: Path) -> None:
+    """Show what read_file returns if a PDF falls back to the tool path."""
+    from crewai.tools.agent_tools.read_file_tool import ReadFileTool
+
+    tool = ReadFileTool()
+    tool.set_files({"document": PDFFile(source=str(pdf_path))})
+    result = tool._run("document")
+
+    print("\n== read_file Fallback ==")
+    print(f"Returned {len(result)} chars")
+    print(f"Contains Base64 marker: {'YES' if 'Base64:' in result else 'NO'}")
+    print("\nPreview:")
+    print(result[:1200])
+    if len(result) > 1200:
+        print("...")
+
+
+def run_crew_kickoff(
+    pdf_path: Path,
+    model: str,
+    api: str | None,
+    prompt: str,
+    *,
+    payload_only: bool = False,
+) -> None:
+    """Run a real Crew kickoff against the supplied model."""
+    from crewai import LLM, Agent, Crew, Task
+
+    if model.startswith("openai/") and not os.getenv("OPENAI_API_KEY") and not payload_only:
+        raise SystemExit(
+            "OPENAI_API_KEY is not set. Export it before running --mode kickoff."
+        )
+
+    kwargs: dict[str, Any] = {"model": model, "temperature": 0}
+    if api:
+        kwargs["api"] = api
+
+    llm = LLM(**kwargs)
+    agent = Agent(
+        role="PDF Analyst",
+        goal="Read the provided PDF and answer accurately from its contents",
+        backstory="You inspect uploaded files carefully and avoid guessing.",
+        llm=llm,
+        verbose=True,
+    )
+    task = Task(
+        description=prompt,
+        expected_output="A concise answer grounded in the uploaded PDF.",
+        agent=agent,
+    )
+    crew = Crew(agents=[agent], tasks=[task], verbose=True)
+
+    print("\n== Crew Kickoff ==")
+    print(f"Model/API: {model} / {api or 'default'}")
+    print(f"PDF: {pdf_path}")
+
+    context = nullcontext()
+    if payload_only:
+        from crewai.llms.providers.openai.completion import OpenAICompletion
+
+        def print_payload_and_stop(
+            self: OpenAICompletion,
+            params: dict[str, Any],
+            *_args: Any,
+            **_kwargs: Any,
+        ) -> str:
+            print("\n== Sanitized Responses Payload ==")
+            print(_sanitize_payload(params))
+            return "Payload debug complete."
+
+        context = patch.object(
+            OpenAICompletion,
+            "_handle_responses",
+            print_payload_and_stop,
+        )
+
+    with context:
+        result = crew.kickoff(input_files={"document": PDFFile(source=str(pdf_path))})
+
+    print("\n== Final Output ==")
+    print(result.raw)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--mode",
+        choices=("inspect", "fallback", "payload", "kickoff", "all"),
+        default="inspect",
+        help="What to run. 'inspect', 'fallback', and 'payload' do not call an LLM.",
+    )
+    parser.add_argument(
+        "--pdf",
+        type=Path,
+        default=DEFAULT_PDF,
+        help="PDF file to test.",
+    )
+    parser.add_argument(
+        "--provider",
+        default="gpt-4o-mini",
+        help="Provider/model string for file formatting inspection.",
+    )
+    parser.add_argument(
+        "--model",
+        default="openai/gpt-4o-mini",
+        help="CrewAI model for real kickoff mode.",
+    )
+    parser.add_argument(
+        "--api",
+        default="responses",
+        help="API variant. Use '' to omit.",
+    )
+    parser.add_argument(
+        "--prompt",
+        default="Summarize the uploaded PDF in 3 bullet points. Do not guess.",
+        help="Task prompt for kickoff mode.",
+    )
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    pdf_path = args.pdf.expanduser().resolve()
+    api = args.api or None
+
+    if not pdf_path.exists():
+        raise SystemExit(f"PDF not found: {pdf_path}")
+
+    if args.mode in ("inspect", "all"):
+        inspect_native_path(pdf_path, args.provider, api)
+    if args.mode in ("fallback", "all"):
+        inspect_fallback_tool(pdf_path)
+    if args.mode == "payload":
+        run_crew_kickoff(pdf_path, args.model, api, args.prompt, payload_only=True)
+    if args.mode in ("kickoff", "all"):
+        run_crew_kickoff(
+            pdf_path,
+            args.model,
+            api,
+            args.prompt,
+            payload_only=args.mode == "all",
+        )
+
+
+if __name__ == "__main__":
+    main()

uv.lock

@@ -13,7 +13,7 @@ resolution-markers = [
 ]
 
 [options]
-exclude-newer = "2026-06-28T20:06:34.114646Z"
+exclude-newer = "2026-06-20T16:46:21.117658Z"
 exclude-newer-span = "P3D"
 
 [options.exclude-newer-package]
@@ -1371,7 +1371,6 @@ azure-ai-inference = [
     { name = "azure-identity" },
 ]
 bedrock = [
-    { name = "aiobotocore" },
     { name = "boto3" },
 ]
 docling = [
@@ -1419,7 +1418,6 @@ watson = [
 requires-dist = [
     { name = "a2a-sdk", marker = "extra == 'a2a'", specifier = "~=0.3.10" },
     { name = "aiobotocore", marker = "extra == 'aws'", specifier = "~=3.5.0" },
-    { name = "aiobotocore", marker = "extra == 'bedrock'", specifier = "~=3.5.0" },
     { name = "aiocache", extras = ["memcached", "redis"], marker = "extra == 'a2a'", specifier = "~=0.12.3" },
     { name = "aiofiles", specifier = "~=24.1.0" },
     { name = "aiosqlite", specifier = "~=0.21.0" },