ccproxy.core.codex_transformers

ccproxy.core.codex_transformers

Codex-specific transformers for request/response transformation.

CodexRequestData

Bases: TypedDict

Typed structure for transformed Codex request data.

CodexRequestTransformer

CodexRequestTransformer()

Bases: RequestTransformer

Codex request transformer for header and instructions field injection.

Source code in ccproxy/core/codex_transformers.py

def __init__(self) -> None:
    """Initialize Codex request transformer."""
    super().__init__()

transform_codex_request async

transform_codex_request(
    method,
    path,
    headers,
    body,
    access_token,
    session_id,
    account_id,
    codex_detection_data=None,
    target_base_url="https://chatgpt.com/backend-api/codex",
)

Transform Codex request using direct parameters from ProxyService.

Parameters:

Name	Type	Description	Default
method	str	HTTP method	required
path	str	Request path	required
headers	dict[str, str]	Request headers	required
body	bytes | None	Request body	required
access_token	str	OAuth access token	required
session_id	str	Codex session ID	required
account_id	str	ChatGPT account ID	required
codex_detection_data	CodexCacheData | None	Optional Codex detection data	None
target_base_url	str	Base URL for the Codex API	'https://chatgpt.com/backend-api/codex'

Returns:

Type	Description
CodexRequestData	Dictionary with transformed request data (method, url, headers, body)

Source code in ccproxy/core/codex_transformers.py

async def transform_codex_request(
    self,
    method: str,
    path: str,
    headers: dict[str, str],
    body: bytes | None,
    access_token: str,
    session_id: str,
    account_id: str,
    codex_detection_data: CodexCacheData | None = None,
    target_base_url: str = "https://chatgpt.com/backend-api/codex",
) -> CodexRequestData:
    """Transform Codex request using direct parameters from ProxyService.

    Args:
        method: HTTP method
        path: Request path
        headers: Request headers
        body: Request body
        access_token: OAuth access token
        session_id: Codex session ID
        account_id: ChatGPT account ID
        codex_detection_data: Optional Codex detection data
        target_base_url: Base URL for the Codex API

    Returns:
        Dictionary with transformed request data (method, url, headers, body)
    """
    # Transform URL path
    transformed_path = self._transform_codex_path(path)
    target_url = f"{target_base_url.rstrip('/')}{transformed_path}"

    # Transform body first (inject instructions)
    codex_body = None
    if body:
        # body is guaranteed to be bytes due to parameter type
        codex_body = self.transform_codex_body(body, codex_detection_data)

    # Transform headers with Codex CLI identity and authentication
    codex_headers = self.create_codex_headers(
        headers, access_token, session_id, account_id, body, codex_detection_data
    )

    # Update Content-Length if body was transformed and size changed
    if codex_body and body and len(codex_body) != len(body):
        # Remove any existing content-length headers (case-insensitive)
        codex_headers = {
            k: v for k, v in codex_headers.items() if k.lower() != "content-length"
        }
        codex_headers["Content-Length"] = str(len(codex_body))
    elif codex_body and not body:
        # New body was created where none existed
        codex_headers["Content-Length"] = str(len(codex_body))

    return CodexRequestData(
        method=method,
        url=target_url,
        headers=codex_headers,
        body=codex_body,
    )

create_codex_headers

create_codex_headers(
    headers,
    access_token,
    session_id,
    account_id,
    body=None,
    codex_detection_data=None,
)

Create Codex headers with CLI identity and authentication.

Source code in ccproxy/core/codex_transformers.py

def create_codex_headers(
    self,
    headers: dict[str, str],
    access_token: str,
    session_id: str,
    account_id: str,
    body: bytes | None = None,
    codex_detection_data: CodexCacheData | None = None,
) -> dict[str, str]:
    """Create Codex headers with CLI identity and authentication."""
    codex_headers = {}

    # Strip potentially problematic headers
    excluded_headers = {
        "host",
        "x-forwarded-for",
        "x-forwarded-proto",
        "x-forwarded-host",
        "forwarded",
        # Authentication headers to be replaced
        "authorization",
        "x-api-key",
        # Compression headers to avoid decompression issues
        "accept-encoding",
        "content-encoding",
        # CORS headers - should not be forwarded to upstream
        "origin",
        "access-control-request-method",
        "access-control-request-headers",
        "access-control-allow-origin",
        "access-control-allow-methods",
        "access-control-allow-headers",
        "access-control-allow-credentials",
        "access-control-max-age",
        "access-control-expose-headers",
    }

    # Copy important headers (excluding problematic ones)
    for key, value in headers.items():
        lower_key = key.lower()
        if lower_key not in excluded_headers:
            codex_headers[key] = value

    # Set authentication with OAuth token
    if access_token:
        codex_headers["Authorization"] = f"Bearer {access_token}"

    # Set defaults for essential headers
    if "content-type" not in [k.lower() for k in codex_headers]:
        codex_headers["Content-Type"] = "application/json"
    if "accept" not in [k.lower() for k in codex_headers]:
        codex_headers["Accept"] = "application/json"

    # Use detected Codex CLI headers when available
    if codex_detection_data:
        detected_headers = codex_detection_data.headers.to_headers_dict()
        # Override with session-specific values
        detected_headers["session_id"] = session_id
        if account_id:
            detected_headers["chatgpt-account-id"] = account_id
        codex_headers.update(detected_headers)
        logger.debug(
            "using_detected_codex_headers",
            version=codex_detection_data.codex_version,
        )
    else:
        # Fallback to hardcoded Codex headers
        codex_headers.update(
            {
                "session_id": session_id,
                "originator": "codex_cli_rs",
                "openai-beta": "responses=experimental",
                "version": "0.21.0",
            }
        )
        if account_id:
            codex_headers["chatgpt-account-id"] = account_id
        logger.debug("using_fallback_codex_headers")

    # Don't set Accept header - let the backend handle it based on stream parameter
    # Setting Accept: text/event-stream with stream:true in body causes 400 Bad Request
    # The backend will determine the response format based on the stream parameter

    return codex_headers

transform_codex_body

transform_codex_body(body, codex_detection_data=None)

Transform request body to inject Codex CLI instructions.

Source code in ccproxy/core/codex_transformers.py

def transform_codex_body(
    self, body: bytes, codex_detection_data: CodexCacheData | None = None
) -> bytes:
    """Transform request body to inject Codex CLI instructions."""
    if not body:
        return body

    try:
        data = json.loads(body.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError) as e:
        # Return original if not valid JSON
        logger.warning(
            "codex_transform_json_decode_failed",
            error=str(e),
            body_preview=body[:200].decode("utf-8", errors="replace")
            if body
            else None,
            body_length=len(body) if body else 0,
        )
        return body

    # Check if this request already has the full Codex instructions
    # If instructions field exists and is longer than 1000 chars, it's already set
    if (
        "instructions" in data
        and data["instructions"]
        and len(data["instructions"]) > 1000
    ):
        # This already has full Codex instructions, don't replace them
        logger.debug("skipping_codex_transform_has_full_instructions")
        return body

    # Get the instructions to inject
    detected_instructions = None
    if codex_detection_data:
        detected_instructions = codex_detection_data.instructions.instructions_field
    else:
        # Fallback instructions from req.json
        detected_instructions = (
            "You are a coding agent running in the Codex CLI, a terminal-based coding assistant. "
            "Codex CLI is an open source project led by OpenAI. You are expected to be precise, safe, and helpful.\n\n"
            "Your capabilities:\n"
            "- Receive user prompts and other context provided by the harness, such as files in the workspace.\n"
            "- Communicate with the user by streaming thinking & responses, and by making & updating plans.\n"
            "- Emit function calls to run terminal commands and apply patches. Depending on how this specific run is configured, "
            "you can request that these function calls be escalated to the user for approval before running. "
            'More on this in the "Sandbox and approvals" section.\n\n'
            "Within this context, Codex refers to the open-source agentic coding interface "
            "(not the old Codex language model built by OpenAI)."
        )

    # Always inject/override the instructions field
    data["instructions"] = detected_instructions

    # Only inject stream: true if user explicitly requested streaming or didn't specify
    # For now, we'll inject stream: true by default since Codex seems to expect it
    if "stream" not in data:
        data["stream"] = True

    return json.dumps(data, separators=(",", ":")).encode("utf-8")