Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-04-02 00:20:30 +01:00
Code Activity

Add support for web search dynamic filtering for Anthropic (#164116)

Browse Source
This commit is contained in:
Denis Shulyaka
2026-03-30 20:40:56 +03:00
committed by GitHub
parent 14cb42349a
commit dc111a475e
6 changed files with 631 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
homeassistant/components/anthropic/const.py
View File

@@ -71,6 +71,16 @@ CODE_EXECUTION_UNSUPPORTED_MODELS = [
"claude-3-haiku",
]
PROGRAMMATIC_TOOL_CALLING_UNSUPPORTED_MODELS = [
"claude-haiku-4-5",
"claude-opus-4-1",
"claude-opus-4-0",
"claude-opus-4-20250514",
"claude-sonnet-4-0",
"claude-sonnet-4-20250514",
"claude-3-haiku",
]
DEPRECATED_MODELS = [
"claude-3",
]

70
homeassistant/components/anthropic/entity.py
View File

@@ -19,6 +19,8 @@ from anthropic.types import (
CitationsWebSearchResultLocation,
CitationWebSearchResultLocationParam,
CodeExecutionTool20250825Param,
CodeExecutionToolResultBlock,
CodeExecutionToolResultBlockParamContentParam,
Container,
ContentBlockParam,
DocumentBlockParam,
@@ -61,15 +63,16 @@ from anthropic.types import (
ToolUseBlockParam,
Usage,
WebSearchTool20250305Param,
WebSearchTool20260209Param,
WebSearchToolResultBlock,
WebSearchToolResultBlockParamContentParam,
)
from anthropic.types.bash_code_execution_tool_result_block_param import (
Content as BashCodeExecutionToolResultContentParam,
Content as BashCodeExecutionToolResultBlockParamContentParam,
)
from anthropic.types.message_create_params import MessageCreateParamsStreaming
from anthropic.types.text_editor_code_execution_tool_result_block_param import (
Content as TextEditorCodeExecutionToolResultContentParam,
Content as TextEditorCodeExecutionToolResultBlockParamContentParam,
)
import voluptuous as vol
from voluptuous_openapi import convert
@@ -105,6 +108,7 @@ from .const import (
MIN_THINKING_BUDGET,
NON_ADAPTIVE_THINKING_MODELS,
NON_THINKING_MODELS,
PROGRAMMATIC_TOOL_CALLING_UNSUPPORTED_MODELS,
UNSUPPORTED_STRUCTURED_OUTPUT_MODELS,
)
@@ -224,12 +228,22 @@ def _convert_content(
},
),
}
elif content.tool_name == "code_execution":
tool_result_block = {
"type": "code_execution_tool_result",
"tool_use_id": content.tool_call_id,
"content": cast(
CodeExecutionToolResultBlockParamContentParam,
content.tool_result,
),
}
elif content.tool_name == "bash_code_execution":
tool_result_block = {
"type": "bash_code_execution_tool_result",
"tool_use_id": content.tool_call_id,
"content": cast(
BashCodeExecutionToolResultContentParam, content.tool_result
BashCodeExecutionToolResultBlockParamContentParam,
content.tool_result,
),
}
elif content.tool_name == "text_editor_code_execution":
@@ -237,7 +251,7 @@ def _convert_content(
"type": "text_editor_code_execution_tool_result",
"tool_use_id": content.tool_call_id,
"content": cast(
TextEditorCodeExecutionToolResultContentParam,
TextEditorCodeExecutionToolResultBlockParamContentParam,
content.tool_result,
),
}
@@ -368,6 +382,7 @@ def _convert_content(
name=cast(
Literal[
"web_search",
"code_execution",
"bash_code_execution",
"text_editor_code_execution",
],
@@ -379,6 +394,7 @@ def _convert_content(
and tool_call.tool_name
in [
"web_search",
"code_execution",
"bash_code_execution",
"text_editor_code_execution",
]
@@ -470,7 +486,7 @@ async def _transform_stream( # noqa: C901 - This is complex, but better to have
type="tool_use",
id=response.content_block.id,
name=response.content_block.name,
input={},
input=response.content_block.input or {},
)
current_tool_args = ""
if response.content_block.name == output_tool:
@@ -532,13 +548,14 @@ async def _transform_stream( # noqa: C901 - This is complex, but better to have
type="server_tool_use",
id=response.content_block.id,
name=response.content_block.name,
input={},
input=response.content_block.input or {},
)
current_tool_args = ""
elif isinstance(
response.content_block,
(
WebSearchToolResultBlock,
CodeExecutionToolResultBlock,
BashCodeExecutionToolResultBlock,
TextEditorCodeExecutionToolResultBlock,
),
@@ -594,13 +611,13 @@ async def _transform_stream( # noqa: C901 - This is complex, but better to have
current_tool_block = None
continue
tool_args = json.loads(current_tool_args) if current_tool_args else {}
current_tool_block["input"] = tool_args
current_tool_block["input"] |= tool_args
yield {
"tool_calls": [
llm.ToolInput(
id=current_tool_block["id"],
tool_name=current_tool_block["name"],
tool_args=tool_args,
tool_args=current_tool_block["input"],
external=current_tool_block["type"] == "server_tool_use",
)
]
@@ -735,19 +752,34 @@ class AnthropicBaseLLMEntity(Entity):
]
if options.get(CONF_CODE_EXECUTION):
tools.append(
CodeExecutionTool20250825Param(
name="code_execution",
type="code_execution_20250825",
),
)
# The `web_search_20260209` tool automatically enables `code_execution_20260120` tool
if model.startswith(
tuple(PROGRAMMATIC_TOOL_CALLING_UNSUPPORTED_MODELS)
) or not options.get(CONF_WEB_SEARCH):
tools.append(
CodeExecutionTool20250825Param(
name="code_execution",
type="code_execution_20250825",
),
)
if options.get(CONF_WEB_SEARCH):
web_search = WebSearchTool20250305Param(
name="web_search",
type="web_search_20250305",
max_uses=options.get(CONF_WEB_SEARCH_MAX_USES),
)
if model.startswith(
tuple(PROGRAMMATIC_TOOL_CALLING_UNSUPPORTED_MODELS)
) or not options.get(CONF_CODE_EXECUTION):
web_search: WebSearchTool20250305Param | WebSearchTool20260209Param = (
WebSearchTool20250305Param(
name="web_search",
type="web_search_20250305",
max_uses=options.get(CONF_WEB_SEARCH_MAX_USES),
)
)
else:
web_search = WebSearchTool20260209Param(
name="web_search",
type="web_search_20260209",
max_uses=options.get(CONF_WEB_SEARCH_MAX_USES),
)
if options.get(CONF_WEB_SEARCH_USER_LOCATION):
web_search["user_location"] = {
"type": "approximate",

72
tests/components/anthropic/__init__.py
View File

@@ -1,5 +1,7 @@
"""Tests for the Anthropic integration."""
from typing import Any
from anthropic.types import (
BashCodeExecutionOutputBlock,
BashCodeExecutionResultBlock,
@@ -7,6 +9,9 @@ from anthropic.types import (
BashCodeExecutionToolResultError,
BashCodeExecutionToolResultErrorCode,
CitationsDelta,
CodeExecutionToolResultBlock,
CodeExecutionToolResultBlockContent,
DirectCaller,
InputJSONDelta,
RawContentBlockDeltaEvent,
RawContentBlockStartEvent,
@@ -24,7 +29,9 @@ from anthropic.types import (
ToolUseBlock,
WebSearchResultBlock,
WebSearchToolResultBlock,
WebSearchToolResultError,
)
from anthropic.types.server_tool_use_block import Caller
from anthropic.types.text_editor_code_execution_tool_result_block import (
Content as TextEditorCodeExecutionToolResultBlockContent,
)
@@ -138,45 +145,58 @@ def create_tool_use_block(
def create_server_tool_use_block(
index: int, id: str, name: str, args_parts: list[str]
index: int,
id: str,
name: str,
input_parts: list[str] | dict[str, Any],
caller: Caller | None = None,
) -> list[RawMessageStreamEvent]:
"""Create a server tool use block."""
if caller is None:
caller = DirectCaller(type="direct")
return [
RawContentBlockStartEvent(
type="content_block_start",
content_block=ServerToolUseBlock(
type="server_tool_use", id=id, input={}, name=name
type="server_tool_use",
id=id,
input=input_parts if isinstance(input_parts, dict) else {},
name=name,
caller=caller,
),
index=index,
),
*[
RawContentBlockDeltaEvent(
delta=InputJSONDelta(type="input_json_delta", partial_json=args_part),
delta=InputJSONDelta(type="input_json_delta", partial_json=input_part),
index=index,
type="content_block_delta",
)
for args_part in args_parts
for input_part in (input_parts if isinstance(input_parts, list) else [])
],
RawContentBlockStopEvent(index=index, type="content_block_stop"),
]
def create_web_search_block(
index: int, id: str, query_parts: list[str]
) -> list[RawMessageStreamEvent]:
"""Create a server tool use block for web search."""
return create_server_tool_use_block(index, id, "web_search", query_parts)
def create_web_search_result_block(
index: int, id: str, results: list[WebSearchResultBlock]
index: int,
id: str,
results: list[WebSearchResultBlock] | WebSearchToolResultError,
caller: Caller | None = None,
) -> list[RawMessageStreamEvent]:
"""Create a server tool result block for web search results."""
if caller is None:
caller = DirectCaller(type="direct")
return [
RawContentBlockStartEvent(
type="content_block_start",
content_block=WebSearchToolResultBlock(
type="web_search_tool_result", tool_use_id=id, content=results
type="web_search_tool_result",
tool_use_id=id,
content=results,
caller=caller,
),
index=index,
),
@@ -184,11 +204,20 @@ def create_web_search_result_block(
]
def create_bash_code_execution_block(
index: int, id: str, command_parts: list[str]
def create_code_execution_result_block(
index: int, id: str, content: CodeExecutionToolResultBlockContent
) -> list[RawMessageStreamEvent]:
"""Create a server tool use block for bash code execution."""
return create_server_tool_use_block(index, id, "bash_code_execution", command_parts)
"""Create a server tool result block for code execution results."""
return [
RawContentBlockStartEvent(
type="content_block_start",
content_block=CodeExecutionToolResultBlock(
type="code_execution_tool_result", tool_use_id=id, content=content
),
index=index,
),
RawContentBlockStopEvent(index=index, type="content_block_stop"),
]
def create_bash_code_execution_result_block(
@@ -226,15 +255,6 @@ def create_bash_code_execution_result_block(
]
def create_text_editor_code_execution_block(
index: int, id: str, command_parts: list[str]
) -> list[RawMessageStreamEvent]:
"""Create a server tool use block for text editor code execution."""
return create_server_tool_use_block(
index, id, "text_editor_code_execution", command_parts
)
def create_text_editor_code_execution_result_block(
index: int,
id: str,

6
tests/components/anthropic/conftest.py
View File

@@ -215,7 +215,11 @@ def mock_create_stream() -> Generator[AsyncMock]:
isinstance(event, RawContentBlockStartEvent)
and isinstance(event.content_block, ServerToolUseBlock)
and event.content_block.name
in ["bash_code_execution", "text_editor_code_execution"]
in [
"code_execution",
"bash_code_execution",
"text_editor_code_execution",
]
):
container = Container(
id=kwargs.get("container_id", "container_1234567890ABCDEFGHIJKLMN"),

283
tests/components/anthropic/snapshots/test_conversation.ambr
View File

@@ -1508,3 +1508,286 @@
}),
])
# ---
# name: test_web_search_dynamic_filtering
list([
dict({
'attachments': None,
'content': 'Who won the Nobel for Chemistry in 2025?',
'created': HAFakeDatetime(2025, 10, 31, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'user',
}),
dict({
'agent_id': 'conversation.claude_conversation',
'content': None,
'created': HAFakeDatetime(2025, 10, 31, 12, 0, tzinfo=datetime.timezone.utc),
'native': dict({
'citation_details': list([
]),
'container': None,
'redacted_thinking': None,
'thinking_signature': 'ErUBCkYIARgCIkCYXaVNJShe3A86Hp7XUzh9YsCYBbJTbQsrklTAPtJ2sP/NoB6tSzpK/nTL6CjSo2R6n0KNBIg5MH6asM2R/kmaEgyB/X1FtZq5OQAC7jUaDEPWCdcwGQ4RaBy5wiIwmRxExIlDhoY6tILoVPnOExkC/0igZxHEwxK8RU/fmw0b+o+TwAarzUitwzbo21E5Kh3pa3I6yqVROf1t2F8rFocNUeCegsWV/ytwYV+ayA==',
}),
'role': 'assistant',
'thinking_content': 'Let me search for this information.',
'tool_calls': list([
dict({
'external': True,
'id': 'srvtoolu_01Qh4oTgPkNfRxjJA3N6jgzT',
'tool_args': dict({
'code': '''
import json
result = await web_search({"query": "Nobel Prize chemistry 2025 winner"})
parsed = json.loads(result)
for r in parsed[:3]:
print(r.get("title", ""))
print(r.get("content", "")[:300])
print("---")
''',
}),
'tool_name': 'code_execution',
}),
dict({
'external': True,
'id': 'srvtoolu_016vjte6G4Lj6yzLc2ak1vY4',
'tool_args': dict({
'query': 'Nobel Prize chemistry 2025 winner',
}),
'tool_name': 'web_search',
}),
]),
}),
dict({
'agent_id': 'conversation.claude_conversation',
'created': HAFakeDatetime(2025, 10, 31, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'tool_result',
'tool_call_id': 'srvtoolu_016vjte6G4Lj6yzLc2ak1vY4',
'tool_name': 'web_search',
'tool_result': dict({
'content': list([
dict({
'encrypted_content': 'ABCDEFG',
'page_age': None,
'title': 'Press release: Nobel Prize in Chemistry 2025 - Example.com',
'type': 'web_search_result',
'url': 'https://www.example.com/prizes/chemistry/2025/press-release/',
}),
dict({
'encrypted_content': 'ABCDEFG',
'page_age': None,
'title': 'Nobel Prize in Chemistry 2025 - NewsSite.com',
'type': 'web_search_result',
'url': 'https://www.newssite.com/prizes/chemistry/2025/summary/',
}),
]),
}),
}),
dict({
'agent_id': 'conversation.claude_conversation',
'created': HAFakeDatetime(2025, 10, 31, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'tool_result',
'tool_call_id': 'srvtoolu_01Qh4oTgPkNfRxjJA3N6jgzT',
'tool_name': 'code_execution',
'tool_result': dict({
'content': list([
]),
'encrypted_stdout': 'EuQJCioIDRgCIiRj',
'return_code': 0,
'stderr': '',
'type': 'encrypted_code_execution_result',
}),
}),
dict({
'agent_id': 'conversation.claude_conversation',
'content': 'The 2025 Nobel Prize in Chemistry was awarded jointly to **Susumu Kitagawa**, **Richard Robson**, and **Omar M. Yaghi** "for the development of metal–organic frameworks."',
'created': HAFakeDatetime(2025, 10, 31, 12, 0, tzinfo=datetime.timezone.utc),
'native': dict({
'citation_details': list([
]),
'container': Container(id='container_1234567890ABCDEFGHIJKLMN', expires_at=HAFakeDatetime(2025, 10, 31, 12, 5, tzinfo=datetime.timezone.utc)),
'redacted_thinking': None,
'thinking_signature': None,
}),
'role': 'assistant',
'thinking_content': None,
'tool_calls': None,
}),
])
# ---
# name: test_web_search_dynamic_filtering.1
list([
dict({
'content': 'Who won the Nobel for Chemistry in 2025?',
'role': 'user',
}),
dict({
'content': list([
dict({
'signature': 'ErUBCkYIARgCIkCYXaVNJShe3A86Hp7XUzh9YsCYBbJTbQsrklTAPtJ2sP/NoB6tSzpK/nTL6CjSo2R6n0KNBIg5MH6asM2R/kmaEgyB/X1FtZq5OQAC7jUaDEPWCdcwGQ4RaBy5wiIwmRxExIlDhoY6tILoVPnOExkC/0igZxHEwxK8RU/fmw0b+o+TwAarzUitwzbo21E5Kh3pa3I6yqVROf1t2F8rFocNUeCegsWV/ytwYV+ayA==',
'thinking': 'Let me search for this information.',
'type': 'thinking',
}),
dict({
'id': 'srvtoolu_01Qh4oTgPkNfRxjJA3N6jgzT',
'input': dict({
'code': '''
import json
result = await web_search({"query": "Nobel Prize chemistry 2025 winner"})
parsed = json.loads(result)
for r in parsed[:3]:
print(r.get("title", ""))
print(r.get("content", "")[:300])
print("---")
''',
}),
'name': 'code_execution',
'type': 'server_tool_use',
}),
dict({
'id': 'srvtoolu_016vjte6G4Lj6yzLc2ak1vY4',
'input': dict({
'query': 'Nobel Prize chemistry 2025 winner',
}),
'name': 'web_search',
'type': 'server_tool_use',
}),
dict({
'content': list([
dict({
'encrypted_content': 'ABCDEFG',
'page_age': None,
'title': 'Press release: Nobel Prize in Chemistry 2025 - Example.com',
'type': 'web_search_result',
'url': 'https://www.example.com/prizes/chemistry/2025/press-release/',
}),
dict({
'encrypted_content': 'ABCDEFG',
'page_age': None,
'title': 'Nobel Prize in Chemistry 2025 - NewsSite.com',
'type': 'web_search_result',
'url': 'https://www.newssite.com/prizes/chemistry/2025/summary/',
}),
]),
'tool_use_id': 'srvtoolu_016vjte6G4Lj6yzLc2ak1vY4',
'type': 'web_search_tool_result',
}),
dict({
'content': dict({
'content': list([
]),
'encrypted_stdout': 'EuQJCioIDRgCIiRj',
'return_code': 0,
'stderr': '',
'type': 'encrypted_code_execution_result',
}),
'tool_use_id': 'srvtoolu_01Qh4oTgPkNfRxjJA3N6jgzT',
'type': 'code_execution_tool_result',
}),
dict({
'text': 'The 2025 Nobel Prize in Chemistry was awarded jointly to **Susumu Kitagawa**, **Richard Robson**, and **Omar M. Yaghi** "for the development of metal–organic frameworks."',
'type': 'text',
}),
]),
'role': 'assistant',
}),
])
# ---
# name: test_web_search_error
list([
dict({
'attachments': None,
'content': "What's on the news today?",
'created': HAFakeDatetime(2025, 10, 31, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'user',
}),
dict({
'agent_id': 'conversation.claude_conversation',
'content': "To get today's news, I'll perform a web search",
'created': HAFakeDatetime(2025, 10, 31, 12, 0, tzinfo=datetime.timezone.utc),
'native': dict({
'citation_details': list([
]),
'container': None,
'redacted_thinking': None,
'thinking_signature': 'ErUBCkYIARgCIkCYXaVNJShe3A86Hp7XUzh9YsCYBbJTbQsrklTAPtJ2sP/NoB6tSzpK/nTL6CjSo2R6n0KNBIg5MH6asM2R/kmaEgyB/X1FtZq5OQAC7jUaDEPWCdcwGQ4RaBy5wiIwmRxExIlDhoY6tILoVPnOExkC/0igZxHEwxK8RU/fmw0b+o+TwAarzUitwzbo21E5Kh3pa3I6yqVROf1t2F8rFocNUeCegsWV/ytwYV+ayA==',
}),
'role': 'assistant',
'thinking_content': "The user is asking about today's news, which requires current, real-time information. This is clearly something that requires recent information beyond my knowledge cutoff. I should use the web_search tool to find today's news.",
'tool_calls': list([
dict({
'external': True,
'id': 'srvtoolu_12345ABC',
'tool_args': dict({
'query': "today's news",
}),
'tool_name': 'web_search',
}),
]),
}),
dict({
'agent_id': 'conversation.claude_conversation',
'created': HAFakeDatetime(2025, 10, 31, 12, 0, tzinfo=datetime.timezone.utc),
'role': 'tool_result',
'tool_call_id': 'srvtoolu_12345ABC',
'tool_name': 'web_search',
'tool_result': dict({
'error_code': 'too_many_requests',
'type': 'web_search_tool_result_error',
}),
}),
dict({
'agent_id': 'conversation.claude_conversation',
'content': 'I am unable to perform the web search at this time.',
'created': HAFakeDatetime(2025, 10, 31, 12, 0, tzinfo=datetime.timezone.utc),
'native': None,
'role': 'assistant',
'thinking_content': None,
'tool_calls': None,
}),
])
# ---
# name: test_web_search_error.1
list([
dict({
'content': "What's on the news today?",
'role': 'user',
}),
dict({
'content': list([
dict({
'signature': 'ErUBCkYIARgCIkCYXaVNJShe3A86Hp7XUzh9YsCYBbJTbQsrklTAPtJ2sP/NoB6tSzpK/nTL6CjSo2R6n0KNBIg5MH6asM2R/kmaEgyB/X1FtZq5OQAC7jUaDEPWCdcwGQ4RaBy5wiIwmRxExIlDhoY6tILoVPnOExkC/0igZxHEwxK8RU/fmw0b+o+TwAarzUitwzbo21E5Kh3pa3I6yqVROf1t2F8rFocNUeCegsWV/ytwYV+ayA==',
'thinking': "The user is asking about today's news, which requires current, real-time information. This is clearly something that requires recent information beyond my knowledge cutoff. I should use the web_search tool to find today's news.",
'type': 'thinking',
}),
dict({
'text': "To get today's news, I'll perform a web search",
'type': 'text',
}),
dict({
'id': 'srvtoolu_12345ABC',
'input': dict({
'query': "today's news",
}),
'name': 'web_search',
'type': 'server_tool_use',
}),
dict({
'content': dict({
'error_code': 'too_many_requests',
'type': 'web_search_tool_result_error',
}),
'tool_use_id': 'srvtoolu_12345ABC',
'type': 'web_search_tool_result',
}),
dict({
'text': 'I am unable to perform the web search at this time.',
'type': 'text',
}),
]),
'role': 'assistant',
}),
])
# ---

246
tests/components/anthropic/test_conversation.py
View File

@@ -8,7 +8,9 @@ from anthropic import AuthenticationError, RateLimitError
from anthropic.types import (
CitationsWebSearchResultLocation,
CitationWebSearchResultLocationParam,
EncryptedCodeExecutionResultBlock,
Message,
ServerToolCaller20260120,
TextBlock,
TextEditorCodeExecutionCreateResultBlock,
TextEditorCodeExecutionStrReplaceResultBlock,
@@ -16,6 +18,7 @@ from anthropic.types import (
TextEditorCodeExecutionViewResultBlock,
Usage,
WebSearchResultBlock,
WebSearchToolResultError,
)
from anthropic.types.text_editor_code_execution_tool_result_block import (
Content as TextEditorCodeExecutionToolResultBlockContent,
@@ -51,15 +54,14 @@ from homeassistant.setup import async_setup_component
from homeassistant.util import ulid as ulid_util
from . import (
create_bash_code_execution_block,
create_bash_code_execution_result_block,
create_code_execution_result_block,
create_content_block,
create_redacted_thinking_block,
create_text_editor_code_execution_block,
create_server_tool_use_block,
create_text_editor_code_execution_result_block,
create_thinking_block,
create_tool_use_block,
create_web_search_block,
create_web_search_result_block,
)
@@ -864,7 +866,7 @@ async def test_web_search(
next(iter(mock_config_entry.subentries.values())),
data={
CONF_LLM_HASS_API: llm.LLM_API_ASSIST,
CONF_CHAT_MODEL: "claude-sonnet-4-5",
CONF_CHAT_MODEL: "claude-sonnet-4-0",
CONF_WEB_SEARCH: True,
CONF_WEB_SEARCH_MAX_USES: 5,
CONF_WEB_SEARCH_USER_LOCATION: True,
@@ -909,9 +911,10 @@ async def test_web_search(
*create_content_block(
1, ["To get today's news, I'll perform a web search"]
),
*create_web_search_block(
*create_server_tool_use_block(
2,
"srvtoolu_12345ABC",
"web_search",
["", '{"que', 'ry"', ": \"today's", ' news"}'],
),
*create_web_search_result_block(3, "srvtoolu_12345ABC", web_search_results),
@@ -984,6 +987,226 @@ async def test_web_search(
assert mock_create_stream.call_args.kwargs["messages"] == snapshot
@freeze_time("2025-10-31 12:00:00")
async def test_web_search_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_init_component,
mock_create_stream: AsyncMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test web search error."""
hass.config_entries.async_update_subentry(
mock_config_entry,
next(iter(mock_config_entry.subentries.values())),
data={
CONF_LLM_HASS_API: llm.LLM_API_ASSIST,
CONF_CHAT_MODEL: "claude-sonnet-4-0",
CONF_WEB_SEARCH: True,
CONF_WEB_SEARCH_MAX_USES: 5,
CONF_WEB_SEARCH_USER_LOCATION: True,
CONF_WEB_SEARCH_CITY: "San Francisco",
CONF_WEB_SEARCH_REGION: "California",
CONF_WEB_SEARCH_COUNTRY: "US",
CONF_WEB_SEARCH_TIMEZONE: "America/Los_Angeles",
},
)
web_search_results = WebSearchToolResultError(
type="web_search_tool_result_error",
error_code="too_many_requests",
)
mock_create_stream.return_value = [
(
*create_thinking_block(
0,
[
"The user is",
" asking about today's news, which",
" requires current, real-time information",
". This is clearly something that requires recent",
" information beyond my knowledge cutoff.",
" I should use the web",
"_search tool to fin",
"d today's news.",
],
),
*create_content_block(
1, ["To get today's news, I'll perform a web search"]
),
*create_server_tool_use_block(
2,
"srvtoolu_12345ABC",
"web_search",
["", '{"que', 'ry"', ": \"today's", ' news"}'],
),
*create_web_search_result_block(3, "srvtoolu_12345ABC", web_search_results),
*create_content_block(
4,
["I am unable to perform the web search at this time."],
),
)
]
result = await conversation.async_converse(
hass,
"What's on the news today?",
None,
Context(),
agent_id="conversation.claude_conversation",
)
chat_log = hass.data.get(conversation.chat_log.DATA_CHAT_LOGS).get(
result.conversation_id
)
# Don't test the prompt because it's not deterministic
assert chat_log.content[1:] == snapshot
assert mock_create_stream.call_args.kwargs["messages"] == snapshot
@freeze_time("2025-10-31 12:00:00")
async def test_web_search_dynamic_filtering(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_init_component,
mock_create_stream: AsyncMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test web search with dynamic filtering of the results."""
hass.config_entries.async_update_subentry(
mock_config_entry,
next(iter(mock_config_entry.subentries.values())),
data={
CONF_LLM_HASS_API: llm.LLM_API_ASSIST,
CONF_CHAT_MODEL: "claude-opus-4-6",
CONF_CODE_EXECUTION: True,
CONF_WEB_SEARCH: True,
CONF_WEB_SEARCH_MAX_USES: 5,
CONF_WEB_SEARCH_USER_LOCATION: True,
CONF_WEB_SEARCH_CITY: "San Francisco",
CONF_WEB_SEARCH_REGION: "California",
CONF_WEB_SEARCH_COUNTRY: "US",
CONF_WEB_SEARCH_TIMEZONE: "America/Los_Angeles",
},
)
web_search_results = [
WebSearchResultBlock(
type="web_search_result",
title="Press release: Nobel Prize in Chemistry 2025 - Example.com",
url="https://www.example.com/prizes/chemistry/2025/press-release/",
page_age=None,
encrypted_content="ABCDEFG",
),
WebSearchResultBlock(
type="web_search_result",
title="Nobel Prize in Chemistry 2025 - NewsSite.com",
url="https://www.newssite.com/prizes/chemistry/2025/summary/",
page_age=None,
encrypted_content="ABCDEFG",
),
]
content = EncryptedCodeExecutionResultBlock(
type="encrypted_code_execution_result",
content=[],
encrypted_stdout="EuQJCioIDRgCIiRj",
return_code=0,
stderr="",
)
mock_create_stream.return_value = [
(
*create_thinking_block(
0, ["Let", " me search", " for this", " information.", ""]
),
*create_server_tool_use_block(
1,
"srvtoolu_01Qh4oTgPkNfRxjJA3N6jgzT",
"code_execution",
[
"",
'{"code": "\\nimport',
" json",
"\\nresult",
" = await",
" web",
'_search({\\"',
'query\\": \\"Nobel Prize chemistry',
" 2025 ",
'winner\\"})\\nparsed',
" = json.loads(result)",
"\\nfor",
" r",
" in parsed[:",
"3",
"]:\\n print(r.",
'get(\\"title',
'\\", \\"\\"))',
'\\n print(r.get(\\"',
"content",
'\\", \\"\\")',
"[:300",
'])\\n print(\\"---\\")',
"\\n",
'"}',
],
),
*create_server_tool_use_block(
2,
"srvtoolu_016vjte6G4Lj6yzLc2ak1vY4",
"web_search",
{"query": "Nobel Prize chemistry 2025 winner"},
caller=ServerToolCaller20260120(
type="code_execution_20260120",
tool_id="srvtoolu_01Qh4oTgPkNfRxjJA3N6jgzT",
),
),
*create_web_search_result_block(
3,
"srvtoolu_016vjte6G4Lj6yzLc2ak1vY4",
web_search_results,
caller=ServerToolCaller20260120(
type="code_execution_20260120",
tool_id="srvtoolu_01Qh4oTgPkNfRxjJA3N6jgzT",
),
),
*create_code_execution_result_block(
4, "srvtoolu_01Qh4oTgPkNfRxjJA3N6jgzT", content
),
*create_content_block(
5,
[
"The ",
"2025 Nobel Prize in Chemistry was",
" awarded jointly to **",
"Susumu Kitagawa**,",
" **",
"Richard Robson**, and **Omar",
' M. Yaghi** "',
"for the development of metal–organic frameworks",
'."',
],
),
)
]
result = await conversation.async_converse(
hass,
"Who won the Nobel for Chemistry in 2025?",
None,
Context(),
agent_id="conversation.claude_conversation",
)
chat_log = hass.data.get(conversation.chat_log.DATA_CHAT_LOGS).get(
result.conversation_id
)
# Don't test the prompt because it's not deterministic
assert chat_log.content[1:] == snapshot
assert mock_create_stream.call_args.kwargs["messages"] == snapshot
@freeze_time("2025-10-31 12:00:00")
async def test_bash_code_execution(
hass: HomeAssistant,
@@ -1014,9 +1237,10 @@ async def test_bash_code_execution(
"tmp/number.txt'.",
],
),
*create_bash_code_execution_block(
*create_server_tool_use_block(
1,
"srvtoolu_12345ABC",
"bash_code_execution",
[
"",
'{"c',
@@ -1093,9 +1317,10 @@ async def test_bash_code_execution_error(
"tmp/number.txt'.",
],
),
*create_bash_code_execution_block(
*create_server_tool_use_block(
1,
"srvtoolu_12345ABC",
"bash_code_execution",
[
"",
'{"c',
@@ -1252,8 +1477,8 @@ async def test_text_editor_code_execution(
mock_create_stream.return_value = [
(
*create_content_block(0, ["I'll do it", "."]),
*create_text_editor_code_execution_block(
1, "srvtoolu_12345ABC", args_parts
*create_server_tool_use_block(
1, "srvtoolu_12345ABC", "text_editor_code_execution", args_parts
),
*create_text_editor_code_execution_result_block(
2, "srvtoolu_12345ABC", content=content
@@ -1287,9 +1512,10 @@ async def test_container_reused(
"""Test that container is reused."""
mock_create_stream.return_value = [
(
*create_bash_code_execution_block(
*create_server_tool_use_block(
0,
"srvtoolu_12345ABC",
"bash_code_execution",
['{"command": "echo $RANDOM"}'],
),
*create_bash_code_execution_result_block(