fix: handle missing anthropic usage on filtered responses

astrbot/core/provider/sources/anthropic_source.py

@@ -302,12 +302,14 @@ class ProviderAnthropic(Provider):
 
         return system_prompt, new_messages
 
-    def _extract_usage(self, usage: Usage) -> TokenUsage:
+    def _extract_usage(self, usage: Usage | None) -> TokenUsage:
+        if usage is None:
+            return TokenUsage()
         # https://docs.claude.com/en/docs/build-with-claude/prompt-caching#tracking-cache-performance
         return TokenUsage(
             input_other=usage.input_tokens or 0,
             input_cached=usage.cache_read_input_tokens or 0,
-            output=usage.output_tokens,
+            output=usage.output_tokens or 0,
         )
 
     def _update_usage(self, token_usage: TokenUsage, usage: MessageDeltaUsage) -> None:

tests/test_anthropic_kimi_code_provider.py

@@ -483,6 +483,40 @@ def _setup_provider_with_mock_client(monkeypatch) -> anthropic_source.ProviderAn
     return provider
 
 
+@pytest.mark.asyncio
+async def test_query_handles_none_usage_when_content_filtered(monkeypatch):
+    provider = _setup_provider_with_mock_client(monkeypatch)
+    content_filter_message = (
+        "The request was rejected because it was considered high risk"
+    )
+
+    class _FakeMessageBlock:
+        def __init__(self, text: str):
+            self.type = "text"
+            self.text = text
+
+    class _FakeMessage:
+        def __init__(self):
+            self.id = "msg_content_filter"
+            self.content = [_FakeMessageBlock(content_filter_message)]
+            self.stop_reason = "content_filter"
+            self.usage = None
+
+    async def fake_create(**kwargs):
+        return _FakeMessage()
+
+    monkeypatch.setattr(anthropic_source, "Message", _FakeMessage)
+    provider.client.messages.create = fake_create
+
+    llm_response = await provider.text_chat(prompt="test")
+
+    assert llm_response.completion_text == content_filter_message
+    assert llm_response.usage is not None
+    assert llm_response.usage.input_other == 0
+    assert llm_response.usage.input_cached == 0
+    assert llm_response.usage.output == 0
+
+
 @pytest.mark.asyncio
 async def test_tool_choice_auto_converts_to_dict(monkeypatch):
     """tool_choice='auto' 应转换为 {'type': 'auto'}"""