diff --git a/TASKS.md b/TASKS.md
index 64c6dfb..35b5bb0 100644
--- a/TASKS.md
+++ b/TASKS.md
@@ -4,7 +4,7 @@
 
 ### `feat/streaming-e2e`
 
-Status: in progress
+Status: completed
 
 DoD:
 - local fixture stack includes:
diff --git a/apps/mcp-server/src/app.rs b/apps/mcp-server/src/app.rs
index 21b4541..b48c1cd 100644
--- a/apps/mcp-server/src/app.rs
+++ b/apps/mcp-server/src/app.rs
@@ -520,8 +520,35 @@ async fn handle_base_tool_call(
     let operation = runtime_operation(&tool);
     let request_preview = build_request_preview(&state.runtime, &operation, &arguments);
     let started_at = Instant::now();
+    let is_window_mode = matches!(
+        operation
+            .execution_config
+            .streaming
+            .as_ref()
+            .map(|streaming| streaming.mode),
+        Some(crank_core::ExecutionMode::Window)
+    );
 
-    match state.runtime.execute(&operation, &arguments).await {
+    let result = if is_window_mode {
+        state
+            .runtime
+            .execute_window(&operation, &arguments)
+            .await
+            .map(|output| {
+                json!({
+                    "summary": output.summary,
+                    "items": output.items,
+                    "cursor": output.cursor,
+                    "window_complete": output.window_complete,
+                    "truncated": output.truncated,
+                    "has_more": output.has_more,
+                })
+            })
+    } else {
+        state.runtime.execute(&operation, &arguments).await
+    };
+
+    match result {
         Ok(output) => {
             let _ = persist_invocation(
                 &state,
diff --git a/apps/mcp-server/src/main.rs b/apps/mcp-server/src/main.rs
index 8c2c0b3..39b6c6b 100644
--- a/apps/mcp-server/src/main.rs
+++ b/apps/mcp-server/src/main.rs
@@ -48,7 +48,12 @@ mod tests {
         time::{Duration, SystemTime, UNIX_EPOCH},
     };
 
-    use axum::{Json, Router, http::header, routing::post};
+    use axum::{
+        Json, Router,
+        http::header,
+        response::sse::{Event, KeepAlive, Sse},
+        routing::{get, post},
+    };
     use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
     use crank_adapter_grpc::test_support as grpc_test_support;
     use crank_core::{
@@ -64,6 +69,7 @@ mod tests {
         PublishAgentRequest, PublishRequest,
     };
     use crank_schema::{Schema, SchemaKind};
+    use futures_util::stream;
     use serde_json::{Value, json};
     use sha2::{Digest, Sha256};
     use sqlx::{Executor, postgres::PgPoolOptions};
@@ -1114,6 +1120,72 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn executes_window_operation_via_mcp() {
+        let registry = test_registry().await;
+        let upstream_base_url = spawn_upstream_server().await;
+        let operation = test_rest_window_operation(&upstream_base_url, "crm_window_logs");
+
+        registry
+            .create_operation(&test_workspace_id(), &operation, Some("alice"))
+            .await
+            .unwrap();
+        registry
+            .publish_operation(PublishRequest {
+                workspace_id: &test_workspace_id(),
+                operation_id: &operation.id,
+                version: 1,
+                published_at: "2026-03-26T10:00:00Z",
+                published_by: Some("alice"),
+            })
+            .await
+            .unwrap();
+        publish_agent_for_operation(&registry, &operation, "sales-window").await;
+        let api_key = create_platform_api_key(
+            &registry,
+            "mcp-window",
+            &[PlatformApiKeyScope::Read, PlatformApiKeyScope::Write],
+        )
+        .await;
+
+        let base_url = spawn_mcp_server(build_app(
+            registry,
+            Duration::from_millis(0),
+            Some("https://crank.example.com".to_owned()),
+        ))
+        .await;
+        let client = reqwest::Client::new();
+        let mcp_url = agent_mcp_url(&base_url, "sales-window");
+        let initialized_session = initialize_session(&client, &mcp_url, &api_key).await;
+
+        let call_result = post_jsonrpc(
+            &client,
+            &mcp_url,
+            &api_key,
+            Some(&initialized_session),
+            json!({
+                "jsonrpc": "2.0",
+                "id": 3,
+                "method": "tools/call",
+                "params": {
+                    "name": "crm_window_logs",
+                    "arguments": {}
+                }
+            }),
+        )
+        .await;
+
+        assert_eq!(call_result["result"]["isError"], false);
+        assert_eq!(
+            call_result["result"]["structuredContent"]["items"][0]["message"],
+            json!("billing started")
+        );
+        assert_eq!(
+            call_result["result"]["structuredContent"]["window_complete"],
+            json!(true)
+        );
+    }
+
     async fn initialize_session(client: &reqwest::Client, mcp_url: &str, api_key: &str) -> String {
         let initialize_response = client
             .post(mcp_url)
@@ -1220,6 +1292,7 @@ mod tests {
 
     async fn spawn_upstream_server() -> String {
         let app = Router::new()
+            .route("/sse/logs", get(stream_logs))
             .route("/crm/leads", post(create_lead))
             .route("/crm/slow-leads", post(create_slow_lead));
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -1343,6 +1416,19 @@ mod tests {
         }))
     }
 
+    async fn stream_logs()
+    -> Sse<impl futures_util::Stream<Item = Result<Event, std::convert::Infallible>>> {
+        let events = vec![
+            json!({ "level": "info", "message": "billing started" }),
+            json!({ "level": "warn", "message": "cache warmup slow" }),
+            json!({ "level": "error", "message": "invoice timeout" }),
+        ];
+        let stream = stream::iter(events.into_iter().map(|payload| {
+            Ok::<_, std::convert::Infallible>(Event::default().data(payload.to_string()))
+        }));
+        Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
+    }
+
     async fn graphql_handler(Json(payload): Json<Value>) -> Json<Value> {
         let email = payload
             .get("variables")
@@ -1670,6 +1756,65 @@ mod tests {
         operation
     }
 
+    fn test_rest_window_operation(base_url: &str, name: &str) -> Operation<Schema, MappingSet> {
+        let mut operation = test_operation(base_url, name);
+        operation.display_name = "Window Logs".to_owned();
+        operation.target = Target::Rest(RestTarget {
+            base_url: base_url.to_owned(),
+            method: HttpMethod::Get,
+            path_template: "/sse/logs".to_owned(),
+            static_headers: BTreeMap::new(),
+        });
+        operation.input_schema = optional_object_schema("window");
+        operation.output_schema = empty_object_schema();
+        operation.input_mapping = MappingSet {
+            rules: vec![MappingRule {
+                source: "$.mcp.window".to_owned(),
+                target: "$.request.query.window".to_owned(),
+                required: false,
+                default_value: Some(json!("recent")),
+                transform: None,
+                condition: None,
+                notes: None,
+            }],
+        };
+        operation.output_mapping = MappingSet { rules: Vec::new() };
+        operation.tool_description = ToolDescription {
+            title: "Window Logs".to_owned(),
+            description: "Collects a bounded SSE log window".to_owned(),
+            tags: vec![
+                "rest".to_owned(),
+                "streaming".to_owned(),
+                "window".to_owned(),
+            ],
+            examples: Vec::new(),
+        };
+        operation.execution_config.streaming = Some(StreamingConfig {
+            mode: ExecutionMode::Window,
+            transport_behavior: TransportBehavior::ServerStream,
+            window_duration_ms: Some(1_000),
+            poll_interval_ms: None,
+            upstream_timeout_ms: Some(1_000),
+            idle_timeout_ms: None,
+            max_session_lifetime_ms: None,
+            max_items: Some(3),
+            max_bytes: Some(16 * 1024),
+            aggregation_mode: AggregationMode::SummaryPlusSamples,
+            summary_path: None,
+            items_path: Some("$.items".to_owned()),
+            cursor_path: None,
+            status_path: None,
+            done_path: Some("$.done".to_owned()),
+            redacted_paths: Vec::new(),
+            truncate_item_fields: false,
+            max_field_length: None,
+            drop_duplicates: false,
+            sampling_rate: None,
+            tool_family: ToolFamilyConfig::default(),
+        });
+        operation
+    }
+
     fn object_schema(field_name: &str) -> Schema {
         Schema {
             kind: SchemaKind::Object,
@@ -1696,4 +1841,45 @@ mod tests {
             variants: Vec::new(),
         }
     }
+
+    fn empty_object_schema() -> Schema {
+        Schema {
+            kind: SchemaKind::Object,
+            description: None,
+            required: true,
+            nullable: false,
+            default_value: None,
+            fields: BTreeMap::new(),
+            items: None,
+            enum_values: Vec::new(),
+            variants: Vec::new(),
+        }
+    }
+
+    fn optional_object_schema(field_name: &str) -> Schema {
+        Schema {
+            kind: SchemaKind::Object,
+            description: None,
+            required: true,
+            nullable: false,
+            default_value: None,
+            fields: BTreeMap::from([(
+                field_name.to_owned(),
+                Schema {
+                    kind: SchemaKind::String,
+                    description: None,
+                    required: false,
+                    nullable: false,
+                    default_value: None,
+                    fields: BTreeMap::new(),
+                    items: None,
+                    enum_values: Vec::new(),
+                    variants: Vec::new(),
+                },
+            )]),
+            items: None,
+            enum_values: Vec::new(),
+            variants: Vec::new(),
+        }
+    }
 }
diff --git a/apps/ui/scripts/playwright-stack.sh b/apps/ui/scripts/playwright-stack.sh
index 13589a4..30e2a98 100644
--- a/apps/ui/scripts/playwright-stack.sh
+++ b/apps/ui/scripts/playwright-stack.sh
@@ -100,6 +100,7 @@ export CRANK_LOG_LEVEL="info"
 export CRANK_SECRET_PROVIDER="env"
 export CRANK_SESSION_SECRET="e2e-session-secret"
 export CRANK_PASSWORD_PEPPER="e2e-password-pepper"
+export CRANK_MASTER_KEY="0000000000000000000000000000000000000000000000000000000000000000"
 export CRANK_SESSION_TTL_HOURS="24"
 export CRANK_BOOTSTRAP_ADMIN_EMAIL="$ADMIN_EMAIL"
 export CRANK_BOOTSTRAP_ADMIN_PASSWORD="$ADMIN_PASSWORD"
diff --git a/apps/ui/scripts/stream-fixture-server.js b/apps/ui/scripts/stream-fixture-server.js
index 7580144..f08d0fd 100644
--- a/apps/ui/scripts/stream-fixture-server.js
+++ b/apps/ui/scripts/stream-fixture-server.js
@@ -11,13 +11,37 @@ function writeJson(response, status, payload) {
   response.end(JSON.stringify(payload));
 }
 
+function readJsonBody(request) {
+  return new Promise((resolve, reject) => {
+    let buffer = '';
+    request.setEncoding('utf8');
+    request.on('data', (chunk) => {
+      buffer += chunk;
+    });
+    request.on('end', () => {
+      if (!buffer) {
+        resolve({});
+        return;
+      }
+      try {
+        resolve(JSON.parse(buffer));
+      } catch (error) {
+        reject(error);
+      }
+    });
+    request.on('error', reject);
+  });
+}
+
 const server = http.createServer((request, response) => {
-  if (request.url === '/health') {
+  const url = new URL(request.url, `http://127.0.0.1:${PORT}`);
+
+  if (url.pathname === '/health') {
     writeJson(response, 200, { ok: true });
     return;
   }
 
-  if (request.url === '/sse/logs') {
+  if (url.pathname === '/sse/logs') {
     response.writeHead(200, {
       'Content-Type': 'text/event-stream; charset=utf-8',
       'Cache-Control': 'no-cache',
@@ -33,8 +57,6 @@ const server = http.createServer((request, response) => {
     let index = 0;
     const timer = setInterval(() => {
       if (index >= events.length) {
-        response.write('event: done\n');
-        response.write('data: {"done":true}\n\n');
         clearInterval(timer);
         response.end();
         return;
@@ -52,7 +74,7 @@ const server = http.createServer((request, response) => {
     return;
   }
 
-  if (request.url === '/snapshot/metrics') {
+  if (url.pathname === '/snapshot/metrics') {
     writeJson(response, 200, {
       summary: {
         service: 'billing',
@@ -67,6 +89,20 @@ const server = http.createServer((request, response) => {
     return;
   }
 
+  if (url.pathname === '/crm/leads' && request.method === 'POST') {
+    readJsonBody(request)
+      .then((payload) => {
+        writeJson(response, 200, {
+          id: 'lead_123',
+          email: payload.email || null,
+        });
+      })
+      .catch(() => {
+        writeJson(response, 400, { error: 'invalid_json' });
+      });
+    return;
+  }
+
   writeJson(response, 404, { error: 'not_found' });
 });
 
diff --git a/apps/ui/tests/e2e/helpers.js b/apps/ui/tests/e2e/helpers.js
index d7b074f..bc0cf7a 100644
--- a/apps/ui/tests/e2e/helpers.js
+++ b/apps/ui/tests/e2e/helpers.js
@@ -1,7 +1,16 @@
 const { expect } = require('@playwright/test');
+const fs = require('fs');
+const path = require('path');
+const { execFileSync } = require('child_process');
 
 const ADMIN_EMAIL = process.env.CRANK_E2E_ADMIN_EMAIL || 'owner@crank.local';
 const ADMIN_PASSWORD = process.env.CRANK_E2E_ADMIN_PASSWORD || 'change-me-admin-password';
+const MCP_PORT = Number(process.env.CRANK_E2E_MCP_PORT || 3302);
+const STREAM_FIXTURE_PORT = Number(process.env.CRANK_E2E_STREAM_FIXTURE_PORT || 3310);
+const GRPC_FIXTURE_BIND = process.env.CRANK_E2E_GRPC_FIXTURE_BIND || '127.0.0.1:3311';
+const REPO_ROOT = path.resolve(__dirname, '../../../..');
+
+let cachedEchoDescriptorSetB64 = null;
 
 function localized(en, ru) {
   return new RegExp(`(?:${en}|${ru})`, 'i');
@@ -17,9 +26,470 @@ async function login(page) {
   await expect(page.locator('.page-title, .page-heading').first()).toBeVisible();
 }
 
+async function browserJson(page, method, urlPath, body, okStatuses = [200]) {
+  const response = await page.evaluate(async ({ method, urlPath, body }) => {
+    const request = {
+      method,
+      credentials: 'same-origin',
+      headers: {},
+    };
+    if (body !== undefined) {
+      request.headers['Content-Type'] = 'application/json';
+      request.body = JSON.stringify(body);
+    }
+
+    const result = await fetch(urlPath, request);
+    const text = await result.text();
+    let json = null;
+    try {
+      json = text ? JSON.parse(text) : null;
+    } catch (_error) {
+      json = null;
+    }
+
+    return {
+      status: result.status,
+      text,
+      json,
+    };
+  }, { method, urlPath, body });
+
+  if (!okStatuses.includes(response.status)) {
+    throw new Error(`HTTP ${response.status} ${urlPath}: ${response.text}`);
+  }
+
+  return response.json;
+}
+
+async function getSession(page) {
+  return browserJson(page, 'GET', '/api/auth/session');
+}
+
+async function getCurrentWorkspace(page) {
+  const session = await getSession(page);
+  const workspaceId = session.current_workspace_id
+    || (session.memberships && session.memberships[0] && session.memberships[0].workspace.id);
+  const membership = (session.memberships || []).find((item) => item.workspace.id === workspaceId)
+    || (session.memberships || [])[0];
+  return membership ? membership.workspace : null;
+}
+
+async function createOperation(page, workspaceId, payload) {
+  return browserJson(page, 'POST', `/api/admin/workspaces/${encodeURIComponent(workspaceId)}/operations`, payload);
+}
+
+async function publishOperation(page, workspaceId, operationId, version = 1) {
+  return browserJson(
+    page,
+    'POST',
+    `/api/admin/workspaces/${encodeURIComponent(workspaceId)}/operations/${encodeURIComponent(operationId)}/publish`,
+    { version },
+  );
+}
+
+async function createPlatformApiKey(page, workspaceId, name, scopes) {
+  return browserJson(
+    page,
+    'POST',
+    `/api/admin/workspaces/${encodeURIComponent(workspaceId)}/platform-api-keys`,
+    { name, scopes },
+  );
+}
+
+async function createAgent(page, workspaceId, payload) {
+  return browserJson(page, 'POST', `/api/admin/workspaces/${encodeURIComponent(workspaceId)}/agents`, payload);
+}
+
+async function saveAgentBindings(page, workspaceId, agentId, bindings) {
+  return browserJson(
+    page,
+    'POST',
+    `/api/admin/workspaces/${encodeURIComponent(workspaceId)}/agents/${encodeURIComponent(agentId)}/bindings`,
+    bindings,
+  );
+}
+
+async function publishAgent(page, workspaceId, agentId, version = 1) {
+  return browserJson(
+    page,
+    'POST',
+    `/api/admin/workspaces/${encodeURIComponent(workspaceId)}/agents/${encodeURIComponent(agentId)}/publish`,
+    { version },
+  );
+}
+
+function schemaObject(fields) {
+  return {
+    type: 'object',
+    required: true,
+    fields: fields || {},
+  };
+}
+
+function schemaString(required = true) {
+  return {
+    type: 'string',
+    required,
+  };
+}
+
+function uniqueName(prefix) {
+  return `${prefix}_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
+}
+
+function findEchoDescriptorSetB64() {
+  if (cachedEchoDescriptorSetB64) {
+    return cachedEchoDescriptorSetB64;
+  }
+
+  const escapedRoot = REPO_ROOT.replace(/'/g, "'\\''");
+  const descriptorPath = execFileSync(
+    'bash',
+    ['-lc', `find '${escapedRoot}/target/debug/build' -path '*/out/echo_descriptor.bin' -print -quit`],
+    { encoding: 'utf8' },
+  ).trim();
+
+  if (!descriptorPath) {
+    throw new Error('echo_descriptor.bin was not found under target/debug/build');
+  }
+
+  cachedEchoDescriptorSetB64 = fs.readFileSync(descriptorPath).toString('base64');
+  return cachedEchoDescriptorSetB64;
+}
+
+function buildRestWindowOperationPayload(name) {
+  return {
+    name,
+    display_name: 'Playwright Window Logs',
+    category: 'streaming',
+    protocol: 'rest',
+    target: {
+      kind: 'rest',
+      base_url: `http://127.0.0.1:${STREAM_FIXTURE_PORT}`,
+      method: 'GET',
+      path_template: '/sse/logs',
+      static_headers: {},
+    },
+    input_schema: schemaObject({
+      window: schemaString(false),
+    }),
+    output_schema: schemaObject({}),
+    input_mapping: {
+      rules: [
+        {
+          source: '$.mcp.window',
+          target: '$.request.query.window',
+          required: false,
+          default_value: 'recent',
+        },
+      ],
+    },
+    output_mapping: { rules: [] },
+    execution_config: {
+      timeout_ms: 1000,
+      headers: {},
+      streaming: {
+        mode: 'window',
+        transport_behavior: 'server_stream',
+        window_duration_ms: 1000,
+        upstream_timeout_ms: 1000,
+        max_items: 3,
+        max_bytes: 16384,
+        aggregation_mode: 'summary_plus_samples',
+        items_path: '$.items',
+        done_path: '$.done',
+        redacted_paths: [],
+        truncate_item_fields: false,
+        drop_duplicates: false,
+        tool_family: {},
+      },
+    },
+    tool_description: {
+      title: 'Playwright Window Logs',
+      description: 'Collects a bounded SSE log window from the local fixture.',
+      tags: ['playwright', 'streaming', 'window'],
+      examples: [{ input: {} }],
+    },
+  };
+}
+
+function buildGrpcSessionOperationPayload(name) {
+  return {
+    name,
+    display_name: 'Playwright gRPC Session',
+    category: 'streaming',
+    protocol: 'grpc',
+    target: {
+      kind: 'grpc',
+      server_addr: `http://${GRPC_FIXTURE_BIND}`,
+      package: 'echo',
+      service: 'EchoService',
+      method: 'ServerEcho',
+      descriptor_ref: 'desc_echo_playwright',
+      descriptor_set_b64: findEchoDescriptorSetB64(),
+    },
+    input_schema: schemaObject({
+      message: schemaString(true),
+    }),
+    output_schema: schemaObject({}),
+    input_mapping: {
+      rules: [
+        {
+          source: '$.mcp.message',
+          target: '$.request.grpc.message',
+          required: true,
+        },
+      ],
+    },
+    output_mapping: { rules: [] },
+    execution_config: {
+      timeout_ms: 1000,
+      headers: {},
+      streaming: {
+        mode: 'session',
+        transport_behavior: 'server_stream',
+        window_duration_ms: 1000,
+        poll_interval_ms: 250,
+        upstream_timeout_ms: 1000,
+        idle_timeout_ms: 5000,
+        max_session_lifetime_ms: 60000,
+        max_items: 1,
+        max_bytes: 16384,
+        aggregation_mode: 'summary_plus_samples',
+        items_path: '$.items',
+        done_path: '$.done',
+        redacted_paths: [],
+        truncate_item_fields: false,
+        drop_duplicates: false,
+        tool_family: {
+          start_tool_name: `${name}_start`,
+          poll_tool_name: `${name}_poll`,
+          stop_tool_name: `${name}_stop`,
+        },
+      },
+    },
+    tool_description: {
+      title: 'Playwright gRPC Session',
+      description: 'Streams gRPC echo messages through a bounded MCP session tool family.',
+      tags: ['playwright', 'streaming', 'session', 'grpc'],
+      examples: [{ input: { message: 'hello' } }],
+    },
+  };
+}
+
+function buildRestAsyncJobOperationPayload(name) {
+  return {
+    name,
+    display_name: 'Playwright Async Lead',
+    category: 'streaming',
+    protocol: 'rest',
+    target: {
+      kind: 'rest',
+      base_url: `http://127.0.0.1:${STREAM_FIXTURE_PORT}`,
+      method: 'POST',
+      path_template: '/crm/leads',
+      static_headers: {},
+    },
+    input_schema: schemaObject({
+      email: schemaString(true),
+    }),
+    output_schema: schemaObject({
+      id: schemaString(true),
+    }),
+    input_mapping: {
+      rules: [
+        {
+          source: '$.mcp.email',
+          target: '$.request.body.email',
+          required: true,
+        },
+      ],
+    },
+    output_mapping: {
+      rules: [
+        {
+          source: '$.response.body.id',
+          target: '$.output.id',
+          required: true,
+        },
+      ],
+    },
+    execution_config: {
+      timeout_ms: 2000,
+      headers: {},
+      streaming: {
+        mode: 'async_job',
+        transport_behavior: 'deferred_result',
+        poll_interval_ms: 250,
+        upstream_timeout_ms: 2000,
+        max_session_lifetime_ms: 300000,
+        max_bytes: 16384,
+        aggregation_mode: 'summary_only',
+        redacted_paths: [],
+        truncate_item_fields: false,
+        drop_duplicates: false,
+        tool_family: {
+          start_tool_name: `${name}_start`,
+          status_tool_name: `${name}_status`,
+          result_tool_name: `${name}_result`,
+          cancel_tool_name: `${name}_cancel`,
+        },
+      },
+    },
+    tool_description: {
+      title: 'Playwright Async Lead',
+      description: 'Creates a lead through the async job MCP tool family.',
+      tags: ['playwright', 'streaming', 'async_job'],
+      examples: [{ input: { email: 'user@example.com' } }],
+    },
+  };
+}
+
+async function setupPublishedAgent(page, { operationPayload, agentSlug, toolName, toolTitle, toolDescription }) {
+  const workspace = await getCurrentWorkspace(page);
+  if (!workspace) {
+    throw new Error('current workspace was not resolved');
+  }
+
+  const createdOperation = await createOperation(page, workspace.id, operationPayload);
+  await publishOperation(page, workspace.id, createdOperation.operation_id, createdOperation.version);
+
+  const createdAgent = await createAgent(page, workspace.id, {
+    slug: agentSlug,
+    display_name: toolTitle,
+    description: toolDescription,
+    instructions: {},
+    tool_selection_policy: {},
+  });
+
+  await saveAgentBindings(page, workspace.id, createdAgent.agent_id, [
+    {
+      operation_id: createdOperation.operation_id,
+      operation_version: createdOperation.version,
+      tool_name: toolName,
+      tool_title: toolTitle,
+      tool_description_override: toolDescription,
+      enabled: true,
+    },
+  ]);
+  await publishAgent(page, workspace.id, createdAgent.agent_id, createdAgent.version);
+
+  const key = await createPlatformApiKey(page, workspace.id, uniqueName('playwright_key'), ['read', 'write']);
+
+  return {
+    workspace,
+    operationId: createdOperation.operation_id,
+    agentId: createdAgent.agent_id,
+    apiKeySecret: key.secret,
+  };
+}
+
+function mcpUrl(workspaceSlug, agentSlug) {
+  return `http://127.0.0.1:${MCP_PORT}/v1/${workspaceSlug}/${agentSlug}`;
+}
+
+async function initializeMcpSession({ workspaceSlug, agentSlug, apiKey }) {
+  const response = await fetch(mcpUrl(workspaceSlug, agentSlug), {
+    method: 'POST',
+    headers: {
+      Accept: 'application/json, text/event-stream',
+      Authorization: `Bearer ${apiKey}`,
+      'Content-Type': 'application/json',
+    },
+    body: JSON.stringify({
+      jsonrpc: '2.0',
+      id: 1,
+      method: 'initialize',
+      params: {
+        protocolVersion: '2025-11-25',
+      },
+    }),
+  });
+
+  if (!response.ok) {
+    throw new Error(`initialize failed: ${response.status} ${await response.text()}`);
+  }
+
+  const sessionId = response.headers.get('MCP-Session-Id');
+  if (!sessionId) {
+    throw new Error('initialize response did not include MCP-Session-Id');
+  }
+
+  const initialized = await fetch(mcpUrl(workspaceSlug, agentSlug), {
+    method: 'POST',
+    headers: {
+      Accept: 'application/json, text/event-stream',
+      Authorization: `Bearer ${apiKey}`,
+      'Content-Type': 'application/json',
+      'MCP-Session-Id': sessionId,
+      'MCP-Protocol-Version': '2025-11-25',
+    },
+    body: JSON.stringify({
+      jsonrpc: '2.0',
+      method: 'notifications/initialized',
+      params: {},
+    }),
+  });
+
+  if (initialized.status !== 202) {
+    throw new Error(`initialized notification failed: ${initialized.status} ${await initialized.text()}`);
+  }
+
+  return {
+    sessionId,
+    workspaceSlug,
+    agentSlug,
+    apiKey,
+  };
+}
+
+async function mcpToolCall(session, toolName, argumentsValue) {
+  const response = await fetch(mcpUrl(session.workspaceSlug, session.agentSlug), {
+    method: 'POST',
+    headers: {
+      Accept: 'application/json, text/event-stream',
+      Authorization: `Bearer ${session.apiKey}`,
+      'Content-Type': 'application/json',
+      'MCP-Session-Id': session.sessionId,
+      'MCP-Protocol-Version': '2025-11-25',
+    },
+    body: JSON.stringify({
+      jsonrpc: '2.0',
+      id: Date.now(),
+      method: 'tools/call',
+      params: {
+        name: toolName,
+        arguments: argumentsValue || {},
+      },
+    }),
+  });
+
+  const payload = await response.json();
+  if (!response.ok || payload.error) {
+    throw new Error(`tools/call failed for ${toolName}: ${JSON.stringify(payload)}`);
+  }
+  return payload.result.structuredContent;
+}
+
 module.exports = {
   ADMIN_EMAIL,
   ADMIN_PASSWORD,
+  browserJson,
+  buildGrpcSessionOperationPayload,
+  buildRestAsyncJobOperationPayload,
+  buildRestWindowOperationPayload,
+  createAgent,
+  createOperation,
+  createPlatformApiKey,
+  getCurrentWorkspace,
+  getSession,
+  initializeMcpSession,
   login,
   localized,
+  mcpToolCall,
+  publishAgent,
+  publishOperation,
+  saveAgentBindings,
+  setupPublishedAgent,
+  uniqueName,
 };
diff --git a/apps/ui/tests/e2e/streaming.spec.js b/apps/ui/tests/e2e/streaming.spec.js
new file mode 100644
index 0000000..3a1109f
--- /dev/null
+++ b/apps/ui/tests/e2e/streaming.spec.js
@@ -0,0 +1,125 @@
+const { test, expect } = require('@playwright/test');
+const {
+  buildGrpcSessionOperationPayload,
+  buildRestAsyncJobOperationPayload,
+  buildRestWindowOperationPayload,
+  initializeMcpSession,
+  login,
+  mcpToolCall,
+  setupPublishedAgent,
+  uniqueName,
+} = require('./helpers');
+
+test('rest window tool executes through MCP', async ({ page }) => {
+  await login(page);
+
+  const operationName = uniqueName('playwright_window_logs');
+  const agentSlug = uniqueName('sales_window');
+  const payload = buildRestWindowOperationPayload(operationName);
+
+  const bundle = await setupPublishedAgent(page, {
+    operationPayload: payload,
+    agentSlug,
+    toolName: operationName,
+    toolTitle: 'Playwright Window Logs',
+    toolDescription: 'Collects a bounded SSE log window from the local fixture.',
+  });
+
+  const mcp = await initializeMcpSession({
+    workspaceSlug: bundle.workspace.slug,
+    agentSlug,
+    apiKey: bundle.apiKeySecret,
+  });
+  const result = await mcpToolCall(mcp, operationName, {});
+
+  expect(result.window_complete).toBe(true);
+  expect(result.has_more).toBe(false);
+  expect(Array.isArray(result.items)).toBe(true);
+  expect(result.items).toHaveLength(3);
+  expect(result.items[0].message).toBe('billing started');
+  expect(result.items[2].message).toBe('invoice timeout');
+});
+
+test('grpc session tools create persisted stream sessions visible in UI', async ({ page }) => {
+  await login(page);
+
+  const operationName = uniqueName('playwright_echo_session');
+  const agentSlug = uniqueName('sales_session');
+  const payload = buildGrpcSessionOperationPayload(operationName);
+
+  const bundle = await setupPublishedAgent(page, {
+    operationPayload: payload,
+    agentSlug,
+    toolName: operationName,
+    toolTitle: 'Playwright gRPC Session',
+    toolDescription: 'Streams gRPC echo messages through a bounded MCP session tool family.',
+  });
+
+  const mcp = await initializeMcpSession({
+    workspaceSlug: bundle.workspace.slug,
+    agentSlug,
+    apiKey: bundle.apiKeySecret,
+  });
+
+  const started = await mcpToolCall(mcp, `${operationName}_start`, { message: 'hello' });
+  const sessionId = started.session_id;
+  expect(started.status).toBe('running');
+
+  const polled = await mcpToolCall(mcp, `${operationName}_poll`, { session_id: sessionId });
+  expect(polled.session_id).toBe(sessionId);
+  expect(Array.isArray(polled.items)).toBe(true);
+  expect(polled.items[0].message).toContain('hello');
+
+  await page.goto('/stream-sessions');
+  const sessionCard = page.locator(`.resource-card[data-session-id="${sessionId}"]`);
+  await expect(sessionCard).toBeVisible();
+  await sessionCard.locator('[data-action="toggle"]').click();
+  await expect(sessionCard.locator('.resource-detail-pre')).toBeVisible();
+  await sessionCard.locator('[data-action="stop"]').click();
+  await expect(sessionCard.locator('[data-action="stop"]')).toHaveCount(0);
+});
+
+test('async job tools complete and expose results in UI', async ({ page }) => {
+  await login(page);
+
+  const operationName = uniqueName('playwright_async_lead');
+  const agentSlug = uniqueName('sales_async');
+  const payload = buildRestAsyncJobOperationPayload(operationName);
+
+  const bundle = await setupPublishedAgent(page, {
+    operationPayload: payload,
+    agentSlug,
+    toolName: operationName,
+    toolTitle: 'Playwright Async Lead',
+    toolDescription: 'Creates a lead through the async job MCP tool family.',
+  });
+
+  const mcp = await initializeMcpSession({
+    workspaceSlug: bundle.workspace.slug,
+    agentSlug,
+    apiKey: bundle.apiKeySecret,
+  });
+
+  const started = await mcpToolCall(mcp, `${operationName}_start`, { email: 'user@example.com' });
+  const jobId = started.job_id;
+  expect(started.status).toBe('running');
+
+  let status = null;
+  for (let attempt = 0; attempt < 20; attempt += 1) {
+    status = await mcpToolCall(mcp, `${operationName}_status`, { job_id: jobId });
+    if (status.status === 'completed') {
+      break;
+    }
+    await page.waitForTimeout(50);
+  }
+
+  expect(status.status).toBe('completed');
+  const result = await mcpToolCall(mcp, `${operationName}_result`, { job_id: jobId });
+  expect(result.id).toBe('lead_123');
+
+  await page.goto('/async-jobs');
+  const jobCard = page.locator(`.resource-card[data-job-id="${jobId}"]`);
+  await expect(jobCard).toBeVisible();
+  await jobCard.locator('[data-action="toggle"]').click();
+  await expect(page.locator(`[data-result-for="${jobId}"]`)).toContainText('lead_123');
+});