diff --git a/apps/api/engine_gateway/provider_registry.py b/apps/api/engine_gateway/provider_registry.py
index c8a066e..cdb6da7 100644
--- a/apps/api/engine_gateway/provider_registry.py
+++ b/apps/api/engine_gateway/provider_registry.py
@@ -44,6 +44,14 @@ CLI_TIMEOUT_SECONDS = float(os.environ.get("ENGINE_CLI_TIMEOUT_SECONDS", "300"))
 ANTHROPIC_API_BASE = os.environ.get(
     "ANTHROPIC_API_BASE", "https://api.anthropic.com"
 ).rstrip("/")
+OPENAI_API_BASE = os.environ.get(
+    "OPENAI_BASE_URL", "https://api.openai.com/v1"
+).rstrip("/")
+_OPENAI_DEFAULT_ENGINE_MODELS = (
+    "gpt-5.6-terra",
+    "gpt-5.6-luna",
+    "gpt-4.1",
+)
 
 
 class ProviderError(RuntimeError):
@@ -90,6 +98,38 @@ def _efforts(values: Iterable[str]) -> list[ReasoningEffort]:
     return [cast(ReasoningEffort, value) for value in values if value in allowed]
 
 
+def _configured_openai_models() -> list[str]:
+    """운영 엔진에 노출할 OpenAI 텍스트 모델을 명시 allowlist로 제한한다."""
+
+    raw = os.environ.get("OPENAI_ENGINE_MODELS", "").strip()
+    candidates = (
+        [item.strip() for item in raw.split(",")]
+        if raw
+        else list(_OPENAI_DEFAULT_ENGINE_MODELS)
+    )
+    configured_default = os.environ.get("OPENAI_ENGINE_MODEL", "").strip()
+    if configured_default:
+        candidates.insert(0, configured_default)
+    result: list[str] = []
+    for model in candidates:
+        if not model or model in result:
+            continue
+        if any(character.isspace() for character in model) or "/" in model:
+            raise ProviderError("OPENAI_ENGINE_MODELS에 유효하지 않은 모델 식별자가 있습니다.")
+        result.append(model)
+    if not result:
+        raise ProviderError("OPENAI_ENGINE_MODELS가 비어 있습니다.")
+    return result
+
+
+def _openai_reasoning_efforts(model: str) -> list[ReasoningEffort]:
+    # API 모델 목록은 추론 강도 메타데이터를 제공하지 않는다. 운영 기본값으로 쓰는
+    # GPT-5 계열에는 여러 세대가 공통 지원하는 보수적 교집합만 노출한다.
+    if model.startswith("gpt-5"):
+        return ["low", "medium", "high"]
+    return []
+
+
 def _binary(env_name: str, fallback: str) -> str | None:
     configured = os.environ.get(env_name, "").strip()
     if configured:
@@ -518,11 +558,78 @@ async def _discover_claude_api() -> EngineCapabilitiesResponse:
     )
 
 
+async def _discover_openai() -> EngineCapabilitiesResponse:
+    api_key = os.environ.get("OPENAI_API_KEY", "").strip()
+    if not api_key:
+        return _unavailable("openai", "OPENAI_API_KEY가 설정되지 않았습니다.")
+    try:
+        allowed_models = _configured_openai_models()
+        async with httpx.AsyncClient(timeout=20) as client:
+            response = await client.get(
+                f"{OPENAI_API_BASE}/models",
+                headers={"Authorization": f"Bearer {api_key}"},
+            )
+            response.raise_for_status()
+            payload = response.json()
+    except ProviderError as exc:
+        return _unavailable("openai", str(exc))
+    except (httpx.HTTPError, ValueError) as exc:
+        return _unavailable("openai", f"OpenAI 모델 조회 실패: {exc}")
+
+    live_ids = {
+        str(item.get("id") or "").strip()
+        for item in payload.get("data", [])
+        if isinstance(item, dict)
+    } if isinstance(payload, dict) else set()
+    available_ids = [model for model in allowed_models if model in live_ids]
+    if not available_ids:
+        return _unavailable(
+            "openai",
+            "OpenAI가 allowlist의 텍스트 모델을 반환하지 않았습니다.",
+        )
+
+    configured_default = os.environ.get("OPENAI_ENGINE_MODEL", "").strip()
+    default_model = (
+        configured_default
+        if configured_default in available_ids
+        else available_ids[0]
+    )
+    models: list[EngineModelOption] = []
+    for model_id in available_ids:
+        efforts = _openai_reasoning_efforts(model_id)
+        default_effort: ReasoningEffort | None = (
+            "medium" if "medium" in efforts else None
+        )
+        models.append(
+            EngineModelOption(
+                id=model_id,
+                label=model_id,
+                description="OpenAI Models API와 운영 allowlist가 함께 허용한 모델입니다.",
+                reasoning_efforts=efforts,
+                default_reasoning_effort=default_effort,
+                is_default=model_id == default_model,
+            )
+        )
+    selected = next(model for model in models if model.id == default_model)
+    return EngineCapabilitiesResponse(
+        provider="openai",
+        available=True,
+        source="live_api",
+        models=models,
+        default_model=default_model,
+        default_reasoning_effort=selected.default_reasoning_effort,
+        detail="OpenAI /v1/models와 운영 allowlist를 교차 확인했습니다.",
+        fetched_at=_now(),
+    )
+
+
 async def _discover(provider: EngineProvider) -> EngineCapabilitiesResponse:
     if provider == "claude_cli":
         return await _discover_claude_cli()
     if provider == "claude_api":
         return await _discover_claude_api()
+    if provider == "openai":
+        return await _discover_openai()
     if provider == "codex_cli":
         return await _discover_codex_cli()
     if provider == "agy_cli":
@@ -899,6 +1006,102 @@ async def _generate_claude_api(
     )
 
 
+def _openai_response_text(body: dict[str, Any]) -> str:
+    direct = body.get("output_text")
+    if isinstance(direct, str) and direct.strip():
+        return direct
+    parts: list[str] = []
+    for item in body.get("output", []) if isinstance(body, dict) else []:
+        if not isinstance(item, dict) or item.get("type") != "message":
+            continue
+        for content in item.get("content", []):
+            if (
+                isinstance(content, dict)
+                and content.get("type") == "output_text"
+                and isinstance(content.get("text"), str)
+            ):
+                parts.append(content["text"])
+    return "".join(parts)
+
+
+async def _generate_openai(
+    req: GenerateRequest,
+    system_prompt: str,
+    user_payload: str,
+) -> ProviderGenerateResult:
+    pricing_started_at = _utcnow()
+    api_key = os.environ.get("OPENAI_API_KEY", "").strip()
+    if not api_key:
+        raise ProviderError("OPENAI_API_KEY가 설정되지 않았습니다.")
+    model, effort = await _resolve_selection(req, "openai")
+    if req.ai_role == "client":
+        messages: list[dict[str, str]] = [
+            {"role": "user", "content": user_payload}
+        ]
+    else:
+        messages = [
+            {"role": message.role, "content": message.content}
+            for message in req.messages
+            if message.role != "system"
+        ]
+    payload: dict[str, Any] = {
+        "model": model,
+        "input": messages,
+        "max_output_tokens": req.max_tokens,
+        # 상담 시뮬레이션 입력을 OpenAI의 응답 상태 저장소에 남기지 않는다.
+        # 회기 기록의 SSOT는 Vignette의 NAS PostgreSQL뿐이다.
+        "store": False,
+    }
+    if system_prompt:
+        payload["instructions"] = system_prompt
+    if effort:
+        payload["reasoning"] = {"effort": effort}
+    else:
+        payload["temperature"] = req.temperature
+    try:
+        async with httpx.AsyncClient(timeout=CLI_TIMEOUT_SECONDS) as client:
+            response = await client.post(
+                f"{OPENAI_API_BASE}/responses",
+                headers={
+                    "Authorization": f"Bearer {api_key}",
+                    "Content-Type": "application/json",
+                },
+                json=payload,
+            )
+            response.raise_for_status()
+            body = response.json()
+    except (httpx.HTTPError, ValueError) as exc:
+        raise ProviderError(f"OpenAI Responses API 호출 실패: {exc}") from exc
+    if not isinstance(body, dict):
+        raise ProviderError("OpenAI Responses API가 객체 응답을 반환하지 않았습니다.")
+    text = _openai_response_text(body).strip()
+    if not text:
+        raise ProviderError("OpenAI Responses API가 텍스트 응답을 반환하지 않았습니다.")
+    usage = body.get("usage") or {}
+    tokens_in = int(usage.get("input_tokens") or 0)
+    tokens_out = int(usage.get("output_tokens") or 0)
+    input_details = usage.get("input_tokens_details") or {}
+    estimate = estimate_reference_cost(
+        provider="openai",
+        model=str(body.get("model") or model),
+        tokens_in=tokens_in,
+        tokens_out=tokens_out,
+        priced_at=pricing_started_at,
+        cached_input_tokens=int(input_details.get("cached_tokens") or 0),
+    )
+    inference_geo = body.get("inference_geo")
+    return ProviderGenerateResult(
+        text=text,
+        model=str(body.get("model") or model),
+        provider="openai",
+        tokens_in=tokens_in,
+        tokens_out=tokens_out,
+        cost_usd=estimate.cost_usd if estimate is not None else 0.0,
+        inference_geo=str(inference_geo) if inference_geo else None,
+        structured=_structured_or_none(text, req),
+    )
+
+
 async def generate_with_provider(
     req: GenerateRequest,
     *,
@@ -912,6 +1115,8 @@ async def generate_with_provider(
         return await _generate_agy(req, system_prompt, user_payload)
     if provider == "claude_api":
         return await _generate_claude_api(req, system_prompt)
+    if provider == "openai":
+        return await _generate_openai(req, system_prompt, user_payload)
     raise ProviderError(f"이 게이트웨이에서 실행할 수 없는 provider입니다: {provider}")
 
 
diff --git a/apps/api/engine_gateway/test_provider_registry.py b/apps/api/engine_gateway/test_provider_registry.py
index d7c13b6..576b9b7 100644
--- a/apps/api/engine_gateway/test_provider_registry.py
+++ b/apps/api/engine_gateway/test_provider_registry.py
@@ -1,7 +1,7 @@
 import json
 import unittest
 from datetime import datetime, timezone
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, Mock, patch
 
 from app.contracts.engine_gateway import EngineMessage, GenerateRequest
 from engine_gateway import provider_registry
@@ -221,6 +221,137 @@ class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(result.models, [])
         self.assertIn("ANTHROPIC_API_KEY", result.detail)
 
+    async def test_openai_catalog_intersects_live_models_with_explicit_allowlist(self):
+        response = Mock()
+        response.raise_for_status.return_value = None
+        response.json.return_value = {
+            "data": [
+                {"id": "gpt-5.6-sol"},
+                {"id": "gpt-5.6-terra"},
+                {"id": "gpt-4.1"},
+                {"id": "gpt-4o-mini-tts"},
+            ]
+        }
+        client = AsyncMock()
+        client.__aenter__.return_value = client
+        client.__aexit__.return_value = False
+        client.get.return_value = response
+        with (
+            patch.dict(
+                provider_registry.os.environ,
+                {
+                    "OPENAI_API_KEY": "test-openai-key",
+                    "OPENAI_ENGINE_MODEL": "gpt-5.6-terra",
+                    "OPENAI_ENGINE_MODELS": "gpt-5.6-terra,gpt-4.1",
+                },
+                clear=True,
+            ),
+            patch.object(
+                provider_registry.httpx,
+                "AsyncClient",
+                return_value=client,
+            ),
+        ):
+            result = await provider_registry.discover_capabilities(
+                "openai", force=True
+            )
+
+        self.assertTrue(result.available)
+        self.assertEqual(result.source, "live_api")
+        self.assertEqual(result.default_model, "gpt-5.6-terra")
+        self.assertEqual(
+            [model.id for model in result.models],
+            ["gpt-5.6-terra", "gpt-4.1"],
+        )
+        self.assertEqual(result.models[0].reasoning_efforts, ["low", "medium", "high"])
+        self.assertEqual(result.models[1].reasoning_efforts, [])
+        request = client.get.await_args
+        self.assertEqual(request.args[0], f"{provider_registry.OPENAI_API_BASE}/models")
+        self.assertEqual(
+            request.kwargs["headers"]["Authorization"],
+            "Bearer test-openai-key",
+        )
+
+    async def test_openai_generation_uses_responses_api_without_temperature_for_reasoning(self):
+        capabilities = provider_registry.EngineCapabilitiesResponse(
+            provider="openai",
+            available=True,
+            source="live_api",
+            models=[
+                provider_registry.EngineModelOption(
+                    id="gpt-5.6-terra",
+                    label="gpt-5.6-terra",
+                    reasoning_efforts=["low", "medium", "high"],
+                    default_reasoning_effort="medium",
+                    is_default=True,
+                )
+            ],
+            default_model="gpt-5.6-terra",
+            default_reasoning_effort="medium",
+            fetched_at=1,
+        )
+        response = Mock()
+        response.raise_for_status.return_value = None
+        response.json.return_value = {
+            "model": "gpt-5.6-terra",
+            "output": [
+                {
+                    "type": "message",
+                    "content": [{"type": "output_text", "text": "OK"}],
+                }
+            ],
+            "usage": {
+                "input_tokens": 12,
+                "output_tokens": 2,
+                "input_tokens_details": {"cached_tokens": 3},
+            },
+        }
+        client = AsyncMock()
+        client.__aenter__.return_value = client
+        client.__aexit__.return_value = False
+        client.post.return_value = response
+        request = GenerateRequest(
+            provider="openai",
+            model="gpt-5.6-terra",
+            reasoning_effort="medium",
+            messages=[EngineMessage(role="user", content="hello")],
+        )
+        with (
+            patch.dict(
+                provider_registry.os.environ,
+                {"OPENAI_API_KEY": "test-openai-key"},
+                clear=True,
+            ),
+            patch.object(
+                provider_registry,
+                "discover_capabilities",
+                AsyncMock(return_value=capabilities),
+            ),
+            patch.object(
+                provider_registry.httpx,
+                "AsyncClient",
+                return_value=client,
+            ),
+        ):
+            result = await provider_registry.generate_with_provider(
+                request,
+                system_prompt="system",
+                user_payload="current turn",
+            )
+
+        self.assertEqual(result.text, "OK")
+        self.assertEqual(result.provider, "openai")
+        self.assertEqual(result.tokens_in, 12)
+        self.assertEqual(result.tokens_out, 2)
+        call = client.post.await_args
+        self.assertEqual(call.args[0], f"{provider_registry.OPENAI_API_BASE}/responses")
+        payload = call.kwargs["json"]
+        self.assertEqual(payload["model"], "gpt-5.6-terra")
+        self.assertEqual(payload["instructions"], "system")
+        self.assertEqual(payload["input"], [{"role": "user", "content": "current turn"}])
+        self.assertIs(payload["store"], False)
+        self.assertEqual(payload["reasoning"], {"effort": "medium"})
+
     async def test_codex_generation_uses_model_and_reasoning_from_selection(self):
         capabilities = provider_registry.EngineCapabilitiesResponse(
             provider="codex_cli",