"""Pluggable adapter for OpenAI Responses-style `/v1/responses` APIs. This module is intentionally independent from any HTTP client or provider implementation. It focuses purely on **format conversion** between: - nanobot's internal `messages` representation (list of `{role, content}`) - OpenResponses / OpenAI Responses request bodies - OpenResponses-style response JSON Usage (pseudo-code): from nanobot.providers.responses_adapter import ResponsesAdapter adapter = ResponsesAdapter(model="gpt-5.4") # 1) Build request body for `/v1/responses` body = adapter.build_request( messages=messages, # nanobot-style max_output_tokens=max_tokens, stream=False, ) # 2) Send HTTP POST with your own client # e.g. client.post(f"{api_base}/responses", json=body, headers=...) # 3) Parse JSON result (non-streaming) content = adapter.parse_response(json_response) The actual HTTP transport is left to the caller so this adapter can be used by different providers (codeproxy, gateway, future services). """ from __future__ import annotations from dataclasses import dataclass import json from typing import Any, Dict, List, Optional from .openresponses_format import ( build_openresponses_body, extract_assistant_text_from_response, ) @dataclass class ResponsesAdapterConfig: """Configuration for the ResponsesAdapter. Attributes ---------- model: Default model identifier to use when none is supplied at call time (e.g. "gpt-5.4"). send_instructions: Whether to send the `instructions` field based on a separate system prompt. For now we keep it simple and embed all system/developer messages as normal items; callers that need a dedicated `instructions` string can pass it explicitly. """ model: str = "gpt-5.4" send_instructions: bool = False @dataclass class ParsedToolCall: """One function tool call parsed from a Responses output item. This is provider-agnostic; nanobot's ToolCallRequest can be constructed from it by the concrete provider. """ name: str arguments: Dict[str, Any] call_id: Optional[str] = None raw_item: Dict[str, Any] | None = None @dataclass class ParsedResponsesResult: """Parsed result from a non-streaming `/v1/responses` JSON. - When `tool_calls` is non-empty, callers should treat this as a tool-calling turn (content may be None). - When `tool_calls` is empty, `content` holds the assistant text (if any). """ content: Optional[str] tool_calls: List[ParsedToolCall] raw_output: List[Dict[str, Any]] | None = None class ResponsesAdapter: """Adapter that converts between nanobot messages and `/v1/responses`. This class is provider-agnostic and can be reused wherever an OpenAI Responses or OpenResponses-compatible endpoint is exposed (e.g. codeproxy, custom gateways, etc.). """ def __init__(self, config: ResponsesAdapterConfig | None = None): self.config = config or ResponsesAdapterConfig() # ------------------------------------------------------------------ # Request-side helpers # ------------------------------------------------------------------ def build_request( self, *, messages: List[Dict[str, Any]], model: Optional[str] = None, max_output_tokens: Optional[int] = None, stream: bool = False, instructions: Optional[str] = None, tools: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, Any]: """Build a `/v1/responses` request body from nanobot messages. Parameters ---------- messages: nanobot-style messages with `role` and `content` fields. model: Optional model override; defaults to the adapter's configured model if not provided. max_output_tokens: Optional output token cap (mapped to `max_output_tokens`). stream: Whether to request streaming (SSE) from the endpoint. instructions: Optional extra instructions string; when provided it is passed through to the `instructions` field in the OpenResponses body. """ effective_model = model or self.config.model request_messages: List[Dict[str, Any]] = [] instruction_parts: List[str] = [] for msg in messages: role = (msg.get("role") or "").lower() if role in {"system", "developer"}: content = msg.get("content") if isinstance(content, str): text = content.strip() elif isinstance(content, list): text = "".join( str(item.get("text", "")) if isinstance(item, dict) else str(item) for item in content ).strip() else: text = str(content).strip() if content is not None else "" if text: instruction_parts.append(text) continue request_messages.append(msg) if instructions: instruction_parts.append(instructions) merged_instructions = "\n\n".join(part for part in instruction_parts if part) body = build_openresponses_body( model=effective_model, messages=request_messages, max_output_tokens=max_output_tokens, stream=stream, instructions=merged_instructions or None, ) # Tools: we expect OpenAI-style function tools, which are # already compatible with OpenResponses tool definitions. if tools: body["tools"] = list(tools) return body # ------------------------------------------------------------------ # Response-side helpers # ------------------------------------------------------------------ def parse_response(self, response_json: Dict[str, Any]) -> ParsedResponsesResult: """Parse a non-streaming Responses JSON into text + tool calls. - If any `function_call` items are present in the `output` array, they are parsed into `ParsedToolCall` instances. - If no tool calls are present, `content` is extracted from the first assistant `message` item (via `extract_assistant_text_from_response`). """ output = response_json.get("output") tool_calls: List[ParsedToolCall] = [] if isinstance(output, list): for idx, item in enumerate(output): if not isinstance(item, dict): continue if item.get("type") != "function_call": continue name = item.get("name") or "" if not name: continue args_raw = item.get("arguments") args: Dict[str, Any] if isinstance(args_raw, str): try: args = json.loads(args_raw) if not isinstance(args, dict): args = {"raw": args_raw} except json.JSONDecodeError: args = {"raw": args_raw} elif isinstance(args_raw, dict): args = args_raw else: args = {} call_id = item.get("call_id") or item.get("id") or f"call_{idx}" tool_calls.append( ParsedToolCall( name=name, arguments=args, call_id=str(call_id), raw_item=item, ) ) # When there are tool calls, we treat this as a tool-calling # turn and do not require a message content. if tool_calls: return ParsedResponsesResult(content=None, tool_calls=tool_calls, raw_output=output) # Fallback: extract assistant text content = extract_assistant_text_from_response(response_json) return ParsedResponsesResult(content=content, tool_calls=[], raw_output=output) __all__ = [ "ResponsesAdapter", "ResponsesAdapterConfig", "ParsedResponsesResult", "ParsedToolCall", ]