mirror of
https://github.com/agno-agi/agno.git
synced 2026-10-12 02:10:54 +00:00
feat: add support for codex harness external agent (#10901)
## Summary Adds `CodexAgent`, an external-agent adapter for OpenAI Codex, alongside the existing Claude Agent SDK, LangGraph, DSPy and Antigravity adapters. Codex agents can now be served through AgentOS or used standalone via `.run()` / `.print_response()`. The adapter wraps the official [Codex Python SDK](https://github.com/openai/codex/tree/main/sdk/python) (`pip install openai-codex`), which bundles the Codex CLI and runs it locally as an app-server subprocess. Codex runs the full agent loop itself (shell commands, file edits, web search, MCP tools); the adapter translates its notifications into Agno run events and maps each Agno session to one Codex thread. **Adapter** (`libs/agno/agno/agents/codex/`) - Subclasses `BaseExternalAgent`, implementing `_arun_adapter` and `_arun_adapter_stream`. - Options map onto the SDK: `model`, `instructions` (developer instructions), `base_instructions`, `sandbox` (`read-only` / `workspace-write` / `full-access`), `approval_mode` (`auto_review` / `deny_all`), `reasoning_effort`, `cwd`, `ephemeral`, `output_schema`, `config` (Codex config overrides, e.g. `mcp_servers`), `codex_bin`, `env`, plus `thread_kwargs` and `turn_kwargs` escape hatches. - Streaming translation: `item/agentMessage/delta` -> `RunContentEvent`; `commandExecution`, `fileChange`, `mcpToolCall`, `webSearch` and `dynamicToolCall` items -> `ToolCallStartedEvent` / `ToolCallCompletedEvent` (tool names `shell`, `apply_patch`, `mcp__<server>__<tool>`, `web_search`); reasoning deltas -> `reasoning_content`; a failed `turn/completed` or non-retryable `error` -> `RunErrorEvent`. - Sessions: the Codex thread id is stored in the Agno session's `session_data["codex_thread_id"]` when a `db` is configured (survives restarts and new app-server processes), with an in-memory fallback otherwise. If a thread cannot be resumed (`ephemeral=True`, rollout files removed) the adapter starts a fresh thread and prepends the persisted chat history to the prompt. - One `AsyncCodex` client per run keeps the client bound to the current event loop, so the sync `run()` path works unchanged. - `login_api_key` / `alogin_api_key` helpers for environments where the `codex` CLI is not on PATH. **Cookbooks** (`cookbook/frameworks/codex/`, plus `00_quickstart/codex_agent.py`) - `codex_basic.py`, `codex_tools.py`, `codex_session.py`, `codex_mcp_tools.py`, `codex_structured_output.py`, `codex_agentos.py`, `codex_session_agentos.py`, with `README.md` and `TEST_LOG.md`. - Quickstart README updated to list Codex. **Packaging** - `openai-codex` added to the `demo` extra; `openai_codex.*` added to the mypy ignore list. **Docs** are in a separate agno-docs-v2 change: new `agent-os/multi-framework/codex.mdx`, overview table/card and nav entry. ## Type of change - [ ] Bug fix - [x] New feature - [ ] Breaking change - [ ] Improvement - [ ] Model update - [ ] Other: --- ## Checklist - [x] Code complies with style guidelines - [x] Ran format/validation scripts (`./scripts/format.sh` and `./scripts/validate.sh`) - [x] Self-review completed - [x] Documentation updated (comments, docstrings) - [x] Examples and guides: Relevant cookbook examples have been included or updated (if applicable) - [x] Tested in clean environment - [x] Tests added/updated (if applicable) ### Duplicate and AI-Generated PR Check - [x] I have searched existing [open pull requests](https://github.com/agno-agi/agno/pulls) and confirmed that no other PR already addresses this issue - [ ] If a similar PR exists, I have explained below why this PR is a better approach - [x] Check if this PR was entirely AI-generated (by Copilot, Claude Code, Cursor, etc.) --- ## Additional Notes **Testing** - 16 new unit tests in `libs/agno/tests/unit/agents/test_codex.py` run against a fake `openai_codex` module (option mapping, notification translation, session/thread persistence, resume fallback, ephemeral handling). No live calls in CI. - All cookbooks were run live against openai-codex 0.161.0 (bundled Codex CLI 0.161.0) with `gpt-5.6-luna`; results are recorded in `cookbook/frameworks/codex/TEST_LOG.md`. Verified: thread resume across turns and across separate app-server processes, MCP tool calls rendering in the Tool Calls panel, structured output parsing as JSON, SSE streaming and non-streaming runs through AgentOS, and session recall across requests. - `ruff` clean; `mypy` reports no errors in the new code (the remaining errors on `main` are pre-existing in unrelated files). **Things to know** - The Codex Python SDK requires Python 3.10+. The adapter imports it lazily, so Agno on 3.9 is unaffected. - Authentication: `codex login` (ChatGPT) or `CODEX_API_KEY` / `OPENAI_API_KEY`. There is no interactive approval prompt in this context, so `approval_mode="deny_all"` is the safe choice for unattended servers. - `ephemeral=True` threads cannot be resumed by Codex, so multi-turn context falls back to history injection in that mode. - Same scope as the other external adapters: registration, streaming, sessions and tool visibility. Background runs, durable queue and workflow step replay are not wired up for external agents. 🤖 Generated with [Claude Code](https://claude.com/claude-code)
This commit is contained in:
Kaustubh
@@ -1,6 +1,6 @@
|
||||
# Frameworks Quickstart: Agno, Claude Code, LangGraph, DSPy
|
||||
# Frameworks Quickstart: Agno, Claude Code, Codex, LangGraph, DSPy
|
||||
|
||||
AgentOS supports agents built with **Agno**, the **Claude Agent SDK** (Claude Code), **LangGraph**, and **DSPy** — served through one runtime, one API, and one UI.
|
||||
AgentOS supports agents built with **Agno**, the **Claude Agent SDK** (Claude Code), **OpenAI Codex**, **LangGraph**, and **DSPy** — served through one runtime, one API, and one UI.
|
||||
|
||||
This quickstart shows each framework on its own and all together.
|
||||
|
||||
@@ -10,6 +10,7 @@ This quickstart shows each framework on its own and all together.
|
||||
|:-----|:--------------|
|
||||
| `agno_agent.py` | Native Agno agent with the `Workspace` tool (read/edit/search/shell, with confirmation gates) |
|
||||
| `claude_agent.py` | Claude Code agent via the Claude Agent SDK, with `Read`, `Edit`, `Bash` tools |
|
||||
| `codex_agent.py` | OpenAI Codex agent via the Codex Python SDK, in a `workspace-write` sandbox |
|
||||
| `langgraph_agent.py` | A LangGraph chatbot wrapped for AgentOS |
|
||||
| `dspy_agent.py` | A DSPy `ChainOfThought` program wrapped for AgentOS |
|
||||
| `multi_framework_agentos.py` | Agno + Claude Code in one AgentOS — the example from the launch blog post |
|
||||
@@ -65,6 +66,9 @@ Install the dependencies for the framework(s) you want:
|
||||
# Claude Agent SDK
|
||||
pip install claude-agent-sdk
|
||||
|
||||
# Codex (bundles the Codex CLI; then `codex login` or set CODEX_API_KEY)
|
||||
pip install openai-codex
|
||||
|
||||
# LangGraph
|
||||
pip install langgraph langchain-openai
|
||||
|
||||
@@ -102,10 +106,11 @@ curl -X POST http://localhost:7777/agents/agno-agent/runs \
|
||||
## Where to Go Next
|
||||
|
||||
- `../claude-agent-sdk/` — sessions, custom MCP tools, and more Claude SDK patterns
|
||||
- `../codex/` — sessions, MCP servers, structured output, sandboxes
|
||||
- `../langgraph/` — tool calls, sessions, time travel
|
||||
- `../dspy/` — sessions, ReAct with tools
|
||||
- [AgentOS docs](https://docs.agno.com/agent-os/introduction)
|
||||
|
||||
## Status
|
||||
|
||||
Claude Agent SDK, LangGraph, and DSPy support is in early alpha. Not every feature is wired up yet — please file issues for rough edges.
|
||||
Claude Agent SDK, Codex, LangGraph, and DSPy support is in early alpha. Not every feature is wired up yet — please file issues for rough edges.
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
"""
|
||||
Codex Agent on AgentOS
|
||||
======================
|
||||
An OpenAI Codex agent (via the Codex Python SDK), served through AgentOS.
|
||||
|
||||
The Codex SDK runs the Codex CLI locally as a subprocess. Tool execution
|
||||
(shell, file edits, web search, MCP) is handled by Codex -- you configure
|
||||
the sandbox level and the working directory.
|
||||
|
||||
Requirements:
|
||||
pip install openai-codex
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/00_quickstart/codex_agent.py
|
||||
"""
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
from agno.db.sqlite import SqliteDb
|
||||
from agno.os import AgentOS
|
||||
|
||||
agent = CodexAgent(
|
||||
name="Codex Agent",
|
||||
model="gpt-5.6-luna",
|
||||
sandbox="workspace-write",
|
||||
cwd=".",
|
||||
)
|
||||
|
||||
agent_os = AgentOS(
|
||||
agents=[agent],
|
||||
db=SqliteDb(db_file="tmp/agentos.db"),
|
||||
)
|
||||
app = agent_os.get_app()
|
||||
|
||||
if __name__ == "__main__":
|
||||
agent_os.serve(app="codex_agent:app", reload=True)
|
||||
@@ -0,0 +1,57 @@
|
||||
# Codex
|
||||
|
||||
Examples for running [OpenAI Codex](https://developers.openai.com/codex) as an Agno agent.
|
||||
|
||||
`CodexAgent` wraps the official [Codex Python SDK](https://github.com/openai/codex/tree/main/sdk/python)
|
||||
(`pip install openai-codex`). The SDK ships the Codex CLI, which runs locally as an
|
||||
app-server subprocess and executes the full agent loop: shell commands, file edits,
|
||||
web search and MCP tool calls. Agno adds sessions, streaming, the AgentOS API and the UI.
|
||||
|
||||
## Setup
|
||||
|
||||
```bash
|
||||
pip install openai-codex
|
||||
|
||||
# Authenticate once. Either sign in with ChatGPT through the Codex CLI...
|
||||
codex login
|
||||
# ...or use an API key
|
||||
export CODEX_API_KEY=sk-... # OPENAI_API_KEY also works
|
||||
```
|
||||
|
||||
If the `codex` CLI is not on your PATH you can also authenticate from Python:
|
||||
|
||||
```python
|
||||
from agno.agents.codex import CodexAgent
|
||||
|
||||
CodexAgent(name="Codex").login_api_key("sk-...")
|
||||
```
|
||||
|
||||
## Files
|
||||
|
||||
- `codex_basic.py` — minimal standalone run with `.print_response()`
|
||||
- `codex_tools.py` — shell commands in a read-only sandbox, surfaced as Agno tool calls
|
||||
- `codex_session.py` — multi-turn session; the Codex thread is resumed across turns via Agno's DB
|
||||
- `codex_mcp_tools.py` — connect Codex to an MCP server through `config` overrides
|
||||
- `codex_structured_output.py` — constrain the final answer with a JSON Schema
|
||||
- `codex_agentos.py` — serve Codex through AgentOS
|
||||
- `codex_session_agentos.py` — same with SQLite-backed sessions
|
||||
|
||||
## How sessions work
|
||||
|
||||
Each Agno `session_id` maps to one Codex thread. The thread id is stored on the
|
||||
Agno session (`session_data["codex_thread_id"]`) when a `db` is configured, so
|
||||
the mapping survives restarts. Without a `db` the mapping is kept in memory for
|
||||
the lifetime of the agent object.
|
||||
|
||||
If a thread cannot be resumed (for example `ephemeral=True`, or the Codex
|
||||
session files were removed), the adapter starts a fresh thread and prepends the
|
||||
persisted chat history to the prompt so context is not lost.
|
||||
|
||||
## Sandbox and approvals
|
||||
|
||||
| Setting | Values | Notes |
|
||||
|---|---|---|
|
||||
| `sandbox` | `read-only`, `workspace-write`, `full-access` | Filesystem access for commands and edits |
|
||||
| `approval_mode` | `auto_review`, `deny_all` | How escalated permission requests are handled. There is no interactive prompt, so `deny_all` is the safe choice for unattended servers |
|
||||
|
||||
Test results are tracked in `TEST_LOG.md`.
|
||||
@@ -0,0 +1,81 @@
|
||||
# Codex cookbook test log
|
||||
|
||||
Tested 2026-10-08 with openai-codex 0.161.0 (bundled Codex CLI 0.161.0), model gpt-5.6-luna, ChatGPT login.
|
||||
|
||||
### codex_basic.py
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Smoke test -- `CodexAgent.print_response(..., stream=True)` in a read-only sandbox.
|
||||
|
||||
**Result:** Streamed a three-sentence answer; Response panel rendered with framework `codex`.
|
||||
|
||||
---
|
||||
|
||||
### codex_tools.py
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Streaming run where Codex executes shell commands; verifies command executions show up as Agno tool calls.
|
||||
|
||||
**Result:** Codex ran `rg`/`ls` style commands in the read-only sandbox and produced a repository summary. Commands surfaced as `shell` tool calls with command and cwd args and aggregated output as the result.
|
||||
|
||||
---
|
||||
|
||||
### codex_session.py
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Two-turn session with SqliteDb; verifies the Codex thread id is stored on the session and the second turn remembers the first.
|
||||
|
||||
**Result:** Turn 2 answered "Rust" from turn 1 context. `agno_sessions.session_data` holds `{"codex_thread_id": "<uuid>"}` and turn 2 resumed that thread.
|
||||
|
||||
---
|
||||
|
||||
### codex_mcp_tools.py
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Connect Codex to the Agno docs MCP server via `config` overrides; verifies MCP tool calls surface as Agno tool calls.
|
||||
|
||||
**Result:** Tool Calls panel showed `mcp__agno_docs__search_docs(query=..., ...)` and the answer cited the Agno documentation.
|
||||
|
||||
---
|
||||
|
||||
### codex_structured_output.py
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Non-streaming run with `output_schema`; verifies the content parses as JSON matching the schema.
|
||||
|
||||
**Result:** `json.loads(run_output.content)` returned an object with title, year, genres and one_line_pitch.
|
||||
|
||||
---
|
||||
|
||||
### codex_agentos.py
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Serve a CodexAgent through AgentOS; verify /agents list and /runs.
|
||||
|
||||
**Result:** `GET /agents` listed `codex-assistant`; `POST /agents/codex-assistant/runs` (stream=false) returned status COMPLETED with the expected content.
|
||||
|
||||
---
|
||||
|
||||
### codex_session_agentos.py
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Same as codex_agentos.py with SqliteDb-backed sessions; verify sessions persist across requests.
|
||||
|
||||
**Result:** Streaming run emitted RunStarted, RunContent deltas and RunCompleted over SSE. A second non-streaming request with the same `session_id` recalled the magic number from turn 1. `GET /sessions?type=agent` listed the session.
|
||||
|
||||
---
|
||||
|
||||
### ../00_quickstart/codex_agent.py
|
||||
|
||||
**Status:** PASS
|
||||
|
||||
**Description:** Quickstart: CodexAgent in a workspace-write sandbox on AgentOS with SQLite storage and tracing.
|
||||
|
||||
**Result:** `POST /agents/codex-agent/runs` returned status COMPLETED.
|
||||
@@ -0,0 +1,61 @@
|
||||
"""
|
||||
Codex on AgentOS
|
||||
================
|
||||
Serve a Codex agent through AgentOS -- the same runtime used for native
|
||||
Agno agents.
|
||||
|
||||
The agent is available at the standard /agents/{agent_id}/runs endpoint,
|
||||
supports streaming (SSE) and non-streaming responses, and appears in
|
||||
the AgentOS UI alongside any native agents.
|
||||
|
||||
Requirements:
|
||||
pip install openai-codex
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/codex/codex_agentos.py
|
||||
|
||||
Then call the API:
|
||||
# List agents
|
||||
curl http://localhost:7777/agents
|
||||
|
||||
# Streaming
|
||||
curl -X POST http://localhost:7777/agents/codex-assistant/runs \
|
||||
-F "message=What is quantum computing?" \
|
||||
-F "stream=true" \
|
||||
--no-buffer
|
||||
|
||||
# Non-streaming
|
||||
curl -X POST http://localhost:7777/agents/codex-assistant/runs \
|
||||
-F "message=What is quantum computing?" \
|
||||
-F "stream=false"
|
||||
"""
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
from agno.os import AgentOS
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Create the Codex agent
|
||||
# ---------------------------------------------------------------------------
|
||||
codex_agent = CodexAgent(
|
||||
name="Codex Assistant",
|
||||
description="A Codex-powered coding assistant served through AgentOS",
|
||||
model="gpt-5.6-luna",
|
||||
sandbox="read-only",
|
||||
cwd=".",
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Setup AgentOS
|
||||
# ---------------------------------------------------------------------------
|
||||
agent_os = AgentOS(
|
||||
name="Codex Example",
|
||||
description="AgentOS serving a Codex agent",
|
||||
agents=[codex_agent],
|
||||
)
|
||||
app = agent_os.get_app()
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Run
|
||||
# ---------------------------------------------------------------------------
|
||||
if __name__ == "__main__":
|
||||
agent_os.serve(app="codex_agentos:app", reload=True)
|
||||
@@ -0,0 +1,27 @@
|
||||
"""
|
||||
Standalone usage of the Codex SDK with Agno's .run() and .print_response() methods.
|
||||
|
||||
Codex (OpenAI's coding agent) runs locally through the Codex CLI that ships
|
||||
with the openai-codex Python package. Authenticate once with the Codex CLI
|
||||
(`codex login`) or set CODEX_API_KEY / OPENAI_API_KEY.
|
||||
|
||||
Requirements:
|
||||
pip install openai-codex
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/codex/codex_basic.py
|
||||
"""
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
|
||||
# ----- Wrap Codex for Agno -----
|
||||
agent = CodexAgent(
|
||||
name="Codex Assistant",
|
||||
model="gpt-5.6-luna",
|
||||
sandbox="read-only",
|
||||
)
|
||||
|
||||
# Use .print_response() just like a native Agno agent
|
||||
agent.print_response(
|
||||
"What is quantum computing? Explain in 2-3 sentences.", stream=True
|
||||
)
|
||||
@@ -0,0 +1,35 @@
|
||||
"""
|
||||
Codex with MCP tools, wrapped in Agno's CodexAgent.
|
||||
|
||||
Codex connects to MCP servers through its config. Pass `config` overrides to
|
||||
CodexAgent and they apply to every thread the agent starts, the same way an
|
||||
`mcp_servers` table in ~/.codex/config.toml would.
|
||||
|
||||
MCP tool calls are surfaced as Agno tool call events named
|
||||
`mcp__<server>__<tool>`.
|
||||
|
||||
Requirements:
|
||||
pip install openai-codex
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/codex/codex_mcp_tools.py
|
||||
"""
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
|
||||
# ----- Agent with an MCP server -----
|
||||
agent = CodexAgent(
|
||||
name="Codex Docs Agent",
|
||||
model="gpt-5.6-luna",
|
||||
sandbox="read-only",
|
||||
config={
|
||||
"mcp_servers": {
|
||||
"agno_docs": {"url": "https://docs.agno.com/mcp"},
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
agent.print_response(
|
||||
"Use the agno_docs MCP server to find out what AgentOS is. Answer in two sentences.",
|
||||
stream=True,
|
||||
)
|
||||
@@ -0,0 +1,43 @@
|
||||
"""
|
||||
Codex with session persistence.
|
||||
|
||||
Demonstrates multi-turn conversations where chat history is persisted
|
||||
to Agno's DB. The Codex thread id is stored on the Agno session, so the
|
||||
second turn resumes the same Codex thread and keeps its full context.
|
||||
|
||||
Requirements:
|
||||
pip install openai-codex
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/codex/codex_session.py
|
||||
"""
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
from agno.db.sqlite import SqliteDb
|
||||
|
||||
db = SqliteDb(db_file="tmp/codex_sessions.db")
|
||||
|
||||
agent = CodexAgent(
|
||||
name="Codex Chat",
|
||||
model="gpt-5.6-luna",
|
||||
sandbox="read-only",
|
||||
db=db,
|
||||
)
|
||||
|
||||
SESSION_ID = "codex-demo-session-1"
|
||||
|
||||
# Turn 1
|
||||
agent.print_response(
|
||||
"My favourite programming language is Rust. Reply with one sentence acknowledging that.",
|
||||
stream=True,
|
||||
session_id=SESSION_ID,
|
||||
)
|
||||
|
||||
# Turn 2 -- same session, same Codex thread
|
||||
agent.print_response(
|
||||
"What is my favourite programming language? Answer in one word.",
|
||||
stream=True,
|
||||
session_id=SESSION_ID,
|
||||
)
|
||||
|
||||
print(f"\n--- Session {SESSION_ID} persisted to SQLite ---")
|
||||
@@ -0,0 +1,44 @@
|
||||
"""
|
||||
Codex on AgentOS with session persistence.
|
||||
|
||||
Chat history is persisted to Agno's DB and the Codex thread id is stored on
|
||||
the Agno session, so follow-up messages in the AgentOS UI resume the same
|
||||
Codex thread.
|
||||
|
||||
Requirements:
|
||||
pip install openai-codex
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/codex/codex_session_agentos.py
|
||||
"""
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
from agno.db.sqlite import SqliteDb
|
||||
from agno.os import AgentOS
|
||||
|
||||
db = SqliteDb(db_file="tmp/codex_agentos.db")
|
||||
|
||||
agent = CodexAgent(
|
||||
name="Codex Chat",
|
||||
model="gpt-5.6-luna",
|
||||
sandbox="read-only",
|
||||
cwd=".",
|
||||
db=db,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Setup AgentOS
|
||||
# ---------------------------------------------------------------------------
|
||||
agent_os = AgentOS(
|
||||
id="codex-session-agentos",
|
||||
name="Codex Example",
|
||||
description="AgentOS serving a Codex agent with persisted sessions",
|
||||
agents=[agent],
|
||||
)
|
||||
app = agent_os.get_app()
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Run
|
||||
# ---------------------------------------------------------------------------
|
||||
if __name__ == "__main__":
|
||||
agent_os.serve(app="codex_session_agentos:app", reload=True)
|
||||
@@ -0,0 +1,43 @@
|
||||
"""
|
||||
Codex with structured output.
|
||||
|
||||
Pass a JSON Schema as `output_schema` and Codex constrains its final answer
|
||||
to that schema. The run content is the JSON string, ready for json.loads().
|
||||
|
||||
Requirements:
|
||||
pip install openai-codex
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/codex/codex_structured_output.py
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
from agno.run.agent import RunOutput
|
||||
|
||||
# ----- JSON Schema for the final answer -----
|
||||
movie_schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"title": {"type": "string"},
|
||||
"year": {"type": "integer"},
|
||||
"genres": {"type": "array", "items": {"type": "string"}},
|
||||
"one_line_pitch": {"type": "string"},
|
||||
},
|
||||
"required": ["title", "year", "genres", "one_line_pitch"],
|
||||
"additionalProperties": False,
|
||||
}
|
||||
|
||||
agent = CodexAgent(
|
||||
name="Codex Movie Writer",
|
||||
model="gpt-5.6-luna",
|
||||
sandbox="read-only",
|
||||
output_schema=movie_schema,
|
||||
)
|
||||
|
||||
run_output = agent.run("Invent a science fiction movie set on Mars.")
|
||||
assert isinstance(run_output, RunOutput)
|
||||
|
||||
movie = json.loads(str(run_output.content))
|
||||
print(json.dumps(movie, indent=2))
|
||||
@@ -0,0 +1,37 @@
|
||||
"""
|
||||
Codex with built-in tool calls, wrapped in Agno's CodexAgent.
|
||||
|
||||
Codex has built-in tools (shell commands, file edits, web search, MCP) that
|
||||
are executed by the Codex runtime. You control what it may touch with the
|
||||
sandbox setting:
|
||||
|
||||
read-only read files and run read-only commands (default)
|
||||
workspace-write also edit files inside the working directory
|
||||
full-access no filesystem restrictions
|
||||
|
||||
Shell commands, file changes and MCP calls show up as Agno tool call events,
|
||||
so they render in print_response() and stream over AgentOS.
|
||||
|
||||
Requirements:
|
||||
pip install openai-codex
|
||||
|
||||
Usage:
|
||||
.venvs/demo/bin/python cookbook/frameworks/codex/codex_tools.py
|
||||
"""
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
|
||||
# ----- Agent with built-in tools -----
|
||||
agent = CodexAgent(
|
||||
name="Codex Coder",
|
||||
model="gpt-5.6-luna",
|
||||
sandbox="read-only",
|
||||
cwd=".",
|
||||
reasoning_effort="low",
|
||||
)
|
||||
|
||||
# Streaming with tool calls visible
|
||||
agent.print_response(
|
||||
"List the Python files in the current directory and summarize what this project does",
|
||||
stream=True,
|
||||
)
|
||||
@@ -0,0 +1,3 @@
|
||||
from agno.agents.codex.agent import CodexAgent
|
||||
|
||||
__all__ = ["CodexAgent"]
|
||||
@@ -0,0 +1,568 @@
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from importlib import import_module
|
||||
from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Set, Tuple
|
||||
from uuid import uuid4
|
||||
|
||||
from agno.agents.base import BaseExternalAgent
|
||||
from agno.models.response import ToolExecution
|
||||
from agno.run.agent import (
|
||||
RunContentEvent,
|
||||
RunOutputEvent,
|
||||
ToolCallCompletedEvent,
|
||||
ToolCallStartedEvent,
|
||||
)
|
||||
from agno.utils.log import log_debug, log_warning
|
||||
|
||||
|
||||
def _sdk() -> Any:
|
||||
"""Lazy-import the openai_codex module."""
|
||||
try:
|
||||
import openai_codex # type: ignore
|
||||
|
||||
return openai_codex
|
||||
except ImportError as e:
|
||||
raise ImportError("openai-codex is required: pip install openai-codex") from e
|
||||
|
||||
|
||||
# Accept the CLI spellings as well as the SDK enum names for the sandbox setting.
|
||||
_SANDBOX_ALIASES: Dict[str, str] = {
|
||||
"read_only": "read-only",
|
||||
"readonly": "read-only",
|
||||
"workspace_write": "workspace-write",
|
||||
"full_access": "full-access",
|
||||
"danger-full-access": "full-access",
|
||||
"danger_full_access": "full-access",
|
||||
}
|
||||
|
||||
# thread_start-only options that thread_resume does not accept.
|
||||
_START_ONLY_KEYS = {"ephemeral", "service_name", "session_start_source", "thread_source"}
|
||||
|
||||
|
||||
def _item_root(item: Any) -> Any:
|
||||
"""Unwrap a pydantic RootModel (ThreadItem) to the concrete item."""
|
||||
return getattr(item, "root", item)
|
||||
|
||||
|
||||
def _coerce_args(raw: Any) -> Optional[Dict[str, Any]]:
|
||||
if raw is None:
|
||||
return None
|
||||
if isinstance(raw, dict):
|
||||
return raw
|
||||
if isinstance(raw, str):
|
||||
try:
|
||||
parsed = json.loads(raw)
|
||||
return parsed if isinstance(parsed, dict) else {"input": parsed}
|
||||
except json.JSONDecodeError:
|
||||
return {"input": raw}
|
||||
return {"input": raw}
|
||||
|
||||
|
||||
def _text_from_content(content: Any) -> Optional[str]:
|
||||
"""Join the text parts of an MCP-style content list."""
|
||||
if not isinstance(content, list):
|
||||
return None
|
||||
texts: List[str] = []
|
||||
for part in content:
|
||||
text = part.get("text") if isinstance(part, dict) else getattr(part, "text", None)
|
||||
if text:
|
||||
texts.append(str(text))
|
||||
return "\n".join(texts) if texts else None
|
||||
|
||||
|
||||
@dataclass
|
||||
class _StreamState:
|
||||
"""Bookkeeping for one streamed turn."""
|
||||
|
||||
text_item_ids: Set[str] = field(default_factory=set)
|
||||
current_text_item: Optional[str] = None
|
||||
emitted_text: bool = False
|
||||
tool_info: Dict[str, Dict[str, Any]] = field(default_factory=dict)
|
||||
error: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class CodexAgent(BaseExternalAgent):
|
||||
"""Adapter for OpenAI Codex via the official Codex Python SDK (openai-codex).
|
||||
|
||||
Wraps the Codex SDK so a Codex agent can be used with AgentOS endpoints or
|
||||
standalone via .run() / .print_response().
|
||||
|
||||
The SDK bundles the Codex CLI and runs it locally as an app-server
|
||||
subprocess. Codex runs the full agent loop itself (shell commands, file
|
||||
edits, web search, MCP tools); this adapter translates its notifications
|
||||
into Agno run events and maps each Agno session to one Codex thread.
|
||||
|
||||
Args:
|
||||
name: Display name for this agent.
|
||||
id: Unique identifier (auto-generated from name if not set).
|
||||
model: Codex model id (e.g. "gpt-5.6-luna"). Defaults to the Codex default.
|
||||
instructions: Developer instructions appended to the Codex system prompt.
|
||||
base_instructions: Replaces the Codex built-in system prompt entirely.
|
||||
sandbox: "read-only", "workspace-write" or "full-access". Defaults to the Codex default.
|
||||
approval_mode: "auto_review" or "deny_all" for escalated permission requests.
|
||||
reasoning_effort: "minimal", "low", "medium", "high" or "xhigh".
|
||||
cwd: Working directory for the agent.
|
||||
ephemeral: Do not persist Codex thread files. Ephemeral threads cannot be resumed.
|
||||
output_schema: JSON Schema constraining the final answer. Content is the JSON string.
|
||||
config: Codex config overrides applied to every thread (e.g. {"mcp_servers": {...}}).
|
||||
codex_bin: Path to a specific Codex executable. Defaults to the bundled one.
|
||||
env: Environment variables for the Codex process.
|
||||
thread_kwargs: Extra kwargs forwarded to thread_start / thread_resume.
|
||||
turn_kwargs: Extra kwargs forwarded to each turn.
|
||||
|
||||
Example:
|
||||
from agno.agents.codex import CodexAgent
|
||||
|
||||
agent = CodexAgent(
|
||||
name="Codex Coder",
|
||||
sandbox="workspace-write",
|
||||
cwd=".",
|
||||
)
|
||||
|
||||
# Standalone usage
|
||||
agent.print_response("Summarize this repository", stream=True)
|
||||
|
||||
# Or deploy with AgentOS
|
||||
from agno.os import AgentOS
|
||||
AgentOS(agents=[agent])
|
||||
"""
|
||||
|
||||
model: Optional[str] = None
|
||||
instructions: Optional[str] = None
|
||||
base_instructions: Optional[str] = None
|
||||
sandbox: Optional[str] = None
|
||||
approval_mode: Optional[str] = None
|
||||
reasoning_effort: Optional[str] = None
|
||||
cwd: Optional[str] = None
|
||||
ephemeral: Optional[bool] = None
|
||||
output_schema: Optional[Dict[str, Any]] = None
|
||||
config: Optional[Dict[str, Any]] = None
|
||||
codex_bin: Optional[str] = None
|
||||
env: Optional[Dict[str, str]] = None
|
||||
thread_kwargs: Dict[str, Any] = field(default_factory=dict)
|
||||
turn_kwargs: Dict[str, Any] = field(default_factory=dict)
|
||||
framework: str = "codex"
|
||||
|
||||
# Key under which the Codex thread id is stored in the Agno session's session_data.
|
||||
_THREAD_KEY = "codex_thread_id"
|
||||
|
||||
# Fallback Agno session_id -> Codex thread id map, used when no db is configured.
|
||||
_thread_ids: Dict[str, str] = field(default_factory=dict, init=False, repr=False)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Authentication helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def login_api_key(self, api_key: str) -> None:
|
||||
"""Store an API key in the Codex CLI auth so future runs are authenticated."""
|
||||
sdk = _sdk()
|
||||
with sdk.Codex(self._codex_config(sdk)) as codex:
|
||||
codex.login_api_key(api_key)
|
||||
|
||||
async def alogin_api_key(self, api_key: str) -> None:
|
||||
"""Async version of login_api_key."""
|
||||
sdk = _sdk()
|
||||
async with sdk.AsyncCodex(self._codex_config(sdk)) as codex:
|
||||
await codex.login_api_key(api_key)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SDK option builders
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _codex_config(self, sdk: Any) -> Any:
|
||||
kwargs: Dict[str, Any] = {}
|
||||
if self.codex_bin:
|
||||
kwargs["codex_bin"] = self.codex_bin
|
||||
if self.env:
|
||||
kwargs["env"] = dict(self.env)
|
||||
return sdk.CodexConfig(**kwargs)
|
||||
|
||||
def _new_client(self) -> Any:
|
||||
"""Create an AsyncCodex client. One app-server process per run keeps the
|
||||
client bound to the current event loop."""
|
||||
sdk = _sdk()
|
||||
return sdk.AsyncCodex(self._codex_config(sdk))
|
||||
|
||||
@staticmethod
|
||||
def _to_sandbox(sdk: Any, value: str) -> Any:
|
||||
normalized = _SANDBOX_ALIASES.get(value.lower(), value.lower())
|
||||
try:
|
||||
return sdk.Sandbox(normalized)
|
||||
except ValueError as e:
|
||||
raise ValueError(
|
||||
f"Unknown sandbox {value!r}. Expected one of: read-only, workspace-write, full-access"
|
||||
) from e
|
||||
|
||||
@staticmethod
|
||||
def _to_approval_mode(sdk: Any, value: str) -> Any:
|
||||
normalized = value.lower().replace("-", "_")
|
||||
try:
|
||||
return sdk.ApprovalMode(normalized)
|
||||
except ValueError as e:
|
||||
raise ValueError(f"Unknown approval_mode {value!r}. Expected one of: auto_review, deny_all") from e
|
||||
|
||||
@staticmethod
|
||||
def _to_reasoning_effort(sdk: Any, value: str) -> Any:
|
||||
types_module = getattr(sdk, "types", None)
|
||||
if types_module is None:
|
||||
types_module = import_module(f"{sdk.__name__}.types")
|
||||
return types_module.ReasoningEffort(value.lower())
|
||||
|
||||
def _thread_kwargs(self, sdk: Any, *, resume: bool) -> Dict[str, Any]:
|
||||
"""Build kwargs for thread_start (resume=False) or thread_resume (resume=True)."""
|
||||
kwargs: Dict[str, Any] = {}
|
||||
if self.model:
|
||||
kwargs["model"] = self.model
|
||||
if self.instructions:
|
||||
kwargs["developer_instructions"] = self.instructions
|
||||
if self.base_instructions:
|
||||
kwargs["base_instructions"] = self.base_instructions
|
||||
if self.sandbox:
|
||||
kwargs["sandbox"] = self._to_sandbox(sdk, self.sandbox)
|
||||
if self.approval_mode:
|
||||
kwargs["approval_mode"] = self._to_approval_mode(sdk, self.approval_mode)
|
||||
if self.cwd:
|
||||
kwargs["cwd"] = self.cwd
|
||||
if self.config:
|
||||
kwargs["config"] = self.config
|
||||
if self.ephemeral is not None and not resume:
|
||||
kwargs["ephemeral"] = self.ephemeral
|
||||
for key, value in self.thread_kwargs.items():
|
||||
if resume and key in _START_ONLY_KEYS:
|
||||
continue
|
||||
kwargs[key] = value
|
||||
return kwargs
|
||||
|
||||
def _turn_kwargs(self, sdk: Any) -> Dict[str, Any]:
|
||||
kwargs: Dict[str, Any] = {}
|
||||
if self.reasoning_effort:
|
||||
kwargs["effort"] = self._to_reasoning_effort(sdk, self.reasoning_effort)
|
||||
if self.output_schema:
|
||||
kwargs["output_schema"] = self.output_schema
|
||||
kwargs.update(self.turn_kwargs)
|
||||
return kwargs
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Session <-> Codex thread mapping
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _get_thread_id(self, session: Any, session_id: Optional[str]) -> Optional[str]:
|
||||
if session is not None and session.session_data:
|
||||
thread_id = session.session_data.get(self._THREAD_KEY)
|
||||
if thread_id:
|
||||
return str(thread_id)
|
||||
if session_id:
|
||||
return self._thread_ids.get(session_id)
|
||||
return None
|
||||
|
||||
def _remember_thread(self, session: Any, session_id: Optional[str], thread_id: Optional[str]) -> None:
|
||||
"""Stash the Codex thread id on the session (persisted by the base class)
|
||||
and in memory. Ephemeral threads are never resumable, so skip them."""
|
||||
if not thread_id or self.ephemeral:
|
||||
return
|
||||
if session is not None:
|
||||
if session.session_data is None:
|
||||
session.session_data = {}
|
||||
session.session_data[self._THREAD_KEY] = thread_id
|
||||
if session_id:
|
||||
self._thread_ids[session_id] = thread_id
|
||||
|
||||
def _forget_thread(self, session: Any, session_id: Optional[str]) -> None:
|
||||
if session is not None and session.session_data:
|
||||
session.session_data.pop(self._THREAD_KEY, None)
|
||||
if session_id:
|
||||
self._thread_ids.pop(session_id, None)
|
||||
|
||||
async def _open_thread(self, codex: Any, sdk: Any, session: Any, session_id: Optional[str]) -> Tuple[Any, bool]:
|
||||
"""Resume the thread tied to this session, or start a new one.
|
||||
|
||||
Returns (thread, resumed). A failed resume (thread files removed,
|
||||
ephemeral thread) falls back to a fresh thread.
|
||||
"""
|
||||
thread_id = self._get_thread_id(session, session_id)
|
||||
if thread_id:
|
||||
try:
|
||||
thread = await codex.thread_resume(thread_id, **self._thread_kwargs(sdk, resume=True))
|
||||
log_debug(f"Codex: resumed thread {thread_id} for session {session_id}")
|
||||
return thread, True
|
||||
except Exception as e:
|
||||
log_warning(
|
||||
f"Codex: could not resume thread {thread_id} for session {session_id}: {e}. Starting a new one."
|
||||
)
|
||||
self._forget_thread(session, session_id)
|
||||
|
||||
thread = await codex.thread_start(**self._thread_kwargs(sdk, resume=False))
|
||||
log_debug(f"Codex: started thread {thread.id} for session {session_id}")
|
||||
self._remember_thread(session, session_id, getattr(thread, "id", None))
|
||||
return thread, False
|
||||
|
||||
@staticmethod
|
||||
def _build_prompt(input: Any, history: Optional[List[Dict[str, Any]]], resumed: bool) -> str:
|
||||
"""Plain prompt when the Codex thread carries its own context; otherwise
|
||||
prepend the persisted chat history so a fresh thread does not lose it."""
|
||||
text = str(input)
|
||||
if resumed or not history:
|
||||
return text
|
||||
lines = ["Previous conversation (for context, do not repeat it):"]
|
||||
for message in history:
|
||||
role = message.get("role")
|
||||
content = message.get("content")
|
||||
if role in ("user", "assistant") and content:
|
||||
lines.append(f"{role}: {content}")
|
||||
if len(lines) == 1:
|
||||
return text
|
||||
lines.extend(["", "Current message:", text])
|
||||
return "\n".join(lines)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Adapter hooks
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
async def _arun_adapter(self, input: Any, *, history: Optional[List[Dict[str, Any]]] = None, **kwargs: Any) -> str:
|
||||
"""Non-streaming: run one turn and return the final answer."""
|
||||
sdk = _sdk()
|
||||
session = kwargs.get("session")
|
||||
session_id = kwargs.get("session_id")
|
||||
|
||||
async with self._new_client() as codex:
|
||||
thread, resumed = await self._open_thread(codex, sdk, session, session_id)
|
||||
prompt = self._build_prompt(input, history, resumed)
|
||||
result = await thread.run(prompt, **self._turn_kwargs(sdk))
|
||||
|
||||
return self._final_text(result)
|
||||
|
||||
@staticmethod
|
||||
def _final_text(result: Any) -> str:
|
||||
final_response = getattr(result, "final_response", None)
|
||||
if final_response:
|
||||
return str(final_response)
|
||||
parts: List[str] = []
|
||||
for item in getattr(result, "items", None) or []:
|
||||
root = _item_root(item)
|
||||
if getattr(root, "type", None) == "agentMessage" and getattr(root, "text", None):
|
||||
parts.append(str(root.text))
|
||||
return "\n\n".join(parts)
|
||||
|
||||
async def _arun_adapter_stream(
|
||||
self, input: Any, *, history: Optional[List[Dict[str, Any]]] = None, **kwargs: Any
|
||||
) -> AsyncIterator[RunOutputEvent]:
|
||||
"""Streaming: translate Codex app-server notifications into Agno events.
|
||||
|
||||
Text arrives as item/agentMessage/delta notifications. Tool activity
|
||||
(shell commands, file changes, MCP calls, web search) arrives as
|
||||
item/started and item/completed notifications. turn/completed closes
|
||||
the stream and carries the failure status, if any.
|
||||
"""
|
||||
sdk = _sdk()
|
||||
run_id = kwargs.get("run_id", str(uuid4()))
|
||||
session = kwargs.get("session")
|
||||
session_id = kwargs.get("session_id")
|
||||
state = _StreamState()
|
||||
|
||||
async with self._new_client() as codex:
|
||||
thread, resumed = await self._open_thread(codex, sdk, session, session_id)
|
||||
prompt = self._build_prompt(input, history, resumed)
|
||||
handle = await thread.turn(prompt, **self._turn_kwargs(sdk))
|
||||
async for notification in handle.stream():
|
||||
for event in self._translate_notification(notification, run_id=run_id, state=state):
|
||||
yield event
|
||||
|
||||
if state.error:
|
||||
raise RuntimeError(f"Codex turn failed: {state.error}")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Notification translation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _content_event(self, run_id: str, content: str, reasoning: Optional[str] = None) -> RunContentEvent:
|
||||
return RunContentEvent(
|
||||
run_id=run_id,
|
||||
agent_id=self.get_id(),
|
||||
agent_name=self.name or "",
|
||||
content=content,
|
||||
reasoning_content=reasoning,
|
||||
)
|
||||
|
||||
def _translate_notification(
|
||||
self, notification: Any, *, run_id: str, state: _StreamState
|
||||
) -> Iterator[RunOutputEvent]:
|
||||
method = getattr(notification, "method", "") or ""
|
||||
payload = getattr(notification, "payload", None)
|
||||
if payload is None:
|
||||
return
|
||||
|
||||
if method == "item/agentMessage/delta":
|
||||
delta = getattr(payload, "delta", "") or ""
|
||||
if not delta:
|
||||
return
|
||||
item_id = getattr(payload, "item_id", None) or ""
|
||||
if item_id != state.current_text_item:
|
||||
# Separate consecutive assistant messages (commentary, then final answer)
|
||||
if state.emitted_text:
|
||||
yield self._content_event(run_id, "\n\n")
|
||||
state.current_text_item = item_id
|
||||
state.text_item_ids.add(item_id)
|
||||
state.emitted_text = True
|
||||
yield self._content_event(run_id, str(delta))
|
||||
|
||||
elif method in ("item/reasoning/summaryTextDelta", "item/reasoning/textDelta"):
|
||||
delta = getattr(payload, "delta", "") or ""
|
||||
if delta:
|
||||
yield self._content_event(run_id, "", reasoning=str(delta))
|
||||
|
||||
elif method == "item/started":
|
||||
item = _item_root(getattr(payload, "item", None))
|
||||
if getattr(item, "type", None) == "agentMessage":
|
||||
return # text follows as deltas
|
||||
tool = self._tool_from_item(item)
|
||||
if tool is not None:
|
||||
state.tool_info[tool.tool_call_id or ""] = {"name": tool.tool_name, "args": tool.tool_args}
|
||||
yield ToolCallStartedEvent(run_id=run_id, agent_id=self.get_id(), agent_name=self.name or "", tool=tool)
|
||||
|
||||
elif method == "item/completed":
|
||||
item = _item_root(getattr(payload, "item", None))
|
||||
if getattr(item, "type", None) == "agentMessage":
|
||||
item_id = getattr(item, "id", "") or ""
|
||||
text = getattr(item, "text", "") or ""
|
||||
# Only emit the full text when no deltas were streamed for this item
|
||||
if text and item_id not in state.text_item_ids:
|
||||
if state.emitted_text:
|
||||
yield self._content_event(run_id, "\n\n")
|
||||
state.emitted_text = True
|
||||
state.current_text_item = item_id
|
||||
yield self._content_event(run_id, str(text))
|
||||
return
|
||||
tool = self._tool_from_item(item)
|
||||
if tool is None:
|
||||
return
|
||||
if (tool.tool_call_id or "") not in state.tool_info:
|
||||
# No item/started arrived for this tool; emit it so args are visible
|
||||
state.tool_info[tool.tool_call_id or ""] = {"name": tool.tool_name, "args": tool.tool_args}
|
||||
yield ToolCallStartedEvent(
|
||||
run_id=run_id,
|
||||
agent_id=self.get_id(),
|
||||
agent_name=self.name or "",
|
||||
tool=ToolExecution(
|
||||
tool_call_id=tool.tool_call_id, tool_name=tool.tool_name, tool_args=tool.tool_args
|
||||
),
|
||||
)
|
||||
tool.result = self._tool_result_from_item(item)
|
||||
yield ToolCallCompletedEvent(run_id=run_id, agent_id=self.get_id(), agent_name=self.name or "", tool=tool)
|
||||
|
||||
elif method == "turn/completed":
|
||||
turn = getattr(payload, "turn", None)
|
||||
status = getattr(turn, "status", None)
|
||||
status_value = getattr(status, "value", status)
|
||||
if status_value == "failed":
|
||||
error = getattr(turn, "error", None)
|
||||
state.error = getattr(error, "message", None) or "turn failed"
|
||||
elif status_value == "interrupted":
|
||||
log_warning("Codex: turn was interrupted before completion")
|
||||
|
||||
elif method == "error":
|
||||
error = getattr(payload, "error", None)
|
||||
message = getattr(error, "message", None) or str(error)
|
||||
if getattr(payload, "will_retry", False):
|
||||
log_debug(f"Codex: transient error, retrying: {message}")
|
||||
else:
|
||||
state.error = message
|
||||
|
||||
@staticmethod
|
||||
def _tool_from_item(item: Any) -> Optional[ToolExecution]:
|
||||
"""Map a Codex thread item to an Agno ToolExecution (without result)."""
|
||||
item_type = getattr(item, "type", None)
|
||||
item_id = getattr(item, "id", None) or str(uuid4())
|
||||
|
||||
if item_type == "commandExecution":
|
||||
args: Dict[str, Any] = {"command": str(getattr(item, "command", "") or "")}
|
||||
cwd = getattr(item, "cwd", None)
|
||||
if cwd:
|
||||
args["cwd"] = str(cwd)
|
||||
return ToolExecution(tool_call_id=item_id, tool_name="shell", tool_args=args)
|
||||
|
||||
if item_type == "mcpToolCall":
|
||||
server = getattr(item, "server", "") or ""
|
||||
tool = getattr(item, "tool", "") or ""
|
||||
return ToolExecution(
|
||||
tool_call_id=item_id,
|
||||
tool_name=f"mcp__{server}__{tool}",
|
||||
tool_args=_coerce_args(getattr(item, "arguments", None)),
|
||||
)
|
||||
|
||||
if item_type == "dynamicToolCall":
|
||||
return ToolExecution(
|
||||
tool_call_id=item_id,
|
||||
tool_name=str(getattr(item, "tool", None) or "tool"),
|
||||
tool_args=_coerce_args(getattr(item, "arguments", None)),
|
||||
)
|
||||
|
||||
if item_type == "fileChange":
|
||||
changes: List[Dict[str, Any]] = []
|
||||
for change in getattr(item, "changes", None) or []:
|
||||
kind = _item_root(getattr(change, "kind", None))
|
||||
kind_name = getattr(kind, "type", None) or (str(kind) if kind is not None else None)
|
||||
changes.append({"path": str(getattr(change, "path", "") or ""), "kind": kind_name})
|
||||
return ToolExecution(tool_call_id=item_id, tool_name="apply_patch", tool_args={"changes": changes})
|
||||
|
||||
if item_type == "webSearch":
|
||||
return ToolExecution(
|
||||
tool_call_id=item_id,
|
||||
tool_name="web_search",
|
||||
tool_args={"query": str(getattr(item, "query", "") or "")},
|
||||
)
|
||||
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _tool_result_from_item(item: Any) -> str:
|
||||
"""Render a completed Codex thread item's outcome as the tool result string."""
|
||||
item_type = getattr(item, "type", None)
|
||||
status = getattr(item, "status", None)
|
||||
status_value = getattr(status, "value", status)
|
||||
|
||||
if item_type == "commandExecution":
|
||||
output = str(getattr(item, "aggregated_output", None) or "")
|
||||
exit_code = getattr(item, "exit_code", None)
|
||||
if exit_code not in (None, 0):
|
||||
output = f"{output}\n[exit code {exit_code}]".strip()
|
||||
if not output and status_value in ("failed", "declined"):
|
||||
output = f"command {status_value}"
|
||||
return output
|
||||
|
||||
if item_type in ("mcpToolCall", "dynamicToolCall"):
|
||||
error = getattr(item, "error", None)
|
||||
if error is not None:
|
||||
return f"error: {getattr(error, 'message', error)}"
|
||||
result = getattr(item, "result", None)
|
||||
if result is not None:
|
||||
text = _text_from_content(getattr(result, "content", None))
|
||||
if text:
|
||||
return text
|
||||
structured = getattr(result, "structured_content", None)
|
||||
if structured is not None:
|
||||
return json.dumps(structured, default=str)
|
||||
return str(result)
|
||||
text = _text_from_content(getattr(item, "content_items", None))
|
||||
if text:
|
||||
return text
|
||||
return str(status_value or "completed")
|
||||
|
||||
if item_type == "fileChange":
|
||||
lines: List[str] = []
|
||||
if status_value:
|
||||
lines.append(f"status: {status_value}")
|
||||
for change in getattr(item, "changes", None) or []:
|
||||
kind = _item_root(getattr(change, "kind", None))
|
||||
kind_name = getattr(kind, "type", None) or (str(kind) if kind is not None else "update")
|
||||
lines.append(f"{kind_name} {getattr(change, 'path', '')}")
|
||||
diff = getattr(change, "diff", None)
|
||||
if diff:
|
||||
lines.append(str(diff))
|
||||
return "\n".join(lines)
|
||||
|
||||
if item_type == "webSearch":
|
||||
results = getattr(item, "results", None)
|
||||
return json.dumps(results, default=str) if results else "completed"
|
||||
|
||||
return str(status_value or "completed")
|
||||
@@ -490,6 +490,7 @@ demo = [
|
||||
"yfinance",
|
||||
"youtube-transcript-api",
|
||||
"claude-agent-sdk",
|
||||
"openai-codex",
|
||||
"langgraph",
|
||||
"langchain-openai",
|
||||
"dspy",
|
||||
@@ -632,6 +633,7 @@ module = [
|
||||
"newspaper.*",
|
||||
"numpy.*",
|
||||
"ollama.*",
|
||||
"openai_codex.*",
|
||||
"openfga_sdk.*",
|
||||
"openpyxl.*",
|
||||
"openai.*",
|
||||
|
||||
@@ -0,0 +1,517 @@
|
||||
"""Unit tests for CodexAgent.
|
||||
|
||||
These exercise option mapping, session-to-thread persistence and notification
|
||||
translation against a fake openai_codex module. They do not launch Codex.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from types import ModuleType, SimpleNamespace
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
import pytest
|
||||
|
||||
from agno.agents.codex import CodexAgent
|
||||
from agno.agents.codex import agent as codex_module
|
||||
from agno.db.sqlite import SqliteDb
|
||||
from agno.run.agent import (
|
||||
RunCompletedEvent,
|
||||
RunContentEvent,
|
||||
RunErrorEvent,
|
||||
RunStartedEvent,
|
||||
ToolCallCompletedEvent,
|
||||
ToolCallStartedEvent,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fake openai_codex SDK
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class Sandbox(str, Enum):
|
||||
read_only = "read-only"
|
||||
workspace_write = "workspace-write"
|
||||
full_access = "full-access"
|
||||
|
||||
|
||||
class ApprovalMode(str, Enum):
|
||||
deny_all = "deny_all"
|
||||
auto_review = "auto_review"
|
||||
|
||||
|
||||
class ReasoningEffort(str, Enum):
|
||||
low = "low"
|
||||
medium = "medium"
|
||||
high = "high"
|
||||
|
||||
|
||||
@dataclass
|
||||
class CodexConfig:
|
||||
codex_bin: Optional[str] = None
|
||||
env: Optional[Dict[str, str]] = None
|
||||
|
||||
|
||||
class FakeState:
|
||||
"""Shared, scripted behaviour for the fake SDK."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.notifications: List[Any] = []
|
||||
self.final_response: Optional[str] = "final"
|
||||
self.unresumable: set = set()
|
||||
self.thread_counter = 0
|
||||
self.calls: List[Dict[str, Any]] = []
|
||||
|
||||
|
||||
class FakeHandle:
|
||||
def __init__(self, state: FakeState) -> None:
|
||||
self._state = state
|
||||
|
||||
async def stream(self):
|
||||
for notification in self._state.notifications:
|
||||
yield notification
|
||||
|
||||
|
||||
class FakeThread:
|
||||
def __init__(self, state: FakeState, thread_id: str) -> None:
|
||||
self._state = state
|
||||
self.id = thread_id
|
||||
|
||||
async def turn(self, prompt: str, **kwargs: Any) -> FakeHandle:
|
||||
self._state.calls.append({"op": "turn", "thread_id": self.id, "prompt": prompt, "kwargs": kwargs})
|
||||
return FakeHandle(self._state)
|
||||
|
||||
async def run(self, prompt: str, **kwargs: Any) -> Any:
|
||||
self._state.calls.append({"op": "run", "thread_id": self.id, "prompt": prompt, "kwargs": kwargs})
|
||||
return SimpleNamespace(final_response=self._state.final_response, items=[], status="completed", usage=None)
|
||||
|
||||
|
||||
class FakeAsyncCodex:
|
||||
state: FakeState
|
||||
|
||||
def __init__(self, config: Any = None) -> None:
|
||||
self.config = config
|
||||
self.state.calls.append({"op": "client", "config": config})
|
||||
|
||||
async def __aenter__(self) -> "FakeAsyncCodex":
|
||||
return self
|
||||
|
||||
async def __aexit__(self, *exc: Any) -> None:
|
||||
return None
|
||||
|
||||
async def thread_start(self, **kwargs: Any) -> FakeThread:
|
||||
self.state.thread_counter += 1
|
||||
thread_id = f"thread-{self.state.thread_counter}"
|
||||
self.state.calls.append({"op": "thread_start", "thread_id": thread_id, "kwargs": kwargs})
|
||||
return FakeThread(self.state, thread_id)
|
||||
|
||||
async def thread_resume(self, thread_id: str, **kwargs: Any) -> FakeThread:
|
||||
if thread_id in self.state.unresumable:
|
||||
raise RuntimeError(f"no rollout found for thread id {thread_id}")
|
||||
self.state.calls.append({"op": "thread_resume", "thread_id": thread_id, "kwargs": kwargs})
|
||||
return FakeThread(self.state, thread_id)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_sdk(monkeypatch) -> FakeState:
|
||||
state = FakeState()
|
||||
FakeAsyncCodex.state = state
|
||||
module = ModuleType("openai_codex")
|
||||
module.Sandbox = Sandbox # type: ignore[attr-defined]
|
||||
module.ApprovalMode = ApprovalMode # type: ignore[attr-defined]
|
||||
module.CodexConfig = CodexConfig # type: ignore[attr-defined]
|
||||
module.AsyncCodex = FakeAsyncCodex # type: ignore[attr-defined]
|
||||
module.types = SimpleNamespace(ReasoningEffort=ReasoningEffort) # type: ignore[attr-defined]
|
||||
monkeypatch.setattr(codex_module, "_sdk", lambda: module)
|
||||
return state
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_db():
|
||||
fd, path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(fd)
|
||||
try:
|
||||
yield SqliteDb(db_file=path)
|
||||
finally:
|
||||
if os.path.exists(path):
|
||||
os.unlink(path)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Notification builders (shapes mirror openai_codex app-server notifications)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _item(**fields: Any) -> Any:
|
||||
return SimpleNamespace(root=SimpleNamespace(**fields))
|
||||
|
||||
|
||||
def _started(item: Any) -> Any:
|
||||
return SimpleNamespace(method="item/started", payload=SimpleNamespace(item=item))
|
||||
|
||||
|
||||
def _completed(item: Any) -> Any:
|
||||
return SimpleNamespace(method="item/completed", payload=SimpleNamespace(item=item))
|
||||
|
||||
|
||||
def _delta(item_id: str, text: str) -> Any:
|
||||
return SimpleNamespace(method="item/agentMessage/delta", payload=SimpleNamespace(item_id=item_id, delta=text))
|
||||
|
||||
|
||||
def _turn_completed(status: str = "completed", error: Optional[str] = None) -> Any:
|
||||
turn = SimpleNamespace(
|
||||
status=SimpleNamespace(value=status),
|
||||
error=SimpleNamespace(message=error) if error else None,
|
||||
)
|
||||
return SimpleNamespace(method="turn/completed", payload=SimpleNamespace(turn=turn))
|
||||
|
||||
|
||||
def _collect(agent: CodexAgent, text: str, **kwargs: Any) -> List[Any]:
|
||||
async def consume():
|
||||
return [event async for event in agent._arun_stream(text, **kwargs)]
|
||||
|
||||
return asyncio.run(consume())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Option mapping
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_thread_kwargs_map_to_sdk_options(fake_sdk):
|
||||
agent = CodexAgent(
|
||||
name="Codex",
|
||||
model="gpt-5.6-luna",
|
||||
instructions="Be terse.",
|
||||
base_instructions="You are Codex.",
|
||||
sandbox="workspace-write",
|
||||
approval_mode="deny_all",
|
||||
cwd="/repo",
|
||||
ephemeral=True,
|
||||
config={"mcp_servers": {"docs": {"url": "https://example.com/mcp"}}},
|
||||
thread_kwargs={"service_tier": "fast", "thread_source": "agno"},
|
||||
)
|
||||
sdk = codex_module._sdk()
|
||||
|
||||
start = agent._thread_kwargs(sdk, resume=False)
|
||||
assert start["model"] == "gpt-5.6-luna"
|
||||
assert start["developer_instructions"] == "Be terse."
|
||||
assert start["base_instructions"] == "You are Codex."
|
||||
assert start["sandbox"] is Sandbox.workspace_write
|
||||
assert start["approval_mode"] is ApprovalMode.deny_all
|
||||
assert start["cwd"] == "/repo"
|
||||
assert start["ephemeral"] is True
|
||||
assert start["config"] == {"mcp_servers": {"docs": {"url": "https://example.com/mcp"}}}
|
||||
assert start["service_tier"] == "fast"
|
||||
assert start["thread_source"] == "agno"
|
||||
|
||||
resume = agent._thread_kwargs(sdk, resume=True)
|
||||
assert "ephemeral" not in resume
|
||||
assert "thread_source" not in resume
|
||||
assert resume["service_tier"] == "fast"
|
||||
|
||||
|
||||
def test_sandbox_aliases_and_validation(fake_sdk):
|
||||
sdk = codex_module._sdk()
|
||||
assert CodexAgent._to_sandbox(sdk, "read_only") is Sandbox.read_only
|
||||
assert CodexAgent._to_sandbox(sdk, "danger-full-access") is Sandbox.full_access
|
||||
with pytest.raises(ValueError, match="Unknown sandbox"):
|
||||
CodexAgent._to_sandbox(sdk, "yolo")
|
||||
with pytest.raises(ValueError, match="Unknown approval_mode"):
|
||||
CodexAgent._to_approval_mode(sdk, "ask")
|
||||
|
||||
|
||||
def test_turn_kwargs_include_effort_and_schema(fake_sdk):
|
||||
schema = {"type": "object", "properties": {"a": {"type": "string"}}}
|
||||
agent = CodexAgent(name="Codex", reasoning_effort="high", output_schema=schema, turn_kwargs={"source": "test"})
|
||||
turn = agent._turn_kwargs(codex_module._sdk())
|
||||
assert turn["effort"] is ReasoningEffort.high
|
||||
assert turn["output_schema"] == schema
|
||||
assert turn["source"] == "test"
|
||||
|
||||
|
||||
def test_codex_config_passes_bin_and_env(fake_sdk):
|
||||
agent = CodexAgent(name="Codex", codex_bin="/usr/local/bin/codex", env={"CODEX_API_KEY": "sk-test"})
|
||||
config = agent._codex_config(codex_module._sdk())
|
||||
assert config.codex_bin == "/usr/local/bin/codex"
|
||||
assert config.env == {"CODEX_API_KEY": "sk-test"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Streaming translation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_stream_translates_text_deltas_and_shell_commands(fake_sdk):
|
||||
fake_sdk.notifications = [
|
||||
SimpleNamespace(method="turn/started", payload=SimpleNamespace()),
|
||||
_started(_item(type="userMessage", id="u1")),
|
||||
_started(_item(type="agentMessage", id="m1", text="")),
|
||||
_delta("m1", "I'll "),
|
||||
_delta("m1", "look."),
|
||||
_completed(_item(type="agentMessage", id="m1", text="I'll look.")),
|
||||
_started(
|
||||
_item(
|
||||
type="commandExecution",
|
||||
id="c1",
|
||||
command="ls",
|
||||
cwd="/repo",
|
||||
aggregated_output=None,
|
||||
exit_code=None,
|
||||
status=SimpleNamespace(value="inProgress"),
|
||||
)
|
||||
),
|
||||
_completed(
|
||||
_item(
|
||||
type="commandExecution",
|
||||
id="c1",
|
||||
command="ls",
|
||||
cwd="/repo",
|
||||
aggregated_output="a.py\nb.py",
|
||||
exit_code=0,
|
||||
status=SimpleNamespace(value="completed"),
|
||||
)
|
||||
),
|
||||
_started(_item(type="agentMessage", id="m2", text="")),
|
||||
_delta("m2", "Two files."),
|
||||
_completed(_item(type="agentMessage", id="m2", text="Two files.")),
|
||||
SimpleNamespace(method="thread/tokenUsage/updated", payload=SimpleNamespace()),
|
||||
_turn_completed("completed"),
|
||||
]
|
||||
agent = CodexAgent(name="Codex", id="codex")
|
||||
|
||||
events = _collect(agent, "What is here?", session_id="s1")
|
||||
|
||||
assert isinstance(events[0], RunStartedEvent)
|
||||
assert isinstance(events[-1], RunCompletedEvent)
|
||||
assert events[-1].content == "I'll look.\n\nTwo files."
|
||||
|
||||
content = "".join(e.content for e in events if isinstance(e, RunContentEvent) and e.content)
|
||||
assert content == "I'll look.\n\nTwo files."
|
||||
|
||||
started = [e for e in events if isinstance(e, ToolCallStartedEvent)]
|
||||
completed = [e for e in events if isinstance(e, ToolCallCompletedEvent)]
|
||||
assert len(started) == 1 and len(completed) == 1
|
||||
assert started[0].tool.tool_call_id == "c1"
|
||||
assert started[0].tool.tool_name == "shell"
|
||||
assert started[0].tool.tool_args == {"command": "ls", "cwd": "/repo"}
|
||||
assert completed[0].tool.result == "a.py\nb.py"
|
||||
|
||||
turn_call = next(c for c in fake_sdk.calls if c["op"] == "turn")
|
||||
assert turn_call["prompt"] == "What is here?"
|
||||
|
||||
|
||||
def test_stream_emits_full_message_when_no_deltas_and_pairs_mcp_without_started(fake_sdk):
|
||||
mcp_result = SimpleNamespace(content=[{"type": "text", "text": "AgentOS is a runtime."}], structured_content=None)
|
||||
fake_sdk.notifications = [
|
||||
_completed(
|
||||
_item(
|
||||
type="mcpToolCall",
|
||||
id="t1",
|
||||
server="agno_docs",
|
||||
tool="search_docs",
|
||||
arguments={"query": "AgentOS"},
|
||||
result=mcp_result,
|
||||
error=None,
|
||||
status=SimpleNamespace(value="completed"),
|
||||
)
|
||||
),
|
||||
_completed(_item(type="agentMessage", id="m1", text="Done.")),
|
||||
_turn_completed("completed"),
|
||||
]
|
||||
agent = CodexAgent(name="Codex", id="codex")
|
||||
|
||||
events = _collect(agent, "hi")
|
||||
|
||||
started = [e for e in events if isinstance(e, ToolCallStartedEvent)]
|
||||
completed = [e for e in events if isinstance(e, ToolCallCompletedEvent)]
|
||||
assert len(started) == 1 and len(completed) == 1
|
||||
assert started[0].tool.tool_name == "mcp__agno_docs__search_docs"
|
||||
assert started[0].tool.tool_args == {"query": "AgentOS"}
|
||||
assert completed[0].tool.result == "AgentOS is a runtime."
|
||||
assert events[-1].content == "Done."
|
||||
|
||||
|
||||
def test_stream_file_change_web_search_and_failed_command_results(fake_sdk):
|
||||
kind = SimpleNamespace(root=SimpleNamespace(type="update"))
|
||||
change = SimpleNamespace(path="app.py", kind=kind, diff="-a\n+b")
|
||||
file_change = _item(type="fileChange", id="f1", changes=[change], status=SimpleNamespace(value="completed"))
|
||||
search = _item(type="webSearch", id="w1", query="agno", results=[{"title": "Agno"}])
|
||||
failed_cmd = _item(
|
||||
type="commandExecution",
|
||||
id="c2",
|
||||
command="false",
|
||||
cwd="/repo",
|
||||
aggregated_output="",
|
||||
exit_code=1,
|
||||
status=SimpleNamespace(value="failed"),
|
||||
)
|
||||
fake_sdk.notifications = [
|
||||
_started(file_change),
|
||||
_completed(file_change),
|
||||
_started(search),
|
||||
_completed(search),
|
||||
_started(failed_cmd),
|
||||
_completed(failed_cmd),
|
||||
_turn_completed("completed"),
|
||||
]
|
||||
agent = CodexAgent(name="Codex", id="codex")
|
||||
|
||||
events = _collect(agent, "edit")
|
||||
completed = {e.tool.tool_call_id: e.tool for e in events if isinstance(e, ToolCallCompletedEvent)}
|
||||
|
||||
assert completed["f1"].tool_name == "apply_patch"
|
||||
assert completed["f1"].tool_args == {"changes": [{"path": "app.py", "kind": "update"}]}
|
||||
assert completed["f1"].result == "status: completed\nupdate app.py\n-a\n+b"
|
||||
assert completed["w1"].tool_name == "web_search"
|
||||
assert completed["w1"].tool_args == {"query": "agno"}
|
||||
assert '"title": "Agno"' in (completed["w1"].result or "")
|
||||
assert completed["c2"].result == "[exit code 1]"
|
||||
|
||||
|
||||
def test_stream_reasoning_deltas_become_reasoning_content(fake_sdk):
|
||||
fake_sdk.notifications = [
|
||||
SimpleNamespace(method="item/reasoning/summaryTextDelta", payload=SimpleNamespace(delta="thinking")),
|
||||
_delta("m1", "answer"),
|
||||
_turn_completed("completed"),
|
||||
]
|
||||
agent = CodexAgent(name="Codex", id="codex")
|
||||
|
||||
events = _collect(agent, "q")
|
||||
reasoning = [e for e in events if isinstance(e, RunContentEvent) and e.reasoning_content]
|
||||
assert len(reasoning) == 1
|
||||
assert reasoning[0].reasoning_content == "thinking"
|
||||
assert reasoning[0].content == ""
|
||||
assert events[-1].content == "answer"
|
||||
|
||||
|
||||
def test_failed_turn_surfaces_as_run_error(fake_sdk):
|
||||
fake_sdk.notifications = [
|
||||
_delta("m1", "partial"),
|
||||
_turn_completed("failed", error="model overloaded"),
|
||||
]
|
||||
agent = CodexAgent(name="Codex", id="codex")
|
||||
|
||||
events = _collect(agent, "q")
|
||||
assert isinstance(events[-1], RunErrorEvent)
|
||||
assert "model overloaded" in (events[-1].content or "")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Non-streaming
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_non_stream_returns_final_response(fake_sdk):
|
||||
fake_sdk.final_response = "pong"
|
||||
agent = CodexAgent(name="Codex", id="codex", sandbox="read-only")
|
||||
|
||||
out = asyncio.run(agent._arun_non_stream("ping", session_id="s1"))
|
||||
|
||||
assert out.content == "pong"
|
||||
assert out.status.value == "COMPLETED" or str(out.status).endswith("completed")
|
||||
start = next(c for c in fake_sdk.calls if c["op"] == "thread_start")
|
||||
assert start["kwargs"]["sandbox"] is Sandbox.read_only
|
||||
|
||||
|
||||
def test_final_text_falls_back_to_agent_messages():
|
||||
result = SimpleNamespace(
|
||||
final_response=None,
|
||||
items=[
|
||||
SimpleNamespace(root=SimpleNamespace(type="userMessage", text="ignored")),
|
||||
SimpleNamespace(root=SimpleNamespace(type="agentMessage", text="first")),
|
||||
SimpleNamespace(root=SimpleNamespace(type="agentMessage", text="second")),
|
||||
],
|
||||
)
|
||||
assert CodexAgent._final_text(result) == "first\n\nsecond"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Session <-> thread mapping
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_thread_id_persisted_on_session_and_resumed(fake_sdk, tmp_db):
|
||||
fake_sdk.notifications = [_delta("m1", "ok"), _turn_completed()]
|
||||
agent = CodexAgent(name="Codex", id="codex", db=tmp_db)
|
||||
|
||||
_collect(agent, "first", session_id="s1", user_id="u1")
|
||||
session = agent.read_or_create_session("s1")
|
||||
assert session.session_data["codex_thread_id"] == "thread-1"
|
||||
|
||||
# A fresh agent object (simulating a restart) resumes the same thread from the DB
|
||||
agent2 = CodexAgent(name="Codex", id="codex", db=tmp_db)
|
||||
_collect(agent2, "second", session_id="s1", user_id="u1")
|
||||
|
||||
ops = [c["op"] for c in fake_sdk.calls if c["op"] in ("thread_start", "thread_resume")]
|
||||
assert ops == ["thread_start", "thread_resume"]
|
||||
resume = next(c for c in fake_sdk.calls if c["op"] == "thread_resume")
|
||||
assert resume["thread_id"] == "thread-1"
|
||||
# Resumed threads carry their own context: prompt is sent verbatim
|
||||
second_turn = [c for c in fake_sdk.calls if c["op"] == "turn"][-1]
|
||||
assert second_turn["prompt"] == "second"
|
||||
|
||||
|
||||
def test_unresumable_thread_starts_fresh_with_history(fake_sdk, tmp_db):
|
||||
fake_sdk.notifications = [_delta("m1", "ok"), _turn_completed()]
|
||||
agent = CodexAgent(name="Codex", id="codex", db=tmp_db)
|
||||
|
||||
_collect(agent, "first", session_id="s1", user_id="u1")
|
||||
fake_sdk.unresumable.add("thread-1")
|
||||
_collect(agent, "second", session_id="s1", user_id="u1")
|
||||
|
||||
ops = [c["op"] for c in fake_sdk.calls if c["op"] in ("thread_start", "thread_resume")]
|
||||
assert ops == ["thread_start", "thread_start"]
|
||||
second_turn = [c for c in fake_sdk.calls if c["op"] == "turn"][-1]
|
||||
assert "Previous conversation" in second_turn["prompt"]
|
||||
assert "user: first" in second_turn["prompt"]
|
||||
assert "assistant: ok" in second_turn["prompt"]
|
||||
assert second_turn["prompt"].endswith("Current message:\nsecond")
|
||||
|
||||
session = agent.read_or_create_session("s1")
|
||||
assert session.session_data["codex_thread_id"] == "thread-2"
|
||||
|
||||
|
||||
def test_in_memory_thread_mapping_without_db(fake_sdk):
|
||||
fake_sdk.notifications = [_delta("m1", "ok"), _turn_completed()]
|
||||
agent = CodexAgent(name="Codex", id="codex")
|
||||
|
||||
_collect(agent, "first", session_id="s1")
|
||||
_collect(agent, "second", session_id="s1")
|
||||
_collect(agent, "other", session_id="s2")
|
||||
|
||||
ops = [(c["op"], c["thread_id"]) for c in fake_sdk.calls if c["op"] in ("thread_start", "thread_resume")]
|
||||
assert ops == [("thread_start", "thread-1"), ("thread_resume", "thread-1"), ("thread_start", "thread-2")]
|
||||
|
||||
|
||||
def test_ephemeral_threads_are_not_remembered(fake_sdk, tmp_db):
|
||||
fake_sdk.notifications = [_delta("m1", "ok"), _turn_completed()]
|
||||
agent = CodexAgent(name="Codex", id="codex", db=tmp_db, ephemeral=True)
|
||||
|
||||
_collect(agent, "first", session_id="s1", user_id="u1")
|
||||
_collect(agent, "second", session_id="s1", user_id="u1")
|
||||
|
||||
ops = [c["op"] for c in fake_sdk.calls if c["op"] in ("thread_start", "thread_resume")]
|
||||
assert ops == ["thread_start", "thread_start"]
|
||||
session = agent.read_or_create_session("s1")
|
||||
assert "codex_thread_id" not in (session.session_data or {})
|
||||
# History is injected because the new thread has no context of its own
|
||||
second_turn = [c for c in fake_sdk.calls if c["op"] == "turn"][-1]
|
||||
assert "user: first" in second_turn["prompt"]
|
||||
|
||||
|
||||
def test_missing_sdk_raises_helpful_import_error(monkeypatch):
|
||||
import builtins
|
||||
|
||||
real_import = builtins.__import__
|
||||
|
||||
def fake_import(name, *args, **kwargs):
|
||||
if name == "openai_codex":
|
||||
raise ImportError("No module named 'openai_codex'")
|
||||
return real_import(name, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(builtins, "__import__", fake_import)
|
||||
with pytest.raises(ImportError, match="pip install openai-codex"):
|
||||
codex_module._sdk()
|
||||
Reference in New Issue
Block a user