Compare commits

..

5 Commits

Author SHA1 Message Date
Vinicius Brasil
854c67d21c docs: snapshot and changelog for v1.14.8a2 (#6230)
Some checks are pending
CodeQL Advanced / Analyze (actions) (push) Waiting to run
CodeQL Advanced / Analyze (python) (push) Waiting to run
Check Documentation Broken Links / Check broken links (push) Waiting to run
Vulnerability Scan / pip-audit (push) Waiting to run
2026-06-18 16:42:17 -07:00
Vinicius Brasil
f48a6389f1 feat: bump versions to 1.14.8a2 (#6229) 2026-06-18 16:37:27 -07:00
Vinicius Brasil
bc2c2a858c Add single agent action to Flow definitions (#6226)
Some checks failed
Build uv cache / build-cache (3.10) (push) Waiting to run
Build uv cache / build-cache (3.11) (push) Waiting to run
Build uv cache / build-cache (3.12) (push) Waiting to run
Build uv cache / build-cache (3.13) (push) Waiting to run
CodeQL Advanced / Analyze (actions) (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Check Documentation Broken Links / Check broken links (push) Has been cancelled
Vulnerability Scan / pip-audit (push) Has been cancelled
* Add single agent action to Flow definitions

Lets a flow method build and run a single CrewAI agent directly, without
wrapping it in a crew. Same idea as the existing `crew` action, but for
one agent.

  methods:
    answer:
      do:
        call: agent
        with:
          role: Analyst
          goal: Answer questions
          backstory: Knows things.
          input: "${state.question}"
      start: true

* `input` is required and interpolated from flow state, like
  `${state.question}` or `${item}` inside an `each` loop
* optional `response_format` points at a Pydantic model (`{"python":
  "models.AnswerModel"}`) to get structured output
* `input` must be a string and its CEL is validated at load time, so bad
  expressions like `${state.}` fail early

* Simplify test code
2026-06-18 14:53:33 -07:00
Lucas Gomide
fa89ac428e docs: add Datadog integration guide with importable operations dashboard (#6225)
Adds a consolidated `datadog.mdx` under `docs/edge/{en,pt-BR,ko,ar}/enterprise/guides/`
covering both the Datadog Agent path (stdout JSON logs via `CREWAI_LOG_FORMAT=json`)
and the Datadog OTLP intake, with a JSON log schema reference and a ready-to-import
operations dashboard (`datadog_dashboard.json`). Reframes `capture_telemetry_logs.mdx`
to lead with OpenTelemetry as the vendor-neutral path and point readers to the new
Datadog page for that ecosystem's setup.
2026-06-18 16:18:42 -04:00
Vinicius Brasil
b0816e00b6 Validate flow CEL expressions at definition load time (#6224)
* Validate flow CEL expressions at definition load time

Promote CEL expression handling to a public Expression API and validate expressions when a FlowDefinition is built instead of when it executes.

Invalid CEL syntax or unknown roots now raise ValidationError from FlowDefinition.from_yaml() and FlowDefinition.from_dict(). Expressions may reference state and outputs, plus item inside each.do; bare identifiers are rejected as unknown roots.

For with values, the CEL contract is intentionally simple: after trimming whitespace, a string is evaluated as CEL only if it starts with ${ and ends with }. Anything else is treated as a literal value, so partial interpolation is not supported. If the content inside the wrapper is not valid CEL, validation fails.

Examples:

```text
"${state.topic}"          -> evaluated, returns state.topic
"topic is ${state.topic}" -> literal string
"${state.topic} suffix"   -> literal string
"${'a'}${'b'}"              -> invalid CEL
```

* Honor explicit empty-context overrides in evaluate() / render_template()
2026-06-18 12:18:22 -07:00
38 changed files with 2797 additions and 255 deletions

View File

@@ -515,6 +515,7 @@
"edge/en/enterprise/guides/update-crew",
"edge/en/enterprise/guides/enable-crew-studio",
"edge/en/enterprise/guides/capture_telemetry_logs",
"edge/en/enterprise/guides/datadog",
"edge/en/enterprise/guides/azure-openai-setup",
"edge/en/enterprise/guides/vertex-ai-workload-identity-setup",
"edge/en/enterprise/guides/tool-repository",
@@ -8647,6 +8648,7 @@
"edge/pt-BR/enterprise/guides/update-crew",
"edge/pt-BR/enterprise/guides/enable-crew-studio",
"edge/pt-BR/enterprise/guides/capture_telemetry_logs",
"edge/pt-BR/enterprise/guides/datadog",
"edge/pt-BR/enterprise/guides/azure-openai-setup",
"edge/pt-BR/enterprise/guides/tool-repository",
"edge/pt-BR/enterprise/guides/custom-mcp-server",
@@ -16510,6 +16512,7 @@
"edge/ko/enterprise/guides/update-crew",
"edge/ko/enterprise/guides/enable-crew-studio",
"edge/ko/enterprise/guides/capture_telemetry_logs",
"edge/ko/enterprise/guides/datadog",
"edge/ko/enterprise/guides/azure-openai-setup",
"edge/ko/enterprise/guides/tool-repository",
"edge/ko/enterprise/guides/custom-mcp-server",
@@ -24565,6 +24568,7 @@
"edge/ar/enterprise/guides/update-crew",
"edge/ar/enterprise/guides/enable-crew-studio",
"edge/ar/enterprise/guides/capture_telemetry_logs",
"edge/ar/enterprise/guides/datadog",
"edge/ar/enterprise/guides/azure-openai-setup",
"edge/ar/enterprise/guides/tool-repository",
"edge/ar/enterprise/guides/custom-mcp-server",

View File

@@ -4,6 +4,27 @@ description: "تحديثات المنتج والتحسينات وإصلاحات
icon: "clock"
mode: "wide"
---
<Update label="18 يونيو 2026">
## v1.14.8a2
[عرض الإصدار على GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.14.8a2)
## ما الذي تغير
### الميزات
- إضافة إجراء عميل واحد إلى تعريفات التدفق
- التحقق من تعبيرات CEL للتدفق عند تحميل التعريف
### الوثائق
- إضافة دليل تكامل Datadog مع لوحة عمليات قابلة للاستيراد
- تحديث اللقطة وسجل التغييرات للإصدار v1.14.8a1
## المساهمون
@joaomdmoura, @lucasgomide, @vinibrsl
</Update>
<Update label="18 يونيو 2026">
## v1.14.8a1

View File

@@ -9,6 +9,10 @@ mode: "wide"
تتبع بيانات القياس [اتفاقيات OpenTelemetry GenAI الدلالية](https://opentelemetry.io/docs/specs/semconv/gen-ai/) بالإضافة إلى سمات خاصة بـ CrewAI.
<Tip>
تُعدّ OpenTelemetry **مسار المراقبة الموصى به** — محايدة تجاه الموردين، وتعمل مع أي خلفية متوافقة مع OTLP (Grafana, Honeycomb, NewRelic، أو مجمّعك الخاص). إذا كنت تستخدم Datadog تحديدًا، فراجع دليل [تكامل Datadog](./datadog) المخصص، الذي يغطي كلًا من مسار وكيل Datadog واستيعاب OTLP من Datadog.
</Tip>
## المتطلبات المسبقة
<CardGroup cols={2}>
@@ -41,17 +45,7 @@ mode: "wide"
<Frame>![تهيئة مجمّع OpenTelemetry](/images/crewai-otel-collector-opentelemetry.png)</Frame>
</Tab>
<Tab title="Datadog">
- **Datadog Site Domain** — مضيف OTLP لموقع Datadog الخاص بك فقط، دون بروتوكول أو مسار. يقوم CrewAI ببناء نقطة نهاية HTTPS OTLP الكاملة نيابةً عنك. استخدم المضيف المطابق لـ [موقع Datadog](https://docs.datadoghq.com/getting_started/site/) الخاص بك:
- `otlp.datadoghq.com` (US1)
- `otlp.us3.datadoghq.com` (US3)
- `otlp.us5.datadoghq.com` (US5)
- `otlp.datadoghq.eu` (EU1)
- `otlp.ap1.datadoghq.com` (AP1)
- **API Key** — مفتاح واجهة برمجة تطبيقات Datadog الخاص بك. راجع [كيفية إنشاء واحد](https://docs.datadoghq.com/account_management/api-app-keys/#api-keys).
يصدّر تكامل Datadog **التتبعات**.
<Frame>![تهيئة مجمّع Datadog](/images/crewai-otel-collector-datadog.png)</Frame>
لإعداد Datadog، راجع دليل [تكامل Datadog](./datadog) المخصص — فهو يغطي كلًا من مسار وكيل Datadog (الموصى به، أرخص لحجم السجلات الكبير) واستيعاب OTLP من Datadog، مع خطوات تهيئة كاملة للمجمّع.
</Tab>
</Tabs>

View File

@@ -0,0 +1,295 @@
---
title: "تكامل Datadog"
description: "راقب عمليات نشر CrewAI AMP المُستضافة ذاتيًا في Datadog عبر وكيل Datadog أو استيعاب OTLP من Datadog — يوفر كلا المسارين نفس الواجهات المهيكلة لاستيراد لوحة معلومات العمليات الجاهزة."
icon: "dog"
mode: "wide"
---
<Note>
**الترجمة قيد التقدم** — يتم عرض المحتوى باللغة الإنجليزية.
</Note>
CrewAI ships first-class support for Datadog: two log-ingestion paths, a JSON log schema designed for cheap indexing, and a ready-made operations dashboard you can import in under five minutes.
<Note>
For vendor-neutral observability via any OTLP backend (Grafana, Honeycomb, your own collector), see [OpenTelemetry Export](./capture_telemetry_logs).
</Note>
## Choose a path
CrewAI supports two log-ingestion paths to Datadog — both are first-class and produce the same structured facets that power the dashboard. Pick the one that fits your infrastructure.
<Tabs>
<Tab title="Datadog Agent">
The Datadog Agent runs alongside your CrewAI containers (typically as a DaemonSet on Kubernetes) and tails their stdout. With `CREWAI_LOG_FORMAT=json` set, each log event ships as a single billable line with structured attributes.
**Setup:**
1. Run the Datadog Agent next to your CrewAI containers — see [Datadog's deployment docs](https://docs.datadoghq.com/agent/) for Kubernetes, ECS, or VM setup. Enable log collection (`logs_enabled: true`) and container log collection (`logs_config.container_collect_all: true`).
2. Set `CREWAI_LOG_FORMAT=json` as an **automation environment variable** in CrewAI AMP (open your automation → **Settings → Environment Variables**) so each log event is a single line instead of a multi-line traceback. AMP propagates the value to every container in the deployment (API + workers) — don't set it on the container or host directly. See [Enabling JSON output](#enabling-json-output) below for the AMP UI walkthrough and the [log schema reference](#log-schema-reference) for the full field contract.
3. Confirm logs arrive in Datadog Logs with the JSON fields parsed — see [Verify ingestion](#verify-ingestion).
**Pick this path if** you already operate Datadog Agents (e.g. for infrastructure metrics), or your log volume makes per-event ingestion cost a real concern — collapsing tracebacks into single events keeps Agent ingestion cheap at scale.
</Tab>
<Tab title="Datadog OTLP intake">
CrewAI AMP exports OpenTelemetry traffic directly to Datadog's OTLP endpoint with no Agent required. Logs and traces ride a single export pipeline configured in AMP's UI, using the same protocol you'd use for any other OTLP backend.
**Setup:**
1. In CrewAI AMP, go to **Settings → OpenTelemetry Collectors → Add Collector** and pick **Datadog**.
2. Configure the connection:
- **Datadog Site Domain** — your Datadog site's OTLP host only, no protocol or path. CrewAI builds the full HTTPS OTLP endpoint for you. Use the host that matches your [Datadog site](https://docs.datadoghq.com/getting_started/site/):
- `otlp.datadoghq.com` (US1)
- `otlp.us3.datadoghq.com` (US3)
- `otlp.us5.datadoghq.com` (US5)
- `otlp.datadoghq.eu` (EU1)
- `otlp.ap1.datadoghq.com` (AP1)
- **API Key** — your Datadog API key. See [how to create one](https://docs.datadoghq.com/account_management/api-app-keys/#api-keys).
3. The Datadog template provisions **both signals at once** — when you save, AMP creates a traces collector at `/v1/traces` and a logs collector at `/v1/logs`, both sharing the same Datadog OTLP host and API key. You'll see them as two separate rows in your OTel collectors list.
4. *(optional)* Click **Test Connection** to verify CrewAI can reach the endpoint with the credentials you provided. Then click **Save** — both collectors are created in one step.
<Frame>![Datadog collector configuration](/images/crewai-otel-collector-datadog.png)</Frame>
**Pick this path if** you'd rather not operate a Datadog Agent, you already use OTLP for traces and want one export pipeline, or you may later want to fan out the same telemetry to other backends (Grafana, Honeycomb, etc.) without changing your application setup.
</Tab>
</Tabs>
Either path lands the same structured facets in Datadog (`@automation_id`, `@kickoff_id`, `@execution_id`, `@automation_name`, `@crewai_version`, `@exception.type`, `@gen_ai.*`), so the dashboard works identically with either choice.
## Log schema reference
<Info>
This schema applies to the **Datadog Agent path** — stdout JSON logs produced when `CREWAI_LOG_FORMAT=json` is set. Logs delivered via the **Datadog OTLP intake** use OpenTelemetry attribute names and may differ; see [OpenTelemetry Export](./capture_telemetry_logs).
</Info>
When `CREWAI_LOG_FORMAT=json` is set, every log event is emitted as a **single JSON object per line** to stdout, with internal newlines escaped. The format is plain JSON — Datadog parses it natively, and the same payload is also consumable by Splunk, Loki, Elasticsearch, and CloudWatch without custom log pipelines.
### Why JSON output
<CardGroup cols={2}>
<Card title="Lower ingestion cost" icon="dollar-sign">
Most managed log backends bill per event. A Python traceback in text format is counted as one event per line — 30+ events for a single error. JSON output collapses each traceback into a single event with the stack trace as an escaped string field.
</Card>
<Card title="Structured search" icon="magnifying-glass">
Search by `@automation_id`, `@exception.type`, `@kickoff_id` instead of grepping free-text. Build dashboards on typed facets without parser configuration.
</Card>
<Card title="APM ↔ logs correlation" icon="link">
Every event carries `trace_id` and `span_id` when fired inside a recording span, so backends auto-link logs to traces.
</Card>
<Card title="Stable contract" icon="file-shield">
The `schema` field gates compatibility — within `v1`, fields are added but never renamed or removed.
</Card>
</CardGroup>
### Enabling JSON output
`CREWAI_LOG_FORMAT=json` must be set as an **automation environment variable** in CrewAI AMP — it is **not** a container, host, or Docker setting. Open your automation in AMP, click the **Settings** icon, and add the variable under the **Environment Variables** section. AMP applies the value to every container in the deployment (API + workers) on the next restart. See [Update Your Crew](./update-crew) for the full UI walkthrough with screenshots.
```shell
CREWAI_LOG_FORMAT=json
```
Restart the deployment to pick up the change. Every log line on stdout from that point on is a single JSON object.
<Note>
The default value is `text`, which preserves the legacy human-readable line format byte-for-byte. Setting any value other than `json` falls back to text mode. There is no migration step — the variable is read at process start and the format switches immediately.
</Note>
### Example events
A single info-level log inside an active automation kickoff:
```json
{
"schema": "v1",
"ts": "2026-06-17T16:14:23.482914Z",
"level": "INFO",
"logger": "crewai_enterprise.utilities.pii_redaction",
"crewai_version": "1.14.7",
"msg": "PII tracking state reset (engines preserved)",
"automation_id": "12",
"task_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"kickoff_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"execution_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"automation_name": "research_flow"
}
```
An error with a Python exception is collapsed into a single event with the traceback as a string:
```json
{
"schema": "v1",
"ts": "2026-06-17T16:14:31.218450Z",
"level": "ERROR",
"logger": "api.tasks.flow_run_task",
"crewai_version": "1.14.7",
"msg": "Flow execution failed",
"automation_id": "12",
"kickoff_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"execution_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"automation_name": "research_flow",
"exception": {
"type": "ValueError",
"message": "Topic cannot be empty",
"stacktrace": "Traceback (most recent call last):\n File \"/app/flow.py\", line 42, in summarize\n ...\nValueError: Topic cannot be empty\n"
}
}
```
The same error in legacy text mode would have produced ~25 separate log events (one per traceback line) — all of which the backend would bill and index individually.
### Schema v1 fields
Within the `v1` schema, fields are only added, never renamed or removed. New fields will appear as soon as a deployment is upgraded.
| Field | Type | Always present | Source |
|-------|------|----------------|--------|
| `schema` | string | Yes | Constant `"v1"`. Increment indicates a breaking schema change. |
| `ts` | string (ISO-8601 UTC, microseconds) | Yes | Record creation time, e.g. `2026-06-17T16:14:23.482914Z`. |
| `level` | string | Yes | Python log level name: `DEBUG` / `INFO` / `WARNING` / `ERROR` / `CRITICAL`. |
| `logger` | string | Yes | Dotted logger name, e.g. `api.tasks.flow_run_task`. |
| `crewai_version` | string | Yes (when `crewai` package metadata is resolvable) | Installed `crewai` package version, e.g. `"1.14.7"`. |
| `msg` | string | Yes | Rendered log message (after `%`-formatting / `{}`-formatting). |
| `automation_id` | string | When `CREWAI_PLUS_ID` env var is set | Numeric deployment ID (AMP provisions this on every container). |
| `task_id` | string | On Celery worker logs | Celery task UUID, or `"no-task"` for non-task contexts. |
| `kickoff_id` | string | Inside an automation kickoff | UUID of the current kickoff. |
| `execution_id` | string | Inside an automation kickoff | UUID of the current sub-execution. Equal to `kickoff_id` at the top level; differs for nested flow methods that spawn sub-executions. |
| `automation_name` | string | Inside an automation kickoff | Human-readable automation/flow name, e.g. `"research_flow"`. |
| `trace_id` | string (32-hex) | Inside a recording OpenTelemetry span | Hex trace ID. Omitted when no span is active. |
| `span_id` | string (16-hex) | Inside a recording OpenTelemetry span | Hex span ID. Omitted when no span is active. |
| `exception` | object | When the log record has `exc_info` | `{type, message, stacktrace}` — full traceback as a single escaped string. |
<Tip>
Any additional `extra={...}` kwargs passed to a logger call appear as top-level JSON fields verbatim. Reserved field names above always win to keep the schema stable.
</Tip>
### Stability promise
The `schema` field declares the contract. Within `v1`, CrewAI commits to:
- **Never removing a field** that customers may have built queries or dashboards against.
- **Never renaming a field** in place — renames happen via a schema bump (e.g. `v2`), with the old name kept as a deprecated alias for at least one release cycle.
- **Adding new fields** at any time. Consumers should ignore unknown top-level keys.
When a `v2` is introduced, both the `schema` field and the migration guide will be published in advance, and `v1` will continue to be emitted for one release cycle so dashboards and queries have time to migrate.
## Prerequisite: promote facets
Datadog auto-discovers fields the first time it sees them but doesn't make them queryable in widgets until they're promoted to **facets**. This is a one-time setup in your Datadog account.
<Steps>
<Step title="Search for a CrewAI log">
Open [Logs Explorer](https://app.datadoghq.com/logs) and search `service:crewai*`. You should see at least one log event.
</Step>
<Step title="Promote each field">
Click any log entry to open the right-hand details panel. For each field below, hover the field name → click the gear icon → **Create facet**.
- `automation_id`, `automation_name`, `execution_id`, `kickoff_id`, `task_id`
- `crewai_version`, `model_id`
- `exception.type`, `exception.message`
Skip any field that already shows a star icon next to its name — that means it's already a facet. The `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, and `gen_ai.request.model` facets are typically promoted automatically by Datadog's LLM Observability auto-discovery, but verify they exist before importing the dashboard.
</Step>
</Steps>
## Import the dashboard
<Steps>
<Step title="Download the dashboard JSON">
Save [`datadog_dashboard.json`](https://raw.githubusercontent.com/crewAIInc/crewAI/main/docs/edge/en/enterprise/guides/datadog_dashboard.json) to your machine.
</Step>
<Step title="Open the import dialog in Datadog">
Navigate to **Dashboards → New Dashboard**. Click the **gear icon** in the top right of the empty dashboard and select **Import Dashboard JSON**.
</Step>
<Step title="Paste or upload the JSON">
Paste the contents of `datadog_dashboard.json` into the import dialog (or drag the file in). Click **Import**.
Datadog creates the dashboard immediately and lands you on it. The first load may show empty widgets for a few seconds while queries execute against the time range.
</Step>
</Steps>
<Tip>
Datadog's [Dashboard API](https://docs.datadoghq.com/api/latest/dashboards/#create-a-new-dashboard) accepts the same JSON via `POST /api/v1/dashboard`. Use it if you manage dashboards through Terraform, Pulumi, or CI.
</Tip>
## What you get
The dashboard is organized into four sections plus a placeholder for a custom drill-down widget:
| Section | Widgets | Useful for |
|---------|---------|------------|
| **Header** | Total Executions · Error Rate (%) · Active Automations · CrewAI Versions in Use | At-a-glance health for the last hour. Error Rate is conditionally formatted (green ≤ 5%, yellow ≤ 10%, red > 10%). |
| **Throughput** | Executions per Hour by Automation (top 10, stacked bars) | Spotting traffic shifts, surfacing busy automations, validating that a rollout didn't change baseline volume. |
| **Errors** | Errors by Exception Type (top 5, stacked bars) · Top Exception Types by Count (toplist) | Triaging failures — which exception types are spiking, which automations they're hitting. |
| **Cost** | Total Tokens per Hour by Model (input + output, stacked area) | Tracking LLM token spend by model. Useful for catching cost regressions when an automation switches model or starts looping. |
| **Drill-Down** | _(empty placeholder)_ | See [Customization](#customize) for adding a recent-errors log stream here. |
Three template variables at the top of the dashboard re-scope every widget at once:
- **`$automation`** — filter to a single automation by name.
- **`$version`** — filter to a single `crewai` SDK version (useful for comparing pre- and post-upgrade behavior).
- **`$service`** — filter to a specific Datadog `service` tag (useful when multiple CrewAI deployments share one Datadog account).
## Verify ingestion
Open [Logs Explorer](https://app.datadoghq.com/logs) and run a query that matches your ingestion path:
<Tabs>
<Tab title="Datadog Agent">
Search `service:crewai* @schema:v1`. You should see structured logs with the JSON fields parsed into Datadog facets. Pick a recent event and verify it has `@automation_id`, `@kickoff_id`, `@execution_id`, `@crewai_version`, and (when running inside a span) `@trace_id` / `@span_id` populated.
If nothing appears, confirm `CREWAI_LOG_FORMAT=json` is set under your automation's **Environment Variables** in AMP, the deployment was restarted after the change, and the Datadog Agent is tailing container stdout.
</Tab>
<Tab title="Datadog OTLP intake">
Search `source:otlp service:crewai*`. OTLP attributes land with their OpenTelemetry names (`automation_id`, `crewai.kickoff.id`, etc.) rather than the stdout JSON keys, but they map to the same dashboard facets after [facet promotion](#prerequisite-promote-facets).
If nothing appears, verify the collector endpoint is correct (`/v1/logs` for logs, `/v1/traces` for traces) and **Test Connection** succeeded when the collector was saved.
</Tab>
</Tabs>
## Customize
The dashboard ships with deliberate gaps so you can extend it without uninstalling and re-importing.
### Add a Recent Errors log stream
The **Drill-Down** section is intentionally empty. Add a Log Stream widget to it for an inline view of recent failures:
1. Edit the dashboard and click **+ Add Widgets** inside the Drill-Down group.
2. Drag in a **Log Stream** widget.
3. Set the filter query to `status:error $automation $version $service`.
4. Choose columns: `@timestamp`, `@automation_name`, `@exception.type`, `@exception.message`, `@execution_id`.
5. Sort by most recent, limit to 25 entries.
Clicking any row jumps to Logs Explorer with the same filter pre-applied.
### Add p95 latency
Logs don't include execution duration by default. Two ways to add a latency widget:
- **From APM traces** — if you also export OTLP traces to Datadog, add a Timeseries widget with data source **Traces**, query `service:crewai*`, aggregation `p95 of @duration`. Datadog APM auto-tracks span duration.
- **From metric extraction** — extract a `flow.duration_ms` metric from logs via [Datadog's log-to-metric pipeline](https://docs.datadoghq.com/logs/log_configuration/logs_to_metrics/), then chart it like any other metric. Useful if you don't run APM.
### Re-scope to multiple deployments
The `$service` template variable defaults to `*` and will catch every CrewAI deployment in your Datadog account. Change the default to a specific service name in **Configure → Template Variables** if you want the dashboard to focus on one deployment by default.
## Troubleshooting
| Symptom | Likely cause | Fix |
|---------|--------------|-----|
| All widgets show "No data" | Facets aren't promoted | Re-do the [Promote facets](#prerequisite-promote-facets) step. Datadog won't query against an un-promoted field. |
| Error Rate widget shows `NaN` | No executions in the time window | Either no traffic, or `@execution_id` isn't faceted. Expand the time range and re-check facets. |
| Throughput chart is flat at the same value | Logs aren't reaching Datadog | Search `service:crewai*` in Logs Explorer. If nothing shows, verify the Datadog Agent is running (Agent path) or the OTel collector endpoint is correct (OTLP path). |
| `crewai_version` shows fewer values than expected | Some containers predate the structured-logs work | The `crewai_version` field was added alongside JSON output. Older deployments running text mode (or older AMP builds) won't emit it. Upgrade those deployments to pick up the field. See the [log schema reference](#log-schema-reference) for the full field contract. |
| Template variables don't filter widgets | The widget's filter line doesn't reference the template variable | Edit the widget and confirm the search includes `$automation $version $service`. |
## Next steps
<CardGroup cols={2}>
<Card title="OpenTelemetry Export" icon="magnifying-glass-chart" href="./capture_telemetry_logs">
Vendor-neutral observability for non-Datadog stacks (Grafana, Honeycomb, your own collector) — or as a Datadog complement when you want to fan out telemetry to multiple backends.
</Card>
<Card title="Datadog Log Search Syntax" icon="magnifying-glass" href="https://docs.datadoghq.com/logs/explorer/search_syntax/">
Reference for customizing widget queries against the structured facets above.
</Card>
</CardGroup>

View File

@@ -4,6 +4,27 @@ description: "Product updates, improvements, and bug fixes for CrewAI"
icon: "clock"
mode: "wide"
---
<Update label="Jun 18, 2026">
## v1.14.8a2
[View release on GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.14.8a2)
## What's Changed
### Features
- Add single agent action to Flow definitions
- Validate flow CEL expressions at definition load time
### Documentation
- Add Datadog integration guide with importable operations dashboard
- Update snapshot and changelog for v1.14.8a1
## Contributors
@joaomdmoura, @lucasgomide, @vinibrsl
</Update>
<Update label="Jun 18, 2026">
## v1.14.8a1

View File

@@ -9,6 +9,10 @@ CrewAI AMP can export OpenTelemetry **traces** and **logs** from your deployment
Telemetry data follows the [OpenTelemetry GenAI semantic conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/) plus additional CrewAI-specific attributes.
<Tip>
OpenTelemetry is the **recommended observability path** — vendor-neutral, works with any OTLP-compatible backend (Grafana, Honeycomb, NewRelic, your own collector). If you specifically use Datadog, see the dedicated [Datadog Integration](./datadog) guide which covers both the Datadog Agent path and Datadog's OTLP intake.
</Tip>
## Prerequisites
<CardGroup cols={2}>
@@ -41,17 +45,7 @@ Telemetry data follows the [OpenTelemetry GenAI semantic conventions](https://op
<Frame>![OpenTelemetry collector configuration](/images/crewai-otel-collector-opentelemetry.png)</Frame>
</Tab>
<Tab title="Datadog">
- **Datadog Site Domain** — Your Datadog site's OTLP host only, with no protocol or path. CrewAI builds the full HTTPS OTLP endpoint for you. Use the host that matches your [Datadog site](https://docs.datadoghq.com/getting_started/site/):
- `otlp.datadoghq.com` (US1)
- `otlp.us3.datadoghq.com` (US3)
- `otlp.us5.datadoghq.com` (US5)
- `otlp.datadoghq.eu` (EU1)
- `otlp.ap1.datadoghq.com` (AP1)
- **API Key** — Your Datadog API key. See [how to create one](https://docs.datadoghq.com/account_management/api-app-keys/#api-keys).
The Datadog integration exports **traces**.
<Frame>![Datadog collector configuration](/images/crewai-otel-collector-datadog.png)</Frame>
For Datadog setup, see the dedicated [Datadog Integration](./datadog) guide — it covers both the Datadog Agent path (recommended, cheaper for log volume) and Datadog's OTLP intake with full collector configuration steps.
</Tab>
</Tabs>

View File

@@ -0,0 +1,291 @@
---
title: "Datadog Integration"
description: "Monitor self-hosted CrewAI AMP deployments in Datadog via the Datadog Agent or Datadog's OTLP intake — either path lands the same structured facets so you can import the ready-made operations dashboard."
icon: "dog"
mode: "wide"
---
CrewAI ships first-class support for Datadog: two log-ingestion paths, a JSON log schema designed for cheap indexing, and a ready-made operations dashboard you can import in under five minutes.
<Note>
For vendor-neutral observability via any OTLP backend (Grafana, Honeycomb, your own collector), see [OpenTelemetry Export](./capture_telemetry_logs).
</Note>
## Choose a path
CrewAI supports two log-ingestion paths to Datadog — both are first-class and produce the same structured facets that power the dashboard. Pick the one that fits your infrastructure.
<Tabs>
<Tab title="Datadog Agent">
The Datadog Agent runs alongside your CrewAI containers (typically as a DaemonSet on Kubernetes) and tails their stdout. With `CREWAI_LOG_FORMAT=json` set, each log event ships as a single billable line with structured attributes.
**Setup:**
1. Run the Datadog Agent next to your CrewAI containers — see [Datadog's deployment docs](https://docs.datadoghq.com/agent/) for Kubernetes, ECS, or VM setup. Enable log collection (`logs_enabled: true`) and container log collection (`logs_config.container_collect_all: true`).
2. Set `CREWAI_LOG_FORMAT=json` as an **automation environment variable** in CrewAI AMP (open your automation → **Settings → Environment Variables**) so each log event is a single line instead of a multi-line traceback. AMP propagates the value to every container in the deployment (API + workers) — don't set it on the container or host directly. See [Enabling JSON output](#enabling-json-output) below for the AMP UI walkthrough and the [log schema reference](#log-schema-reference) for the full field contract.
3. Confirm logs arrive in Datadog Logs with the JSON fields parsed — see [Verify ingestion](#verify-ingestion).
**Pick this path if** you already operate Datadog Agents (e.g. for infrastructure metrics), or your log volume makes per-event ingestion cost a real concern — collapsing tracebacks into single events keeps Agent ingestion cheap at scale.
</Tab>
<Tab title="Datadog OTLP intake">
CrewAI AMP exports OpenTelemetry traffic directly to Datadog's OTLP endpoint with no Agent required. Logs and traces ride a single export pipeline configured in AMP's UI, using the same protocol you'd use for any other OTLP backend.
**Setup:**
1. In CrewAI AMP, go to **Settings → OpenTelemetry Collectors → Add Collector** and pick **Datadog**.
2. Configure the connection:
- **Datadog Site Domain** — your Datadog site's OTLP host only, no protocol or path. CrewAI builds the full HTTPS OTLP endpoint for you. Use the host that matches your [Datadog site](https://docs.datadoghq.com/getting_started/site/):
- `otlp.datadoghq.com` (US1)
- `otlp.us3.datadoghq.com` (US3)
- `otlp.us5.datadoghq.com` (US5)
- `otlp.datadoghq.eu` (EU1)
- `otlp.ap1.datadoghq.com` (AP1)
- **API Key** — your Datadog API key. See [how to create one](https://docs.datadoghq.com/account_management/api-app-keys/#api-keys).
3. The Datadog template provisions **both signals at once** — when you save, AMP creates a traces collector at `/v1/traces` and a logs collector at `/v1/logs`, both sharing the same Datadog OTLP host and API key. You'll see them as two separate rows in your OTel collectors list.
4. *(optional)* Click **Test Connection** to verify CrewAI can reach the endpoint with the credentials you provided. Then click **Save** — both collectors are created in one step.
<Frame>![Datadog collector configuration](/images/crewai-otel-collector-datadog.png)</Frame>
**Pick this path if** you'd rather not operate a Datadog Agent, you already use OTLP for traces and want one export pipeline, or you may later want to fan out the same telemetry to other backends (Grafana, Honeycomb, etc.) without changing your application setup.
</Tab>
</Tabs>
Either path lands the same structured facets in Datadog (`@automation_id`, `@kickoff_id`, `@execution_id`, `@automation_name`, `@crewai_version`, `@exception.type`, `@gen_ai.*`), so the dashboard works identically with either choice.
## Log schema reference
<Info>
This schema applies to the **Datadog Agent path** — stdout JSON logs produced when `CREWAI_LOG_FORMAT=json` is set. Logs delivered via the **Datadog OTLP intake** use OpenTelemetry attribute names and may differ; see [OpenTelemetry Export](./capture_telemetry_logs).
</Info>
When `CREWAI_LOG_FORMAT=json` is set, every log event is emitted as a **single JSON object per line** to stdout, with internal newlines escaped. The format is plain JSON — Datadog parses it natively, and the same payload is also consumable by Splunk, Loki, Elasticsearch, and CloudWatch without custom log pipelines.
### Why JSON output
<CardGroup cols={2}>
<Card title="Lower ingestion cost" icon="dollar-sign">
Most managed log backends bill per event. A Python traceback in text format is counted as one event per line — 30+ events for a single error. JSON output collapses each traceback into a single event with the stack trace as an escaped string field.
</Card>
<Card title="Structured search" icon="magnifying-glass">
Search by `@automation_id`, `@exception.type`, `@kickoff_id` instead of grepping free-text. Build dashboards on typed facets without parser configuration.
</Card>
<Card title="APM ↔ logs correlation" icon="link">
Every event carries `trace_id` and `span_id` when fired inside a recording span, so backends auto-link logs to traces.
</Card>
<Card title="Stable contract" icon="file-shield">
The `schema` field gates compatibility — within `v1`, fields are added but never renamed or removed.
</Card>
</CardGroup>
### Enabling JSON output
`CREWAI_LOG_FORMAT=json` must be set as an **automation environment variable** in CrewAI AMP — it is **not** a container, host, or Docker setting. Open your automation in AMP, click the **Settings** icon, and add the variable under the **Environment Variables** section. AMP applies the value to every container in the deployment (API + workers) on the next restart. See [Update Your Crew](./update-crew) for the full UI walkthrough with screenshots.
```shell
CREWAI_LOG_FORMAT=json
```
Restart the deployment to pick up the change. Every log line on stdout from that point on is a single JSON object.
<Note>
The default value is `text`, which preserves the legacy human-readable line format byte-for-byte. Setting any value other than `json` falls back to text mode. There is no migration step — the variable is read at process start and the format switches immediately.
</Note>
### Example events
A single info-level log inside an active automation kickoff:
```json
{
"schema": "v1",
"ts": "2026-06-17T16:14:23.482914Z",
"level": "INFO",
"logger": "crewai_enterprise.utilities.pii_redaction",
"crewai_version": "1.14.7",
"msg": "PII tracking state reset (engines preserved)",
"automation_id": "12",
"task_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"kickoff_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"execution_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"automation_name": "research_flow"
}
```
An error with a Python exception is collapsed into a single event with the traceback as a string:
```json
{
"schema": "v1",
"ts": "2026-06-17T16:14:31.218450Z",
"level": "ERROR",
"logger": "api.tasks.flow_run_task",
"crewai_version": "1.14.7",
"msg": "Flow execution failed",
"automation_id": "12",
"kickoff_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"execution_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"automation_name": "research_flow",
"exception": {
"type": "ValueError",
"message": "Topic cannot be empty",
"stacktrace": "Traceback (most recent call last):\n File \"/app/flow.py\", line 42, in summarize\n ...\nValueError: Topic cannot be empty\n"
}
}
```
The same error in legacy text mode would have produced ~25 separate log events (one per traceback line) — all of which the backend would bill and index individually.
### Schema v1 fields
Within the `v1` schema, fields are only added, never renamed or removed. New fields will appear as soon as a deployment is upgraded.
| Field | Type | Always present | Source |
|-------|------|----------------|--------|
| `schema` | string | Yes | Constant `"v1"`. Increment indicates a breaking schema change. |
| `ts` | string (ISO-8601 UTC, microseconds) | Yes | Record creation time, e.g. `2026-06-17T16:14:23.482914Z`. |
| `level` | string | Yes | Python log level name: `DEBUG` / `INFO` / `WARNING` / `ERROR` / `CRITICAL`. |
| `logger` | string | Yes | Dotted logger name, e.g. `api.tasks.flow_run_task`. |
| `crewai_version` | string | Yes (when `crewai` package metadata is resolvable) | Installed `crewai` package version, e.g. `"1.14.7"`. |
| `msg` | string | Yes | Rendered log message (after `%`-formatting / `{}`-formatting). |
| `automation_id` | string | When `CREWAI_PLUS_ID` env var is set | Numeric deployment ID (AMP provisions this on every container). |
| `task_id` | string | On Celery worker logs | Celery task UUID, or `"no-task"` for non-task contexts. |
| `kickoff_id` | string | Inside an automation kickoff | UUID of the current kickoff. |
| `execution_id` | string | Inside an automation kickoff | UUID of the current sub-execution. Equal to `kickoff_id` at the top level; differs for nested flow methods that spawn sub-executions. |
| `automation_name` | string | Inside an automation kickoff | Human-readable automation/flow name, e.g. `"research_flow"`. |
| `trace_id` | string (32-hex) | Inside a recording OpenTelemetry span | Hex trace ID. Omitted when no span is active. |
| `span_id` | string (16-hex) | Inside a recording OpenTelemetry span | Hex span ID. Omitted when no span is active. |
| `exception` | object | When the log record has `exc_info` | `{type, message, stacktrace}` — full traceback as a single escaped string. |
<Tip>
Any additional `extra={...}` kwargs passed to a logger call appear as top-level JSON fields verbatim. Reserved field names above always win to keep the schema stable.
</Tip>
### Stability promise
The `schema` field declares the contract. Within `v1`, CrewAI commits to:
- **Never removing a field** that customers may have built queries or dashboards against.
- **Never renaming a field** in place — renames happen via a schema bump (e.g. `v2`), with the old name kept as a deprecated alias for at least one release cycle.
- **Adding new fields** at any time. Consumers should ignore unknown top-level keys.
When a `v2` is introduced, both the `schema` field and the migration guide will be published in advance, and `v1` will continue to be emitted for one release cycle so dashboards and queries have time to migrate.
## Prerequisite: promote facets
Datadog auto-discovers fields the first time it sees them but doesn't make them queryable in widgets until they're promoted to **facets**. This is a one-time setup in your Datadog account.
<Steps>
<Step title="Search for a CrewAI log">
Open [Logs Explorer](https://app.datadoghq.com/logs) and search `service:crewai*`. You should see at least one log event.
</Step>
<Step title="Promote each field">
Click any log entry to open the right-hand details panel. For each field below, hover the field name → click the gear icon → **Create facet**.
- `automation_id`, `automation_name`, `execution_id`, `kickoff_id`, `task_id`
- `crewai_version`, `model_id`
- `exception.type`, `exception.message`
Skip any field that already shows a star icon next to its name — that means it's already a facet. The `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, and `gen_ai.request.model` facets are typically promoted automatically by Datadog's LLM Observability auto-discovery, but verify they exist before importing the dashboard.
</Step>
</Steps>
## Import the dashboard
<Steps>
<Step title="Download the dashboard JSON">
Save [`datadog_dashboard.json`](https://raw.githubusercontent.com/crewAIInc/crewAI/main/docs/edge/en/enterprise/guides/datadog_dashboard.json) to your machine.
</Step>
<Step title="Open the import dialog in Datadog">
Navigate to **Dashboards → New Dashboard**. Click the **gear icon** in the top right of the empty dashboard and select **Import Dashboard JSON**.
</Step>
<Step title="Paste or upload the JSON">
Paste the contents of `datadog_dashboard.json` into the import dialog (or drag the file in). Click **Import**.
Datadog creates the dashboard immediately and lands you on it. The first load may show empty widgets for a few seconds while queries execute against the time range.
</Step>
</Steps>
<Tip>
Datadog's [Dashboard API](https://docs.datadoghq.com/api/latest/dashboards/#create-a-new-dashboard) accepts the same JSON via `POST /api/v1/dashboard`. Use it if you manage dashboards through Terraform, Pulumi, or CI.
</Tip>
## What you get
The dashboard is organized into four sections plus a placeholder for a custom drill-down widget:
| Section | Widgets | Useful for |
|---------|---------|------------|
| **Header** | Total Executions · Error Rate (%) · Active Automations · CrewAI Versions in Use | At-a-glance health for the last hour. Error Rate is conditionally formatted (green ≤ 5%, yellow ≤ 10%, red > 10%). |
| **Throughput** | Executions per Hour by Automation (top 10, stacked bars) | Spotting traffic shifts, surfacing busy automations, validating that a rollout didn't change baseline volume. |
| **Errors** | Errors by Exception Type (top 5, stacked bars) · Top Exception Types by Count (toplist) | Triaging failures — which exception types are spiking, which automations they're hitting. |
| **Cost** | Total Tokens per Hour by Model (input + output, stacked area) | Tracking LLM token spend by model. Useful for catching cost regressions when an automation switches model or starts looping. |
| **Drill-Down** | _(empty placeholder)_ | See [Customization](#customize) for adding a recent-errors log stream here. |
Three template variables at the top of the dashboard re-scope every widget at once:
- **`$automation`** — filter to a single automation by name.
- **`$version`** — filter to a single `crewai` SDK version (useful for comparing pre- and post-upgrade behavior).
- **`$service`** — filter to a specific Datadog `service` tag (useful when multiple CrewAI deployments share one Datadog account).
## Verify ingestion
Open [Logs Explorer](https://app.datadoghq.com/logs) and run a query that matches your ingestion path:
<Tabs>
<Tab title="Datadog Agent">
Search `service:crewai* @schema:v1`. You should see structured logs with the JSON fields parsed into Datadog facets. Pick a recent event and verify it has `@automation_id`, `@kickoff_id`, `@execution_id`, `@crewai_version`, and (when running inside a span) `@trace_id` / `@span_id` populated.
If nothing appears, confirm `CREWAI_LOG_FORMAT=json` is set under your automation's **Environment Variables** in AMP, the deployment was restarted after the change, and the Datadog Agent is tailing container stdout.
</Tab>
<Tab title="Datadog OTLP intake">
Search `source:otlp service:crewai*`. OTLP attributes land with their OpenTelemetry names (`automation_id`, `crewai.kickoff.id`, etc.) rather than the stdout JSON keys, but they map to the same dashboard facets after [facet promotion](#prerequisite-promote-facets).
If nothing appears, verify the collector endpoint is correct (`/v1/logs` for logs, `/v1/traces` for traces) and **Test Connection** succeeded when the collector was saved.
</Tab>
</Tabs>
## Customize
The dashboard ships with deliberate gaps so you can extend it without uninstalling and re-importing.
### Add a Recent Errors log stream
The **Drill-Down** section is intentionally empty. Add a Log Stream widget to it for an inline view of recent failures:
1. Edit the dashboard and click **+ Add Widgets** inside the Drill-Down group.
2. Drag in a **Log Stream** widget.
3. Set the filter query to `status:error $automation $version $service`.
4. Choose columns: `@timestamp`, `@automation_name`, `@exception.type`, `@exception.message`, `@execution_id`.
5. Sort by most recent, limit to 25 entries.
Clicking any row jumps to Logs Explorer with the same filter pre-applied.
### Add p95 latency
Logs don't include execution duration by default. Two ways to add a latency widget:
- **From APM traces** — if you also export OTLP traces to Datadog, add a Timeseries widget with data source **Traces**, query `service:crewai*`, aggregation `p95 of @duration`. Datadog APM auto-tracks span duration.
- **From metric extraction** — extract a `flow.duration_ms` metric from logs via [Datadog's log-to-metric pipeline](https://docs.datadoghq.com/logs/log_configuration/logs_to_metrics/), then chart it like any other metric. Useful if you don't run APM.
### Re-scope to multiple deployments
The `$service` template variable defaults to `*` and will catch every CrewAI deployment in your Datadog account. Change the default to a specific service name in **Configure → Template Variables** if you want the dashboard to focus on one deployment by default.
## Troubleshooting
| Symptom | Likely cause | Fix |
|---------|--------------|-----|
| All widgets show "No data" | Facets aren't promoted | Re-do the [Promote facets](#prerequisite-promote-facets) step. Datadog won't query against an un-promoted field. |
| Error Rate widget shows `NaN` | No executions in the time window | Either no traffic, or `@execution_id` isn't faceted. Expand the time range and re-check facets. |
| Throughput chart is flat at the same value | Logs aren't reaching Datadog | Search `service:crewai*` in Logs Explorer. If nothing shows, verify the Datadog Agent is running (Agent path) or the OTel collector endpoint is correct (OTLP path). |
| `crewai_version` shows fewer values than expected | Some containers predate the structured-logs work | The `crewai_version` field was added alongside JSON output. Older deployments running text mode (or older AMP builds) won't emit it. Upgrade those deployments to pick up the field. See the [log schema reference](#log-schema-reference) for the full field contract. |
| Template variables don't filter widgets | The widget's filter line doesn't reference the template variable | Edit the widget and confirm the search includes `$automation $version $service`. |
## Next steps
<CardGroup cols={2}>
<Card title="OpenTelemetry Export" icon="magnifying-glass-chart" href="./capture_telemetry_logs">
Vendor-neutral observability for non-Datadog stacks (Grafana, Honeycomb, your own collector) — or as a Datadog complement when you want to fan out telemetry to multiple backends.
</Card>
<Card title="Datadog Log Search Syntax" icon="magnifying-glass" href="https://docs.datadoghq.com/logs/explorer/search_syntax/">
Reference for customizing widget queries against the structured facets above.
</Card>
</CardGroup>

View File

@@ -0,0 +1,582 @@
{
"title": "crewAI -- Operations",
"description": "Monitoring dashboard for self-hosted crewAI deployments running structured JSON logs. Tracks executions, errors, token usage, and automation health.",
"widgets": [
{
"id": 8810001,
"definition": {
"title": "Header",
"background_color": "vivid_blue",
"show_title": true,
"type": "group",
"layout_type": "ordered",
"widgets": [
{
"id": 9910001,
"definition": {
"title": "Total Executions",
"time": {
"live_span": "1h"
},
"type": "query_value",
"requests": [
{
"response_format": "scalar",
"queries": [
{
"data_source": "logs",
"name": "query1",
"search": {
"query": "$automation $version $service"
},
"compute": {
"aggregation": "cardinality",
"metric": "@execution_id"
},
"indexes": [
"*"
]
}
],
"formulas": [
{
"formula": "query1"
}
]
}
],
"autoscale": true,
"precision": 0
},
"layout": {
"x": 0,
"y": 0,
"width": 3,
"height": 2
}
},
{
"id": 9910002,
"definition": {
"title": "Error Rate (%)",
"time": {
"live_span": "1h"
},
"type": "query_value",
"requests": [
{
"response_format": "scalar",
"queries": [
{
"data_source": "logs",
"name": "query1",
"search": {
"query": "status:error $automation $version $service"
},
"compute": {
"aggregation": "count"
},
"indexes": [
"*"
]
},
{
"data_source": "logs",
"name": "query2",
"search": {
"query": "$automation $version $service"
},
"compute": {
"aggregation": "cardinality",
"metric": "@execution_id"
},
"indexes": [
"*"
]
}
],
"formulas": [
{
"formula": "query1 / query2 * 100"
}
],
"conditional_formats": [
{
"comparator": ">",
"value": 10,
"palette": "white_on_red"
},
{
"comparator": ">",
"value": 5,
"palette": "white_on_yellow"
},
{
"comparator": ">=",
"value": 0,
"palette": "white_on_green"
}
]
}
],
"autoscale": false,
"custom_unit": "%",
"precision": 2
},
"layout": {
"x": 3,
"y": 0,
"width": 3,
"height": 2
}
},
{
"id": 9910003,
"definition": {
"title": "Active Automations",
"time": {
"live_span": "1h"
},
"type": "query_value",
"requests": [
{
"response_format": "scalar",
"queries": [
{
"data_source": "logs",
"name": "query1",
"search": {
"query": "$automation $version $service"
},
"compute": {
"aggregation": "cardinality",
"metric": "@automation_id"
},
"indexes": [
"*"
]
}
],
"formulas": [
{
"formula": "query1"
}
]
}
],
"autoscale": true,
"precision": 0
},
"layout": {
"x": 6,
"y": 0,
"width": 3,
"height": 2
}
},
{
"id": 9910004,
"definition": {
"title": "CrewAI Versions in Use",
"time": {
"live_span": "1h"
},
"type": "query_value",
"requests": [
{
"response_format": "scalar",
"queries": [
{
"data_source": "logs",
"name": "query1",
"search": {
"query": "$automation $version $service"
},
"compute": {
"aggregation": "cardinality",
"metric": "@crewai_version"
},
"indexes": [
"*"
]
}
],
"formulas": [
{
"formula": "query1"
}
]
}
],
"autoscale": true,
"precision": 0
},
"layout": {
"x": 9,
"y": 0,
"width": 3,
"height": 2
}
}
]
},
"layout": {
"x": 0,
"y": 0,
"width": 12,
"height": 3
}
},
{
"id": 8820001,
"definition": {
"title": "Throughput",
"background_color": "vivid_green",
"show_title": true,
"type": "group",
"layout_type": "ordered",
"widgets": [
{
"id": 9920001,
"definition": {
"title": "Executions per Hour by Automation (top 10)",
"show_legend": false,
"type": "timeseries",
"requests": [
{
"response_format": "timeseries",
"queries": [
{
"data_source": "logs",
"name": "query1",
"search": {
"query": "$automation $version $service"
},
"compute": {
"aggregation": "cardinality",
"metric": "@execution_id",
"interval": 3600000
},
"group_by": [
{
"facet": "@automation_name",
"limit": 10,
"sort": {
"aggregation": "cardinality",
"metric": "@execution_id",
"order": "desc"
}
}
],
"indexes": [
"*"
]
}
],
"formulas": [
{
"formula": "query1"
}
],
"style": {
"palette": "semantic"
},
"display_type": "bars"
}
]
},
"layout": {
"x": 0,
"y": 0,
"width": 12,
"height": 3
}
}
]
},
"layout": {
"x": 0,
"y": 3,
"width": 12,
"height": 4
}
},
{
"id": 8830001,
"definition": {
"title": "Errors",
"background_color": "vivid_orange",
"show_title": true,
"type": "group",
"layout_type": "ordered",
"widgets": [
{
"id": 9930001,
"definition": {
"title": "Errors by Exception Type (top 5)",
"show_legend": false,
"type": "timeseries",
"requests": [
{
"response_format": "timeseries",
"queries": [
{
"data_source": "logs",
"name": "query1",
"search": {
"query": "status:error $automation $version $service"
},
"compute": {
"aggregation": "count"
},
"group_by": [
{
"facet": "@exception.type",
"limit": 5,
"sort": {
"aggregation": "count",
"order": "desc"
}
}
],
"indexes": [
"*"
]
}
],
"formulas": [
{
"formula": "query1"
}
],
"style": {
"palette": "warm"
},
"display_type": "bars"
}
]
},
"layout": {
"x": 0,
"y": 0,
"width": 6,
"height": 3
}
},
{
"id": 9930002,
"definition": {
"title": "Top Exception Types by Count",
"type": "toplist",
"requests": [
{
"response_format": "scalar",
"queries": [
{
"data_source": "logs",
"name": "query1",
"search": {
"query": "status:error $automation $version $service"
},
"compute": {
"aggregation": "count"
},
"group_by": [
{
"facet": "@exception.type",
"limit": 10,
"sort": {
"aggregation": "count",
"order": "desc"
}
}
],
"indexes": [
"*"
]
}
],
"formulas": [
{
"formula": "query1"
}
],
"sort": {
"count": 10,
"order_by": [
{
"type": "formula",
"index": 0,
"order": "desc"
}
]
}
}
],
"style": {
"palette": "datadog16"
}
},
"layout": {
"x": 6,
"y": 0,
"width": 6,
"height": 3
}
}
]
},
"layout": {
"x": 0,
"y": 7,
"width": 12,
"height": 4
}
},
{
"id": 8840001,
"definition": {
"title": "Cost",
"background_color": "vivid_purple",
"show_title": true,
"type": "group",
"layout_type": "ordered",
"widgets": [
{
"id": 9940001,
"definition": {
"title": "Total Tokens per Hour by Model (input + output)",
"show_legend": false,
"type": "timeseries",
"requests": [
{
"response_format": "timeseries",
"queries": [
{
"data_source": "logs",
"name": "query1",
"search": {
"query": "$automation $version $service"
},
"compute": {
"aggregation": "sum",
"metric": "@gen_ai.usage.input_tokens",
"interval": 3600000
},
"group_by": [
{
"facet": "@gen_ai.request.model",
"limit": 10,
"sort": {
"aggregation": "sum",
"metric": "@gen_ai.usage.input_tokens",
"order": "desc"
}
}
],
"indexes": [
"*"
]
},
{
"data_source": "logs",
"name": "query2",
"search": {
"query": "$automation $version $service"
},
"compute": {
"aggregation": "sum",
"metric": "@gen_ai.usage.output_tokens",
"interval": 3600000
},
"group_by": [
{
"facet": "@gen_ai.request.model",
"limit": 10,
"sort": {
"aggregation": "sum",
"metric": "@gen_ai.usage.output_tokens",
"order": "desc"
}
}
],
"indexes": [
"*"
]
}
],
"formulas": [
{
"formula": "query1 + query2",
"alias": "Total Tokens"
}
],
"style": {
"palette": "cool"
},
"display_type": "area"
}
]
},
"layout": {
"x": 0,
"y": 0,
"width": 12,
"height": 3
}
}
]
},
"layout": {
"x": 0,
"y": 11,
"width": 12,
"height": 4
}
},
{
"id": 8850002,
"definition": {
"title": "Drill-Down",
"background_color": "gray",
"show_title": true,
"type": "group",
"layout_type": "ordered",
"widgets": []
},
"layout": {
"x": 0,
"y": 15,
"width": 12,
"height": 1
}
}
],
"template_variables": [
{
"name": "automation",
"prefix": "@automation_name",
"available_values": [],
"default": "*"
},
{
"name": "version",
"prefix": "@crewai_version",
"available_values": [],
"default": "*"
},
{
"name": "service",
"prefix": "service",
"available_values": [],
"default": "*"
}
],
"layout_type": "ordered",
"notify_list": [],
"pause_auto_refresh": false,
"reflow_type": "fixed",
"tags": [
"ai:created_with_ai"
]
}

View File

@@ -4,6 +4,27 @@ description: "CrewAI의 제품 업데이트, 개선 사항 및 버그 수정"
icon: "clock"
mode: "wide"
---
<Update label="2026년 6월 18일">
## v1.14.8a2
[GitHub 릴리스 보기](https://github.com/crewAIInc/crewAI/releases/tag/1.14.8a2)
## 변경 사항
### 기능
- Flow 정의에 단일 에이전트 작업 추가
- 정의 로드 시 흐름 CEL 표현식 검증
### 문서
- 가져올 수 있는 운영 대시보드와 함께 Datadog 통합 가이드 추가
- v1.14.8a1의 스냅샷 및 변경 로그 업데이트
## 기여자
@joaomdmoura, @lucasgomide, @vinibrsl
</Update>
<Update label="2026년 6월 18일">
## v1.14.8a1

View File

@@ -9,6 +9,10 @@ CrewAI AMP는 배포에서 OpenTelemetry **트레이스**와 **로그**를 자
텔레메트리 데이터는 [OpenTelemetry GenAI 시맨틱 규칙](https://opentelemetry.io/docs/specs/semconv/gen-ai/)과 추가적인 CrewAI 전용 속성을 따릅니다.
<Tip>
OpenTelemetry는 **권장되는 관측 가능성 경로**입니다 — 벤더 중립적이며, OTLP 호환 백엔드(Grafana, Honeycomb, NewRelic, 자체 수집기)에서 작동합니다. Datadog을 사용하는 경우, Datadog Agent 경로와 Datadog의 OTLP 수집을 모두 다루는 전용 [Datadog 통합](./datadog) 가이드를 참조하세요.
</Tip>
## 사전 요구 사항
<CardGroup cols={2}>
@@ -41,17 +45,7 @@ CrewAI AMP는 배포에서 OpenTelemetry **트레이스**와 **로그**를 자
<Frame>![OpenTelemetry 수집기 구성](/images/crewai-otel-collector-opentelemetry.png)</Frame>
</Tab>
<Tab title="Datadog">
- **Datadog Site Domain** — Datadog 사이트의 OTLP 호스트만 입력합니다 (프로토콜이나 경로 제외). CrewAI가 전체 HTTPS OTLP 엔드포인트를 자동으로 구성합니다. [Datadog 사이트](https://docs.datadoghq.com/getting_started/site/)에 맞는 호스트를 사용하세요:
- `otlp.datadoghq.com` (US1)
- `otlp.us3.datadoghq.com` (US3)
- `otlp.us5.datadoghq.com` (US5)
- `otlp.datadoghq.eu` (EU1)
- `otlp.ap1.datadoghq.com` (AP1)
- **API Key** — Datadog API 키입니다. [키 생성 방법](https://docs.datadoghq.com/account_management/api-app-keys/#api-keys)을 참고하세요.
Datadog 통합은 **트레이스**를 내보냅니다.
<Frame>![Datadog 수집기 구성](/images/crewai-otel-collector-datadog.png)</Frame>
Datadog 설정은 전용 [Datadog 통합](./datadog) 가이드를 참조하세요 — Datadog Agent 경로(권장, 로그 볼륨에 더 저렴)와 Datadog의 OTLP 수집을 모두 다루며, 수집기 구성 단계를 완전히 설명합니다.
</Tab>
</Tabs>

View File

@@ -0,0 +1,295 @@
---
title: "Datadog 통합"
description: "Datadog Agent 또는 Datadog의 OTLP 수집을 통해 자체 호스팅 CrewAI AMP 배포를 Datadog에서 모니터링하세요 — 두 경로 모두 동일한 구조화된 패싯을 생성하므로 기성 운영 대시보드를 가져올 수 있습니다."
icon: "dog"
mode: "wide"
---
<Note>
**번역 진행 중** — 콘텐츠가 영어로 표시됩니다.
</Note>
CrewAI ships first-class support for Datadog: two log-ingestion paths, a JSON log schema designed for cheap indexing, and a ready-made operations dashboard you can import in under five minutes.
<Note>
For vendor-neutral observability via any OTLP backend (Grafana, Honeycomb, your own collector), see [OpenTelemetry Export](./capture_telemetry_logs).
</Note>
## Choose a path
CrewAI supports two log-ingestion paths to Datadog — both are first-class and produce the same structured facets that power the dashboard. Pick the one that fits your infrastructure.
<Tabs>
<Tab title="Datadog Agent">
The Datadog Agent runs alongside your CrewAI containers (typically as a DaemonSet on Kubernetes) and tails their stdout. With `CREWAI_LOG_FORMAT=json` set, each log event ships as a single billable line with structured attributes.
**Setup:**
1. Run the Datadog Agent next to your CrewAI containers — see [Datadog's deployment docs](https://docs.datadoghq.com/agent/) for Kubernetes, ECS, or VM setup. Enable log collection (`logs_enabled: true`) and container log collection (`logs_config.container_collect_all: true`).
2. Set `CREWAI_LOG_FORMAT=json` as an **automation environment variable** in CrewAI AMP (open your automation → **Settings → Environment Variables**) so each log event is a single line instead of a multi-line traceback. AMP propagates the value to every container in the deployment (API + workers) — don't set it on the container or host directly. See [Enabling JSON output](#enabling-json-output) below for the AMP UI walkthrough and the [log schema reference](#log-schema-reference) for the full field contract.
3. Confirm logs arrive in Datadog Logs with the JSON fields parsed — see [Verify ingestion](#verify-ingestion).
**Pick this path if** you already operate Datadog Agents (e.g. for infrastructure metrics), or your log volume makes per-event ingestion cost a real concern — collapsing tracebacks into single events keeps Agent ingestion cheap at scale.
</Tab>
<Tab title="Datadog OTLP intake">
CrewAI AMP exports OpenTelemetry traffic directly to Datadog's OTLP endpoint with no Agent required. Logs and traces ride a single export pipeline configured in AMP's UI, using the same protocol you'd use for any other OTLP backend.
**Setup:**
1. In CrewAI AMP, go to **Settings → OpenTelemetry Collectors → Add Collector** and pick **Datadog**.
2. Configure the connection:
- **Datadog Site Domain** — your Datadog site's OTLP host only, no protocol or path. CrewAI builds the full HTTPS OTLP endpoint for you. Use the host that matches your [Datadog site](https://docs.datadoghq.com/getting_started/site/):
- `otlp.datadoghq.com` (US1)
- `otlp.us3.datadoghq.com` (US3)
- `otlp.us5.datadoghq.com` (US5)
- `otlp.datadoghq.eu` (EU1)
- `otlp.ap1.datadoghq.com` (AP1)
- **API Key** — your Datadog API key. See [how to create one](https://docs.datadoghq.com/account_management/api-app-keys/#api-keys).
3. The Datadog template provisions **both signals at once** — when you save, AMP creates a traces collector at `/v1/traces` and a logs collector at `/v1/logs`, both sharing the same Datadog OTLP host and API key. You'll see them as two separate rows in your OTel collectors list.
4. *(optional)* Click **Test Connection** to verify CrewAI can reach the endpoint with the credentials you provided. Then click **Save** — both collectors are created in one step.
<Frame>![Datadog collector configuration](/images/crewai-otel-collector-datadog.png)</Frame>
**Pick this path if** you'd rather not operate a Datadog Agent, you already use OTLP for traces and want one export pipeline, or you may later want to fan out the same telemetry to other backends (Grafana, Honeycomb, etc.) without changing your application setup.
</Tab>
</Tabs>
Either path lands the same structured facets in Datadog (`@automation_id`, `@kickoff_id`, `@execution_id`, `@automation_name`, `@crewai_version`, `@exception.type`, `@gen_ai.*`), so the dashboard works identically with either choice.
## Log schema reference
<Info>
This schema applies to the **Datadog Agent path** — stdout JSON logs produced when `CREWAI_LOG_FORMAT=json` is set. Logs delivered via the **Datadog OTLP intake** use OpenTelemetry attribute names and may differ; see [OpenTelemetry Export](./capture_telemetry_logs).
</Info>
When `CREWAI_LOG_FORMAT=json` is set, every log event is emitted as a **single JSON object per line** to stdout, with internal newlines escaped. The format is plain JSON — Datadog parses it natively, and the same payload is also consumable by Splunk, Loki, Elasticsearch, and CloudWatch without custom log pipelines.
### Why JSON output
<CardGroup cols={2}>
<Card title="Lower ingestion cost" icon="dollar-sign">
Most managed log backends bill per event. A Python traceback in text format is counted as one event per line — 30+ events for a single error. JSON output collapses each traceback into a single event with the stack trace as an escaped string field.
</Card>
<Card title="Structured search" icon="magnifying-glass">
Search by `@automation_id`, `@exception.type`, `@kickoff_id` instead of grepping free-text. Build dashboards on typed facets without parser configuration.
</Card>
<Card title="APM ↔ logs correlation" icon="link">
Every event carries `trace_id` and `span_id` when fired inside a recording span, so backends auto-link logs to traces.
</Card>
<Card title="Stable contract" icon="file-shield">
The `schema` field gates compatibility — within `v1`, fields are added but never renamed or removed.
</Card>
</CardGroup>
### Enabling JSON output
`CREWAI_LOG_FORMAT=json` must be set as an **automation environment variable** in CrewAI AMP — it is **not** a container, host, or Docker setting. Open your automation in AMP, click the **Settings** icon, and add the variable under the **Environment Variables** section. AMP applies the value to every container in the deployment (API + workers) on the next restart. See [Update Your Crew](./update-crew) for the full UI walkthrough with screenshots.
```shell
CREWAI_LOG_FORMAT=json
```
Restart the deployment to pick up the change. Every log line on stdout from that point on is a single JSON object.
<Note>
The default value is `text`, which preserves the legacy human-readable line format byte-for-byte. Setting any value other than `json` falls back to text mode. There is no migration step — the variable is read at process start and the format switches immediately.
</Note>
### Example events
A single info-level log inside an active automation kickoff:
```json
{
"schema": "v1",
"ts": "2026-06-17T16:14:23.482914Z",
"level": "INFO",
"logger": "crewai_enterprise.utilities.pii_redaction",
"crewai_version": "1.14.7",
"msg": "PII tracking state reset (engines preserved)",
"automation_id": "12",
"task_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"kickoff_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"execution_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"automation_name": "research_flow"
}
```
An error with a Python exception is collapsed into a single event with the traceback as a string:
```json
{
"schema": "v1",
"ts": "2026-06-17T16:14:31.218450Z",
"level": "ERROR",
"logger": "api.tasks.flow_run_task",
"crewai_version": "1.14.7",
"msg": "Flow execution failed",
"automation_id": "12",
"kickoff_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"execution_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"automation_name": "research_flow",
"exception": {
"type": "ValueError",
"message": "Topic cannot be empty",
"stacktrace": "Traceback (most recent call last):\n File \"/app/flow.py\", line 42, in summarize\n ...\nValueError: Topic cannot be empty\n"
}
}
```
The same error in legacy text mode would have produced ~25 separate log events (one per traceback line) — all of which the backend would bill and index individually.
### Schema v1 fields
Within the `v1` schema, fields are only added, never renamed or removed. New fields will appear as soon as a deployment is upgraded.
| Field | Type | Always present | Source |
|-------|------|----------------|--------|
| `schema` | string | Yes | Constant `"v1"`. Increment indicates a breaking schema change. |
| `ts` | string (ISO-8601 UTC, microseconds) | Yes | Record creation time, e.g. `2026-06-17T16:14:23.482914Z`. |
| `level` | string | Yes | Python log level name: `DEBUG` / `INFO` / `WARNING` / `ERROR` / `CRITICAL`. |
| `logger` | string | Yes | Dotted logger name, e.g. `api.tasks.flow_run_task`. |
| `crewai_version` | string | Yes (when `crewai` package metadata is resolvable) | Installed `crewai` package version, e.g. `"1.14.7"`. |
| `msg` | string | Yes | Rendered log message (after `%`-formatting / `{}`-formatting). |
| `automation_id` | string | When `CREWAI_PLUS_ID` env var is set | Numeric deployment ID (AMP provisions this on every container). |
| `task_id` | string | On Celery worker logs | Celery task UUID, or `"no-task"` for non-task contexts. |
| `kickoff_id` | string | Inside an automation kickoff | UUID of the current kickoff. |
| `execution_id` | string | Inside an automation kickoff | UUID of the current sub-execution. Equal to `kickoff_id` at the top level; differs for nested flow methods that spawn sub-executions. |
| `automation_name` | string | Inside an automation kickoff | Human-readable automation/flow name, e.g. `"research_flow"`. |
| `trace_id` | string (32-hex) | Inside a recording OpenTelemetry span | Hex trace ID. Omitted when no span is active. |
| `span_id` | string (16-hex) | Inside a recording OpenTelemetry span | Hex span ID. Omitted when no span is active. |
| `exception` | object | When the log record has `exc_info` | `{type, message, stacktrace}` — full traceback as a single escaped string. |
<Tip>
Any additional `extra={...}` kwargs passed to a logger call appear as top-level JSON fields verbatim. Reserved field names above always win to keep the schema stable.
</Tip>
### Stability promise
The `schema` field declares the contract. Within `v1`, CrewAI commits to:
- **Never removing a field** that customers may have built queries or dashboards against.
- **Never renaming a field** in place — renames happen via a schema bump (e.g. `v2`), with the old name kept as a deprecated alias for at least one release cycle.
- **Adding new fields** at any time. Consumers should ignore unknown top-level keys.
When a `v2` is introduced, both the `schema` field and the migration guide will be published in advance, and `v1` will continue to be emitted for one release cycle so dashboards and queries have time to migrate.
## Prerequisite: promote facets
Datadog auto-discovers fields the first time it sees them but doesn't make them queryable in widgets until they're promoted to **facets**. This is a one-time setup in your Datadog account.
<Steps>
<Step title="Search for a CrewAI log">
Open [Logs Explorer](https://app.datadoghq.com/logs) and search `service:crewai*`. You should see at least one log event.
</Step>
<Step title="Promote each field">
Click any log entry to open the right-hand details panel. For each field below, hover the field name → click the gear icon → **Create facet**.
- `automation_id`, `automation_name`, `execution_id`, `kickoff_id`, `task_id`
- `crewai_version`, `model_id`
- `exception.type`, `exception.message`
Skip any field that already shows a star icon next to its name — that means it's already a facet. The `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, and `gen_ai.request.model` facets are typically promoted automatically by Datadog's LLM Observability auto-discovery, but verify they exist before importing the dashboard.
</Step>
</Steps>
## Import the dashboard
<Steps>
<Step title="Download the dashboard JSON">
Save [`datadog_dashboard.json`](https://raw.githubusercontent.com/crewAIInc/crewAI/main/docs/edge/en/enterprise/guides/datadog_dashboard.json) to your machine.
</Step>
<Step title="Open the import dialog in Datadog">
Navigate to **Dashboards → New Dashboard**. Click the **gear icon** in the top right of the empty dashboard and select **Import Dashboard JSON**.
</Step>
<Step title="Paste or upload the JSON">
Paste the contents of `datadog_dashboard.json` into the import dialog (or drag the file in). Click **Import**.
Datadog creates the dashboard immediately and lands you on it. The first load may show empty widgets for a few seconds while queries execute against the time range.
</Step>
</Steps>
<Tip>
Datadog's [Dashboard API](https://docs.datadoghq.com/api/latest/dashboards/#create-a-new-dashboard) accepts the same JSON via `POST /api/v1/dashboard`. Use it if you manage dashboards through Terraform, Pulumi, or CI.
</Tip>
## What you get
The dashboard is organized into four sections plus a placeholder for a custom drill-down widget:
| Section | Widgets | Useful for |
|---------|---------|------------|
| **Header** | Total Executions · Error Rate (%) · Active Automations · CrewAI Versions in Use | At-a-glance health for the last hour. Error Rate is conditionally formatted (green ≤ 5%, yellow ≤ 10%, red > 10%). |
| **Throughput** | Executions per Hour by Automation (top 10, stacked bars) | Spotting traffic shifts, surfacing busy automations, validating that a rollout didn't change baseline volume. |
| **Errors** | Errors by Exception Type (top 5, stacked bars) · Top Exception Types by Count (toplist) | Triaging failures — which exception types are spiking, which automations they're hitting. |
| **Cost** | Total Tokens per Hour by Model (input + output, stacked area) | Tracking LLM token spend by model. Useful for catching cost regressions when an automation switches model or starts looping. |
| **Drill-Down** | _(empty placeholder)_ | See [Customization](#customize) for adding a recent-errors log stream here. |
Three template variables at the top of the dashboard re-scope every widget at once:
- **`$automation`** — filter to a single automation by name.
- **`$version`** — filter to a single `crewai` SDK version (useful for comparing pre- and post-upgrade behavior).
- **`$service`** — filter to a specific Datadog `service` tag (useful when multiple CrewAI deployments share one Datadog account).
## Verify ingestion
Open [Logs Explorer](https://app.datadoghq.com/logs) and run a query that matches your ingestion path:
<Tabs>
<Tab title="Datadog Agent">
Search `service:crewai* @schema:v1`. You should see structured logs with the JSON fields parsed into Datadog facets. Pick a recent event and verify it has `@automation_id`, `@kickoff_id`, `@execution_id`, `@crewai_version`, and (when running inside a span) `@trace_id` / `@span_id` populated.
If nothing appears, confirm `CREWAI_LOG_FORMAT=json` is set under your automation's **Environment Variables** in AMP, the deployment was restarted after the change, and the Datadog Agent is tailing container stdout.
</Tab>
<Tab title="Datadog OTLP intake">
Search `source:otlp service:crewai*`. OTLP attributes land with their OpenTelemetry names (`automation_id`, `crewai.kickoff.id`, etc.) rather than the stdout JSON keys, but they map to the same dashboard facets after [facet promotion](#prerequisite-promote-facets).
If nothing appears, verify the collector endpoint is correct (`/v1/logs` for logs, `/v1/traces` for traces) and **Test Connection** succeeded when the collector was saved.
</Tab>
</Tabs>
## Customize
The dashboard ships with deliberate gaps so you can extend it without uninstalling and re-importing.
### Add a Recent Errors log stream
The **Drill-Down** section is intentionally empty. Add a Log Stream widget to it for an inline view of recent failures:
1. Edit the dashboard and click **+ Add Widgets** inside the Drill-Down group.
2. Drag in a **Log Stream** widget.
3. Set the filter query to `status:error $automation $version $service`.
4. Choose columns: `@timestamp`, `@automation_name`, `@exception.type`, `@exception.message`, `@execution_id`.
5. Sort by most recent, limit to 25 entries.
Clicking any row jumps to Logs Explorer with the same filter pre-applied.
### Add p95 latency
Logs don't include execution duration by default. Two ways to add a latency widget:
- **From APM traces** — if you also export OTLP traces to Datadog, add a Timeseries widget with data source **Traces**, query `service:crewai*`, aggregation `p95 of @duration`. Datadog APM auto-tracks span duration.
- **From metric extraction** — extract a `flow.duration_ms` metric from logs via [Datadog's log-to-metric pipeline](https://docs.datadoghq.com/logs/log_configuration/logs_to_metrics/), then chart it like any other metric. Useful if you don't run APM.
### Re-scope to multiple deployments
The `$service` template variable defaults to `*` and will catch every CrewAI deployment in your Datadog account. Change the default to a specific service name in **Configure → Template Variables** if you want the dashboard to focus on one deployment by default.
## Troubleshooting
| Symptom | Likely cause | Fix |
|---------|--------------|-----|
| All widgets show "No data" | Facets aren't promoted | Re-do the [Promote facets](#prerequisite-promote-facets) step. Datadog won't query against an un-promoted field. |
| Error Rate widget shows `NaN` | No executions in the time window | Either no traffic, or `@execution_id` isn't faceted. Expand the time range and re-check facets. |
| Throughput chart is flat at the same value | Logs aren't reaching Datadog | Search `service:crewai*` in Logs Explorer. If nothing shows, verify the Datadog Agent is running (Agent path) or the OTel collector endpoint is correct (OTLP path). |
| `crewai_version` shows fewer values than expected | Some containers predate the structured-logs work | The `crewai_version` field was added alongside JSON output. Older deployments running text mode (or older AMP builds) won't emit it. Upgrade those deployments to pick up the field. See the [log schema reference](#log-schema-reference) for the full field contract. |
| Template variables don't filter widgets | The widget's filter line doesn't reference the template variable | Edit the widget and confirm the search includes `$automation $version $service`. |
## Next steps
<CardGroup cols={2}>
<Card title="OpenTelemetry Export" icon="magnifying-glass-chart" href="./capture_telemetry_logs">
Vendor-neutral observability for non-Datadog stacks (Grafana, Honeycomb, your own collector) — or as a Datadog complement when you want to fan out telemetry to multiple backends.
</Card>
<Card title="Datadog Log Search Syntax" icon="magnifying-glass" href="https://docs.datadoghq.com/logs/explorer/search_syntax/">
Reference for customizing widget queries against the structured facets above.
</Card>
</CardGroup>

View File

@@ -4,6 +4,27 @@ description: "Atualizações de produto, melhorias e correções do CrewAI"
icon: "clock"
mode: "wide"
---
<Update label="18 jun 2026">
## v1.14.8a2
[Ver release no GitHub](https://github.com/crewAIInc/crewAI/releases/tag/1.14.8a2)
## O que Mudou
### Funcionalidades
- Adicionar ação de agente único às definições de Fluxo
- Validar expressões CEL de fluxo no momento do carregamento da definição
### Documentação
- Adicionar guia de integração do Datadog com painel de operações importável
- Atualizar snapshot e changelog para v1.14.8a1
## Contributors
@joaomdmoura, @lucasgomide, @vinibrsl
</Update>
<Update label="18 jun 2026">
## v1.14.8a1

View File

@@ -9,6 +9,10 @@ O CrewAI AMP pode exportar **traces** e **logs** do OpenTelemetry das suas impla
Os dados de telemetria seguem as [convenções semânticas GenAI do OpenTelemetry](https://opentelemetry.io/docs/specs/semconv/gen-ai/) além de atributos adicionais específicos do CrewAI.
<Tip>
OpenTelemetry é o **caminho de observabilidade recomendado** — neutro em relação a fornecedores, funciona com qualquer backend compatível com OTLP (Grafana, Honeycomb, NewRelic, seu próprio coletor). Se você usa especificamente o Datadog, veja o guia dedicado [Integração com Datadog](./datadog), que cobre tanto o caminho do Datadog Agent quanto o ingest OTLP do Datadog.
</Tip>
## Pré-requisitos
<CardGroup cols={2}>
@@ -41,17 +45,7 @@ Os dados de telemetria seguem as [convenções semânticas GenAI do OpenTelemetr
<Frame>![Configuração do coletor OpenTelemetry](/images/crewai-otel-collector-opentelemetry.png)</Frame>
</Tab>
<Tab title="Datadog">
- **Datadog Site Domain** — Apenas o host OTLP do seu site Datadog, sem protocolo ou caminho. O CrewAI monta o endpoint HTTPS OTLP completo para você. Use o host correspondente ao seu [site Datadog](https://docs.datadoghq.com/getting_started/site/):
- `otlp.datadoghq.com` (US1)
- `otlp.us3.datadoghq.com` (US3)
- `otlp.us5.datadoghq.com` (US5)
- `otlp.datadoghq.eu` (EU1)
- `otlp.ap1.datadoghq.com` (AP1)
- **API Key** — Sua chave de API do Datadog. Veja [como criar uma](https://docs.datadoghq.com/account_management/api-app-keys/#api-keys).
A integração com o Datadog exporta **traces**.
<Frame>![Configuração do coletor Datadog](/images/crewai-otel-collector-datadog.png)</Frame>
Para configurar o Datadog, veja o guia dedicado [Integração com Datadog](./datadog) — ele cobre tanto o caminho do Datadog Agent (recomendado, mais barato para volumes altos de log) quanto o ingest OTLP do Datadog, com os passos completos de configuração do coletor.
</Tab>
</Tabs>

View File

@@ -0,0 +1,295 @@
---
title: "Integração com Datadog"
description: "Monitore implantações CrewAI AMP auto-hospedadas no Datadog via Datadog Agent ou ingest OTLP do Datadog — ambos os caminhos entregam as mesmas facetas estruturadas para importar o dashboard de operações pronto."
icon: "dog"
mode: "wide"
---
<Note>
**Tradução em andamento** — conteúdo exibido em inglês.
</Note>
CrewAI ships first-class support for Datadog: two log-ingestion paths, a JSON log schema designed for cheap indexing, and a ready-made operations dashboard you can import in under five minutes.
<Note>
For vendor-neutral observability via any OTLP backend (Grafana, Honeycomb, your own collector), see [OpenTelemetry Export](./capture_telemetry_logs).
</Note>
## Choose a path
CrewAI supports two log-ingestion paths to Datadog — both are first-class and produce the same structured facets that power the dashboard. Pick the one that fits your infrastructure.
<Tabs>
<Tab title="Datadog Agent">
The Datadog Agent runs alongside your CrewAI containers (typically as a DaemonSet on Kubernetes) and tails their stdout. With `CREWAI_LOG_FORMAT=json` set, each log event ships as a single billable line with structured attributes.
**Setup:**
1. Run the Datadog Agent next to your CrewAI containers — see [Datadog's deployment docs](https://docs.datadoghq.com/agent/) for Kubernetes, ECS, or VM setup. Enable log collection (`logs_enabled: true`) and container log collection (`logs_config.container_collect_all: true`).
2. Set `CREWAI_LOG_FORMAT=json` as an **automation environment variable** in CrewAI AMP (open your automation → **Settings → Environment Variables**) so each log event is a single line instead of a multi-line traceback. AMP propagates the value to every container in the deployment (API + workers) — don't set it on the container or host directly. See [Enabling JSON output](#enabling-json-output) below for the AMP UI walkthrough and the [log schema reference](#log-schema-reference) for the full field contract.
3. Confirm logs arrive in Datadog Logs with the JSON fields parsed — see [Verify ingestion](#verify-ingestion).
**Pick this path if** you already operate Datadog Agents (e.g. for infrastructure metrics), or your log volume makes per-event ingestion cost a real concern — collapsing tracebacks into single events keeps Agent ingestion cheap at scale.
</Tab>
<Tab title="Datadog OTLP intake">
CrewAI AMP exports OpenTelemetry traffic directly to Datadog's OTLP endpoint with no Agent required. Logs and traces ride a single export pipeline configured in AMP's UI, using the same protocol you'd use for any other OTLP backend.
**Setup:**
1. In CrewAI AMP, go to **Settings → OpenTelemetry Collectors → Add Collector** and pick **Datadog**.
2. Configure the connection:
- **Datadog Site Domain** — your Datadog site's OTLP host only, no protocol or path. CrewAI builds the full HTTPS OTLP endpoint for you. Use the host that matches your [Datadog site](https://docs.datadoghq.com/getting_started/site/):
- `otlp.datadoghq.com` (US1)
- `otlp.us3.datadoghq.com` (US3)
- `otlp.us5.datadoghq.com` (US5)
- `otlp.datadoghq.eu` (EU1)
- `otlp.ap1.datadoghq.com` (AP1)
- **API Key** — your Datadog API key. See [how to create one](https://docs.datadoghq.com/account_management/api-app-keys/#api-keys).
3. The Datadog template provisions **both signals at once** — when you save, AMP creates a traces collector at `/v1/traces` and a logs collector at `/v1/logs`, both sharing the same Datadog OTLP host and API key. You'll see them as two separate rows in your OTel collectors list.
4. *(optional)* Click **Test Connection** to verify CrewAI can reach the endpoint with the credentials you provided. Then click **Save** — both collectors are created in one step.
<Frame>![Datadog collector configuration](/images/crewai-otel-collector-datadog.png)</Frame>
**Pick this path if** you'd rather not operate a Datadog Agent, you already use OTLP for traces and want one export pipeline, or you may later want to fan out the same telemetry to other backends (Grafana, Honeycomb, etc.) without changing your application setup.
</Tab>
</Tabs>
Either path lands the same structured facets in Datadog (`@automation_id`, `@kickoff_id`, `@execution_id`, `@automation_name`, `@crewai_version`, `@exception.type`, `@gen_ai.*`), so the dashboard works identically with either choice.
## Log schema reference
<Info>
This schema applies to the **Datadog Agent path** — stdout JSON logs produced when `CREWAI_LOG_FORMAT=json` is set. Logs delivered via the **Datadog OTLP intake** use OpenTelemetry attribute names and may differ; see [OpenTelemetry Export](./capture_telemetry_logs).
</Info>
When `CREWAI_LOG_FORMAT=json` is set, every log event is emitted as a **single JSON object per line** to stdout, with internal newlines escaped. The format is plain JSON — Datadog parses it natively, and the same payload is also consumable by Splunk, Loki, Elasticsearch, and CloudWatch without custom log pipelines.
### Why JSON output
<CardGroup cols={2}>
<Card title="Lower ingestion cost" icon="dollar-sign">
Most managed log backends bill per event. A Python traceback in text format is counted as one event per line — 30+ events for a single error. JSON output collapses each traceback into a single event with the stack trace as an escaped string field.
</Card>
<Card title="Structured search" icon="magnifying-glass">
Search by `@automation_id`, `@exception.type`, `@kickoff_id` instead of grepping free-text. Build dashboards on typed facets without parser configuration.
</Card>
<Card title="APM ↔ logs correlation" icon="link">
Every event carries `trace_id` and `span_id` when fired inside a recording span, so backends auto-link logs to traces.
</Card>
<Card title="Stable contract" icon="file-shield">
The `schema` field gates compatibility — within `v1`, fields are added but never renamed or removed.
</Card>
</CardGroup>
### Enabling JSON output
`CREWAI_LOG_FORMAT=json` must be set as an **automation environment variable** in CrewAI AMP — it is **not** a container, host, or Docker setting. Open your automation in AMP, click the **Settings** icon, and add the variable under the **Environment Variables** section. AMP applies the value to every container in the deployment (API + workers) on the next restart. See [Update Your Crew](./update-crew) for the full UI walkthrough with screenshots.
```shell
CREWAI_LOG_FORMAT=json
```
Restart the deployment to pick up the change. Every log line on stdout from that point on is a single JSON object.
<Note>
The default value is `text`, which preserves the legacy human-readable line format byte-for-byte. Setting any value other than `json` falls back to text mode. There is no migration step — the variable is read at process start and the format switches immediately.
</Note>
### Example events
A single info-level log inside an active automation kickoff:
```json
{
"schema": "v1",
"ts": "2026-06-17T16:14:23.482914Z",
"level": "INFO",
"logger": "crewai_enterprise.utilities.pii_redaction",
"crewai_version": "1.14.7",
"msg": "PII tracking state reset (engines preserved)",
"automation_id": "12",
"task_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"kickoff_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"execution_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"automation_name": "research_flow"
}
```
An error with a Python exception is collapsed into a single event with the traceback as a string:
```json
{
"schema": "v1",
"ts": "2026-06-17T16:14:31.218450Z",
"level": "ERROR",
"logger": "api.tasks.flow_run_task",
"crewai_version": "1.14.7",
"msg": "Flow execution failed",
"automation_id": "12",
"kickoff_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"execution_id": "0843a930-b306-464b-89c8-bfafa78cc711",
"automation_name": "research_flow",
"exception": {
"type": "ValueError",
"message": "Topic cannot be empty",
"stacktrace": "Traceback (most recent call last):\n File \"/app/flow.py\", line 42, in summarize\n ...\nValueError: Topic cannot be empty\n"
}
}
```
The same error in legacy text mode would have produced ~25 separate log events (one per traceback line) — all of which the backend would bill and index individually.
### Schema v1 fields
Within the `v1` schema, fields are only added, never renamed or removed. New fields will appear as soon as a deployment is upgraded.
| Field | Type | Always present | Source |
|-------|------|----------------|--------|
| `schema` | string | Yes | Constant `"v1"`. Increment indicates a breaking schema change. |
| `ts` | string (ISO-8601 UTC, microseconds) | Yes | Record creation time, e.g. `2026-06-17T16:14:23.482914Z`. |
| `level` | string | Yes | Python log level name: `DEBUG` / `INFO` / `WARNING` / `ERROR` / `CRITICAL`. |
| `logger` | string | Yes | Dotted logger name, e.g. `api.tasks.flow_run_task`. |
| `crewai_version` | string | Yes (when `crewai` package metadata is resolvable) | Installed `crewai` package version, e.g. `"1.14.7"`. |
| `msg` | string | Yes | Rendered log message (after `%`-formatting / `{}`-formatting). |
| `automation_id` | string | When `CREWAI_PLUS_ID` env var is set | Numeric deployment ID (AMP provisions this on every container). |
| `task_id` | string | On Celery worker logs | Celery task UUID, or `"no-task"` for non-task contexts. |
| `kickoff_id` | string | Inside an automation kickoff | UUID of the current kickoff. |
| `execution_id` | string | Inside an automation kickoff | UUID of the current sub-execution. Equal to `kickoff_id` at the top level; differs for nested flow methods that spawn sub-executions. |
| `automation_name` | string | Inside an automation kickoff | Human-readable automation/flow name, e.g. `"research_flow"`. |
| `trace_id` | string (32-hex) | Inside a recording OpenTelemetry span | Hex trace ID. Omitted when no span is active. |
| `span_id` | string (16-hex) | Inside a recording OpenTelemetry span | Hex span ID. Omitted when no span is active. |
| `exception` | object | When the log record has `exc_info` | `{type, message, stacktrace}` — full traceback as a single escaped string. |
<Tip>
Any additional `extra={...}` kwargs passed to a logger call appear as top-level JSON fields verbatim. Reserved field names above always win to keep the schema stable.
</Tip>
### Stability promise
The `schema` field declares the contract. Within `v1`, CrewAI commits to:
- **Never removing a field** that customers may have built queries or dashboards against.
- **Never renaming a field** in place — renames happen via a schema bump (e.g. `v2`), with the old name kept as a deprecated alias for at least one release cycle.
- **Adding new fields** at any time. Consumers should ignore unknown top-level keys.
When a `v2` is introduced, both the `schema` field and the migration guide will be published in advance, and `v1` will continue to be emitted for one release cycle so dashboards and queries have time to migrate.
## Prerequisite: promote facets
Datadog auto-discovers fields the first time it sees them but doesn't make them queryable in widgets until they're promoted to **facets**. This is a one-time setup in your Datadog account.
<Steps>
<Step title="Search for a CrewAI log">
Open [Logs Explorer](https://app.datadoghq.com/logs) and search `service:crewai*`. You should see at least one log event.
</Step>
<Step title="Promote each field">
Click any log entry to open the right-hand details panel. For each field below, hover the field name → click the gear icon → **Create facet**.
- `automation_id`, `automation_name`, `execution_id`, `kickoff_id`, `task_id`
- `crewai_version`, `model_id`
- `exception.type`, `exception.message`
Skip any field that already shows a star icon next to its name — that means it's already a facet. The `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, and `gen_ai.request.model` facets are typically promoted automatically by Datadog's LLM Observability auto-discovery, but verify they exist before importing the dashboard.
</Step>
</Steps>
## Import the dashboard
<Steps>
<Step title="Download the dashboard JSON">
Save [`datadog_dashboard.json`](https://raw.githubusercontent.com/crewAIInc/crewAI/main/docs/edge/en/enterprise/guides/datadog_dashboard.json) to your machine.
</Step>
<Step title="Open the import dialog in Datadog">
Navigate to **Dashboards → New Dashboard**. Click the **gear icon** in the top right of the empty dashboard and select **Import Dashboard JSON**.
</Step>
<Step title="Paste or upload the JSON">
Paste the contents of `datadog_dashboard.json` into the import dialog (or drag the file in). Click **Import**.
Datadog creates the dashboard immediately and lands you on it. The first load may show empty widgets for a few seconds while queries execute against the time range.
</Step>
</Steps>
<Tip>
Datadog's [Dashboard API](https://docs.datadoghq.com/api/latest/dashboards/#create-a-new-dashboard) accepts the same JSON via `POST /api/v1/dashboard`. Use it if you manage dashboards through Terraform, Pulumi, or CI.
</Tip>
## What you get
The dashboard is organized into four sections plus a placeholder for a custom drill-down widget:
| Section | Widgets | Useful for |
|---------|---------|------------|
| **Header** | Total Executions · Error Rate (%) · Active Automations · CrewAI Versions in Use | At-a-glance health for the last hour. Error Rate is conditionally formatted (green ≤ 5%, yellow ≤ 10%, red > 10%). |
| **Throughput** | Executions per Hour by Automation (top 10, stacked bars) | Spotting traffic shifts, surfacing busy automations, validating that a rollout didn't change baseline volume. |
| **Errors** | Errors by Exception Type (top 5, stacked bars) · Top Exception Types by Count (toplist) | Triaging failures — which exception types are spiking, which automations they're hitting. |
| **Cost** | Total Tokens per Hour by Model (input + output, stacked area) | Tracking LLM token spend by model. Useful for catching cost regressions when an automation switches model or starts looping. |
| **Drill-Down** | _(empty placeholder)_ | See [Customization](#customize) for adding a recent-errors log stream here. |
Three template variables at the top of the dashboard re-scope every widget at once:
- **`$automation`** — filter to a single automation by name.
- **`$version`** — filter to a single `crewai` SDK version (useful for comparing pre- and post-upgrade behavior).
- **`$service`** — filter to a specific Datadog `service` tag (useful when multiple CrewAI deployments share one Datadog account).
## Verify ingestion
Open [Logs Explorer](https://app.datadoghq.com/logs) and run a query that matches your ingestion path:
<Tabs>
<Tab title="Datadog Agent">
Search `service:crewai* @schema:v1`. You should see structured logs with the JSON fields parsed into Datadog facets. Pick a recent event and verify it has `@automation_id`, `@kickoff_id`, `@execution_id`, `@crewai_version`, and (when running inside a span) `@trace_id` / `@span_id` populated.
If nothing appears, confirm `CREWAI_LOG_FORMAT=json` is set under your automation's **Environment Variables** in AMP, the deployment was restarted after the change, and the Datadog Agent is tailing container stdout.
</Tab>
<Tab title="Datadog OTLP intake">
Search `source:otlp service:crewai*`. OTLP attributes land with their OpenTelemetry names (`automation_id`, `crewai.kickoff.id`, etc.) rather than the stdout JSON keys, but they map to the same dashboard facets after [facet promotion](#prerequisite-promote-facets).
If nothing appears, verify the collector endpoint is correct (`/v1/logs` for logs, `/v1/traces` for traces) and **Test Connection** succeeded when the collector was saved.
</Tab>
</Tabs>
## Customize
The dashboard ships with deliberate gaps so you can extend it without uninstalling and re-importing.
### Add a Recent Errors log stream
The **Drill-Down** section is intentionally empty. Add a Log Stream widget to it for an inline view of recent failures:
1. Edit the dashboard and click **+ Add Widgets** inside the Drill-Down group.
2. Drag in a **Log Stream** widget.
3. Set the filter query to `status:error $automation $version $service`.
4. Choose columns: `@timestamp`, `@automation_name`, `@exception.type`, `@exception.message`, `@execution_id`.
5. Sort by most recent, limit to 25 entries.
Clicking any row jumps to Logs Explorer with the same filter pre-applied.
### Add p95 latency
Logs don't include execution duration by default. Two ways to add a latency widget:
- **From APM traces** — if you also export OTLP traces to Datadog, add a Timeseries widget with data source **Traces**, query `service:crewai*`, aggregation `p95 of @duration`. Datadog APM auto-tracks span duration.
- **From metric extraction** — extract a `flow.duration_ms` metric from logs via [Datadog's log-to-metric pipeline](https://docs.datadoghq.com/logs/log_configuration/logs_to_metrics/), then chart it like any other metric. Useful if you don't run APM.
### Re-scope to multiple deployments
The `$service` template variable defaults to `*` and will catch every CrewAI deployment in your Datadog account. Change the default to a specific service name in **Configure → Template Variables** if you want the dashboard to focus on one deployment by default.
## Troubleshooting
| Symptom | Likely cause | Fix |
|---------|--------------|-----|
| All widgets show "No data" | Facets aren't promoted | Re-do the [Promote facets](#prerequisite-promote-facets) step. Datadog won't query against an un-promoted field. |
| Error Rate widget shows `NaN` | No executions in the time window | Either no traffic, or `@execution_id` isn't faceted. Expand the time range and re-check facets. |
| Throughput chart is flat at the same value | Logs aren't reaching Datadog | Search `service:crewai*` in Logs Explorer. If nothing shows, verify the Datadog Agent is running (Agent path) or the OTel collector endpoint is correct (OTLP path). |
| `crewai_version` shows fewer values than expected | Some containers predate the structured-logs work | The `crewai_version` field was added alongside JSON output. Older deployments running text mode (or older AMP builds) won't emit it. Upgrade those deployments to pick up the field. See the [log schema reference](#log-schema-reference) for the full field contract. |
| Template variables don't filter widgets | The widget's filter line doesn't reference the template variable | Edit the widget and confirm the search includes `$automation $version $service`. |
## Next steps
<CardGroup cols={2}>
<Card title="OpenTelemetry Export" icon="magnifying-glass-chart" href="./capture_telemetry_logs">
Vendor-neutral observability for non-Datadog stacks (Grafana, Honeycomb, your own collector) — or as a Datadog complement when you want to fan out telemetry to multiple backends.
</Card>
<Card title="Datadog Log Search Syntax" icon="magnifying-glass" href="https://docs.datadoghq.com/logs/explorer/search_syntax/">
Reference for customizing widget queries against the structured facets above.
</Card>
</CardGroup>

View File

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

View File

@@ -1 +1 @@
__version__ = "1.14.8a1"
__version__ = "1.14.8a2"

View File

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
authors = [{ name = "Your Name", email = "you@example.com" }]
requires-python = ">=3.10,<3.14"
dependencies = [
"crewai[tools]==1.14.8a1"
"crewai[tools]==1.14.8a2"
]
[project.scripts]

View File

@@ -5,7 +5,7 @@ description = "{{name}} using crewAI"
authors = [{ name = "Your Name", email = "you@example.com" }]
requires-python = ">=3.10,<3.14"
dependencies = [
"crewai[tools]==1.14.8a1"
"crewai[tools]==1.14.8a2"
]
[project.scripts]

View File

@@ -5,7 +5,7 @@ description = "Power up your crews with {{folder_name}}"
readme = "README.md"
requires-python = ">=3.10,<3.14"
dependencies = [
"crewai[tools]==1.14.8a1"
"crewai[tools]==1.14.8a2"
]
[tool.crewai]

View File

@@ -1 +1 @@
__version__ = "1.14.8a1"
__version__ = "1.14.8a2"

View File

@@ -152,4 +152,4 @@ __all__ = [
"wrap_file_source",
]
__version__ = "1.14.8a1"
__version__ = "1.14.8a2"

View File

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

View File

@@ -330,4 +330,4 @@ __all__ = [
"ZapierActionTools",
]
__version__ = "1.14.8a1"
__version__ = "1.14.8a2"

View File

@@ -8,8 +8,8 @@ authors = [
]
requires-python = ">=3.10, <3.14"
dependencies = [
"crewai-core==1.14.8a1",
"crewai-cli==1.14.8a1",
"crewai-core==1.14.8a2",
"crewai-cli==1.14.8a2",
# 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.14.8a1",
"crewai-tools==1.14.8a2",
]
embeddings = [
"tiktoken>=0.8.0,<0.13"

View File

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

View File

@@ -10,6 +10,7 @@ from crewai.flow.conversation import (
ConversationalInputs,
)
from crewai.flow.dsl import HumanFeedbackResult, human_feedback
from crewai.flow.expressions import Expression
from crewai.flow.flow import Flow, and_, listen, or_, router, start
from crewai.flow.flow_config import flow_config
from crewai.flow.input_provider import InputProvider, InputResponse
@@ -26,6 +27,7 @@ __all__ = [
"ConsoleProvider",
"ConversationalConfig",
"ConversationalInputs",
"Expression",
"Flow",
"FlowStructure",
"HumanFeedbackPending",

View File

@@ -0,0 +1,329 @@
"""Runtime expression support for FlowDefinition CEL expressions."""
from __future__ import annotations
from collections.abc import Iterable
import json
from typing import TYPE_CHECKING, Any, TypeAlias, cast
from crewai.utilities.serialization import to_serializable
if TYPE_CHECKING:
from crewai.flow.runtime import Flow
else:
from typing_extensions import TypeAliasType
_CEL_MACROS_WITH_LOCAL_BINDINGS = frozenset(
{"all", "exists", "exists_one", "filter", "map"}
)
if TYPE_CHECKING:
ExpressionData: TypeAlias = (
str
| int
| float
| bool
| None
| list["ExpressionData"]
| dict[str, "ExpressionData"]
)
else:
ExpressionData = TypeAliasType(
"ExpressionData",
str
| int
| float
| bool
| None
| list["ExpressionData"]
| dict[str, "ExpressionData"],
)
__all__ = [
"Expression",
"ExpressionData",
"ExpressionError",
]
class ExpressionError(ValueError):
"""An expression failed to parse, validate, render, or evaluate."""
class Expression:
"""CEL expression helper used for definition-time checks and runtime rendering."""
def __init__(
self, value: ExpressionData, *, context: dict[str, Any] | None = None
) -> None:
self.value = value
self.context = context
@classmethod
def from_flow(
cls,
value: ExpressionData,
flow: Flow[Any],
*,
local_context: dict[str, Any] | None = None,
) -> Expression:
"""Build an expression with the standard Flow runtime context."""
return cls(value, context=cls._flow_context(flow, local_context=local_context))
def validate_expression(
self,
*,
allowed_roots: Iterable[str],
source: str = "CEL expression",
) -> None:
"""Validate a full CEL expression without evaluating it."""
allowed = frozenset(allowed_roots)
expression = self._require_cel_source(cast(str, self.value), source=source)
roots = self._collect_root_identifiers(
self._compile_cel(expression, source=source)
)
unknown = sorted(root for root in roots if root not in allowed)
if unknown:
allowed_list = ", ".join(sorted(allowed))
unknown_list = ", ".join(repr(root) for root in unknown)
raise ExpressionError(
f"unknown CEL root at {source}: {unknown_list}; "
f"allowed roots: {allowed_list}. Reference flow data through one "
"of those roots, for example state.field or outputs.step_name."
)
def validate_template(
self,
*,
allowed_roots: Iterable[str],
source: str = "with block",
) -> None:
"""Validate nested strings fully wrapped in ``${...}`` as CEL."""
self._validate_template_value(
self.value, allowed_roots=allowed_roots, source=source
)
def evaluate(self, context: dict[str, Any] | None = None) -> Any:
"""Evaluate this value as a full CEL expression."""
resolved_context = self.context if context is None else context
return self._evaluate_cel(
self._require_cel_source(cast(str, self.value)),
resolved_context or {},
)
def render_template(self, context: dict[str, Any] | None = None) -> Any:
"""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 {})
@staticmethod
def _validate_template_value(
value: ExpressionData,
*,
allowed_roots: Iterable[str],
source: str,
) -> None:
if isinstance(value, str):
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):
for key, item in value.items():
item_source = f"{source}.{key}" if isinstance(key, str) else source
Expression._validate_template_value(
item, allowed_roots=allowed_roots, source=item_source
)
return
if isinstance(value, list):
for index, item in enumerate(value):
Expression._validate_template_value(
item,
allowed_roots=allowed_roots,
source=f"{source}[{index}]",
)
@staticmethod
def _flow_context(
flow: Flow[Any], local_context: dict[str, Any] | None = None
) -> dict[str, Any]:
from crewai.flow.runtime._outputs import outputs_by_name
local_outputs = local_context.get("outputs") if local_context else None
outputs = outputs_by_name(
flow._method_outputs,
local_outputs=local_outputs,
serialize=True,
)
context: dict[str, Any] = {
"state": flow._copy_and_serialize_state(),
"outputs": outputs,
}
if local_context:
context.update(
{
key: to_serializable(value, max_depth=0)
for key, value in local_context.items()
if key not in {"outputs", "state"}
}
)
return context
@staticmethod
def _render_template_value(value: ExpressionData, context: dict[str, Any]) -> Any:
if isinstance(value, str):
return Expression._render_template_string(value, context)
if isinstance(value, dict):
return {
key: Expression._render_template_value(item, context)
for key, item in value.items()
}
if isinstance(value, list):
return [Expression._render_template_value(item, context) for item in value]
return value
@staticmethod
def _render_template_string(value: str, context: dict[str, Any]) -> Any:
expression = Expression._expression_marker_source(value)
if expression is None:
return value
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:
try:
from celpy import Environment
from celpy.adapter import CELJSONEncoder, json_to_cel
from celpy.evaluation import Context
environment = Environment()
program = environment.program(
Expression._compile_cel(expression, environment=environment)
)
result = program.evaluate(cast(Context, json_to_cel(context)))
return json.loads(json.dumps(result, cls=CELJSONEncoder))
except Exception as e:
raise ExpressionError(
f"failed to evaluate CEL expression {expression!r}: {e}"
) from e
@staticmethod
def _compile_cel(
expression: str,
*,
source: str | None = None,
environment: Any | None = None,
) -> Any:
if environment is None:
from celpy import Environment
environment = Environment()
try:
return environment.compile(expression)
except Exception as e:
if source is None:
raise
raise ExpressionError(
f"invalid CEL expression at {source}: {expression!r}. "
f"Check the CEL syntax. Parser details: {e}"
) from e
@staticmethod
def _require_cel_source(value: str, *, source: str | None = None) -> str:
expression = value.strip()
if expression.startswith("${") and expression.endswith("}"):
expression = expression[2:-1].strip()
if expression:
return expression
if source is None:
raise ExpressionError("empty CEL expression")
raise ExpressionError(
f"empty CEL expression at {source}. Provide a CEL expression such as "
"state.topic or outputs.step_name."
)
@staticmethod
def _collect_root_identifiers(
tree: Any, local_roots: frozenset[str] = frozenset()
) -> set[str]:
"""Collect CEL root identifiers, excluding receiver macro local variables."""
data = getattr(tree, "data", None)
children = list(getattr(tree, "children", []) or [])
if data == "ident" and children:
name = str(children[0])
return set() if name in local_roots else {name}
if data == "ident_arg":
return Expression._collect_root_identifiers_from(
children[1:], local_roots=local_roots
)
if data == "member_dot_arg":
roots = (
Expression._collect_root_identifiers(children[0], local_roots)
if children
else set()
)
nested_locals = frozenset(
{*local_roots, *Expression._receiver_macro_local_roots(children)}
)
roots.update(
Expression._collect_root_identifiers_from(
children[2:], local_roots=nested_locals
)
)
return roots
return Expression._collect_root_identifiers_from(
children, local_roots=local_roots
)
@staticmethod
def _collect_root_identifiers_from(
trees: Iterable[Any], *, local_roots: frozenset[str]
) -> set[str]:
return set().union(
*(Expression._collect_root_identifiers(tree, local_roots) for tree in trees)
)
@staticmethod
def _receiver_macro_local_roots(children: list[Any]) -> set[str]:
if len(children) < 3 or str(children[1]) not in _CEL_MACROS_WITH_LOCAL_BINDINGS:
return set()
exprlist = children[2]
exprs = list(getattr(exprlist, "children", []) or [])
if exprs and (name := Expression._single_identifier_name(exprs[0])):
return {name}
return set()
@staticmethod
def _single_identifier_name(tree: Any) -> str | None:
data = getattr(tree, "data", None)
children = list(getattr(tree, "children", []) or [])
if data == "ident" and children:
return str(children[0])
if len(children) != 1:
return None
return Expression._single_identifier_name(children[0])

View File

@@ -12,7 +12,7 @@ from __future__ import annotations
import json
import logging
import re
from typing import Annotated, Any, Literal, TypeAlias
from typing import Annotated, Any, Literal, TypeAlias, cast
from pydantic import (
BaseModel,
@@ -27,16 +27,20 @@ from crewai.flow.conversational_definition import (
FlowConversationalDefinition,
FlowConversationalRouterDefinition,
)
from crewai.project.crew_definition import CrewDefinition
from crewai.flow.expressions import ExpressionData
from crewai.project.crew_definition import AgentDefinition, CrewDefinition
logger = logging.getLogger(__name__)
FlowDefinitionCondition = str | dict[str, Any]
_STEP_NAME_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
_BASE_CEL_ROOTS = frozenset({"outputs", "state"})
_EACH_STEP_CEL_ROOTS = frozenset({"item", "outputs", "state"})
__all__ = [
"FlowActionDefinition",
"FlowAgentActionDefinition",
"FlowAtomicActionDefinition",
"FlowCodeActionDefinition",
"FlowConfigDefinition",
@@ -353,10 +357,14 @@ class FlowCodeActionDefinition(BaseModel):
description="Import reference for the callable, formatted as module:qualname.",
examples=["my_project.flows:normalize_topic"],
)
with_: dict[str, Any] | None = Field(
with_: dict[str, ExpressionData] | None = Field(
default=None,
alias="with",
description="Keyword arguments passed to the callable after expression rendering.",
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}"}],
)
@@ -377,10 +385,14 @@ class FlowToolActionDefinition(BaseModel):
description="Import reference for a BaseTool class, formatted as module:qualname.",
examples=["my_project.tools:SearchTool"],
)
with_: dict[str, Any] | None = Field(
with_: dict[str, ExpressionData] | None = Field(
default=None,
alias="with",
description="Tool input arguments after expression rendering.",
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}],
)
@@ -424,6 +436,33 @@ class FlowCrewActionDefinition(BaseModel):
)
class FlowAgentActionDefinition(BaseModel):
"""A Flow method action that builds and kicks off a CrewAI agent."""
model_config = ConfigDict(
populate_by_name=True,
extra="forbid",
)
call: Literal["agent"] = Field(
description="Action discriminator. Use agent to run an inline Agent definition.",
examples=["agent"],
)
with_: AgentDefinition = Field(
alias="with",
description="Inline Agent definition to load and execute for this action.",
examples=[
{
"role": "Analyst",
"goal": "Answer user questions",
"backstory": "Precise and concise.",
"settings": {"llm": "openai/gpt-4o-mini"},
"input": "${state.question}",
}
],
)
class FlowExpressionActionDefinition(BaseModel):
"""A Flow method action that evaluates a CEL expression."""
@@ -470,6 +509,7 @@ FlowAtomicActionDefinition: TypeAlias = Annotated[
FlowCodeActionDefinition
| FlowToolActionDefinition
| FlowCrewActionDefinition
| FlowAgentActionDefinition
| FlowExpressionActionDefinition
| FlowScriptActionDefinition,
Field(discriminator="call"),
@@ -562,6 +602,7 @@ FlowActionDefinition: TypeAlias = (
FlowCodeActionDefinition
| FlowToolActionDefinition
| FlowCrewActionDefinition
| FlowAgentActionDefinition
| FlowExpressionActionDefinition
| FlowScriptActionDefinition
| FlowEachActionDefinition
@@ -696,6 +737,16 @@ class FlowDefinition(BaseModel):
_validate_step_name(method_name, field="Flow method names")
return self
@model_validator(mode="after")
def _validate_cel_expressions(self) -> FlowDefinition:
for method_name, method in self.methods.items():
_validate_action_cel(
method.do,
path=f"methods.{method_name}.do",
allowed_roots=_BASE_CEL_ROOTS,
)
return self
def to_dict(self, *, exclude_none: bool = True) -> dict[str, Any]:
"""Serialize the definition to a JSON/YAML-ready dictionary."""
return self.model_dump(by_alias=True, exclude_none=exclude_none, mode="json")
@@ -753,6 +804,69 @@ def _validate_step_list(steps: list[FlowEachStepDefinition], *, field: str) -> N
seen.add(name)
def _validate_action_cel(
action: FlowActionDefinition,
*,
path: str,
allowed_roots: frozenset[str],
) -> None:
from crewai.flow.expressions import Expression
if isinstance(action, FlowExpressionActionDefinition):
Expression(action.expr).validate_expression(
allowed_roots=allowed_roots, source=f"{path}.expr"
)
return
if isinstance(action, (FlowCodeActionDefinition, FlowToolActionDefinition)):
if action.with_ is not None:
Expression(action.with_).validate_template(
allowed_roots=allowed_roots, source=f"{path}.with"
)
return
if isinstance(action, FlowCrewActionDefinition):
Expression(cast(ExpressionData, action.with_.inputs)).validate_template(
allowed_roots=allowed_roots,
source=f"{path}.with.inputs",
)
return
if isinstance(action, FlowAgentActionDefinition):
Expression(cast(ExpressionData, action.with_.input)).validate_template(
allowed_roots=allowed_roots,
source=f"{path}.with.input",
)
return
if isinstance(action, FlowEachActionDefinition):
Expression(action.in_).validate_expression(
allowed_roots=_BASE_CEL_ROOTS,
source=f"{path}.in",
)
for index, step in enumerate(action.do):
step_path = f"{path}.do[{index}]"
if step.if_ is not None:
Expression(step.if_).validate_expression(
allowed_roots=_EACH_STEP_CEL_ROOTS,
source=f"{step_path}.if",
)
_validate_action_cel(
step.action,
path=f"{step_path}.action",
allowed_roots=_EACH_STEP_CEL_ROOTS,
)
return
if isinstance(action, FlowScriptActionDefinition):
return
raise TypeError(
f"no CEL validation defined for action type {type(action).__name__} at "
f"{path}; add a branch to _validate_action_cel for it."
)
def log_flow_definition_issues(definition: FlowDefinition) -> None:
for method_name, method in definition.methods.items():
path = f"methods.{method_name}"

View File

@@ -10,8 +10,10 @@ import inspect
import os
from typing import TYPE_CHECKING, Any, Protocol, cast
from crewai.flow.expressions import Expression, ExpressionData
from crewai.flow.flow_definition import (
FlowActionDefinition,
FlowAgentActionDefinition,
FlowCodeActionDefinition,
FlowCrewActionDefinition,
FlowEachActionDefinition,
@@ -20,7 +22,6 @@ from crewai.flow.flow_definition import (
FlowScriptActionDefinition,
FlowToolActionDefinition,
)
from crewai.flow.runtime._expressions import evaluate_expression, render_with_block
from crewai.flow.runtime._outputs import outputs_by_name
from crewai.flow.runtime._refs import InvalidRefError, resolve_ref
@@ -67,9 +68,9 @@ class CodeAction:
if self.definition.with_ is None:
return self.handler(*args, **kwargs)
return self.handler(
**render_with_block(
self.flow, self.definition.with_, local_context=local_context
)
**Expression.from_flow(
self.definition.with_, self.flow, local_context=local_context
).render_template()
)
def _resolve_handler(self) -> Callable[..., Any]:
@@ -95,7 +96,9 @@ class ToolAction:
def run(self, *_args: Any, **kwargs: Any) -> Any:
local_context = _pop_local_context(kwargs)
return self.tool.run(
**render_with_block(self.flow, self.kwargs, local_context=local_context)
**Expression.from_flow(
self.kwargs, self.flow, local_context=local_context
).render_template()
)
def _build_tool(self) -> Any:
@@ -129,13 +132,44 @@ class CrewAction:
local_context = _pop_local_context(kwargs)
crew_definition = self.definition.with_
inputs = render_with_block(
self.flow, crew_definition.inputs, local_context=local_context
)
inputs = Expression.from_flow(
cast(ExpressionData, crew_definition.inputs),
self.flow,
local_context=local_context,
).render_template()
crew, _ = load_crew_from_definition(crew_definition, source="crew action")
return await crew.kickoff_async(inputs=inputs)
class AgentAction:
definition_type = FlowAgentActionDefinition
def __init__(self, flow: Flow[Any], definition: FlowAgentActionDefinition) -> None:
self.flow = flow
self.definition = definition
async def run(self, *_args: Any, **kwargs: Any) -> Any:
from crewai.project.json_loader import load_agent_from_definition
local_context = _pop_local_context(kwargs)
rendered_input = Expression.from_flow(
cast(ExpressionData, self.definition.with_.input),
self.flow,
local_context=local_context,
).render_template()
if not isinstance(rendered_input, str):
raise ValueError("agent input must render to a string")
agent, response_format = load_agent_from_definition(
self.definition.with_,
source="agent action",
)
return await agent.kickoff_async(
rendered_input,
response_format=response_format,
)
class ExpressionAction:
definition_type = FlowExpressionActionDefinition
@@ -147,9 +181,9 @@ class ExpressionAction:
def run(self, *_args: Any, **kwargs: Any) -> Any:
local_context = _pop_local_context(kwargs)
return evaluate_expression(
self.flow, self.definition.expr, local_context=local_context
)
return Expression.from_flow(
self.definition.expr, self.flow, local_context=local_context
).evaluate()
class ScriptAction:
@@ -225,7 +259,7 @@ class EachAction:
]
async def run(self, *_args: Any, **_kwargs: Any) -> list[Any]:
items = evaluate_expression(self.flow, self.definition.in_)
items = Expression.from_flow(self.definition.in_, self.flow).evaluate()
if not isinstance(items, list):
raise ValueError("each.in must evaluate to an array")
@@ -248,7 +282,9 @@ class EachAction:
return results
def _condition_matches(self, condition: str, local_context: LocalContext) -> bool:
result = evaluate_expression(self.flow, condition, local_context=local_context)
result = Expression.from_flow(
condition, self.flow, local_context=local_context
).evaluate()
if not isinstance(result, bool):
raise ValueError("if expression must evaluate to a boolean")
return result
@@ -278,6 +314,7 @@ _ACTION_TYPES: tuple[_ActionType, ...] = (
EachAction,
CodeAction,
ToolAction,
AgentAction,
CrewAction,
ExpressionAction,
ScriptAction,

View File

@@ -1,146 +0,0 @@
"""Runtime expression support for FlowDefinition CEL expressions."""
from __future__ import annotations
from itertools import pairwise
import json
import re
from typing import TYPE_CHECKING, Any, cast
from crewai.flow.runtime._outputs import outputs_by_name
from crewai.utilities.serialization import to_serializable
if TYPE_CHECKING:
from crewai.flow.runtime import Flow
_EXPRESSION_PATTERN = re.compile(r"\$\{([^{}]*)\}")
__all__ = ["FlowExpressionError", "evaluate_expression", "render_with_block"]
class FlowExpressionError(ValueError):
"""A FlowDefinition expression failed to parse or evaluate."""
def render_with_block(
flow: Flow[Any], value: Any, local_context: dict[str, Any] | None = None
) -> Any:
"""Render CEL expressions inside a FlowDefinition ``with:`` payload."""
context = _expression_context(flow, local_context=local_context)
return _render_value(value, context)
def evaluate_expression(
flow: Flow[Any], expression: str, local_context: dict[str, Any] | None = None
) -> Any:
"""Evaluate a FlowDefinition CEL expression against runtime context."""
expression = expression.strip()
if not expression:
raise FlowExpressionError("empty CEL expression")
return _eval_cel(expression, _expression_context(flow, local_context=local_context))
def _expression_context(
flow: Flow[Any], local_context: dict[str, Any] | None = None
) -> dict[str, Any]:
local_outputs = local_context.get("outputs") if local_context else None
outputs = outputs_by_name(
flow._method_outputs,
local_outputs=local_outputs,
serialize=True,
)
context: dict[str, Any] = {
"state": flow._copy_and_serialize_state(),
"outputs": outputs,
}
if local_context:
local_values = {
key: to_serializable(value, max_depth=0)
for key, value in local_context.items()
if key not in {"outputs", "state"}
}
context.update(local_values)
return context
def _render_value(value: Any, context: dict[str, Any]) -> Any:
if isinstance(value, str):
return _render_string(value, context)
if isinstance(value, dict):
return {key: _render_value(item, context) for key, item in value.items()}
if isinstance(value, list):
return [_render_value(item, context) for item in value]
return value
def _render_string(value: str, context: dict[str, Any]) -> Any:
matches = list(_EXPRESSION_PATTERN.finditer(value))
if not matches:
_raise_for_invalid_interpolation(value)
return value
_raise_for_literal_braces(value[: matches[0].start()])
for previous, current in pairwise(matches):
_raise_for_literal_braces(value[previous.end() : current.start()])
_raise_for_literal_braces(value[matches[-1].end() :])
if len(matches) == 1 and matches[0].span() == (0, len(value)):
expression = matches[0].group(1).strip()
if not expression:
raise FlowExpressionError("empty CEL expression in with block")
return _eval_cel(expression, context)
rendered: list[str] = []
position = 0
for match in matches:
start, end = match.span()
literal = value[position:start]
rendered.append(literal)
expression = match.group(1).strip()
if not expression:
raise FlowExpressionError("empty CEL expression in with block")
result = _eval_cel(expression, context)
rendered.append(result if isinstance(result, str) else json.dumps(result))
position = end
literal = value[position:]
rendered.append(literal)
return "".join(rendered)
def _raise_for_invalid_interpolation(value: str) -> None:
if "${" not in value:
return
raise FlowExpressionError(
"invalid CEL interpolation in with block: expressions must be enclosed "
"as ${...} and cannot contain braces"
)
def _raise_for_literal_braces(value: str) -> None:
if "{" not in value and "}" not in value:
return
raise FlowExpressionError(
"invalid CEL interpolation in with block: expressions must be enclosed "
"as ${...} and cannot contain braces"
)
def _eval_cel(expression: str, context: dict[str, Any]) -> Any:
try:
from celpy import Environment
from celpy.adapter import CELJSONEncoder, json_to_cel
from celpy.evaluation import Context
environment = Environment()
program = environment.program(environment.compile(expression))
result = program.evaluate(cast(Context, json_to_cel(context)))
return json.loads(json.dumps(result, cls=CELJSONEncoder))
except Exception as e:
raise FlowExpressionError(
f"failed to evaluate CEL expression {expression!r}: {e}"
) from e

View File

@@ -15,16 +15,22 @@ from crewai.project.annotations import (
)
from crewai.project.crew_base import CrewBase
from crewai.project.crew_definition import (
AgentDefinition,
CrewAgentDefinition,
CrewDefinition,
CrewTaskDefinition,
PythonReferenceDefinition,
)
from crewai.project.crew_loader import load_crew, load_crew_and_kickoff
from crewai.project.json_loader import load_agent, strip_jsonc_comments
from crewai.project.json_loader import (
load_agent,
load_agent_from_definition,
strip_jsonc_comments,
)
__all__ = [
"AgentDefinition",
"CrewAgentDefinition",
"CrewBase",
"CrewDefinition",
@@ -38,6 +44,7 @@ __all__ = [
"crew",
"llm",
"load_agent",
"load_agent_from_definition",
"load_crew",
"load_crew_and_kickoff",
"output_json",

View File

@@ -8,6 +8,7 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator, model_valida
__all__ = [
"AgentDefinition",
"CrewAgentDefinition",
"CrewDefinition",
"CrewTaskDefinition",
@@ -53,6 +54,20 @@ class CrewAgentDefinition(BaseModel):
return value or {}
class AgentDefinition(CrewAgentDefinition):
"""Inline agent definition used by a Flow agent action."""
input: str
response_format: PythonReferenceDefinition | None = None
@field_validator("input", mode="before")
@classmethod
def _validate_input(cls, value: Any) -> Any:
if not isinstance(value, str):
raise ValueError("agent.input must be a string")
return value
class CrewTaskDefinition(BaseModel):
"""Task definition used by a crew definition."""

View File

@@ -207,19 +207,18 @@ def load_jsonc_file(source: str | Path) -> Any:
return parse_jsonc(path.read_text(encoding="utf-8"), source=path)
def load_agent(source: str | Path) -> Any:
"""Load an existing ``Agent`` from a ``.json`` / ``.jsonc`` definition file."""
path = Path(source)
defn = _expect_object(load_jsonc_file(path), path)
root = path.parent.parent if path.parent.name == "agents" else path.parent
def _instantiate_agent_from_data(
defn: dict[str, Any], source_label: str, root: Path
) -> Any:
"""Resolve the agent class and kwargs from definition data and instantiate it."""
agent_class = _agent_class_from_definition(
defn,
f"{path}: type",
f"{source_label}: type",
project_root=root,
)
agent_kwargs = _agent_kwargs_from_definition(
defn,
path,
source_label,
agent_class=agent_class,
project_root=root,
)
@@ -227,9 +226,50 @@ def load_agent(source: str | Path) -> Any:
try:
return agent_class(**agent_kwargs)
except ValidationError as exc:
raise JSONProjectError(_format_validation_error(path, exc)) from exc
raise JSONProjectError(_format_validation_error(source_label, exc)) from exc
except Exception as exc:
raise JSONProjectError(f"{path}: failed to load agent: {exc}") from exc
raise JSONProjectError(f"{source_label}: failed to load agent: {exc}") from exc
def load_agent(source: str | Path) -> Any:
"""Load an existing ``Agent`` from a ``.json`` / ``.jsonc`` definition file."""
path = Path(source)
defn = _expect_object(load_jsonc_file(path), path)
root = path.parent.parent if path.parent.name == "agents" else path.parent
return _instantiate_agent_from_data(defn, str(path), root)
def load_agent_from_definition(
definition: dict[str, Any] | Any,
*,
source: str | Path = "<inline agent>",
project_root: str | Path | None = None,
) -> tuple[Any, type[BaseModel] | None]:
"""Load an ``Agent`` and optional kickoff response model from an inline definition."""
from crewai.project.crew_definition import AgentDefinition
root = Path(project_root) if project_root is not None else Path.cwd()
source_label = str(source)
agent_definition = (
definition
if isinstance(definition, AgentDefinition)
else AgentDefinition.model_validate(definition)
)
definition_data = agent_definition.model_dump(mode="python", exclude_none=True)
response_format_ref = definition_data.pop("response_format", None)
definition_data.pop("input", None)
agent = _instantiate_agent_from_data(definition_data, source_label, root)
response_format = None
if response_format_ref is not None:
response_format = _resolve_model_class(
response_format_ref,
f"{source_label}: response_format",
root,
)
return agent, response_format
def validate_crew_project(

View File

@@ -7,6 +7,7 @@ from pathlib import Path
import sys
import pytest
from pydantic import BaseModel
from crewai.llms.base_llm import BaseLLM
from crewai.project.json_loader import (
@@ -14,6 +15,7 @@ from crewai.project.json_loader import (
_looks_like_windows_absolute_path,
find_json_project_file,
load_agent,
load_agent_from_definition,
strip_jsonc_comments,
)
@@ -358,6 +360,30 @@ class TestLoadAgent:
load_agent(Path("/nonexistent/agent.json"))
class TestLoadAgentFromDefinition:
def test_resolves_response_format_from_project_module(self, tmp_path: Path):
(tmp_path / "models.py").write_text(
"from pydantic import BaseModel\n"
"class AnswerModel(BaseModel):\n"
" answer: str\n"
)
_, response_format = load_agent_from_definition(
{
"role": "Analyst",
"goal": "Analyze data",
"backstory": "Data expert.",
"input": "Summarize this",
"response_format": {"python": "models.AnswerModel"},
},
source="agent action",
project_root=tmp_path,
)
assert issubclass(response_format, BaseModel)
assert response_format.__name__ == "AnswerModel"
class TestResolveTools:
def test_unknown_tool_raises_with_guidance(self):
from crewai.project.json_loader import JSONProjectError, _resolve_tools

View File

@@ -631,7 +631,7 @@ class TestLegacyMethodOutputsRestore:
assert restored.method_outputs == ["first", "second"]
def test_restore_legacy_outputs_evaluates_expressions(self) -> None:
from crewai.flow.runtime._expressions import _expression_context
from crewai.flow.expressions import Expression
flow = Flow()
flow._method_outputs = ["legacy"]
@@ -642,7 +642,7 @@ class TestLegacyMethodOutputsRestore:
cfg = CheckpointConfig(restore_from=loc)
restored = Flow.from_checkpoint(cfg)
context = _expression_context(restored)
context = Expression._flow_context(restored)
assert context["outputs"] == {"": "legacy"}
def test_raw_legacy_outputs_property_remains_readable(self) -> None:

View File

@@ -37,6 +37,7 @@ def test_flow_public_exports_are_explicit():
}
assert set(flow_definition.__all__) == {
"FlowActionDefinition",
"FlowAgentActionDefinition",
"FlowAtomicActionDefinition",
"FlowCodeActionDefinition",
"FlowConfigDefinition",
@@ -80,6 +81,10 @@ def test_flow_definition_json_schema_carries_reference_descriptions():
assert "not interpolated" in script_properties["code"]["description"]
assert "not sandboxed" in script_properties["code"]["description"]
agent_properties = defs["FlowAgentActionDefinition"]["properties"]
assert "Inline Agent definition" in agent_properties["with"]["description"]
assert "run an inline Agent" in agent_properties["call"]["description"]
state_schema = next(
branch
for branch in schema["properties"]["state"]["anyOf"]
@@ -122,6 +127,7 @@ def test_flow_definition_json_schema_carries_field_examples_only():
"FlowDefinition",
"FlowCodeActionDefinition",
"FlowToolActionDefinition",
"FlowAgentActionDefinition",
"FlowCrewActionDefinition",
"FlowExpressionActionDefinition",
"FlowScriptActionDefinition",
@@ -157,6 +163,10 @@ def test_flow_definition_json_schema_carries_field_examples_only():
]
assert action_properties["with"]["examples"] == [{"topic": "${state.topic}"}]
agent_properties = defs["FlowAgentActionDefinition"]["properties"]
assert agent_properties["call"]["examples"] == ["agent"]
assert agent_properties["with"]["examples"][0]["input"] == "${state.question}"
each_properties = defs["FlowEachActionDefinition"]["properties"]
assert each_properties["in"]["examples"] == ["state.rows"]
assert each_properties["do"]["examples"][0][0]["name"] == "clean"

View File

@@ -644,7 +644,7 @@ methods:
assert flow.kickoff(inputs={"topic": "ai"}) == "found:ai agents"
def test_tool_action_rejects_braces_in_embedded_cel_input():
def test_tool_action_treats_embedded_cel_marker_as_literal():
definition = FlowDefinition.from_dict(
{
"schema": "crewai.flow/v1",
@@ -660,16 +660,62 @@ def test_tool_action_rejects_braces_in_embedded_cel_input():
"prefix": "${'p}x'}",
},
},
}
},
},
}
)
with pytest.raises(ValueError, match="cannot contain braces"):
Flow.from_definition(definition).kickoff()
assert Flow.from_definition(definition).kickoff() == "p}x:wrapped ${'a}b'} value"
def test_tool_action_rejects_braces_in_full_cel_input():
def test_tool_action_treats_marker_with_trailing_text_as_literal():
definition = FlowDefinition.from_dict(
{
"schema": "crewai.flow/v1",
"name": "ToolFlow",
"methods": {
"search": {
"start": True,
"do": {
"call": "tool",
"ref": f"{__name__}:StaticSearchTool",
"with": {
"search_query": "${state.topic} extra",
"prefix": "p",
},
},
},
},
}
)
assert Flow.from_definition(definition).kickoff() == "p:${state.topic} extra"
def test_tool_action_rejects_adjacent_markers_as_invalid_cel():
with pytest.raises(ValidationError, match="invalid CEL expression"):
FlowDefinition.from_dict(
{
"schema": "crewai.flow/v1",
"name": "ToolFlow",
"methods": {
"search": {
"start": True,
"do": {
"call": "tool",
"ref": f"{__name__}:StaticSearchTool",
"with": {
"search_query": "${'a'}${'b'}",
"prefix": "p",
},
},
},
},
}
)
def test_tool_action_accepts_braces_in_full_cel_marker():
definition = FlowDefinition.from_dict(
{
"schema": "crewai.flow/v1",
@@ -682,16 +728,15 @@ def test_tool_action_rejects_braces_in_full_cel_input():
"ref": f"{__name__}:StaticSearchTool",
"with": {
"search_query": "${{'query': 'ai agents'}.query}",
"prefix": "found",
"prefix": "${'p}x'}",
},
},
}
},
},
}
)
with pytest.raises(ValueError, match="cannot contain braces"):
Flow.from_definition(definition).kickoff()
assert Flow.from_definition(definition).kickoff() == "p}x:ai agents"
def test_tool_action_renders_latest_output_by_method_name():
@@ -766,6 +811,166 @@ methods:
)
def test_agent_action_runs_inline_yaml_definition(monkeypatch: pytest.MonkeyPatch):
from crewai import Agent
async def fake_kickoff_async(
self: Agent, messages: str, **_kwargs: Any
) -> dict[str, Any]:
return {"agent": self.role, "input": messages}
monkeypatch.setattr(Agent, "kickoff_async", fake_kickoff_async)
yaml_str = """
schema: crewai.flow/v1
name: AgentFlow
methods:
answer:
do:
call: agent
with:
role: Analyst
goal: Answer questions
backstory: Knows things.
input: "${state.question}"
start: true
"""
flow = Flow.from_definition(FlowDefinition.from_yaml(yaml_str))
assert flow.kickoff(inputs={"question": "What is CrewAI?"}) == {
"agent": "Analyst",
"input": "What is CrewAI?",
}
def test_agent_action_runs_inside_each(monkeypatch: pytest.MonkeyPatch):
from crewai import Agent
async def fake_kickoff_async(
self: Agent, messages: str, **_kwargs: Any
) -> str:
return f"{self.role}:{messages}"
monkeypatch.setattr(Agent, "kickoff_async", fake_kickoff_async)
yaml_str = """
schema: crewai.flow/v1
name: AgentEachFlow
methods:
answer_each:
do:
call: each
in: state.questions
do:
- name: answer
action:
call: agent
with:
role: Analyst
goal: Answer questions
backstory: Knows things.
input: "${item}"
start: true
"""
flow = Flow.from_definition(FlowDefinition.from_yaml(yaml_str))
assert flow.kickoff(inputs={"questions": ["one", "two"]}) == [
"Analyst:one",
"Analyst:two",
]
def test_agent_action_round_trips_with_inline_definition():
definition = FlowDefinition.from_dict(
{
"schema": "crewai.flow/v1",
"name": "AgentFlow",
"methods": {
"answer": {
"start": True,
"do": {
"call": "agent",
"with": {
"role": "Analyst",
"goal": "Answer questions",
"backstory": "Knows things.",
"settings": {"verbose": True},
"input": "${state.question}",
},
},
}
},
}
)
round_trip = FlowDefinition.from_yaml(definition.to_yaml())
action = round_trip.to_dict()["methods"]["answer"]["do"]
assert action["call"] == "agent"
assert action["with"]["role"] == "Analyst"
assert action["with"]["input"] == "${state.question}"
assert action["with"]["settings"] == {"verbose": True}
def test_agent_action_json_schema_describes_inline_agent_definitions():
schema_defs = FlowDefinition.json_schema()["$defs"]
assert set(schema_defs["AgentDefinition"]["properties"]) >= {
"role",
"goal",
"backstory",
"settings",
"input",
"response_format",
}
def test_agent_action_rejects_non_string_input_in_definition():
with pytest.raises(ValidationError, match="agent.input must be a string"):
FlowDefinition.from_dict(
{
"schema": "crewai.flow/v1",
"name": "AgentFlow",
"methods": {
"answer": {
"start": True,
"do": {
"call": "agent",
"with": {
"role": "Analyst",
"goal": "Answer questions",
"backstory": "Knows things.",
"input": 123,
},
},
}
},
}
)
def test_agent_action_reports_invalid_cel_expression():
yaml_str = """
schema: crewai.flow/v1
name: AgentFlow
methods:
answer:
do:
call: agent
with:
role: Analyst
goal: Answer questions
backstory: Knows things.
input: "${state.}"
start: true
"""
with pytest.raises(ValidationError, match="invalid CEL expression"):
FlowDefinition.from_yaml(yaml_str)
def test_crew_action_runs_inline_yaml_definition(monkeypatch: pytest.MonkeyPatch):
from crewai import Crew
@@ -1026,10 +1231,8 @@ methods:
start: true
"""
flow = Flow.from_definition(FlowDefinition.from_yaml(yaml_str))
with pytest.raises(ValueError, match="failed to evaluate CEL expression"):
flow.kickoff()
with pytest.raises(ValidationError, match="invalid CEL expression"):
FlowDefinition.from_yaml(yaml_str)
def test_code_action_renders_keyword_inputs():
@@ -1407,6 +1610,33 @@ methods:
) == ["kept:a", "skipped:b"]
def test_each_action_accepts_expression_markers_in_explicit_cel_fields():
yaml_str = """
schema: crewai.flow/v1
name: EachIfFlow
methods:
process_rows:
do:
call: each
in: "${state.rows}"
do:
- name: kind
action:
call: expression
expr: "${item.kind}"
- name: kept
if: "${outputs.kind == 'keep'}"
action:
call: expression
expr: "${item.value}"
start: true
"""
flow = Flow.from_definition(FlowDefinition.from_yaml(yaml_str))
assert flow.kickoff(inputs={"rows": [{"kind": "keep", "value": "a"}]}) == ["a"]
def test_each_action_skipped_if_keeps_previous_output():
yaml_str = """
schema: crewai.flow/v1
@@ -1690,8 +1920,28 @@ def test_expression_action_round_trips():
assert Flow.from_definition(definition).kickoff(inputs={"score": 90}) == "qualified"
def test_explicit_cel_fields_accept_expression_markers():
definition = FlowDefinition.from_dict(
{
"schema": "crewai.flow/v1",
"name": "ExpressionFlow",
"methods": {
"classify": {
"start": True,
"do": {
"call": "expression",
"expr": "${state.score >= 80 ? 'qualified' : 'nurture'}",
},
}
},
}
)
assert Flow.from_definition(definition).kickoff(inputs={"score": 90}) == "qualified"
def test_expression_local_context_recurses_into_dataclass_values():
from crewai.flow.runtime._expressions import evaluate_expression
from crewai.flow.expressions import Expression
class Payload(BaseModel):
name: str
@@ -1701,15 +1951,37 @@ def test_expression_local_context_recurses_into_dataclass_values():
payload: Payload
assert (
evaluate_expression(
Flow(),
Expression.from_flow(
"item.payload.name",
Flow(),
local_context={"item": Row(payload=Payload(name="qualified"))},
)
).evaluate()
== "qualified"
)
def test_expression_empty_context_overrides_stored_context():
from crewai.flow.expressions import Expression, ExpressionError
expression = Expression("state.score", context={"state": {"score": 90}})
assert expression.evaluate() == 90
with pytest.raises(ExpressionError):
expression.evaluate({})
def test_expression_template_empty_context_overrides_stored_context():
from crewai.flow.expressions import Expression, ExpressionError
expression = Expression(
{"score": "${state.score}"}, context={"state": {"score": 90}}
)
assert expression.render_template() == {"score": 90}
with pytest.raises(ExpressionError):
expression.render_template({})
def test_expression_action_can_route_like_if_else():
yaml_str = f"""
schema: crewai.flow/v1
@@ -1761,10 +2033,24 @@ methods:
start: true
"""
flow = Flow.from_definition(FlowDefinition.from_yaml(yaml_str))
with pytest.raises(ValidationError, match="invalid CEL expression"):
FlowDefinition.from_yaml(yaml_str)
with pytest.raises(ValueError, match="failed to evaluate CEL expression"):
flow.kickoff()
def test_expression_action_rejects_unknown_cel_root():
yaml_str = """
schema: crewai.flow/v1
name: ExpressionFlow
methods:
classify:
do:
call: expression
expr: "score >= 80"
start: true
"""
with pytest.raises(ValidationError, match="unknown CEL root"):
FlowDefinition.from_yaml(yaml_str)
def test_tool_action_requires_module_qualname_ref():

View File

@@ -1,3 +1,3 @@
"""CrewAI development tools."""
__version__ = "1.14.8a1"
__version__ = "1.14.8a2"