diff --git a/pyproject.toml b/pyproject.toml index c87ae524..472fd871 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,7 +26,7 @@ dependencies = [ "fastapi>=0.115.0", "uvicorn[standard]>=0.32.0", "redis>=5.0.0", - "anthropic[vertex]>=0.40.0", + "anthropic[vertex]>=0.77.0", "deepagents>=0.1.0", "langchain-anthropic>=0.3.0", "langchain-google-genai>=4.0.0", diff --git a/src/forge/integrations/agents/agent.py b/src/forge/integrations/agents/agent.py index fc1d32d0..c826bca3 100644 --- a/src/forge/integrations/agents/agent.py +++ b/src/forge/integrations/agents/agent.py @@ -51,7 +51,13 @@ HAS_GOOGLE_GENAI = False try: - from langchain_google_vertexai.model_garden import ChatAnthropicVertex + from forge.integrations.agents.vertex_anthropic import ( + ForgeChatAnthropicVertex as ChatAnthropicVertex, + ) + from forge.integrations.agents.vertex_anthropic import ( + is_transient_vertex_error, + is_unsupported_native_output_error, + ) HAS_ANTHROPIC_VERTEX = True except ImportError: @@ -675,6 +681,12 @@ async def _run_agent( """ # Use async version to load MCP tools response_format = ProviderStrategy(response_schema) if response_schema else None + selected_model = model_target.model if model_target else self.settings.llm_model + selected_backend = model_target.backend if model_target else self.settings.llm_backend + vertex_claude = ( + selected_backend == "vertex-ai" + and Settings.detect_model_provider(selected_model) == "anthropic" + ) agent = await self._create_agent_async( system_prompt=system_prompt, include_tools=include_tools, @@ -737,6 +749,23 @@ async def _run_agent( break # Success, exit retry loop except Exception as e: last_error = e + if vertex_claude and not ( + response_schema is not None + and not used_tool_fallback + and is_unsupported_native_output_error(e) + ): + if is_transient_vertex_error(e) and attempt < self.MAX_RETRIES - 1: + delay = self.INITIAL_BACKOFF_SECONDS * (2**attempt) + logger.warning( + "Transient Vertex error (attempt %s/%s), retrying in %ss: %s", + attempt + 1, + self.MAX_RETRIES, + delay, + e, + ) + await asyncio.sleep(delay) + continue + raise if response_schema is not None and not used_tool_fallback: logger.warning( "Native structured output failed for %s; retrying with validated " diff --git a/src/forge/integrations/agents/vertex_anthropic.py b/src/forge/integrations/agents/vertex_anthropic.py new file mode 100644 index 00000000..cfcd461d --- /dev/null +++ b/src/forge/integrations/agents/vertex_anthropic.py @@ -0,0 +1,214 @@ +"""Translate LangChain native structured output for Claude on Vertex.""" + +from copy import deepcopy +from typing import Any + +import httpx +from anthropic import APIConnectionError, APIStatusError, transform_schema +from langchain_anthropic.chat_models import convert_to_anthropic_tool +from langchain_core.outputs import ChatResult +from langchain_core.tools import BaseTool +from langchain_google_vertexai.model_garden import ChatAnthropicVertex + +_SERVER_TOOL_PREFIXES = ( + "web_search_", + "web_fetch_", + "code_execution_", + "computer_", + "bash_", + "text_editor_", + "tool_search_", + "memory_", +) + + +class VertexResponseStopError(ValueError): + """Claude stopped before a complete, usable response was available.""" + + def __init__(self, stop_reason: str) -> None: + self.stop_reason = stop_reason + super().__init__(f"Vertex Claude response stop_reason={stop_reason}") + + +def _argument_structure(schema: Any) -> Any: + """Retain schema features that determine accepted argument shapes.""" + if not isinstance(schema, dict): + return schema + properties = schema.get("properties") + structure: dict[str, Any] = { + key: schema[key] + for key in ("type", "$ref", "required", "dependentRequired") + if key in schema + } + if properties is not None: + structure["properties"] = { + name: _argument_structure(value) for name, value in properties.items() + } + for key in ( + "items", + "additionalProperties", + "additionalItems", + "unevaluatedProperties", + "unevaluatedItems", + "contains", + "not", + "if", + "then", + "else", + "propertyNames", + ): + if key in schema: + structure[key] = _argument_structure(schema[key]) + if "additionalProperties" not in schema and properties: + structure["additionalProperties"] = True + for key in ("anyOf", "oneOf", "allOf", "prefixItems"): + if key in schema: + structure[key] = [_argument_structure(item) for item in schema[key]] + for key in ("$defs", "definitions", "patternProperties", "dependentSchemas"): + if key in schema: + structure[key] = { + name: _argument_structure(value) for name, value in schema[key].items() + } + return structure + + +def _has_recursive_ref(schema: dict[str, Any]) -> bool: + definitions = schema.get("$defs", {}) + + def visit(value: Any, active: frozenset[str]) -> bool: + if isinstance(value, list): + return any(visit(item, active) for item in value) + if not isinstance(value, dict): + return False + reference = value.get("$ref") + if isinstance(reference, str) and reference.startswith("#/$defs/"): + if reference in active: + return True + target = definitions.get(reference.removeprefix("#/$defs/")) + if target is not None and visit(target, active | {reference}): + return True + return any(visit(item, active) for key, item in value.items() if key != "$ref") + + return visit(schema, frozenset()) + + +def _format_tool(tool: Any) -> dict[str, Any]: + if ( + isinstance(tool, dict) + and "type" in tool + and tool["type"] != "function" + and "input_schema" not in tool + ): + tool_type = tool["type"] + if not ( + isinstance(tool_type, str) + and tool_type.startswith(_SERVER_TOOL_PREFIXES) + and isinstance(tool.get("name"), str) + ): + raise ValueError(f"Unsupported Vertex Claude tool: {tool_type!r}") + return deepcopy(tool) + + formatted = dict(deepcopy(convert_to_anthropic_tool(tool))) + if isinstance(tool, dict) and tool.get("type") == "function": + function = tool.get("function") + if isinstance(function, dict) and isinstance(function.get("strict"), bool): + formatted["strict"] = function["strict"] + extras = tool.extras if isinstance(tool, BaseTool) else None + if isinstance(extras, dict) and "strict" in extras: + formatted["strict"] = extras["strict"] + if formatted.get("strict") is True: + original_schema = formatted.get("input_schema") + if not isinstance(original_schema, dict): + raise ValueError("Explicit strict tool needs an input schema") + if _has_recursive_ref(original_schema): + raise ValueError("Unsupported recursive strict tool input schema") + try: + adapted_schema = transform_schema(original_schema) + except (TypeError, ValueError) as error: + raise ValueError("Explicit strict tool has an unsupported input schema") from error + if _argument_structure(original_schema) != _argument_structure(adapted_schema): + raise ValueError("Explicit strict tool would change its input schema") + formatted["input_schema"] = adapted_schema + return formatted + + +def _explicitly_strict(tool: Any) -> bool: + if isinstance(tool, dict): + if tool.get("strict") is True: + return True + function = tool.get("function") if tool.get("type") == "function" else None + return isinstance(function, dict) and function.get("strict") is True + extras = tool.extras if isinstance(tool, BaseTool) else None + return isinstance(extras, dict) and extras.get("strict") is True + + +class ForgeChatAnthropicVertex(ChatAnthropicVertex): + """Keep LangChain's provider strategy on Anthropic's Vertex wire format.""" + + def __init__(self, **kwargs: Any) -> None: + # Pydantic accepts configured model fields dynamically at runtime. + super().__init__(**kwargs) + + def _format_output(self, data: Any, **kwargs: Any) -> ChatResult: + stop_reason = data.stop_reason + if stop_reason in {"max_tokens", "refusal"}: + raise VertexResponseStopError(stop_reason) + return super()._format_output(data, **kwargs) + + def bind_tools( + self, + tools: Any, + *, + tool_choice: dict[str, str] | str | None = None, + strict: bool | None = None, + response_format: dict[str, Any] | None = None, + **kwargs: Any, + ) -> Any: + del strict # LangChain's binding-wide strictness is not a per-tool opt-in. + if sum(_explicitly_strict(tool) for tool in tools) > 20: + raise ValueError("Claude on Vertex accepts at most 20 strict tools") + formatted = [_format_tool(tool) for tool in tools] + if response_format is not None: + if ( + not isinstance(response_format, dict) + or response_format.get("type") != "json_schema" + or not isinstance(response_format.get("json_schema"), dict) + or not isinstance(response_format["json_schema"].get("schema"), dict) + ): + raise ValueError("Unsupported Vertex Claude response_format") + schema = transform_schema(response_format["json_schema"]["schema"]) + native_format = {"type": "json_schema", "schema": schema} + output_config = dict(kwargs.pop("output_config", {}) or {}) + if "format" in output_config and output_config["format"] != native_format: + raise ValueError("Conflicting response_format and output_config.format") + output_config["format"] = native_format + kwargs["output_config"] = output_config + if isinstance(tool_choice, dict): + kwargs["tool_choice"] = tool_choice + elif tool_choice in ("any", "auto"): + kwargs["tool_choice"] = {"type": tool_choice} + elif isinstance(tool_choice, str): + kwargs["tool_choice"] = {"type": "tool", "name": tool_choice} + elif tool_choice is not None: + raise ValueError(f"Unrecognized tool_choice: {tool_choice!r}") + return self.bind(tools=formatted, **kwargs) + + +def is_unsupported_native_output_error(error: Exception) -> bool: + """Require an explicit model-capability rejection before changing strategy.""" + if not isinstance(error, APIStatusError) or error.status_code != 400: + return False + body = error.body + detail = body.get("error") if isinstance(body, dict) else None + message = detail.get("message") if isinstance(detail, dict) else None + return isinstance(message, str) and ( + "output_config.format" in message.lower() + and "not supported for this model" in message.lower() + ) + + +def is_transient_vertex_error(error: Exception) -> bool: + """Retry only transport and provider rate/server errors on the same strategy.""" + if isinstance(error, APIStatusError): + return error.status_code == 429 or error.status_code >= 500 + return isinstance(error, (APIConnectionError, httpx.TransportError)) diff --git a/tests/unit/integrations/agents/test_agent.py b/tests/unit/integrations/agents/test_agent.py index 73e10ba3..6d015360 100644 --- a/tests/unit/integrations/agents/test_agent.py +++ b/tests/unit/integrations/agents/test_agent.py @@ -10,6 +10,7 @@ from langchain_core.messages import BaseMessage from forge.integrations.agents.agent import ForgeAgent +from forge.models.model_policy import ResolvedModelTarget class MockChatModel(SimpleChatModel): @@ -37,6 +38,7 @@ async def _acall( def _llm_type(self) -> str: return "mock" + def _model_agent(backend: str, model: str) -> ForgeAgent: agent = ForgeAgent.__new__(ForgeAgent) agent.settings = MagicMock( @@ -93,6 +95,32 @@ def test_create_model_uses_vertex_backend_for_anthropic(): ) +def test_create_model_uses_vertex_adapter_for_named_connection(): + agent = _model_agent("anthropic", "claude-sonnet-4-6") + target = ResolvedModelTarget( + connection="vertex-primary", + model="claude-opus-4-8", + backend="vertex-ai", + project="named-project", + location="us-east5", + policy_key="generate_tasks", + policy_source="project", + max_output_tokens=8192, + temperature=0.2, + ) + + with patch("forge.integrations.agents.agent.ChatAnthropicVertex") as model_class: + agent._create_model(model_target=target) + + model_class.assert_called_once_with( + model_name="claude-opus-4-8", + project="named-project", + location="us-east5", + max_tokens=8192, + temperature=0.2, + ) + + def test_create_model_uses_anthropic_backend(): agent = _model_agent("anthropic", "claude-sonnet-4-6") @@ -263,13 +291,13 @@ async def test_revise_draft_with_feedback_success(): """Verify that revise_draft_with_feedback properly renders prompt and parses valid JSON.""" agent = ForgeAgent() - mock_model = MockChatModel(response='{"parent_key": "PROJ-1", "items": [{"id": 1, "summary": "Task 1"}]}') + mock_model = MockChatModel( + response='{"parent_key": "PROJ-1", "items": [{"id": 1, "summary": "Task 1"}]}' + ) with patch.object(agent, "_create_model", return_value=mock_model): result = await agent.revise_draft_with_feedback( - draft_content='{"items": []}', - feedback="Add Task 1", - context={"ticket_key": "PROJ-1"} + draft_content='{"items": []}', feedback="Add Task 1", context={"ticket_key": "PROJ-1"} ) assert json.loads(result) == {"parent_key": "PROJ-1", "items": [{"id": 1, "summary": "Task 1"}]} @@ -296,9 +324,7 @@ async def test_revise_draft_with_feedback_markdown_stripping(): with patch.object(agent, "_create_model", return_value=mock_model): result = await agent.revise_draft_with_feedback( - draft_content='{"items": []}', - feedback="Add Task 1", - context={"ticket_key": "PROJ-1"} + draft_content='{"items": []}', feedback="Add Task 1", context={"ticket_key": "PROJ-1"} ) assert json.loads(result) == {"items": [{"id": 1, "summary": "Task 1"}]} @@ -310,14 +336,14 @@ async def test_revise_draft_with_feedback_preamble_no_codeblock(): """Verify that revise_draft_with_feedback strips preamble and postamble without markdown code block.""" agent = ForgeAgent() - llm_response = 'The corrected draft is: {"items": [{"id": 1, "summary": "Task 1"}]} please review.' + llm_response = ( + 'The corrected draft is: {"items": [{"id": 1, "summary": "Task 1"}]} please review.' + ) mock_model = MockChatModel(response=llm_response) with patch.object(agent, "_create_model", return_value=mock_model): result = await agent.revise_draft_with_feedback( - draft_content='{"items": []}', - feedback="Add Task 1", - context={"ticket_key": "PROJ-1"} + draft_content='{"items": []}', feedback="Add Task 1", context={"ticket_key": "PROJ-1"} ) assert json.loads(result) == {"items": [{"id": 1, "summary": "Task 1"}]} @@ -336,9 +362,7 @@ async def test_revise_draft_with_feedback_invalid_json(): pytest.raises(ValueError, match="Failed to parse revised draft as JSON"), ): await agent.revise_draft_with_feedback( - draft_content='{"items": []}', - feedback="Add Task 1", - context={"ticket_key": "PROJ-1"} + draft_content='{"items": []}', feedback="Add Task 1", context={"ticket_key": "PROJ-1"} ) await agent.close() @@ -351,13 +375,13 @@ async def test_revise_draft_with_feedback_prompt_formatting(): mock_model = MockChatModel(response='{"items": []}') with ( - patch("forge.integrations.agents.agent.load_prompt", return_value="FORMATTED PROMPT") as mock_load_prompt, + patch( + "forge.integrations.agents.agent.load_prompt", return_value="FORMATTED PROMPT" + ) as mock_load_prompt, patch.object(agent, "_create_model", return_value=mock_model), ): await agent.revise_draft_with_feedback( - draft_content='{"some": "json"}', - feedback="Do this", - context={"ticket_key": "PROJ-123"} + draft_content='{"some": "json"}', feedback="Do this", context={"ticket_key": "PROJ-123"} ) mock_load_prompt.assert_called_once_with( @@ -375,9 +399,7 @@ async def test_revise_draft_with_feedback_fallback_matched_delimiters(): agent = ForgeAgent() # Case 1: JSON Object starting with '{' but having a trailing ']' in the postamble - llm_response_object = ( - 'Here is the result: {"parent_key": "PROJ-1", "items": [{"id": 1}]} with an unmatched trailing bracket ]' - ) + llm_response_object = 'Here is the result: {"parent_key": "PROJ-1", "items": [{"id": 1}]} with an unmatched trailing bracket ]' mock_model_object = MockChatModel(response=llm_response_object) with patch.object(agent, "_create_model", return_value=mock_model_object): @@ -387,9 +409,7 @@ async def test_revise_draft_with_feedback_fallback_matched_delimiters(): assert json.loads(result_object) == {"parent_key": "PROJ-1", "items": [{"id": 1}]} # Case 2: JSON List starting with '[' but having a trailing '}' in the postamble - llm_response_list = ( - 'Here is the result: [{"id": 1}] with an unmatched trailing brace }' - ) + llm_response_list = 'Here is the result: [{"id": 1}] with an unmatched trailing brace }' mock_model_list = MockChatModel(response=llm_response_list) with patch.object(agent, "_create_model", return_value=mock_model_list): diff --git a/tests/unit/integrations/agents/test_structured_output.py b/tests/unit/integrations/agents/test_structured_output.py index 125fcb17..45357e53 100644 --- a/tests/unit/integrations/agents/test_structured_output.py +++ b/tests/unit/integrations/agents/test_structured_output.py @@ -1,11 +1,14 @@ -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch +import anthropic +import httpx import pytest from langchain.agents.structured_output import ProviderStrategy, ToolStrategy from pydantic import BaseModel, ConfigDict from forge.integrations.agents.agent import ForgeAgent from forge.integrations.agents.structured_outputs import ArtifactDocument, TaskGeneration +from forge.integrations.agents.vertex_anthropic import VertexResponseStopError def test_artifact_document_requires_repository_names() -> None: @@ -34,6 +37,107 @@ class Decision(BaseModel): reason: str +def _vertex_error(status: int, message: str) -> anthropic.APIStatusError: + request = httpx.Request("POST", "https://aiplatform.googleapis.com/mock") + response = httpx.Response(status, request=request) + error_type = "invalid_request_error" if status == 400 else "permission_error" + error_class = { + 400: anthropic.BadRequestError, + 403: anthropic.PermissionDeniedError, + 429: anthropic.RateLimitError, + 503: anthropic.InternalServerError, + }[status] + return error_class( + message, + response=response, + body={"error": {"type": error_type, "message": message}}, + ) + + +def _vertex_agent() -> ForgeAgent: + forge = ForgeAgent.__new__(ForgeAgent) + forge.settings = MagicMock(llm_backend="vertex-ai", llm_model="claude-opus-4-8") + return forge + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + TypeError("unexpected keyword argument 'output_config'"), + ValueError("Conflicting response_format and output_config.format"), + _vertex_error(400, "schema is too complex for compilation"), + _vertex_error(400, "invalid schema"), + _vertex_error(403, "structured_outputs is disabled by project policy"), + ValueError("invalid JSON in provider response"), + VertexResponseStopError("max_tokens"), + VertexResponseStopError("refusal"), + ], +) +async def test_vertex_native_failure_does_not_downgrade_or_retry(error: Exception) -> None: + forge = _vertex_agent() + native = AsyncMock() + native.ainvoke.side_effect = error + with ( + patch.object(forge, "_create_agent_async", return_value=native) as create, + patch("forge.integrations.agents.agent.asyncio.sleep", new_callable=AsyncMock) as sleep, + pytest.raises(type(error)), + ): + await forge._run_agent("prompt", "system", response_schema=Decision) + + create.assert_awaited_once() + native.ainvoke.assert_awaited_once() + sleep.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_vertex_native_fallback_requires_explicit_unsupported_feature_error() -> None: + forge = _vertex_agent() + native = AsyncMock() + native.ainvoke.side_effect = _vertex_error( + 400, "output_config.format is not supported for this model" + ) + fallback = AsyncMock() + fallback.ainvoke.return_value = { + "messages": [], + "structured_response": {"accepted": True, "reason": "valid"}, + } + with patch.object(forge, "_create_agent_async", side_effect=[native, fallback]) as create: + result = await forge._run_agent("prompt", "system", response_schema=Decision) + + assert result == Decision(accepted=True, reason="valid") + assert isinstance(create.call_args_list[1].kwargs["response_format"], ToolStrategy) + assert create.await_count == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + _vertex_error(429, "Rate limit exceeded"), + _vertex_error(503, "Service unavailable"), + anthropic.APITimeoutError(httpx.Request("POST", "https://aiplatform.googleapis.com/mock")), + ], +) +async def test_vertex_transient_failure_retries_same_native_strategy(error: Exception) -> None: + forge = _vertex_agent() + native = AsyncMock() + native.ainvoke.side_effect = [ + error, + {"messages": [], "structured_response": {"accepted": True, "reason": "valid"}}, + ] + with ( + patch.object(forge, "_create_agent_async", return_value=native) as create, + patch("forge.integrations.agents.agent.asyncio.sleep", new_callable=AsyncMock) as sleep, + ): + result = await forge._run_agent("prompt", "system", response_schema=Decision) + + assert result == Decision(accepted=True, reason="valid") + create.assert_awaited_once() + assert native.ainvoke.await_count == 2 + sleep.assert_awaited_once() + + @pytest.mark.asyncio async def test_tool_loop_returns_validated_structured_response() -> None: forge = ForgeAgent() diff --git a/tests/unit/integrations/agents/test_vertex_structured_output.py b/tests/unit/integrations/agents/test_vertex_structured_output.py new file mode 100644 index 00000000..083aa57a --- /dev/null +++ b/tests/unit/integrations/agents/test_vertex_structured_output.py @@ -0,0 +1,549 @@ +"""Offline wire-level contract for Claude on Vertex structured output.""" + +import json +from copy import deepcopy +from typing import Any +from unittest.mock import MagicMock, patch + +import anthropic +import httpx +import pytest +from anthropic import transform_schema +from langchain.agents import create_agent +from langchain.agents.structured_output import ( + ProviderStrategy, + StructuredOutputValidationError, +) +from langchain_core.tools import StructuredTool, tool +from langchain_google_vertexai.model_garden import ChatAnthropicVertex + +from forge.integrations.agents.agent import ForgeAgent +from forge.integrations.agents.structured_outputs import ( + STRUCTURED_RESPONSE_SCHEMAS, + TaskGeneration, +) +from forge.integrations.agents.vertex_anthropic import VertexResponseStopError + +MODEL = "claude-opus-4-8" +TASKS = { + "tasks": [ + { + "summary": "Offline task", + "description": "Offline acceptance criteria", + "repo": "example/repo", + } + ] +} + + +@tool +async def offline_probe() -> str: + """Return a harmless constant.""" + return "Offline probe finished." + + +def _message(content: list[dict[str, Any]], stop_reason: str) -> httpx.Response: + return httpx.Response( + 200, + json={ + "id": "msg_offline", + "type": "message", + "role": "assistant", + "model": MODEL, + "content": content, + "stop_reason": stop_reason, + "stop_sequence": None, + "usage": {"input_tokens": 1, "output_tokens": 1}, + }, + ) + + +def _forge_model(sync_http: httpx.Client, async_http: httpx.AsyncClient) -> ChatAnthropicVertex: + agent = ForgeAgent.__new__(ForgeAgent) + agent.settings = MagicMock( + llm_backend="vertex-ai", + llm_model=MODEL, + llm_max_tokens=1024, + google_cloud_project="offline-project", + google_cloud_location="global", + ) + original_sync = anthropic.AnthropicVertex + original_async = anthropic.AsyncAnthropicVertex + + def make_sync(**kwargs: Any) -> Any: + return original_sync( + **{**kwargs, "access_token": "offline-token", "http_client": sync_http} + ) + + def make_async(**kwargs: Any) -> Any: + return original_async( + **{**kwargs, "access_token": "offline-token", "http_client": async_http} + ) + + with ( + patch.object(anthropic, "AnthropicVertex", make_sync), + patch.object(anthropic, "AsyncAnthropicVertex", make_async), + ): + return agent._create_model() + + +@pytest.fixture +async def vertex_model() -> Any: + def reject_network(_request: httpx.Request) -> httpx.Response: + raise AssertionError("Binding must not send a provider request") + + transport = httpx.MockTransport(reject_network) + with httpx.Client(transport=transport) as sync_http: + async with httpx.AsyncClient(transport=transport) as async_http: + model = _forge_model(sync_http, async_http) + try: + yield model + finally: + model.client.close() + await model.async_client.close() + + +def _ordinary_tool(index: int, *, strict: bool | None = None) -> dict[str, Any]: + definition: dict[str, Any] = { + "name": f"lookup_{index}", + "description": "Look up a harmless value.", + "input_schema": { + "type": "object", + "properties": {"filters": {"type": "object", "additionalProperties": True}}, + "required": ["filters"], + }, + } + if strict is not None: + definition["strict"] = strict + return definition + + +def _lookup_tool(index: int) -> StructuredTool: + async def lookup(filters: dict[str, Any]) -> str: + """Return a harmless lookup value.""" + return str(filters) + + return StructuredTool.from_function( + coroutine=lookup, + name=f"lookup_{index}", + description="Return a harmless lookup value.", + ) + + +@pytest.mark.asyncio +async def test_many_ordinary_tools_keep_open_dictionary_inputs(vertex_model: Any) -> None: + definitions = [_ordinary_tool(i) for i in range(26)] + original = deepcopy(definitions) + bound = vertex_model.bind_tools( + definitions, + strict=True, + **ProviderStrategy(TaskGeneration).to_model_kwargs(), + ) + + assert len(bound.kwargs["tools"]) == 26 + assert all("strict" not in tool for tool in bound.kwargs["tools"]) + assert bound.kwargs["tools"][0]["input_schema"]["properties"]["filters"] == { + "type": "object", + "additionalProperties": True, + } + assert definitions == original + + +@pytest.mark.asyncio +async def test_more_than_twenty_explicit_strict_tools_are_rejected(vertex_model: Any) -> None: + with pytest.raises(ValueError, match="20 strict tools"): + vertex_model.bind_tools( + [_ordinary_tool(i, strict=True) for i in range(21)], + strict=True, + **ProviderStrategy(TaskGeneration).to_model_kwargs(), + ) + + +@pytest.mark.asyncio +async def test_explicit_strict_tool_must_keep_open_dictionary_arguments(vertex_model: Any) -> None: + with pytest.raises(ValueError, match="strict tool.*input schema"): + vertex_model.bind_tools([_ordinary_tool(1, strict=True)]) + + +@pytest.mark.asyncio +async def test_explicit_strict_nullable_dictionary_branch_is_rejected(vertex_model: Any) -> None: + definition = _ordinary_tool(1, strict=True) + definition["input_schema"]["additionalProperties"] = False + definition["input_schema"]["properties"]["filters"] = { + "anyOf": [ + {"type": "object", "additionalProperties": {"type": "string"}}, + {"type": "null"}, + ] + } + + with pytest.raises(ValueError, match="would change its input schema"): + vertex_model.bind_tools([definition]) + + +@pytest.mark.asyncio +async def test_explicit_strict_marker_in_tool_extras_is_preserved(vertex_model: Any) -> None: + opted_in = deepcopy(offline_probe) + opted_in.extras = { + "strict": True, + "provider_tool_definition": { + "name": "offline_probe", + "description": "Return a harmless constant.", + "input_schema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + }, + } + + bound = vertex_model.bind_tools([opted_in], strict=False) + + assert bound.kwargs["tools"][0]["strict"] is True + + +@pytest.mark.asyncio +async def test_supported_server_tool_keeps_provider_definition(vertex_model: Any) -> None: + server_tool = {"type": "web_search_20250305", "name": "web_search", "max_uses": 1} + + bound = vertex_model.bind_tools([server_tool], strict=True) + + assert bound.kwargs["tools"] == [server_tool] + + +@pytest.mark.asyncio +async def test_unrepresentable_server_tool_fails_locally(vertex_model: Any) -> None: + with pytest.raises(ValueError, match="Unsupported Vertex Claude tool"): + vertex_model.bind_tools([{"type": "unknown_server_tool", "name": "unknown"}]) + + +@pytest.mark.asyncio +async def test_openai_function_tool_definition_remains_ordinary(vertex_model: Any) -> None: + definition = { + "type": "function", + "function": { + "name": "open_lookup", + "description": "Look up a harmless value.", + "parameters": { + "type": "object", + "properties": {"key": {"type": "string"}}, + }, + }, + } + + bound = vertex_model.bind_tools([definition], strict=True) + + assert bound.kwargs["tools"][0]["name"] == "open_lookup" + assert "strict" not in bound.kwargs["tools"][0] + + +@pytest.mark.asyncio +async def test_openai_function_per_tool_strict_marker_is_preserved(vertex_model: Any) -> None: + definition = { + "type": "function", + "function": { + "name": "fixed_lookup", + "description": "Look up one value.", + "parameters": { + "type": "object", + "properties": {"key": {"type": "string"}}, + "required": ["key"], + "additionalProperties": False, + }, + "strict": True, + }, + } + + bound = vertex_model.bind_tools([definition]) + + assert bound.kwargs["tools"][0]["strict"] is True + + +@pytest.mark.asyncio +async def test_recursive_explicit_strict_schema_fails_locally(vertex_model: Any) -> None: + recursive_tool = { + "name": "walk_tree", + "description": "Walk a tree.", + "strict": True, + "input_schema": { + "type": "object", + "properties": {"root": {"$ref": "#/$defs/Node"}}, + "required": ["root"], + "additionalProperties": False, + "$defs": { + "Node": { + "type": "object", + "properties": {"child": {"$ref": "#/$defs/Node"}}, + "additionalProperties": False, + } + }, + }, + } + + with pytest.raises(ValueError, match="recursive strict tool"): + vertex_model.bind_tools([recursive_tool]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("binding_strict", [None, False, True]) +async def test_binding_wide_strict_never_changes_ordinary_tools( + vertex_model: Any, binding_strict: bool | None +) -> None: + bound = vertex_model.bind_tools([_ordinary_tool(1)], strict=binding_strict) + + assert "strict" not in bound.kwargs + assert "strict" not in bound.kwargs["tools"][0] + + +@pytest.mark.asyncio +async def test_compatible_explicit_strict_tool_is_preserved(vertex_model: Any) -> None: + definition = { + "name": "fixed_lookup", + "description": "Look up one value.", + "input_schema": { + "type": "object", + "properties": {"key": {"type": "string"}}, + "required": ["key"], + "additionalProperties": False, + }, + "strict": True, + } + + bound = vertex_model.bind_tools([definition], strict=True) + + assert bound.kwargs["tools"][0]["strict"] is True + assert bound.kwargs["tools"][0]["input_schema"]["properties"]["key"]["type"] == "string" + + +@pytest.mark.asyncio +async def test_existing_output_configuration_merges_without_mutation(vertex_model: Any) -> None: + response_format = ProviderStrategy(TaskGeneration).to_model_kwargs()["response_format"] + original = deepcopy(response_format) + output_config = {"effort": "low"} + + bound = vertex_model.bind_tools( + [], + response_format=response_format, + output_config=output_config, + tool_choice="auto", + ) + + assert bound.kwargs["output_config"] == { + "effort": "low", + "format": {"type": "json_schema", "schema": transform_schema(TaskGeneration)}, + } + assert bound.kwargs["tool_choice"] == {"type": "auto"} + assert "response_format" not in bound.kwargs + assert response_format == original + assert output_config == {"effort": "low"} + + +@pytest.mark.asyncio +async def test_identical_existing_format_and_repeated_binding_do_not_mutate( + vertex_model: Any, +) -> None: + definition = _ordinary_tool(1, strict=False) + original = deepcopy(definition) + response_format = ProviderStrategy(TaskGeneration).to_model_kwargs()["response_format"] + native_format = {"type": "json_schema", "schema": transform_schema(TaskGeneration)} + + first = vertex_model.bind_tools( + [definition], response_format=response_format, output_config={"format": native_format} + ) + second = vertex_model.bind_tools([definition], response_format=response_format) + + assert first.kwargs["output_config"]["format"] == native_format + assert second.kwargs["tools"] == first.kwargs["tools"] + assert first.kwargs["tools"][0]["strict"] is False + assert definition == original + + +@pytest.mark.asyncio +async def test_unstructured_binding_omits_native_output_config(vertex_model: Any) -> None: + bound = vertex_model.bind_tools([]) + + assert bound.kwargs["tools"] == [] + assert "output_config" not in bound.kwargs + + +@pytest.mark.asyncio +async def test_conflicting_existing_output_format_fails(vertex_model: Any) -> None: + with pytest.raises(ValueError, match="Conflicting response_format"): + vertex_model.bind_tools( + [], + **ProviderStrategy(TaskGeneration).to_model_kwargs(), + output_config={"format": {"type": "other"}}, + ) + + +@pytest.mark.asyncio +async def test_unsupported_response_format_fails_locally(vertex_model: Any) -> None: + with pytest.raises(ValueError, match="Unsupported Vertex Claude response_format"): + vertex_model.bind_tools([], response_format={"type": "json_object"}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("schema", list(STRUCTURED_RESPONSE_SCHEMAS.values())) +async def test_every_host_response_schema_uses_native_vertex_format( + vertex_model: Any, schema: Any +) -> None: + bound = vertex_model.bind_tools([], **ProviderStrategy(schema).to_model_kwargs()) + + assert bound.kwargs["output_config"]["format"] == { + "type": "json_schema", + "schema": transform_schema(schema), + } + assert "response_format" not in bound.kwargs + + +@pytest.mark.asyncio +async def test_unadapted_library_class_still_rejects_provider_strategy_before_http() -> None: + requests: list[httpx.Request] = [] + + def reject_network(request: httpx.Request) -> httpx.Response: + requests.append(request) + raise AssertionError("The unadapted class should fail before HTTP") + + transport = httpx.MockTransport(reject_network) + with httpx.Client(transport=transport) as sync_http: + async with httpx.AsyncClient(transport=transport) as async_http: + model = ChatAnthropicVertex( + model_name=MODEL, + project="offline-project", + location="global", + access_token="offline-token", + http_client=sync_http, + async_http_client=async_http, + max_retries=0, + ) + try: + graph = create_agent(model, response_format=ProviderStrategy(TaskGeneration)) + with pytest.raises(TypeError, match="unexpected keyword argument 'strict'"): + await graph.ainvoke({"messages": [("user", "Return tasks")]}) + finally: + model.client.close() + await model.async_client.close() + + assert not requests + + +@pytest.mark.asyncio +async def test_forge_vertex_provider_strategy_preserves_ordinary_tool_and_final_json() -> None: + requests: list[dict[str, Any]] = [] + + def respond(request: httpx.Request) -> httpx.Response: + payload = json.loads(request.content) + requests.append(payload) + assert "strict" not in payload + assert "response_format" not in payload + assert payload["output_config"]["format"] == { + "type": "json_schema", + "schema": transform_schema(TaskGeneration), + } + assert len(payload["tools"]) == 27 + assert payload["tools"][0]["name"] == "offline_probe" + assert all("strict" not in entry for entry in payload["tools"]) + assert payload["tools"][1]["input_schema"]["properties"]["filters"]["type"] == "object" + if len(requests) == 1: + return _message( + [{"type": "tool_use", "id": "probe", "name": "offline_probe", "input": {}}], + "tool_use", + ) + assert len(requests) == 2 + assert "Offline probe finished." in json.dumps(payload["messages"]) + return _message([{"type": "text", "text": json.dumps(TASKS)}], "end_turn") + + transport = httpx.MockTransport(respond) + with httpx.Client(transport=transport) as sync_http: + async with httpx.AsyncClient(transport=transport) as async_http: + model = _forge_model(sync_http, async_http) + model.max_retries = 0 + try: + graph = create_agent( + model, + tools=[offline_probe, *[_lookup_tool(i) for i in range(26)]], + response_format=ProviderStrategy(TaskGeneration), + ) + result = await graph.ainvoke( + {"messages": [("user", "Run offline probe before final tasks")]}, + config={"recursion_limit": 12}, + ) + finally: + model.client.close() + await model.async_client.close() + + assert result["structured_response"] == TaskGeneration.model_validate(TASKS) + assert len(requests) == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("response_text", "stop_reason"), + [ + (json.dumps({"tasks": [{**TASKS["tasks"][0], "summary": ""}]}), "end_turn"), + ( + json.dumps({"tasks": [{k: v for k, v in TASKS["tasks"][0].items() if k != "repo"}]}), + "end_turn", + ), + (json.dumps({"tasks": [{**TASKS["tasks"][0], "repo": "invalid"}]}), "end_turn"), + (json.dumps({"tasks": [{**TASKS["tasks"][0], "unknown": 1}]}), "end_turn"), + ], +) +async def test_native_graph_rejects_invalid_final_task_json( + response_text: str, stop_reason: str +) -> None: + requests: list[dict[str, Any]] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(json.loads(request.content)) + return _message([{"type": "text", "text": response_text}], stop_reason) + + transport = httpx.MockTransport(respond) + with httpx.Client(transport=transport) as sync_http: + async with httpx.AsyncClient(transport=transport) as async_http: + model = _forge_model(sync_http, async_http) + model.max_retries = 0 + try: + graph = create_agent(model, response_format=ProviderStrategy(TaskGeneration)) + with pytest.raises(StructuredOutputValidationError): + await graph.ainvoke( + {"messages": [("user", "Return tasks")]}, + config={"recursion_limit": 8}, + ) + finally: + model.client.close() + await model.async_client.close() + + assert len(requests) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stop_reason", ["max_tokens", "refusal"]) +async def test_native_graph_rejects_valid_json_with_incomplete_stop_reason( + stop_reason: str, +) -> None: + requests: list[dict[str, Any]] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(json.loads(request.content)) + return _message([{"type": "text", "text": json.dumps(TASKS)}], stop_reason) + + transport = httpx.MockTransport(respond) + with httpx.Client(transport=transport) as sync_http: + async with httpx.AsyncClient(transport=transport) as async_http: + model = _forge_model(sync_http, async_http) + model.max_retries = 0 + try: + graph = create_agent(model, response_format=ProviderStrategy(TaskGeneration)) + with pytest.raises( + VertexResponseStopError, match=f"stop_reason={stop_reason}" + ) as stopped: + await graph.ainvoke( + {"messages": [("user", "Return tasks")]}, + config={"recursion_limit": 8}, + ) + assert stopped.value.stop_reason == stop_reason + finally: + model.client.close() + await model.async_client.close() + + assert len(requests) == 1 diff --git a/uv.lock b/uv.lock index daf87289..45b4dfca 100644 --- a/uv.lock +++ b/uv.lock @@ -849,7 +849,7 @@ k8s = [ [package.metadata] requires-dist = [ - { name = "anthropic", extras = ["vertex"], specifier = ">=0.40.0" }, + { name = "anthropic", extras = ["vertex"], specifier = ">=0.77.0" }, { name = "deepagents", specifier = ">=0.1.0" }, { name = "factory-boy", marker = "extra == 'dev'", specifier = ">=3.3.0" }, { name = "fastapi", specifier = ">=0.115.0" },