kaxil opened a new pull request, #73447:
URL: https://github.com/apache/airflow/pull/73447
## Summary
OpenAI shipped a Managed Agents API in `openai` SDK 3.13
(`client.beta.agents`, header `OpenAI-Beta: agents=v1`): persisted agents,
sessions with turns, hosted or self-hosted environments, vaults for MCP
credentials, and subagents. It is the counterpart of the Anthropic Managed
Agents surface this provider's sibling already integrates.
This adds `OpenAIAgentSessionOperator`, which creates a fresh session with
an initial message, waits for the first turn to finish, and returns the session
ID. `deferrable=True` hands the wait to `OpenAIAgentSessionTrigger`. When XCom
pushing is on it also records `session_id`, `turn_id` and the turn's token
`usage`. Five `OpenAIHook` helpers back it (create agent, create session, get
session, cancel session, poll session).
Usage from the example DAG:
```python
OpenAIAgentSessionOperator(
task_id="run_agent",
input="Explain how Airflow retries affect a task that calls an external
API.",
environment={"type": "none"},
session_kwargs={"agent": {"model": "gpt-6-astra", "instructions":
"..."}},
deferrable=True,
)
```
Pass `agent_id` instead of an inline `agent` to run a saved agent, and
`vault_ids` or an environment template reference through `session_kwargs` or
`environment`.
## Design rationale
**The SDK floor stays at `openai>=2.37.0`; the Agents API is
feature-detected.** Raising the floor to 3.13 fails the provider dependency
resolution because `llama-index-llms-openai` (pulled in through the Common AI
provider's LlamaIndex extra) still pins `openai<3`. The hook checks for
`client.beta.agents` at first use and raises an error naming the required
version, so every other OpenAI operator keeps working on older SDKs and the new
test module skips itself when the installed SDK predates the API.
**Each task attempt owns one fresh session and waits only for its first
turn.** A retry creates a new session. Reusing the previous one would mean
picking the right turn out of a session with history the operator did not
write. The poll lists turns in ascending order with `limit=1` and reads that
turn's status. An idle session with no turn yet counts as still waiting,
because the initial input can sit queued before a turn object exists, and
treating idle as done would return success for work that has not started.
**Client-side function tools fail the task.** The worker is deferred or
asleep while the agent runs, and a `function_call` required action would need
Airflow to run arbitrary tool code mid-wait and post results back. The error
message points at service-side tools, MCP servers or hosted environments.
Self-hosted environments are supported but their worker (`codex exec-server`)
has to be run independently; an `environment_connection` required action is
treated as still waiting.
**Failure paths cancel the active turn but keep the session.** Timeout,
polling failure, a bad trigger event, a failed turn and a killed task all send
`agent.session.input.cancel`, so a runaway turn stops consuming quota while the
items and artifacts remain retrievable by session ID. Up to three consecutive
polling errors are retried before the session is cancelled, the same tolerance
the Anthropic operators use. A shorter `execution_timeout` caps the deferral
timeout.
The deferrable path was run against the live API: session creation,
deferral, trigger polling and cancellation of a failed turn all behaved as
described. A mock-transport test against the real SDK covers the wire format
for the success path, including the beta header, request payloads, turn listing
parameters and the cancel event.
## Gotchas
* Managed Agents needs `openai>=3.13.0` on workers and triggerers. Libraries
pinning `openai<3`, including current LlamaIndex OpenAI integrations, cannot
share that environment.
* Cancellation of a killed *deferred* task relies on trigger `on_kill`
cleanup, which needs Airflow 3.3 or newer. On older versions a killed deferred
task leaves the turn running until it finishes or times out server-side.
* Usage pushed to XCom is for the current attempt only and carries
`try_number`; it is not cumulative across retries.
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information. Note: commit author/co-author name and email in commits
become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal
([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
