kagent

Async Python SDK for running Codex-backed coding-agent tasks with a small, memorable thread API.

The PyPI distribution is codex-python-kagent; the import package is kagent.

import asyncio

from kagent import kagent


async def main() -> None:
    agent = kagent()
    run = await agent.thread("main").run("Create hello.txt with exactly: hello")
    print(run.text)


asyncio.run(main())

Status

kagent is experimental. It always runs through the experimental OpenAI Codex Python SDK and the local Codex app-server. There is no CLI runner option.

The SDK still needs a local Codex binary because the Python SDK launches codex app-server under the hood.

Install

In a uv project:

uv add codex-python-kagent
uv add "openai-codex @ git+https://github.com/openai/codex.git@rust-v0.131.0-alpha.9#subdirectory=sdk/python"
uv run kagent login

For API-key auth instead of ChatGPT subscription auth:

export OPENAI_API_KEY="sk-..."
uv run kagent login --api-key-env OPENAI_API_KEY

For a headless server without a browser:

uv run kagent login --device-auth

For local development in this repo:

uv sync --group dev --group codex --group examples

The local machine must also have the Codex CLI installed and authenticated:

codex login status

API

Use kagent(...) when you already authenticated with kagent login or codex login.

from kagent import kagent

agent = kagent(workspace=".", model="gpt-5.5", reasoning_effort="medium")
thread = agent.thread("refactor-auth")
await thread.run("Refactor auth.py without changing behavior.")
await thread.run("Now run the tests and fix any failures.")

Create a fresh Codex-named thread by omitting the id:

thread = agent.thread()
await thread.run("Create hello.txt with exactly: hello")
print(thread.id)

Stream events with async for while still getting the final result:

stream = agent.thread("refactor-auth").stream("Run the tests and fix failures.")

async for event in stream:
    print(event.get("type") or event.get("method"))

print(stream.result.text)
print(stream.events)

Or trigger Codex auth from Python after configuring the agent:

import os

from kagent import kagent

agent = kagent(model="gpt-5.5").login()
api_agent = kagent(model="gpt-5.5").login(api_key=os.environ["OPENAI_API_KEY"])
server_agent = kagent(model="gpt-5.5").login(device=True)

Options:

kagent(
    workspace=".",
    model="gpt-5.5",
    reasoning_effort="medium",
    ask_for_approval="never",
    sandbox="danger-full-access",
    codex_bin=None,
    store_path=None,
    mcp_servers=None,
)

agent.thread(id, ...) accepts a human-readable id. kagent maps that id globally to Codex's generated session id using a local shelve store. If you omit id, kagent starts a new Codex thread and adopts Codex's generated thread id as thread.id after the first run starts. By default, the store lives at ~/.kagent/threads, not inside the workspace. Set KAGENT_HOME to move the whole kagent state directory or KAGENT_STORE_PATH to point directly at a custom thread store.

Thread creation and loading:

agent.thread("billing-main") # create only; raise if saved already
agent.load("billing-main")   # resume only; never create

Thread execution always uses the Codex SDK. kagent stores the friendly id mapping in its user-level store, then resumes the underlying Codex SDK thread id on later runs with the same human-readable Thread id. If you need separate threads for separate projects, choose separate ids such as billing-main and website-main.

reasoning_effort maps to Codex's model_reasoning_effort config. The default is model="gpt-5.5" and reasoning_effort="medium". The examples use model="gpt-5.5" and reasoning_effort="low" for lighter test runs.

SDK runs inherit your Codex user config through app-server by default, so your configured MCPs are available.

You can also provide MCP servers directly from Python. These are passed as Codex config overrides for the run, without editing your global Codex config:

from kagent import McpServer, kagent

agent = kagent(
    mcp_servers={
        "read_ai": McpServer(
            command="npx",
            args=["-y", "mcp-remote@latest", "https://api.read.ai/mcp"],
        ),
        "internal_api": McpServer(
            url="https://internal.example.com/mcp",
            bearer_token_env_var="INTERNAL_MCP_TOKEN",
        ),
    },
)

Use run(verbose=True) when you want the simple API while still printing readable model activity as it arrives:

run = await agent.thread().run("Create hello.txt with exactly: hello", verbose=True)
print(run.text)

Verbose output is grouped by prefix:

event: turn.started
message: Creating result.txt with the requested content.
action: file_change completed: add /path/to/result.txt
event: turn.completed input_tokens=12 output_tokens=4

run() consumes Codex SDK events internally because that is how kagent gets the generated thread id, final output, tool activity, and structured failures. stream() exposes the same event source when you want to handle events yourself.

Examples

Already logged in with kagent login or codex login:

uv run python examples/01_already_logged_in.py

Trigger ChatGPT/Codex subscription login from Python:

uv run python examples/02_subscription_login.py

Log in with an OpenAI API key from Python:

export OPENAI_API_KEY="sk-..."
uv run python examples/03_api_key_login.py

Stream events in real time:

uv run python examples/04_stream_events.py

Run the OpenHands comparison example:

uv run --group examples python examples/openhands_thread.py

OpenHands may require interactive ChatGPT subscription login on first run.

Development

uv sync --group dev --group codex --group examples
uv run pytest
uv run ruff check .
uv run mypy
uv build

Publish:

export UV_PUBLISH_TOKEN="pypi-..."
uv build
uv publish

The package uses:

• src/ layout
• hatchling build backend
• MIT license
• py.typed for typed package consumers
• uv.lock committed for reproducible local development
• pytest, ruff, and mypy for validation

License

MIT