Support Anthropic-routed Claude models

backend/api/server.py

@@ -50,7 +50,8 @@ from src.token_budget import calculate_token_budget
 STATIC_DIR = Path(__file__).resolve().parents[2] / 'frontend' / 'legacy-static'
 
 VALIDATED_CHAT_MODEL_PROVIDERS = {
-    # 这些 provider 已用 owned_by/id 形式实际请求 /chat/completions 验证通过。
+    # 这些 provider 已验证可以在当前 WebUI 中使用。
+    # 部分模型族会在客户端按模型 id 切换到专用兼容路由。
     'azure_openai',
     'hunyuan',
     'minimax',
@@ -65,11 +66,6 @@ VALIDATED_CHAT_MODEL_PROVIDERS = {
     'zhipuai',
 }
 
-UNSUPPORTED_CHAT_MODEL_PREFIXES: tuple[tuple[str, str], ...] = (
-    # /models 会返回这批 ppio 模型，但 /chat/completions 实测不支持。
-    ('ppio', 'pa/claude-'),
-)
-
 # WebUI 的快速入口只展示可以通过对话输入框安全执行的 slash command。
 # 这些命令仍然保留在终端 `/` 列表里，但不适合作为 WebUI 快捷项。
 WEBUI_HIDDEN_SLASH_COMMANDS = {
@@ -1552,8 +1548,6 @@ def _normalize_model_list(payload: Any) -> list[dict[str, str]]:
         )
         if provider and provider not in VALIDATED_CHAT_MODEL_PROVIDERS:
             continue
-        if _is_known_unsupported_chat_model(provider, model_id):
-            continue
         normalized = _normalize_model_id(model_id, provider=provider)
         if normalized is not None:
             entry = {'id': normalized}
@@ -1586,15 +1580,6 @@ def _normalize_model_id(model: str, *, provider: str | None = None) -> str | Non
     return value
 
 
-def _is_known_unsupported_chat_model(provider: str | None, model_id: str) -> bool:
-    provider_key = (provider or '').strip().lower()
-    model_key = model_id.strip().lower()
-    return any(
-        provider_key == blocked_provider and model_key.startswith(blocked_prefix)
-        for blocked_provider, blocked_prefix in UNSUPPORTED_CHAT_MODEL_PREFIXES
-    )
-
-
 def _is_llm_model(model_id: str, model_type: str | None) -> bool:
     lower = model_id.strip().lower()
     blocked_fragments = (

frontend/app/components/assistant-ui/thread.tsx

@@ -761,7 +761,7 @@ function useComposerContextStatus() {
 				if (cancelled) return;
 				setStatus((current) => ({
 					model:
-						contextBudget?.model ?? sessionPayload?.model ?? statePayload.model,
+						statePayload.model ?? contextBudget?.model ?? sessionPayload?.model,
 					contextBudget,
 					sessionId,
 					models: current.models,

src/openai_compat.py

@@ -25,6 +25,13 @@ PROVIDER_MIN_TEMPERATURES = {
     'wenxin': 0.1,
 }
 
+ANTHROPIC_MESSAGES_MODEL_PREFIXES = (
+    'ppio/pa/claude-',
+)
+
+ANTHROPIC_VERSION = '2023-06-01'
+ANTHROPIC_MAX_TOKENS = 4096
+
 
 def _join_url(base_url: str, suffix: str) -> str:
     base = base_url.rstrip('/')
@@ -104,6 +111,24 @@ def _temperature_for_model(model: str, configured: float) -> float:
     return max(configured, minimum)
 
 
+def _uses_anthropic_messages_api(model: str, base_url: str) -> bool:
+    normalized_model = model.strip().lower()
+    normalized_base = base_url.rstrip('/').lower()
+    return normalized_base.endswith('/anthropic') or any(
+        normalized_model.startswith(prefix)
+        for prefix in ANTHROPIC_MESSAGES_MODEL_PREFIXES
+    )
+
+
+def _anthropic_base_url(base_url: str) -> str:
+    base = base_url.rstrip('/')
+    if base.lower().endswith('/anthropic'):
+        return base
+    if base.lower().endswith('/v1'):
+        return f'{base[:-3]}/anthropic'
+    return f'{base}/anthropic'
+
+
 def _parse_usage(payload: Any) -> UsageStats:
     if not isinstance(payload, dict):
         return UsageStats()
@@ -160,6 +185,12 @@ class OpenAICompatClient:
         *,
         output_schema: OutputSchemaConfig | None = None,
     ) -> AssistantTurn:
+        if self._uses_anthropic_messages_api():
+            return self._complete_anthropic_messages(
+                messages=messages,
+                tools=tools,
+                output_schema=output_schema,
+            )
         payload = self._request_json(
             self._build_payload(
                 messages=messages,
@@ -201,6 +232,13 @@ class OpenAICompatClient:
         *,
         output_schema: OutputSchemaConfig | None = None,
     ) -> Iterator[StreamEvent]:
+        if self._uses_anthropic_messages_api():
+            yield from self._stream_anthropic_messages(
+                messages=messages,
+                tools=tools,
+                output_schema=output_schema,
+            )
+            return
         payload = self._build_payload(
             messages=messages,
             tools=tools,
@@ -231,6 +269,9 @@ class OpenAICompatClient:
                 f'Unable to reach local model backend at {self.config.base_url}: {exc.reason}'
             ) from exc
 
+    def _uses_anthropic_messages_api(self) -> bool:
+        return _uses_anthropic_messages_api(self.config.model, self.config.base_url)
+
     def _request_json(self, payload: dict[str, Any]) -> dict[str, Any]:
         body = json.dumps(payload).encode('utf-8')
         req = request.Request(
@@ -287,6 +328,363 @@ class OpenAICompatClient:
             payload['response_format'] = response_format
         return payload
 
+    def _build_anthropic_payload(
+        self,
+        *,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]],
+        stream: bool,
+        output_schema: OutputSchemaConfig | None,
+    ) -> dict[str, Any]:
+        system_parts: list[str] = []
+        anthropic_messages: list[dict[str, Any]] = []
+        for message in messages:
+            role = message.get('role')
+            if role == 'system':
+                content = _normalize_content(message.get('content')).strip()
+                if content:
+                    system_parts.append(content)
+                continue
+            if role == 'assistant':
+                blocks = self._anthropic_assistant_blocks(message)
+                if blocks:
+                    self._append_anthropic_message(anthropic_messages, 'assistant', blocks)
+                continue
+            if role == 'tool':
+                tool_call_id = message.get('tool_call_id')
+                if not isinstance(tool_call_id, str) or not tool_call_id:
+                    tool_call_id = 'toolu_unknown'
+                self._append_anthropic_message(
+                    anthropic_messages,
+                    'user',
+                    [
+                        {
+                            'type': 'tool_result',
+                            'tool_use_id': tool_call_id,
+                            'content': _normalize_content(message.get('content')),
+                        }
+                    ],
+                )
+                continue
+            if role == 'user':
+                blocks = self._anthropic_text_blocks(message.get('content'))
+                if blocks:
+                    self._append_anthropic_message(anthropic_messages, 'user', blocks)
+
+        payload: dict[str, Any] = {
+            'model': self.config.model,
+            'messages': anthropic_messages,
+            'max_tokens': ANTHROPIC_MAX_TOKENS,
+            'stream': stream,
+        }
+        if system_parts:
+            payload['system'] = '\n\n'.join(system_parts)
+        converted_tools = self._anthropic_tools(tools)
+        if converted_tools:
+            payload['tools'] = converted_tools
+        if output_schema is not None:
+            schema_hint = (
+                f'请严格输出符合 JSON Schema `{output_schema.name}` 的 JSON，'
+                '不要输出额外解释。'
+            )
+            payload['system'] = (
+                f'{payload.get("system", "")}\n\n{schema_hint}'.strip()
+            )
+        return payload
+
+    def _anthropic_headers(self) -> dict[str, str]:
+        return {
+            'Authorization': f'Bearer {self.config.api_key}',
+            'Content-Type': 'application/json',
+            'anthropic-version': ANTHROPIC_VERSION,
+            'api-key': self.config.api_key,
+            'x-api-key': self.config.api_key,
+        }
+
+    def _anthropic_messages_url(self) -> str:
+        return _join_url(_anthropic_base_url(self.config.base_url), '/v1/messages')
+
+    def _complete_anthropic_messages(
+        self,
+        *,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]],
+        output_schema: OutputSchemaConfig | None,
+    ) -> AssistantTurn:
+        payload = self._build_anthropic_payload(
+            messages=messages,
+            tools=tools,
+            stream=False,
+            output_schema=output_schema,
+        )
+        body = json.dumps(payload).encode('utf-8')
+        req = request.Request(
+            self._anthropic_messages_url(),
+            data=body,
+            headers=self._anthropic_headers(),
+            method='POST',
+        )
+        try:
+            with request.urlopen(req, timeout=self.config.timeout_seconds) as response:
+                raw = response.read()
+        except error.HTTPError as exc:
+            detail = exc.read().decode('utf-8', errors='replace')
+            raise OpenAICompatError(
+                f'HTTP {exc.code} from local model backend: {detail}'
+            ) from exc
+        except error.URLError as exc:
+            raise OpenAICompatError(
+                f'Unable to reach local model backend at {self.config.base_url}: {exc.reason}'
+            ) from exc
+
+        try:
+            response_payload = json.loads(raw.decode('utf-8'))
+        except json.JSONDecodeError as exc:
+            raise OpenAICompatError('Local model backend returned invalid JSON') from exc
+        if not isinstance(response_payload, dict):
+            raise OpenAICompatError('Local model backend returned malformed JSON payload')
+        return self._parse_anthropic_message_response(response_payload)
+
+    def _stream_anthropic_messages(
+        self,
+        *,
+        messages: list[dict[str, Any]],
+        tools: list[dict[str, Any]],
+        output_schema: OutputSchemaConfig | None,
+    ) -> Iterator[StreamEvent]:
+        payload = self._build_anthropic_payload(
+            messages=messages,
+            tools=tools,
+            stream=True,
+            output_schema=output_schema,
+        )
+        req = request.Request(
+            self._anthropic_messages_url(),
+            data=json.dumps(payload).encode('utf-8'),
+            headers=self._anthropic_headers(),
+            method='POST',
+        )
+        try:
+            with request.urlopen(req, timeout=self.config.timeout_seconds) as response:
+                yield StreamEvent(type='message_start')
+                tool_block_indexes: dict[int, int] = {}
+                next_tool_index = 0
+                finish_reason: str | None = None
+                for event_payload in self._iter_sse_payloads(response):
+                    event_type = event_payload.get('type')
+                    if event_type == 'message_start':
+                        message = event_payload.get('message')
+                        usage = _parse_usage(
+                            message.get('usage') if isinstance(message, dict) else None
+                        )
+                        if usage.total_tokens:
+                            yield StreamEvent(
+                                type='usage',
+                                usage=usage,
+                                raw_event=event_payload,
+                            )
+                        continue
+                    if event_type == 'content_block_start':
+                        block_index = event_payload.get('index')
+                        content_block = event_payload.get('content_block')
+                        if not isinstance(block_index, int) or not isinstance(content_block, dict):
+                            continue
+                        if content_block.get('type') != 'tool_use':
+                            continue
+                        tool_index = next_tool_index
+                        next_tool_index += 1
+                        tool_block_indexes[block_index] = tool_index
+                        tool_input = content_block.get('input')
+                        arguments = (
+                            json.dumps(tool_input, ensure_ascii=True)
+                            if isinstance(tool_input, dict) and tool_input
+                            else ''
+                        )
+                        yield StreamEvent(
+                            type='tool_call_delta',
+                            tool_call_index=tool_index,
+                            tool_call_id=(
+                                content_block.get('id')
+                                if isinstance(content_block.get('id'), str)
+                                else None
+                            ),
+                            tool_name=(
+                                content_block.get('name')
+                                if isinstance(content_block.get('name'), str)
+                                else None
+                            ),
+                            arguments_delta=arguments,
+                            raw_event=event_payload,
+                        )
+                        continue
+                    if event_type == 'content_block_delta':
+                        delta = event_payload.get('delta')
+                        if not isinstance(delta, dict):
+                            continue
+                        if delta.get('type') == 'text_delta':
+                            text = delta.get('text')
+                            if isinstance(text, str) and text:
+                                yield StreamEvent(
+                                    type='content_delta',
+                                    delta=text,
+                                    raw_event=event_payload,
+                                )
+                            continue
+                        if delta.get('type') == 'input_json_delta':
+                            block_index = event_payload.get('index')
+                            partial_json = delta.get('partial_json')
+                            if (
+                                isinstance(block_index, int)
+                                and block_index in tool_block_indexes
+                                and isinstance(partial_json, str)
+                                and partial_json
+                            ):
+                                yield StreamEvent(
+                                    type='tool_call_delta',
+                                    tool_call_index=tool_block_indexes[block_index],
+                                    arguments_delta=partial_json,
+                                    raw_event=event_payload,
+                                )
+                            continue
+                    if event_type == 'message_delta':
+                        delta = event_payload.get('delta')
+                        if isinstance(delta, dict) and isinstance(delta.get('stop_reason'), str):
+                            finish_reason = delta['stop_reason']
+                        usage = _parse_usage(event_payload.get('usage'))
+                        if usage.total_tokens:
+                            yield StreamEvent(
+                                type='usage',
+                                usage=usage,
+                                raw_event=event_payload,
+                            )
+                        continue
+                    if event_type == 'message_stop':
+                        yield StreamEvent(
+                            type='message_stop',
+                            finish_reason=finish_reason,
+                            raw_event=event_payload,
+                        )
+        except error.HTTPError as exc:
+            detail = exc.read().decode('utf-8', errors='replace')
+            raise OpenAICompatError(
+                f'HTTP {exc.code} from local model backend: {detail}'
+            ) from exc
+        except error.URLError as exc:
+            raise OpenAICompatError(
+                f'Unable to reach local model backend at {self.config.base_url}: {exc.reason}'
+            ) from exc
+
+    def _parse_anthropic_message_response(self, payload: dict[str, Any]) -> AssistantTurn:
+        raw_content = payload.get('content')
+        if not isinstance(raw_content, list):
+            raise OpenAICompatError('Anthropic backend returned no content blocks')
+        text_parts: list[str] = []
+        tool_calls: list[ToolCall] = []
+        for index, block in enumerate(raw_content):
+            if not isinstance(block, dict):
+                continue
+            if block.get('type') == 'text':
+                text = block.get('text')
+                if isinstance(text, str):
+                    text_parts.append(text)
+                continue
+            if block.get('type') == 'tool_use':
+                name = block.get('name')
+                if not isinstance(name, str) or not name:
+                    raise OpenAICompatError('Tool call missing function name')
+                call_id = block.get('id')
+                if not isinstance(call_id, str) or not call_id:
+                    call_id = f'call_{index}'
+                arguments = block.get('input')
+                if not isinstance(arguments, dict):
+                    arguments = {}
+                tool_calls.append(ToolCall(id=call_id, name=name, arguments=arguments))
+        finish_reason = payload.get('stop_reason')
+        if finish_reason is not None and not isinstance(finish_reason, str):
+            finish_reason = str(finish_reason)
+        return AssistantTurn(
+            content=''.join(text_parts),
+            tool_calls=tuple(tool_calls),
+            finish_reason=finish_reason,
+            raw_message=payload,
+            usage=_parse_usage(payload.get('usage')),
+        )
+
+    def _anthropic_assistant_blocks(self, message: dict[str, Any]) -> list[dict[str, Any]]:
+        blocks: list[dict[str, Any]] = []
+        content = _normalize_content(message.get('content'))
+        if content:
+            blocks.append({'type': 'text', 'text': content})
+        raw_tool_calls = message.get('tool_calls')
+        if isinstance(raw_tool_calls, list):
+            for index, raw_call in enumerate(raw_tool_calls):
+                if not isinstance(raw_call, dict):
+                    continue
+                function_block = raw_call.get('function')
+                if not isinstance(function_block, dict):
+                    continue
+                name = function_block.get('name')
+                if not isinstance(name, str) or not name:
+                    continue
+                call_id = raw_call.get('id')
+                if not isinstance(call_id, str) or not call_id:
+                    call_id = f'call_{index}'
+                try:
+                    arguments = _parse_tool_arguments(function_block.get('arguments'))
+                except OpenAICompatError:
+                    arguments = {}
+                blocks.append(
+                    {
+                        'type': 'tool_use',
+                        'id': call_id,
+                        'name': name,
+                        'input': arguments,
+                    }
+                )
+        return blocks
+
+    def _anthropic_text_blocks(self, content: Any) -> list[dict[str, Any]]:
+        normalized = _normalize_content(content)
+        if not normalized:
+            return []
+        return [{'type': 'text', 'text': normalized}]
+
+    def _anthropic_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        converted: list[dict[str, Any]] = []
+        for tool in tools:
+            if not isinstance(tool, dict):
+                continue
+            function_block = tool.get('function')
+            if not isinstance(function_block, dict):
+                continue
+            name = function_block.get('name')
+            if not isinstance(name, str) or not name:
+                continue
+            entry: dict[str, Any] = {
+                'name': name,
+                'input_schema': function_block.get('parameters') or {'type': 'object'},
+            }
+            description = function_block.get('description')
+            if isinstance(description, str) and description:
+                entry['description'] = description
+            converted.append(entry)
+        return converted
+
+    def _append_anthropic_message(
+        self,
+        messages: list[dict[str, Any]],
+        role: str,
+        blocks: list[dict[str, Any]],
+    ) -> None:
+        if not blocks:
+            return
+        if messages and messages[-1].get('role') == role:
+            previous = messages[-1].get('content')
+            if isinstance(previous, list):
+                previous.extend(blocks)
+                return
+        messages.append({'role': role, 'content': blocks})
+
     def _parse_tool_calls_from_message(self, message: dict[str, Any]) -> list[ToolCall]:
         tool_calls: list[ToolCall] = []
         raw_tool_calls = message.get('tool_calls')

tests/test_model_list.py

@@ -78,6 +78,7 @@ class ModelListTests(unittest.TestCase):
             [
                 'azure_openai/gpt-5',
                 'ppio/gemini-2.0-flash-20250609',
+                'ppio/pa/claude-opus-4-7',
                 'siliconflow/Pro/deepseek-ai/DeepSeek-V3',
                 'xiaomi/DeepSeek-R1-0528',
             ],

tests/test_openai_compat.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
+import json
 import unittest
+from unittest.mock import patch
 
 from src.openai_compat import (
     OpenAICompatClient,
     OpenAICompatError,
+    _anthropic_base_url,
     _build_response_format,
     _join_url,
     _normalize_content,
@@ -12,10 +15,45 @@ from src.openai_compat import (
     _parse_tool_arguments,
     _parse_usage,
     _temperature_for_model,
+    _uses_anthropic_messages_api,
 )
 from src.agent_types import ModelConfig, OutputSchemaConfig, UsageStats
 
 
+class FakeHTTPResponse:
+    def __init__(self, payload: dict[str, object]) -> None:
+        self.payload = payload
+
+    def read(self) -> bytes:
+        return json.dumps(self.payload).encode('utf-8')
+
+    def __enter__(self) -> 'FakeHTTPResponse':
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        return None
+
+
+class FakeStreamingHTTPResponse:
+    def __init__(self, payloads: list[dict[str, object]]) -> None:
+        self.lines: list[bytes] = []
+        for payload in payloads:
+            self.lines.append(b'event: message\n')
+            self.lines.append(f'data: {json.dumps(payload)}\n'.encode('utf-8'))
+            self.lines.append(b'\n')
+
+    def readline(self) -> bytes:
+        if not self.lines:
+            return b''
+        return self.lines.pop(0)
+
+    def __enter__(self) -> 'FakeStreamingHTTPResponse':
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        return None
+
+
 class TestJoinUrl(unittest.TestCase):
     def test_base_with_trailing_slash(self):
         self.assertEqual(_join_url('http://localhost:8000/', 'v1/chat'), 'http://localhost:8000/v1/chat')
@@ -210,5 +248,187 @@ class TestProviderTemperatureFloor(unittest.TestCase):
         self.assertEqual(payload['tool_choice'], 'auto')
 
 
+class TestAnthropicMessagesRouting(unittest.TestCase):
+    def test_ppio_pa_claude_uses_anthropic_messages_api(self):
+        self.assertTrue(
+            _uses_anthropic_messages_api(
+                'ppio/pa/claude-opus-4-7',
+                'http://model.mify.ai.srv/v1',
+            )
+        )
+        self.assertFalse(
+            _uses_anthropic_messages_api(
+                'ppio/gemini-2.5-pro',
+                'http://model.mify.ai.srv/v1',
+            )
+        )
+
+    def test_anthropic_base_url_is_derived_from_openai_base(self):
+        self.assertEqual(
+            _anthropic_base_url('http://model.mify.ai.srv/v1'),
+            'http://model.mify.ai.srv/anthropic',
+        )
+        self.assertEqual(
+            _anthropic_base_url('http://model.mify.ai.srv/anthropic'),
+            'http://model.mify.ai.srv/anthropic',
+        )
+
+    def test_anthropic_complete_converts_tools_and_parses_tool_use(self):
+        recorded: dict[str, object] = {}
+
+        def fake_urlopen(request_obj, timeout=None):  # noqa: ANN001
+            recorded['url'] = request_obj.full_url
+            recorded['payload'] = json.loads(request_obj.data.decode('utf-8'))
+            return FakeHTTPResponse(
+                {
+                    'id': 'msg_1',
+                    'type': 'message',
+                    'role': 'assistant',
+                    'content': [
+                        {'type': 'text', 'text': '我来读取文件。'},
+                        {
+                            'type': 'tool_use',
+                            'id': 'toolu_1',
+                            'name': 'read_file',
+                            'input': {'path': 'hello.txt'},
+                        },
+                    ],
+                    'stop_reason': 'tool_use',
+                    'usage': {'input_tokens': 10, 'output_tokens': 4},
+                }
+            )
+
+        client = OpenAICompatClient(
+            ModelConfig(
+                model='ppio/pa/claude-opus-4-7',
+                base_url='http://model.mify.ai.srv/v1',
+                api_key='token',
+            )
+        )
+        with patch('src.openai_compat.request.urlopen', side_effect=fake_urlopen):
+            turn = client.complete(
+                messages=[
+                    {'role': 'system', 'content': '你是工具型助手。'},
+                    {'role': 'user', 'content': '读取 hello.txt'},
+                ],
+                tools=[
+                    {
+                        'type': 'function',
+                        'function': {
+                            'name': 'read_file',
+                            'description': '读取文件',
+                            'parameters': {
+                                'type': 'object',
+                                'properties': {'path': {'type': 'string'}},
+                                'required': ['path'],
+                            },
+                        },
+                    }
+                ],
+            )
+
+        self.assertEqual(
+            recorded['url'],
+            'http://model.mify.ai.srv/anthropic/v1/messages',
+        )
+        payload = recorded['payload']
+        self.assertEqual(payload['model'], 'ppio/pa/claude-opus-4-7')
+        self.assertEqual(payload['system'], '你是工具型助手。')
+        self.assertEqual(payload['tools'][0]['name'], 'read_file')
+        self.assertEqual(payload['tools'][0]['input_schema']['required'], ['path'])
+        self.assertEqual(turn.content, '我来读取文件。')
+        self.assertEqual(turn.finish_reason, 'tool_use')
+        self.assertEqual(turn.tool_calls[0].id, 'toolu_1')
+        self.assertEqual(turn.tool_calls[0].arguments, {'path': 'hello.txt'})
+        self.assertEqual(turn.usage.input_tokens, 10)
+
+    def test_anthropic_stream_parses_text_usage_and_tool_use(self):
+        payloads = [
+            {
+                'type': 'message_start',
+                'message': {'usage': {'input_tokens': 7, 'output_tokens': 1}},
+            },
+            {
+                'type': 'content_block_start',
+                'index': 0,
+                'content_block': {'type': 'text', 'text': ''},
+            },
+            {
+                'type': 'content_block_delta',
+                'index': 0,
+                'delta': {'type': 'text_delta', 'text': '读取'},
+            },
+            {
+                'type': 'content_block_start',
+                'index': 1,
+                'content_block': {
+                    'type': 'tool_use',
+                    'id': 'toolu_1',
+                    'name': 'read_file',
+                    'input': {},
+                },
+            },
+            {
+                'type': 'content_block_delta',
+                'index': 1,
+                'delta': {'type': 'input_json_delta', 'partial_json': '{"path":'},
+            },
+            {
+                'type': 'content_block_delta',
+                'index': 1,
+                'delta': {'type': 'input_json_delta', 'partial_json': '"hello.txt"}'},
+            },
+            {
+                'type': 'message_delta',
+                'delta': {'stop_reason': 'tool_use'},
+                'usage': {'output_tokens': 5},
+            },
+            {'type': 'message_stop'},
+        ]
+
+        def fake_urlopen(request_obj, timeout=None):  # noqa: ANN001
+            return FakeStreamingHTTPResponse(payloads)
+
+        client = OpenAICompatClient(
+            ModelConfig(
+                model='ppio/pa/claude-opus-4-7',
+                base_url='http://model.mify.ai.srv/v1',
+                api_key='token',
+            )
+        )
+        with patch('src.openai_compat.request.urlopen', side_effect=fake_urlopen):
+            events = list(
+                client.stream(
+                    messages=[{'role': 'user', 'content': '读取 hello.txt'}],
+                    tools=[
+                        {
+                            'type': 'function',
+                            'function': {
+                                'name': 'read_file',
+                                'parameters': {'type': 'object'},
+                            },
+                        }
+                    ],
+                )
+            )
+
+        self.assertEqual(events[0].type, 'message_start')
+        self.assertEqual(
+            ''.join(event.delta for event in events if event.type == 'content_delta'),
+            '读取',
+        )
+        tool_events = [event for event in events if event.type == 'tool_call_delta']
+        self.assertEqual(tool_events[0].tool_call_index, 0)
+        self.assertEqual(tool_events[0].tool_call_id, 'toolu_1')
+        self.assertEqual(tool_events[0].tool_name, 'read_file')
+        self.assertEqual(
+            ''.join(event.arguments_delta for event in tool_events),
+            '{"path":"hello.txt"}',
+        )
+        self.assertTrue(any(event.type == 'usage' for event in events))
+        self.assertEqual(events[-1].type, 'message_stop')
+        self.assertEqual(events[-1].finish_reason, 'tool_use')
+
+
 if __name__ == '__main__':
     unittest.main()