feat(anthropic): stream the Messages web-search terminal answer

apis/src/web_search/config.rs

@@ -125,6 +125,14 @@ pub(crate) struct WebSearchFilterConfig {
     /// callout. Enable this only for a trusted private provider endpoint.
     #[serde(default)]
     pub(crate) allow_private_base_url: bool,
+
+    /// Select Praxis streaming transport for effective `stream: true`
+    /// Messages requests. When enabled, the terminal inference response is
+    /// streamed incrementally as one coherent client-visible SSE lifecycle
+    /// while intermediate tool/search transitions stay internal. This knob is
+    /// anthropic-only; `openai_web_search` does not accept it.
+    #[serde(default)]
+    pub(crate) terminal_streaming: bool,
 }
 
 // -----------------------------------------------------------------------------
@@ -186,6 +194,10 @@ impl OpenAiWebSearchConfig {
             max_body_bytes: None,
             base_url: self.base_url,
             allow_private_base_url: self.allow_private_base_url,
+            // `openai_web_search` has no terminal_streaming knob; its own
+            // deny_unknown_fields config never accepts the field, so the
+            // shared validated form is always off for it.
+            terminal_streaming: false,
         }
     }
 }
@@ -217,6 +229,9 @@ pub(crate) struct ValidatedConfig {
 
     /// Connect-time private-address policy for the provider target.
     pub allow_private_base_url: bool,
+
+    /// Whether to stream the terminal Messages response incrementally.
+    pub terminal_streaming: bool,
 }
 
 impl std::fmt::Debug for ValidatedConfig {
@@ -229,6 +244,7 @@ impl std::fmt::Debug for ValidatedConfig {
             .field("max_body_bytes", &self.max_body_bytes)
             .field("base_url", &self.base_url)
             .field("allow_private_base_url", &self.allow_private_base_url)
+            .field("terminal_streaming", &self.terminal_streaming)
             .finish()
     }
 }
@@ -267,6 +283,7 @@ fn build_validated_config(
         max_body_bytes: validate_max_body_bytes_field(filter_name, raw.max_body_bytes)?,
         base_url: raw.base_url.clone(),
         allow_private_base_url: raw.allow_private_base_url,
+        terminal_streaming: raw.terminal_streaming,
     })
 }
 
@@ -333,9 +350,36 @@ mod tests {
             max_body_bytes: None,
             base_url: None,
             allow_private_base_url: false,
+            terminal_streaming: false,
         }
     }
 
+    #[test]
+    fn build_config_defaults_terminal_streaming_off() {
+        let validated = build_config("anthropic_web_search", &base_config()).unwrap();
+        assert!(
+            !validated.terminal_streaming,
+            "terminal_streaming must default to off so non-streaming behavior is unchanged"
+        );
+    }
+
+    #[test]
+    fn parse_config_reads_terminal_streaming() {
+        let yaml = serde_yaml::from_str("provider: you\napi_key: k\nterminal_streaming: true").unwrap();
+        let raw = parse_filter_config::<WebSearchFilterConfig>("anthropic_web_search", &yaml).unwrap();
+        let validated = build_config("anthropic_web_search", &raw).unwrap();
+        assert!(validated.terminal_streaming);
+    }
+
+    #[test]
+    fn openai_web_search_rejects_terminal_streaming() {
+        let yaml = serde_yaml::from_str("provider: you\napi_key: k\nterminal_streaming: true").unwrap();
+        assert!(
+            parse_filter_config::<OpenAiWebSearchConfig>("openai_web_search", &yaml).is_err(),
+            "terminal_streaming is anthropic-only; openai_web_search must reject the unknown field"
+        );
+    }
+
     #[test]
     fn build_config_applies_defaults() {
         let cfg = build_config("openai_web_search", &base_config()).unwrap();

apis/src/web_search/provider.rs

@@ -503,6 +503,7 @@ mod tests {
             max_body_bytes: 64 * 1024 * 1024,
             base_url: None,
             allow_private_base_url: false,
+            terminal_streaming: false,
         };
         let client = SearchClient::from_config("test", &config, test_subrequest_client()).unwrap();
 
@@ -560,6 +561,7 @@ mod tests {
             max_body_bytes: 64 * 1024 * 1024,
             base_url: None,
             allow_private_base_url: false,
+            terminal_streaming: false,
         };
         let client = SearchClient::from_config("test", &config, test_subrequest_client());
         assert!(client.is_ok(), "a valid search configuration should build a client");
@@ -575,6 +577,7 @@ mod tests {
             max_body_bytes: 64 * 1024 * 1024,
             base_url: None,
             allow_private_base_url: false,
+            terminal_streaming: false,
         };
 
         let error = SearchClient::from_config("anthropic_web_search", &config, test_subrequest_client()).unwrap_err();
@@ -597,6 +600,7 @@ mod tests {
             max_body_bytes: 64 * 1024 * 1024,
             base_url: Some("http://localhost:9999".into()),
             allow_private_base_url: true,
+            terminal_streaming: false,
         };
         let client = SearchClient::from_config("test", &config, test_subrequest_client()).unwrap();
         let (url, _) = client.build_brave_request("test query", 5);
@@ -616,6 +620,7 @@ mod tests {
             max_body_bytes: 64 * 1024 * 1024,
             base_url: Some("http://localhost:9999".into()),
             allow_private_base_url: true,
+            terminal_streaming: false,
         };
         let client = SearchClient::from_config("test", &config, test_subrequest_client()).unwrap();
         let (url, _) = client.build_tavily_request("test query", SearchContextSize::Medium);
@@ -635,6 +640,7 @@ mod tests {
             max_body_bytes: 64 * 1024 * 1024,
             base_url: Some("http://localhost:9999".into()),
             allow_private_base_url: true,
+            terminal_streaming: false,
         };
         let client = SearchClient::from_config("test", &config, test_subrequest_client()).unwrap();
         let (url, _) = client.build_you_request("test query", 5);
@@ -654,6 +660,7 @@ mod tests {
             max_body_bytes: 64 * 1024 * 1024,
             base_url: None,
             allow_private_base_url: false,
+            terminal_streaming: false,
         };
         let client = SearchClient::from_config("test", &config, test_subrequest_client()).unwrap();
         let outcome = client.parse_response(b"not json");
@@ -673,6 +680,7 @@ mod tests {
             max_body_bytes: 64 * 1024 * 1024,
             base_url: None,
             allow_private_base_url: false,
+            terminal_streaming: false,
         };
         let client = SearchClient::from_config("test", &config, test_subrequest_client()).unwrap();
         let outcome = client.parse_response(br#"{"web":{"results":[]}}"#);
@@ -691,6 +699,7 @@ mod tests {
             max_body_bytes: 64 * 1024 * 1024,
             base_url: None,
             allow_private_base_url: true,
+            terminal_streaming: false,
         };
         SearchClient::from_config("test", &config, test_subrequest_client()).unwrap()
     }

docs/filters/anthropic_web_search.md

@@ -16,6 +16,7 @@ Executes server-owned `WebSearch` tool calls in an Anthropic Messages loop.
 | `max_body_bytes` | integer | no | Maximum request body bytes to buffer. |
 | `base_url` | string | no | Override the provider's default API base URL. |
 | `allow_private_base_url` | bool | no | Allow a `base_url` that targets local-sensitive addresses. DNS names are resolved once per request and every result is checked immediately before the transport connects. By default, any private, loopback, link-local, or otherwise non-public result rejects the callout. Enable this only for a trusted private provider endpoint. |
+| `terminal_streaming` | bool | no | Select Praxis streaming transport for effective `stream: true` Messages requests. When enabled, the terminal inference response is streamed incrementally as one coherent client-visible SSE lifecycle while intermediate tool/search transitions stay internal. This knob is anthropic-only; `openai_web_search` does not accept it. |
 
 ## Examples
 
@@ -43,7 +44,7 @@ max_body_bytes: 67108864
 ```yaml
 # cargo run -p praxis-test-utils --example anthropic_messages_web_search_mock
 # WEB_SEARCH_API_KEY="$WEB_SEARCH_API_KEY" cargo run -p praxis-ai-proxy -- \
-#   -c examples/configs/anthropic/messages-web-search.yaml
+#   -c examples/configs/anthropic/full-flow-agentic.yaml
 # curl http://127.0.0.1:8080/v1/messages \
 #   -H 'content-type: application/json' \
 #   -d '{"model":"openai/gpt-oss-20b","max_tokens":1024,"stream":false,"messages":[{"role":"user","content":"Use web search to look up potato, then summarize in one sentence."}],"tools":[{"name":"WebSearch","description":"Search the web","input_schema":{"type":"object","properties":{"query":{"type":"string"}},"required":["query"]}}]}'

examples/README.md

@@ -52,9 +52,9 @@ before sending requests.
 
 | File | Description |
 | ------ | ------------- |
+| [full-flow-agentic.yaml](configs/anthropic/full-flow-agentic.yaml) | A single Anthropic Messages gateway that runs the server-owned web-search loop through Praxis core's iterative_request_router (IRR) and serves BOTH streaming and buffered clients from one pipeline. `anthropic_web_search` selects the transport per request from the client's `stream` flag (`terminal_streaming: true`) |
 | [messages-protocol.yaml](configs/anthropic/messages-protocol.yaml) | Routes Anthropic Messages API requests to a native `/v1/messages` backend |
 | [messages-to-openai.yaml](configs/anthropic/messages-to-openai.yaml) | Transforms Anthropic Messages API requests and responses for Chat Completions-compatible inference backends |
-| [messages-web-search.yaml](configs/anthropic/messages-web-search.yaml) | Runs a non-streaming native Messages model -> You.com search -> model loop through Praxis core's iterative_request_router |
 | [request-validate.yaml](configs/anthropic/request-validate.yaml) | Rejects empty, malformed, or non-object JSON request bodies |
 | [unified-gateway.yaml](configs/anthropic/unified-gateway.yaml) | Routes traffic by classifier-promoted headers so a single listener handles Anthropic Messages, OpenAI Chat Completions, and OpenAI Responses requests |
 

examples/configs/anthropic/full-flow-agentic.yaml

@@ -0,0 +1,87 @@
+# Anthropic Messages Full-Flow Agentic (Web Search)
+#
+# A single Anthropic Messages gateway that runs the server-owned web-search
+# loop through Praxis core's iterative_request_router (IRR) and serves BOTH
+# streaming and buffered clients from one pipeline. `anthropic_web_search`
+# selects the transport per request from the client's `stream` flag
+# (`terminal_streaming: true`):
+#
+#   * `"stream": true`  -> the terminal answer is streamed to the client as one
+#     coherent Anthropic Messages SSE lifecycle (a single `message_start`,
+#     forwarded text `content_block_*` frames, then one terminal `message_delta`
+#     / `message_stop`). Intermediate model/search transitions stay internal and
+#     the managed `WebSearch` tool-use block is suppressed.
+#   * `"stream": false` -> the same loop runs buffered and returns one final
+#     Anthropic Messages JSON object.
+#
+# Because the inference step may stream its response, every response filter in
+# that step uses `BodyMode::Stream`; `anthropic_web_search` rewrites the SSE body
+# incrementally for a streaming round and accumulates then classifies a buffered
+# round.
+#
+# The model runs natively on the Anthropic Messages wire format (for example
+# vLLM's `/v1/messages` endpoint). The managed `WebSearch` tool is executed by
+# the proxy against the configured provider, so the model only needs to emit
+# `WebSearch` tool-use blocks; the loop performs the search and re-enters.
+#
+# Requires WEB_SEARCH_API_KEY. The Messages backend listens on 127.0.0.1:8000.
+#
+# Run the deterministic model mock:
+#   cargo run -p praxis-test-utils --example anthropic_messages_web_search_mock
+#
+# In a second terminal, run Praxis:
+#   WEB_SEARCH_API_KEY="$WEB_SEARCH_API_KEY" cargo run -p praxis-ai-proxy -- \
+#     -c examples/configs/anthropic/full-flow-agentic.yaml
+#
+# Streaming request (incremental terminal answer):
+#   curl -N http://127.0.0.1:8080/v1/messages \
+#     -H 'content-type: application/json' \
+#     -d '{"model":"openai/gpt-oss-20b","max_tokens":1024,"stream":true,"messages":[{"role":"user","content":"Use web search to look up potato, then summarize in one sentence."}],"tools":[{"name":"WebSearch","description":"Search the web","input_schema":{"type":"object","properties":{"query":{"type":"string"}},"required":["query"]}}]}'
+#
+# Buffered request (single final JSON object): send the same body with
+# "stream": false (or omit "stream").
+
+listeners:
+  - name: anthropic-full-flow-agentic
+    address: "127.0.0.1:8080"
+    filter_chains: [full-flow-agentic]
+
+filter_chains:
+  - name: full-flow-agentic
+    filters:
+      - filter: anthropic_messages_format
+        on_invalid: reject
+      - filter: anthropic_validate
+      - filter: iterative_request_router
+        initial_step: inference
+        max_iterations: 6
+        timeout_ms: 90000
+        steps:
+          - name: inference
+            filters:
+              - filter: anthropic_web_search
+                provider: you
+                api_key: ${WEB_SEARCH_API_KEY}
+                default_context_size: medium
+                timeout_ms: 10000
+                terminal_streaming: true
+              - filter: anthropic_messages_protocol
+                default_version: "2023-06-01"
+              - filter: router
+                routes:
+                  - path_prefix: "/v1/messages"
+                    cluster: messages-backend
+              - filter: load_balancer
+                clusters:
+                  - name: messages-backend
+                    endpoints: ["127.0.0.1:8000"]
+            on_result:
+              - filter: anthropic_web_search
+                key: action
+                value: loop
+                next: inference
+              - default: true
+                done: true
+
+insecure_options:
+  allow_private_endpoints: true # example proxies to local backends