Use OpenAI responses API

nanobot/providers/azure_openai_provider.py

@@ -1,31 +1,37 @@
-"""Azure OpenAI provider implementation with API version 2024-10-21."""
+"""Azure OpenAI provider using the OpenAI SDK Responses API.
+
+Uses ``AsyncOpenAI`` pointed at ``https://{endpoint}/openai/v1/`` which
+routes to the Responses API (``/responses``).  Reuses shared conversion
+helpers from :mod:`nanobot.providers.openai_responses_common`.
+"""
 
 from __future__ import annotations
 
-import json
 import uuid
 from collections.abc import Awaitable, Callable
 from typing import Any
-from urllib.parse import urljoin
 
 import httpx
-import json_repair
+from openai import AsyncOpenAI
 
-from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
-
-_AZURE_MSG_KEYS = frozenset({"role", "content", "tool_calls", "tool_call_id", "name"})
+from nanobot.providers.base import LLMProvider, LLMResponse
+from nanobot.providers.openai_responses_common import (
+    consume_sse,
+    convert_messages,
+    convert_tools,
+    parse_response_output,
+)
 
 
 class AzureOpenAIProvider(LLMProvider):
-    """
-    Azure OpenAI provider with API version 2024-10-21 compliance.
-    
+    """Azure OpenAI provider backed by the Responses API.
+
     Features:
-    - Hardcoded API version 2024-10-21
-    - Uses model field as Azure deployment name in URL path
-    - Uses api-key header instead of Authorization Bearer
-    - Uses max_completion_tokens instead of max_tokens
-    - Direct HTTP calls, bypasses LiteLLM
+    - Uses the OpenAI Python SDK (``AsyncOpenAI``) with
+      ``base_url = {endpoint}/openai/v1/``
+    - Calls ``client.responses.create()`` (Responses API)
+    - Reuses shared message/tool/SSE conversion from
+      ``openai_responses_common``
     """
 
     def __init__(
@@ -36,40 +42,28 @@ class AzureOpenAIProvider(LLMProvider):
     ):
         super().__init__(api_key, api_base)
         self.default_model = default_model
-        self.api_version = "2024-10-21"
-        
-        # Validate required parameters
+
         if not api_key:
             raise ValueError("Azure OpenAI api_key is required")
         if not api_base:
             raise ValueError("Azure OpenAI api_base is required")
-        
-        # Ensure api_base ends with /
-        if not api_base.endswith('/'):
-            api_base += '/'
+
+        # Normalise: ensure trailing slash
+        if not api_base.endswith("/"):
+            api_base += "/"
         self.api_base = api_base
 
-    def _build_chat_url(self, deployment_name: str) -> str:
-        """Build the Azure OpenAI chat completions URL."""
-        # Azure OpenAI URL format:
-        # https://{resource}.openai.azure.com/openai/deployments/{deployment}/chat/completions?api-version={version}
-        base_url = self.api_base
-        if not base_url.endswith('/'):
-            base_url += '/'
-        
-        url = urljoin(
-            base_url, 
-            f"openai/deployments/{deployment_name}/chat/completions"
+        # SDK client targeting the Azure Responses API endpoint
+        base_url = f"{api_base.rstrip('/')}/openai/v1/"
+        self._client = AsyncOpenAI(
+            api_key=api_key,
+            base_url=base_url,
+            default_headers={"x-session-affinity": uuid.uuid4().hex},
         )
-        return f"{url}?api-version={self.api_version}"
 
-    def _build_headers(self) -> dict[str, str]:
-        """Build headers for Azure OpenAI API with api-key header."""
-        return {
-            "Content-Type": "application/json",
-            "api-key": self.api_key,  # Azure OpenAI uses api-key header, not Authorization
-            "x-session-affinity": uuid.uuid4().hex,  # For cache locality
-        }
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
 
     @staticmethod
     def _supports_temperature(
@@ -82,36 +76,50 @@ class AzureOpenAIProvider(LLMProvider):
         name = deployment_name.lower()
         return not any(token in name for token in ("gpt-5", "o1", "o3", "o4"))
 
-    def _prepare_request_payload(
+    def _build_body(
         self,
-        deployment_name: str,
         messages: list[dict[str, Any]],
-        tools: list[dict[str, Any]] | None = None,
-        max_tokens: int = 4096,
-        temperature: float = 0.7,
-        reasoning_effort: str | None = None,
-        tool_choice: str | dict[str, Any] | None = None,
+        tools: list[dict[str, Any]] | None,
+        model: str | None,
+        max_tokens: int,
+        temperature: float,
+        reasoning_effort: str | None,
+        tool_choice: str | dict[str, Any] | None,
     ) -> dict[str, Any]:
-        """Prepare the request payload with Azure OpenAI 2024-10-21 compliance."""
-        payload: dict[str, Any] = {
-            "messages": self._sanitize_request_messages(
-                self._sanitize_empty_content(messages),
-                _AZURE_MSG_KEYS,
-            ),
-            "max_completion_tokens": max(1, max_tokens),  # Azure API 2024-10-21 uses max_completion_tokens
+        """Build the Responses API request body from Chat-Completions-style args."""
+        deployment = model or self.default_model
+        instructions, input_items = convert_messages(messages)
+
+        body: dict[str, Any] = {
+            "model": deployment,
+            "instructions": instructions or None,
+            "input": input_items,
+            "store": False,
+            "stream": False,
         }
 
-        if self._supports_temperature(deployment_name, reasoning_effort):
-            payload["temperature"] = temperature
+        if self._supports_temperature(deployment, reasoning_effort):
+            body["temperature"] = temperature
 
         if reasoning_effort:
-            payload["reasoning_effort"] = reasoning_effort
+            body["reasoning"] = {"effort": reasoning_effort}
+            body["include"] = ["reasoning.encrypted_content"]
 
         if tools:
-            payload["tools"] = tools
-            payload["tool_choice"] = tool_choice or "auto"
+            body["tools"] = convert_tools(tools)
+            body["tool_choice"] = tool_choice or "auto"
 
-        return payload
+        return body
+
+    @staticmethod
+    def _handle_error(e: Exception) -> LLMResponse:
+        body = getattr(e, "body", None) or getattr(getattr(e, "response", None), "text", None)
+        msg = f"Error: {str(body).strip()[:500]}" if body else f"Error calling Azure OpenAI: {e}"
+        return LLMResponse(content=msg, finish_reason="error")
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
 
     async def chat(
         self,
@@ -123,92 +131,15 @@ class AzureOpenAIProvider(LLMProvider):
         reasoning_effort: str | None = None,
         tool_choice: str | dict[str, Any] | None = None,
     ) -> LLMResponse:
-        """
-        Send a chat completion request to Azure OpenAI.
-
-        Args:
-            messages: List of message dicts with 'role' and 'content'.
-            tools: Optional list of tool definitions in OpenAI format.
-            model: Model identifier (used as deployment name).
-            max_tokens: Maximum tokens in response (mapped to max_completion_tokens).
-            temperature: Sampling temperature.
-            reasoning_effort: Optional reasoning effort parameter.
-
-        Returns:
-            LLMResponse with content and/or tool calls.
-        """
-        deployment_name = model or self.default_model
-        url = self._build_chat_url(deployment_name)
-        headers = self._build_headers()
-        payload = self._prepare_request_payload(
-            deployment_name, messages, tools, max_tokens, temperature, reasoning_effort,
-            tool_choice=tool_choice,
+        body = self._build_body(
+            messages, tools, model, max_tokens, temperature,
+            reasoning_effort, tool_choice,
         )
-
         try:
-            async with httpx.AsyncClient(timeout=60.0, verify=True) as client:
-                response = await client.post(url, headers=headers, json=payload)
-                if response.status_code != 200:
-                    return LLMResponse(
-                        content=f"Azure OpenAI API Error {response.status_code}: {response.text}",
-                        finish_reason="error",
-                    )
-                
-                response_data = response.json()
-                return self._parse_response(response_data)
-
+            response = await self._client.responses.create(**body)
+            return parse_response_output(response)
         except Exception as e:
-            return LLMResponse(
-                content=f"Error calling Azure OpenAI: {repr(e)}",
-                finish_reason="error",
-            )
-
-    def _parse_response(self, response: dict[str, Any]) -> LLMResponse:
-        """Parse Azure OpenAI response into our standard format."""
-        try:
-            choice = response["choices"][0]
-            message = choice["message"]
-
-            tool_calls = []
-            if message.get("tool_calls"):
-                for tc in message["tool_calls"]:
-                    # Parse arguments from JSON string if needed
-                    args = tc["function"]["arguments"]
-                    if isinstance(args, str):
-                        args = json_repair.loads(args)
-
-                    tool_calls.append(
-                        ToolCallRequest(
-                            id=tc["id"],
-                            name=tc["function"]["name"],
-                            arguments=args,
-                        )
-                    )
-
-            usage = {}
-            if response.get("usage"):
-                usage_data = response["usage"]
-                usage = {
-                    "prompt_tokens": usage_data.get("prompt_tokens", 0),
-                    "completion_tokens": usage_data.get("completion_tokens", 0),
-                    "total_tokens": usage_data.get("total_tokens", 0),
-                }
-
-            reasoning_content = message.get("reasoning_content") or None
-
-            return LLMResponse(
-                content=message.get("content"),
-                tool_calls=tool_calls,
-                finish_reason=choice.get("finish_reason", "stop"),
-                usage=usage,
-                reasoning_content=reasoning_content,
-            )
-
-        except (KeyError, IndexError) as e:
-            return LLMResponse(
-                content=f"Error parsing Azure OpenAI response: {str(e)}",
-                finish_reason="error",
-            )
+            return self._handle_error(e)
 
     async def chat_stream(
         self,
@@ -221,89 +152,40 @@ class AzureOpenAIProvider(LLMProvider):
         tool_choice: str | dict[str, Any] | None = None,
         on_content_delta: Callable[[str], Awaitable[None]] | None = None,
     ) -> LLMResponse:
-        """Stream a chat completion via Azure OpenAI SSE."""
-        deployment_name = model or self.default_model
-        url = self._build_chat_url(deployment_name)
-        headers = self._build_headers()
-        payload = self._prepare_request_payload(
-            deployment_name, messages, tools, max_tokens, temperature,
-            reasoning_effort, tool_choice=tool_choice,
+        body = self._build_body(
+            messages, tools, model, max_tokens, temperature,
+            reasoning_effort, tool_choice,
         )
-        payload["stream"] = True
+        body["stream"] = True
 
         try:
-            async with httpx.AsyncClient(timeout=60.0, verify=True) as client:
-                async with client.stream("POST", url, headers=headers, json=payload) as response:
+            # Use raw httpx stream via the SDK's base URL so we can reuse
+            # the shared Responses-API SSE parser (same as Codex provider).
+            base_url = str(self._client.base_url).rstrip("/")
+            url = f"{base_url}/responses"
+            headers = {
+                "Authorization": f"Bearer {self._client.api_key}",
+                "Content-Type": "application/json",
+                **(self._client._custom_headers or {}),
+            }
+            async with httpx.AsyncClient(timeout=60.0, verify=True) as http:
+                async with http.stream("POST", url, headers=headers, json=body) as response:
                     if response.status_code != 200:
                         text = await response.aread()
                         return LLMResponse(
                             content=f"Azure OpenAI API Error {response.status_code}: {text.decode('utf-8', 'ignore')}",
                             finish_reason="error",
                         )
-                    return await self._consume_stream(response, on_content_delta)
+                    content, tool_calls, finish_reason = await consume_sse(
+                        response, on_content_delta,
+                    )
+                    return LLMResponse(
+                        content=content or None,
+                        tool_calls=tool_calls,
+                        finish_reason=finish_reason,
+                    )
         except Exception as e:
-            return LLMResponse(content=f"Error calling Azure OpenAI: {repr(e)}", finish_reason="error")
-
-    async def _consume_stream(
-        self,
-        response: httpx.Response,
-        on_content_delta: Callable[[str], Awaitable[None]] | None,
-    ) -> LLMResponse:
-        """Parse Azure OpenAI SSE stream into an LLMResponse."""
-        content_parts: list[str] = []
-        tool_call_buffers: dict[int, dict[str, str]] = {}
-        finish_reason = "stop"
-
-        async for line in response.aiter_lines():
-            if not line.startswith("data: "):
-                continue
-            data = line[6:].strip()
-            if data == "[DONE]":
-                break
-            try:
-                chunk = json.loads(data)
-            except Exception:
-                continue
-
-            choices = chunk.get("choices") or []
-            if not choices:
-                continue
-            choice = choices[0]
-            if choice.get("finish_reason"):
-                finish_reason = choice["finish_reason"]
-            delta = choice.get("delta") or {}
-
-            text = delta.get("content")
-            if text:
-                content_parts.append(text)
-                if on_content_delta:
-                    await on_content_delta(text)
-
-            for tc in delta.get("tool_calls") or []:
-                idx = tc.get("index", 0)
-                buf = tool_call_buffers.setdefault(idx, {"id": "", "name": "", "arguments": ""})
-                if tc.get("id"):
-                    buf["id"] = tc["id"]
-                fn = tc.get("function") or {}
-                if fn.get("name"):
-                    buf["name"] = fn["name"]
-                if fn.get("arguments"):
-                    buf["arguments"] += fn["arguments"]
-
-        tool_calls = [
-            ToolCallRequest(
-                id=buf["id"], name=buf["name"],
-                arguments=json_repair.loads(buf["arguments"]) if buf["arguments"] else {},
-            )
-            for buf in tool_call_buffers.values()
-        ]
-
-        return LLMResponse(
-            content="".join(content_parts) or None,
-            tool_calls=tool_calls,
-            finish_reason=finish_reason,
-        )
+            return self._handle_error(e)
 
     def get_default_model(self) -> str:
-        """Get the default model (also used as default deployment name)."""
         return self.default_model

nanobot/providers/openai_codex_provider.py

@@ -6,13 +6,18 @@ import asyncio
 import hashlib
 import json
 from collections.abc import Awaitable, Callable
-from typing import Any, AsyncGenerator
+from typing import Any
 
 import httpx
 from loguru import logger
 from oauth_cli_kit import get_token as get_codex_token
 
 from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
+from nanobot.providers.openai_responses_common import (
+    consume_sse,
+    convert_messages,
+    convert_tools,
+)
 
 DEFAULT_CODEX_URL = "https://chatgpt.com/backend-api/codex/responses"
 DEFAULT_ORIGINATOR = "nanobot"
@@ -36,7 +41,7 @@ class OpenAICodexProvider(LLMProvider):
     ) -> LLMResponse:
         """Shared request logic for both chat() and chat_stream()."""
         model = model or self.default_model
-        system_prompt, input_items = _convert_messages(messages)
+        system_prompt, input_items = convert_messages(messages)
 
         token = await asyncio.to_thread(get_codex_token)
         headers = _build_headers(token.account_id, token.access)
@@ -56,7 +61,7 @@ class OpenAICodexProvider(LLMProvider):
         if reasoning_effort:
             body["reasoning"] = {"effort": reasoning_effort}
         if tools:
-            body["tools"] = _convert_tools(tools)
+            body["tools"] = convert_tools(tools)
 
         try:
             try:
@@ -127,96 +132,7 @@ async def _request_codex(
             if response.status_code != 200:
                 text = await response.aread()
                 raise RuntimeError(_friendly_error(response.status_code, text.decode("utf-8", "ignore")))
-            return await _consume_sse(response, on_content_delta)
-
-
-def _convert_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Convert OpenAI function-calling schema to Codex flat format."""
-    converted: list[dict[str, Any]] = []
-    for tool in tools:
-        fn = (tool.get("function") or {}) if tool.get("type") == "function" else tool
-        name = fn.get("name")
-        if not name:
-            continue
-        params = fn.get("parameters") or {}
-        converted.append({
-            "type": "function",
-            "name": name,
-            "description": fn.get("description") or "",
-            "parameters": params if isinstance(params, dict) else {},
-        })
-    return converted
-
-
-def _convert_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
-    system_prompt = ""
-    input_items: list[dict[str, Any]] = []
-
-    for idx, msg in enumerate(messages):
-        role = msg.get("role")
-        content = msg.get("content")
-
-        if role == "system":
-            system_prompt = content if isinstance(content, str) else ""
-            continue
-
-        if role == "user":
-            input_items.append(_convert_user_message(content))
-            continue
-
-        if role == "assistant":
-            if isinstance(content, str) and content:
-                input_items.append({
-                    "type": "message", "role": "assistant",
-                    "content": [{"type": "output_text", "text": content}],
-                    "status": "completed", "id": f"msg_{idx}",
-                })
-            for tool_call in msg.get("tool_calls", []) or []:
-                fn = tool_call.get("function") or {}
-                call_id, item_id = _split_tool_call_id(tool_call.get("id"))
-                input_items.append({
-                    "type": "function_call",
-                    "id": item_id or f"fc_{idx}",
-                    "call_id": call_id or f"call_{idx}",
-                    "name": fn.get("name"),
-                    "arguments": fn.get("arguments") or "{}",
-                })
-            continue
-
-        if role == "tool":
-            call_id, _ = _split_tool_call_id(msg.get("tool_call_id"))
-            output_text = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)
-            input_items.append({"type": "function_call_output", "call_id": call_id, "output": output_text})
-
-    return system_prompt, input_items
-
-
-def _convert_user_message(content: Any) -> dict[str, Any]:
-    if isinstance(content, str):
-        return {"role": "user", "content": [{"type": "input_text", "text": content}]}
-    if isinstance(content, list):
-        converted: list[dict[str, Any]] = []
-        for item in content:
-            if not isinstance(item, dict):
-                continue
-            if item.get("type") == "text":
-                converted.append({"type": "input_text", "text": item.get("text", "")})
-            elif item.get("type") == "image_url":
-                url = (item.get("image_url") or {}).get("url")
-                if url:
-                    converted.append({"type": "input_image", "image_url": url, "detail": "auto"})
-        if converted:
-            return {"role": "user", "content": converted}
-    return {"role": "user", "content": [{"type": "input_text", "text": ""}]}
-
-
-def _split_tool_call_id(tool_call_id: Any) -> tuple[str, str | None]:
-    if isinstance(tool_call_id, str) and tool_call_id:
-        if "|" in tool_call_id:
-            call_id, item_id = tool_call_id.split("|", 1)
-            return call_id, item_id or None
-        return tool_call_id, None
-    return "call_0", None
+            return await consume_sse(response, on_content_delta)
 
 
 def _prompt_cache_key(messages: list[dict[str, Any]]) -> str:
@@ -224,96 +140,6 @@ def _prompt_cache_key(messages: list[dict[str, Any]]) -> str:
     return hashlib.sha256(raw.encode("utf-8")).hexdigest()
 
 
-async def _iter_sse(response: httpx.Response) -> AsyncGenerator[dict[str, Any], None]:
-    buffer: list[str] = []
-    async for line in response.aiter_lines():
-        if line == "":
-            if buffer:
-                data_lines = [l[5:].strip() for l in buffer if l.startswith("data:")]
-                buffer = []
-                if not data_lines:
-                    continue
-                data = "\n".join(data_lines).strip()
-                if not data or data == "[DONE]":
-                    continue
-                try:
-                    yield json.loads(data)
-                except Exception:
-                    continue
-            continue
-        buffer.append(line)
-
-
-async def _consume_sse(
-    response: httpx.Response,
-    on_content_delta: Callable[[str], Awaitable[None]] | None = None,
-) -> tuple[str, list[ToolCallRequest], str]:
-    content = ""
-    tool_calls: list[ToolCallRequest] = []
-    tool_call_buffers: dict[str, dict[str, Any]] = {}
-    finish_reason = "stop"
-
-    async for event in _iter_sse(response):
-        event_type = event.get("type")
-        if event_type == "response.output_item.added":
-            item = event.get("item") or {}
-            if item.get("type") == "function_call":
-                call_id = item.get("call_id")
-                if not call_id:
-                    continue
-                tool_call_buffers[call_id] = {
-                    "id": item.get("id") or "fc_0",
-                    "name": item.get("name"),
-                    "arguments": item.get("arguments") or "",
-                }
-        elif event_type == "response.output_text.delta":
-            delta_text = event.get("delta") or ""
-            content += delta_text
-            if on_content_delta and delta_text:
-                await on_content_delta(delta_text)
-        elif event_type == "response.function_call_arguments.delta":
-            call_id = event.get("call_id")
-            if call_id and call_id in tool_call_buffers:
-                tool_call_buffers[call_id]["arguments"] += event.get("delta") or ""
-        elif event_type == "response.function_call_arguments.done":
-            call_id = event.get("call_id")
-            if call_id and call_id in tool_call_buffers:
-                tool_call_buffers[call_id]["arguments"] = event.get("arguments") or ""
-        elif event_type == "response.output_item.done":
-            item = event.get("item") or {}
-            if item.get("type") == "function_call":
-                call_id = item.get("call_id")
-                if not call_id:
-                    continue
-                buf = tool_call_buffers.get(call_id) or {}
-                args_raw = buf.get("arguments") or item.get("arguments") or "{}"
-                try:
-                    args = json.loads(args_raw)
-                except Exception:
-                    args = {"raw": args_raw}
-                tool_calls.append(
-                    ToolCallRequest(
-                        id=f"{call_id}|{buf.get('id') or item.get('id') or 'fc_0'}",
-                        name=buf.get("name") or item.get("name"),
-                        arguments=args,
-                    )
-                )
-        elif event_type == "response.completed":
-            status = (event.get("response") or {}).get("status")
-            finish_reason = _map_finish_reason(status)
-        elif event_type in {"error", "response.failed"}:
-            raise RuntimeError("Codex response failed")
-
-    return content, tool_calls, finish_reason
-
-
-_FINISH_REASON_MAP = {"completed": "stop", "incomplete": "length", "failed": "error", "cancelled": "error"}
-
-
-def _map_finish_reason(status: str | None) -> str:
-    return _FINISH_REASON_MAP.get(status or "completed", "stop")
-
-
 def _friendly_error(status_code: int, raw: str) -> str:
     if status_code == 429:
         return "ChatGPT usage quota exceeded or rate limit triggered. Please try again later."

nanobot/providers/openai_responses_common/__init__.py

@@ -0,0 +1,27 @@
+"""Shared helpers for OpenAI Responses API providers (Codex, Azure OpenAI)."""
+
+from nanobot.providers.openai_responses_common.converters import (
+    convert_messages,
+    convert_tools,
+    convert_user_message,
+    split_tool_call_id,
+)
+from nanobot.providers.openai_responses_common.parsing import (
+    FINISH_REASON_MAP,
+    consume_sse,
+    iter_sse,
+    map_finish_reason,
+    parse_response_output,
+)
+
+__all__ = [
+    "convert_messages",
+    "convert_tools",
+    "convert_user_message",
+    "split_tool_call_id",
+    "iter_sse",
+    "consume_sse",
+    "map_finish_reason",
+    "parse_response_output",
+    "FINISH_REASON_MAP",
+]

nanobot/providers/openai_responses_common/converters.py

@@ -0,0 +1,110 @@
+"""Convert Chat Completions messages/tools to Responses API format."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+def convert_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
+    """Convert Chat Completions messages to Responses API input items.
+
+    Returns ``(system_prompt, input_items)`` where *system_prompt* is extracted
+    from any ``system`` role message and *input_items* is the Responses API
+    ``input`` array.
+    """
+    system_prompt = ""
+    input_items: list[dict[str, Any]] = []
+
+    for idx, msg in enumerate(messages):
+        role = msg.get("role")
+        content = msg.get("content")
+
+        if role == "system":
+            system_prompt = content if isinstance(content, str) else ""
+            continue
+
+        if role == "user":
+            input_items.append(convert_user_message(content))
+            continue
+
+        if role == "assistant":
+            if isinstance(content, str) and content:
+                input_items.append({
+                    "type": "message", "role": "assistant",
+                    "content": [{"type": "output_text", "text": content}],
+                    "status": "completed", "id": f"msg_{idx}",
+                })
+            for tool_call in msg.get("tool_calls", []) or []:
+                fn = tool_call.get("function") or {}
+                call_id, item_id = split_tool_call_id(tool_call.get("id"))
+                input_items.append({
+                    "type": "function_call",
+                    "id": item_id or f"fc_{idx}",
+                    "call_id": call_id or f"call_{idx}",
+                    "name": fn.get("name"),
+                    "arguments": fn.get("arguments") or "{}",
+                })
+            continue
+
+        if role == "tool":
+            call_id, _ = split_tool_call_id(msg.get("tool_call_id"))
+            output_text = content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)
+            input_items.append({"type": "function_call_output", "call_id": call_id, "output": output_text})
+
+    return system_prompt, input_items
+
+
+def convert_user_message(content: Any) -> dict[str, Any]:
+    """Convert a user message's content to Responses API format.
+
+    Handles plain strings, ``text`` blocks → ``input_text``, and
+    ``image_url`` blocks → ``input_image``.
+    """
+    if isinstance(content, str):
+        return {"role": "user", "content": [{"type": "input_text", "text": content}]}
+    if isinstance(content, list):
+        converted: list[dict[str, Any]] = []
+        for item in content:
+            if not isinstance(item, dict):
+                continue
+            if item.get("type") == "text":
+                converted.append({"type": "input_text", "text": item.get("text", "")})
+            elif item.get("type") == "image_url":
+                url = (item.get("image_url") or {}).get("url")
+                if url:
+                    converted.append({"type": "input_image", "image_url": url, "detail": "auto"})
+        if converted:
+            return {"role": "user", "content": converted}
+    return {"role": "user", "content": [{"type": "input_text", "text": ""}]}
+
+
+def convert_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Convert OpenAI function-calling tool schema to Responses API flat format."""
+    converted: list[dict[str, Any]] = []
+    for tool in tools:
+        fn = (tool.get("function") or {}) if tool.get("type") == "function" else tool
+        name = fn.get("name")
+        if not name:
+            continue
+        params = fn.get("parameters") or {}
+        converted.append({
+            "type": "function",
+            "name": name,
+            "description": fn.get("description") or "",
+            "parameters": params if isinstance(params, dict) else {},
+        })
+    return converted
+
+
+def split_tool_call_id(tool_call_id: Any) -> tuple[str, str | None]:
+    """Split a compound ``call_id|item_id`` string.
+
+    Returns ``(call_id, item_id)`` where *item_id* may be ``None``.
+    """
+    if isinstance(tool_call_id, str) and tool_call_id:
+        if "|" in tool_call_id:
+            call_id, item_id = tool_call_id.split("|", 1)
+            return call_id, item_id or None
+        return tool_call_id, None
+    return "call_0", None

nanobot/providers/openai_responses_common/parsing.py

@@ -0,0 +1,173 @@
+"""Parse Responses API SSE streams and SDK response objects."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Awaitable, Callable
+from typing import Any, AsyncGenerator
+
+import httpx
+
+from nanobot.providers.base import LLMResponse, ToolCallRequest
+
+FINISH_REASON_MAP = {
+    "completed": "stop",
+    "incomplete": "length",
+    "failed": "error",
+    "cancelled": "error",
+}
+
+
+def map_finish_reason(status: str | None) -> str:
+    """Map a Responses API status string to a Chat-Completions-style finish_reason."""
+    return FINISH_REASON_MAP.get(status or "completed", "stop")
+
+
+async def iter_sse(response: httpx.Response) -> AsyncGenerator[dict[str, Any], None]:
+    """Yield parsed JSON events from a Responses API SSE stream."""
+    buffer: list[str] = []
+    async for line in response.aiter_lines():
+        if line == "":
+            if buffer:
+                data_lines = [l[5:].strip() for l in buffer if l.startswith("data:")]
+                buffer = []
+                if not data_lines:
+                    continue
+                data = "\n".join(data_lines).strip()
+                if not data or data == "[DONE]":
+                    continue
+                try:
+                    yield json.loads(data)
+                except Exception:
+                    continue
+            continue
+        buffer.append(line)
+
+
+async def consume_sse(
+    response: httpx.Response,
+    on_content_delta: Callable[[str], Awaitable[None]] | None = None,
+) -> tuple[str, list[ToolCallRequest], str]:
+    """Consume a Responses API SSE stream into ``(content, tool_calls, finish_reason)``."""
+    content = ""
+    tool_calls: list[ToolCallRequest] = []
+    tool_call_buffers: dict[str, dict[str, Any]] = {}
+    finish_reason = "stop"
+
+    async for event in iter_sse(response):
+        event_type = event.get("type")
+        if event_type == "response.output_item.added":
+            item = event.get("item") or {}
+            if item.get("type") == "function_call":
+                call_id = item.get("call_id")
+                if not call_id:
+                    continue
+                tool_call_buffers[call_id] = {
+                    "id": item.get("id") or "fc_0",
+                    "name": item.get("name"),
+                    "arguments": item.get("arguments") or "",
+                }
+        elif event_type == "response.output_text.delta":
+            delta_text = event.get("delta") or ""
+            content += delta_text
+            if on_content_delta and delta_text:
+                await on_content_delta(delta_text)
+        elif event_type == "response.function_call_arguments.delta":
+            call_id = event.get("call_id")
+            if call_id and call_id in tool_call_buffers:
+                tool_call_buffers[call_id]["arguments"] += event.get("delta") or ""
+        elif event_type == "response.function_call_arguments.done":
+            call_id = event.get("call_id")
+            if call_id and call_id in tool_call_buffers:
+                tool_call_buffers[call_id]["arguments"] = event.get("arguments") or ""
+        elif event_type == "response.output_item.done":
+            item = event.get("item") or {}
+            if item.get("type") == "function_call":
+                call_id = item.get("call_id")
+                if not call_id:
+                    continue
+                buf = tool_call_buffers.get(call_id) or {}
+                args_raw = buf.get("arguments") or item.get("arguments") or "{}"
+                try:
+                    args = json.loads(args_raw)
+                except Exception:
+                    args = {"raw": args_raw}
+                tool_calls.append(
+                    ToolCallRequest(
+                        id=f"{call_id}|{buf.get('id') or item.get('id') or 'fc_0'}",
+                        name=buf.get("name") or item.get("name"),
+                        arguments=args,
+                    )
+                )
+        elif event_type == "response.completed":
+            status = (event.get("response") or {}).get("status")
+            finish_reason = map_finish_reason(status)
+        elif event_type in {"error", "response.failed"}:
+            raise RuntimeError("Response failed")
+
+    return content, tool_calls, finish_reason
+
+
+def parse_response_output(response: Any) -> LLMResponse:
+    """Parse an SDK ``Response`` object (from ``client.responses.create()``)
+    into an ``LLMResponse``.
+
+    Works with both Pydantic model objects and plain dicts.
+    """
+    # Normalise to dict
+    if not isinstance(response, dict):
+        dump = getattr(response, "model_dump", None)
+        response = dump() if callable(dump) else vars(response)
+
+    output = response.get("output") or []
+    content_parts: list[str] = []
+    tool_calls: list[ToolCallRequest] = []
+
+    for item in output:
+        if not isinstance(item, dict):
+            dump = getattr(item, "model_dump", None)
+            item = dump() if callable(dump) else vars(item)
+
+        item_type = item.get("type")
+        if item_type == "message":
+            for block in item.get("content") or []:
+                if not isinstance(block, dict):
+                    dump = getattr(block, "model_dump", None)
+                    block = dump() if callable(dump) else vars(block)
+                if block.get("type") == "output_text":
+                    content_parts.append(block.get("text") or "")
+        elif item_type == "function_call":
+            call_id = item.get("call_id") or ""
+            item_id = item.get("id") or "fc_0"
+            args_raw = item.get("arguments") or "{}"
+            try:
+                args = json.loads(args_raw) if isinstance(args_raw, str) else args_raw
+            except Exception:
+                args = {"raw": args_raw}
+            tool_calls.append(ToolCallRequest(
+                id=f"{call_id}|{item_id}",
+                name=item.get("name") or "",
+                arguments=args if isinstance(args, dict) else {},
+            ))
+
+    usage_raw = response.get("usage") or {}
+    if not isinstance(usage_raw, dict):
+        dump = getattr(usage_raw, "model_dump", None)
+        usage_raw = dump() if callable(dump) else vars(usage_raw)
+    usage = {}
+    if usage_raw:
+        usage = {
+            "prompt_tokens": int(usage_raw.get("input_tokens") or 0),
+            "completion_tokens": int(usage_raw.get("output_tokens") or 0),
+            "total_tokens": int(usage_raw.get("total_tokens") or 0),
+        }
+
+    status = response.get("status")
+    finish_reason = map_finish_reason(status)
+
+    return LLMResponse(
+        content="".join(content_parts) or None,
+        tool_calls=tool_calls,
+        finish_reason=finish_reason,
+        usage=usage,
+    )