diff --git a/claude_code_api/api/chat.py b/claude_code_api/api/chat.py index 5f41d99..73b595a 100644 --- a/claude_code_api/api/chat.py +++ b/claude_code_api/api/chat.py @@ -90,6 +90,20 @@ async def _log_raw_request(req: Request) -> None: ) +def _extract_json_schema(request: ChatCompletionRequest) -> Optional[Dict[str, Any]]: + response_format = request.response_format + if not response_format or response_format.type != "json_schema": + return None + if not response_format.json_schema: + raise _http_error( + status.HTTP_400_BAD_REQUEST, + "response_format.type is 'json_schema' but no json_schema was provided.", + "invalid_request_error", + "missing_json_schema", + ) + return response_format.json_schema.schema_ + + def _extract_prompts(request: ChatCompletionRequest) -> Tuple[str, str]: if not request.messages: raise _http_error( @@ -145,6 +159,7 @@ async def _collect_non_streaming_response( session_id: str, model: str, project_id: str, + prefer_result_content: bool = False, ) -> Dict[str, Any]: messages, parser = await _gather_claude_messages(claude_process) _log_message_summary(messages) @@ -155,7 +170,12 @@ async def _collect_non_streaming_response( ) response = _build_non_streaming_response( - messages, session_id, model, usage_summary, project_id + messages, + session_id, + model, + usage_summary, + project_id, + prefer_result_content=prefer_result_content, ) _log_response_payload(response) return response @@ -226,9 +246,14 @@ def _build_non_streaming_response( model: str, usage_summary: Dict[str, Any], project_id: str, + prefer_result_content: bool = False, ) -> Dict[str, Any]: response = create_non_streaming_response( - messages=messages, session_id=session_id, model=model, usage=usage_summary + messages=messages, + session_id=session_id, + model=model, + usage=usage_summary, + prefer_result_content=prefer_result_content, ) response["project_id"] = project_id return response @@ -303,6 +328,7 @@ async def create_chat_completion(request: ChatCompletionRequest, req: Request) - response_model = claude_model or get_default_model() user_prompt, system_prompt = _extract_prompts(request) + json_schema = _extract_json_schema(request) # Handle project context project_id = request.project_id or f"default-{client_id}" @@ -330,6 +356,7 @@ def _register_cli_session(cli_session_id: str): model=claude_model, system_prompt=system_prompt, on_cli_session_id=_register_cli_session, + json_schema=json_schema, ) except ClaudeSessionConflictError as e: logger.warning( @@ -381,7 +408,12 @@ def _register_cli_session(cli_session_id: str): # Handle streaming vs non-streaming if request.stream: return StreamingResponse( - create_sse_response(api_session_id, response_model, claude_process), + create_sse_response( + api_session_id, + response_model, + claude_process, + prefer_result_content=json_schema is not None, + ), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", @@ -398,6 +430,7 @@ def _register_cli_session(cli_session_id: str): session_id=api_session_id, model=response_model, project_id=project_id, + prefer_result_content=json_schema is not None, ) except HTTPException: diff --git a/claude_code_api/core/claude_manager.py b/claude_code_api/core/claude_manager.py index 66330e4..0e6b0fb 100644 --- a/claude_code_api/core/claude_manager.py +++ b/claude_code_api/core/claude_manager.py @@ -46,6 +46,7 @@ async def start( prompt: str, model: Optional[str] = None, system_prompt: Optional[str] = None, + json_schema: Optional[Dict[str, Any]] = None, ) -> bool: """Start Claude Code process and wait for completion.""" self.last_error = None @@ -60,6 +61,9 @@ async def start( if model: cmd.extend(["--model", model]) + if json_schema is not None: + cmd.extend(["--json-schema", json.dumps(json_schema)]) + # Always use stream-json output format (exact order from working example) cmd.extend( [ @@ -86,7 +90,7 @@ async def start( safe_cmd.append("") redact_next = False continue - if part in ("-p", "--system-prompt"): + if part in ("-p", "--system-prompt", "--json-schema"): safe_cmd.append(part) redact_next = True continue @@ -440,6 +444,7 @@ async def _start_with_fallback_models( selected_model: Optional[str], system_prompt: Optional[str], on_cli_session_id: Optional[Callable[[str], None]], + json_schema: Optional[Dict[str, Any]] = None, ) -> ClaudeProcess: model_candidates = self._build_model_candidates(selected_model) last_error = "Failed to start Claude process" @@ -450,11 +455,14 @@ async def _start_with_fallback_models( project_path=project_path, on_cli_session_id=on_cli_session_id, ) - success = await process.start( - prompt=prompt, - model=candidate_model, - system_prompt=system_prompt, - ) + start_kwargs: Dict[str, Any] = { + "prompt": prompt, + "model": candidate_model, + "system_prompt": system_prompt, + } + if json_schema is not None: + start_kwargs["json_schema"] = json_schema + success = await process.start(**start_kwargs) if success: self.processes[session_id] = process @@ -502,6 +510,7 @@ async def create_session( model: Optional[str] = None, system_prompt: Optional[str] = None, on_cli_session_id: Optional[Callable[[str], None]] = None, + json_schema: Optional[Dict[str, Any]] = None, ) -> ClaudeProcess: """Create new Claude session.""" async with self._session_lock: @@ -515,6 +524,7 @@ async def create_session( selected_model=model, system_prompt=system_prompt, on_cli_session_id=on_cli_session_id, + json_schema=json_schema, ) async def _stop_session_locked(self, session_id: str) -> None: diff --git a/claude_code_api/models/openai.py b/claude_code_api/models/openai.py index f538096..2f54a88 100644 --- a/claude_code_api/models/openai.py +++ b/claude_code_api/models/openai.py @@ -71,6 +71,33 @@ class ToolCallDelta(BaseModel): ) +class JSONSchemaSpec(BaseModel): + """JSON Schema payload for structured output (OpenAI `response_format.json_schema` shape).""" + + name: Optional[str] = Field(None, description="Schema name") + description: Optional[str] = Field(None, description="Schema description") + schema_: Dict[str, Any] = Field( + ..., alias="schema", description="JSON Schema definition" + ) + strict: Optional[bool] = Field( + None, description="Whether to enforce strict schema adherence" + ) + + model_config = {"populate_by_name": True} + + +class ResponseFormat(BaseModel): + """OpenAI-compatible response_format. Use type='json_schema' to constrain + Claude Code's output via the CLI --json-schema flag.""" + + type: Literal["text", "json_object", "json_schema"] = Field( + ..., description="Response format type" + ) + json_schema: Optional[JSONSchemaSpec] = Field( + None, description="JSON schema definition when type is 'json_schema'" + ) + + class ChatMessage(BaseModel): """Chat message model - accepts any content format.""" @@ -150,6 +177,13 @@ class ChatCompletionRequest(BaseModel): None, description="Tool choice preference (e.g. 'auto', 'none', or a specific tool)", ) + response_format: Optional[ResponseFormat] = Field( + None, + description=( + "OpenAI-compatible response format. Set type='json_schema' with a " + "json_schema.schema to get CLI-validated structured output via --json-schema." + ), + ) # Extension fields for Claude Code project_id: Optional[str] = Field( diff --git a/claude_code_api/utils/streaming.py b/claude_code_api/utils/streaming.py index 1a8db00..62d4fb6 100644 --- a/claude_code_api/utils/streaming.py +++ b/claude_code_api/utils/streaming.py @@ -59,7 +59,9 @@ def format_heartbeat() -> str: class OpenAIStreamConverter: """Converts Claude Code output to OpenAI-compatible streaming format.""" - def __init__(self, model: str, session_id: str): + def __init__( + self, model: str, session_id: str, prefer_result_content: bool = False + ): self.model = model self.session_id = session_id self.completion_id = f"chatcmpl-{uuid.uuid4().hex[:29]}" @@ -67,6 +69,7 @@ def __init__(self, model: str, session_id: str): self.chunk_index = 0 self.parser = ClaudeOutputParser() self.tool_call_index = 0 + self.prefer_result_content = prefer_result_content def _build_chunk( self, delta: Dict[str, Any], finish_reason: Optional[str] = None @@ -93,12 +96,18 @@ def _assistant_chunks(self, message: Any) -> Tuple[List[str], bool, bool]: saw_text = False saw_tool_calls = False - text_content = self.parser.extract_text_content(message).strip() - if text_content: - chunks.append( - SSEFormatter.format_event(self._build_chunk({"content": text_content})) - ) - saw_text = True + # In schema mode, the final `result` payload is the sole authoritative + # content; skip incremental assistant text so clients don't receive it + # concatenated with the schema-validated result. + if not self.prefer_result_content: + text_content = self.parser.extract_text_content(message).strip() + if text_content: + chunks.append( + SSEFormatter.format_event( + self._build_chunk({"content": text_content}) + ) + ) + saw_text = True tool_uses = self.parser.extract_tool_uses(message) if tool_uses: @@ -138,6 +147,11 @@ async def convert_stream( saw_tool_calls = saw_tool_calls or saw_tools if self.parser.is_final_message(message): + if self.prefer_result_content and message.result: + yield SSEFormatter.format_event( + self._build_chunk({"content": message.result.strip()}) + ) + saw_assistant_text = True break # Send final chunk @@ -168,10 +182,16 @@ def __init__(self): self.heartbeat_interval = 30 # seconds async def create_stream( - self, session_id: str, model: str, claude_process: ClaudeProcess + self, + session_id: str, + model: str, + claude_process: ClaudeProcess, + prefer_result_content: bool = False, ) -> AsyncGenerator[str, None]: """Create new streaming connection.""" - converter = OpenAIStreamConverter(model, session_id) + converter = OpenAIStreamConverter( + model, session_id, prefer_result_content=prefer_result_content + ) heartbeat_queue: asyncio.Queue[Optional[str]] = asyncio.Queue() self.active_streams[session_id] = StreamState( converter=converter, heartbeat_queue=heartbeat_queue @@ -318,12 +338,15 @@ async def stream_with_backpressure( async def create_sse_response( - session_id: str, model: str, claude_process: ClaudeProcess + session_id: str, + model: str, + claude_process: ClaudeProcess, + prefer_result_content: bool = False, ) -> AsyncGenerator[str, None]: """Create SSE response for Claude Code output.""" try: async for chunk in streaming_manager.create_stream( - session_id, model, claude_process + session_id, model, claude_process, prefer_result_content=prefer_result_content ): yield chunk except Exception as e: @@ -385,8 +408,21 @@ def _extract_assistant_payload( return content_parts, tool_calls +def _extract_result_content(messages: list) -> Optional[str]: + """Return the CLI's final `result` payload (the --json-schema-validated output).""" + for msg in reversed(messages): + normalized = normalize_claude_message(msg) + if normalized and normalized.type == "result" and normalized.result: + return normalized.result + return None + + def create_non_streaming_response( - messages: list, session_id: str, model: str, usage: Optional[Dict[str, Any]] = None + messages: list, + session_id: str, + model: str, + usage: Optional[Dict[str, Any]] = None, + prefer_result_content: bool = False, ) -> Dict[str, Any]: """Create non-streaming response.""" completion_id = f"chatcmpl-{uuid.uuid4().hex[:29]}" @@ -409,6 +445,11 @@ def create_non_streaming_response( else: complete_content = "" + if prefer_result_content: + result_content = _extract_result_content(messages) + if result_content is not None: + complete_content = result_content.strip() + logger.info( "Final response content", content_parts_count=len(content_parts), diff --git a/tests/test_claude_manager_unit.py b/tests/test_claude_manager_unit.py index 8648fd5..0da824b 100644 --- a/tests/test_claude_manager_unit.py +++ b/tests/test_claude_manager_unit.py @@ -55,7 +55,7 @@ def test_decode_output_line(): async def test_create_session_rejects_duplicate_active_session(monkeypatch, tmp_path): manager = cm.ClaudeManager() - async def fake_start(self, prompt, model=None, system_prompt=None): + async def fake_start(self, prompt, model=None, system_prompt=None, **_kwargs): self.is_running = True return True @@ -83,7 +83,7 @@ async def fake_start(self, prompt, model=None, system_prompt=None): async def test_create_session_replaces_stale_process(monkeypatch, tmp_path): manager = cm.ClaudeManager() - async def fake_start(self, prompt, model=None, system_prompt=None): + async def fake_start(self, prompt, model=None, system_prompt=None, **_kwargs): self.is_running = True return True @@ -126,7 +126,7 @@ async def test_create_session_retries_opus_45_when_opus_46_rejected( ], ) - async def fake_start(self, prompt, model=None, system_prompt=None): + async def fake_start(self, prompt, model=None, system_prompt=None, **_kwargs): attempted_models.append(model) if model == "claude-opus-4-6-20260205": self.last_error = "invalid model: claude-opus-4-6-20260205" @@ -165,7 +165,7 @@ async def test_create_session_raises_when_model_rejected_without_fallback( lambda: [types.SimpleNamespace(id="claude-sonnet-4-5-20250929")], ) - async def fake_start(self, prompt, model=None, system_prompt=None): + async def fake_start(self, prompt, model=None, system_prompt=None, **_kwargs): self.last_error = "unsupported model" self.is_running = False return False @@ -196,7 +196,7 @@ async def test_create_session_raises_for_non_model_start_failure_without_fallbac lambda: [types.SimpleNamespace(id="claude-opus-4-5-20251101")], ) - async def fake_start(self, prompt, model=None, system_prompt=None): + async def fake_start(self, prompt, model=None, system_prompt=None, **_kwargs): attempted_models.append(model) self.last_error = "failed to spawn process" self.is_running = False @@ -223,7 +223,7 @@ async def test_create_session_without_model_does_not_force_model_flag( manager = cm.ClaudeManager() attempted_models = [] - async def fake_start(self, prompt, model=None, system_prompt=None): + async def fake_start(self, prompt, model=None, system_prompt=None, **_kwargs): attempted_models.append(model) self.is_running = True return True