Added json_schema support

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:

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("<redacted>")
                     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:

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(

claude_code_api/utils/streaming.py

@@ -59,14 +59,17 @@ 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]}"
         self.created = utc_timestamp()
         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),

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