"""Browser page context capture endpoints. This router is intentionally isolated behind ``config.yaml``: browser_context: enabled: true When disabled, mutating endpoints behave as if the integration is unavailable. """ from __future__ import annotations import asyncio import logging import re from datetime import UTC, datetime from typing import Any, Literal from urllib.parse import urlencode from uuid import uuid4 from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel, Field from app.gateway.authz import require_permission from app.gateway.deps import get_checkpointer, get_config, get_thread_store from app.gateway.routers.thread_runs import RunCreateRequest from app.gateway.routers.threads import ThreadCreateRequest, create_thread from app.gateway.routers.uploads import _make_file_sandbox_writable from app.gateway.services import start_run from deerflow.browser_context.tasks import claim_next_browser_task, complete_browser_task, prune_browser_tasks, tasks_dir from deerflow.config.app_config import AppConfig from deerflow.config.browser_context_config import BrowserContextConfig from deerflow.runtime import serialize_channel_values from deerflow.runtime.user_context import get_effective_user_id from deerflow.sandbox.sandbox_provider import get_sandbox_provider from deerflow.uploads.manager import ( UnsafeUploadPathError, ensure_uploads_dir, normalize_filename, open_upload_file_no_symlink, upload_artifact_url, upload_virtual_path, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/browser-context", tags=["browser-context"]) DEFAULT_BROWSER_CONTEXT_PROMPT = "\u8bf7\u57fa\u4e8e\u521a\u521a\u53d1\u9001\u7684\u7f51\u9875\u5185\u5bb9\uff0c\u5148\u7528\u4e2d\u6587\u603b\u7ed3\u8981\u70b9\uff0c\u5e76\u7b49\u5f85\u6211\u7ee7\u7eed\u63d0\u95ee\u3002" class BrowserContextStatusResponse(BaseModel): enabled: bool max_content_chars: int max_title_chars: int max_url_chars: int task_result_timeout_seconds: int task_poll_interval_seconds: float task_lease_seconds: int tool_result_max_chars: int class BrowserContextCaptureRequest(BaseModel): thread_id: str | None = Field(default=None, description="Existing or caller-generated zncm thread id") title: str | None = Field(default=None, description="Current browser page title") url: str | None = Field(default=None, description="Current browser page URL") content: str = Field(description="Captured page content, preferably Markdown") content_type: Literal["markdown", "text"] = Field(default="markdown") prompt: str | None = Field(default=None, description="Optional first question to ask in zncm after capture") class BrowserContextFileInfo(BaseModel): filename: str size: int path: str virtual_path: str artifact_url: str class BrowserContextCaptureResponse(BaseModel): success: bool thread_id: str file: BrowserContextFileInfo chat_path: str chat_url: str browser_context_token: str prompt: str truncated: bool = False message: str class BrowserContextAskRequest(BaseModel): thread_id: str | None = Field(default=None, description="Existing zncm thread id for follow-up questions") title: str | None = Field(default=None, description="Current browser page title") url: str | None = Field(default=None, description="Current browser page URL") content: str = Field(description="Captured page content, preferably Markdown") content_type: Literal["markdown", "text"] = Field(default="markdown") question: str = Field(description="Question asked from the in-page assistant") class BrowserContextAskResponse(BaseModel): success: bool thread_id: str run_id: str answer: str truncated: bool = False message: str class BrowserContextTaskPayload(BaseModel): id: str kind: str = "fetch" url: str instruction: str = "" session_id: str = "" action: dict[str, Any] | None = None created_at: str updated_at: str max_content_chars: int class BrowserContextTaskNextResponse(BaseModel): task: BrowserContextTaskPayload | None = None class BrowserContextTaskResultRequest(BaseModel): kind: str = Field(default="fetch", description="Task kind the extension processed: fetch or act") title: str | None = Field(default=None, description="Captured page title") url: str | None = Field(default=None, description="Final page URL after redirects") content: str | None = Field(default=None, description="Captured page content") content_type: Literal["markdown", "text"] = Field(default="markdown") observation: dict[str, Any] | None = Field(default=None, description="Structured observation for an act step (elements, text, status)") error: str | None = Field(default=None, description="Extension-side task error") metadata: dict[str, Any] = Field(default_factory=dict) class BrowserContextTaskResultResponse(BaseModel): success: bool task_id: str status: str truncated: bool = False def _settings(config: AppConfig) -> BrowserContextConfig: value = getattr(config, "browser_context", None) if isinstance(value, BrowserContextConfig): return value if isinstance(value, dict): return BrowserContextConfig.model_validate(value) return BrowserContextConfig() def _require_enabled(config: AppConfig) -> BrowserContextConfig: settings = _settings(config) if not settings.enabled: raise HTTPException(status_code=404, detail="Browser context integration is disabled") return settings def _trim(value: str | None, limit: int) -> str: if not value: return "" return value.strip()[:limit] def _limit_utf8_bytes(value: str, max_bytes: int) -> str: text = value while len(text.encode("utf-8")) > max_bytes: text = text[:-1] return text def _filename_part(value: str, fallback: str, *, max_bytes: int) -> str: text = value.strip() or fallback # Keep unicode word chars so Chinese titles remain readable, while removing # path separators and shell-hostile punctuation. text = re.sub(r"[^\w.-]+", "-", text, flags=re.UNICODE).strip("-._") text = _limit_utf8_bytes(text, max_bytes) return text or fallback def _build_markdown( *, title: str, url: str, content: str, content_type: str, captured_at: str, truncated: bool, ) -> str: lines = [ f"# {title or 'Captured Web Page'}", "", f"- Source URL: {url or '(unknown)'}", f"- Captured at: {captured_at}", ] if truncated: lines.append("- Note: Content was truncated by the browser_context.max_content_chars limit.") lines.extend(["", "---", ""]) if content_type == "markdown": lines.append(content.strip()) else: lines.append("```text") lines.append(content.strip()) lines.append("```") lines.append("") return "\n".join(lines) def _build_ask_prompt( *, title: str, url: str, content: str, content_type: str, captured_at: str, truncated: bool, ) -> str: note = "页面内容已按配置截断，请只基于可见内容回答。" if truncated else "页面内容如下。" body = content.strip() if content_type == "text": body = f"```text\n{body}\n```" return "\n\n".join( [ "你是 zncm 页面问答助手。请直接基于当前网页内容回答用户问题。", "如果网页内容不足以回答，请说明缺少哪些信息；不要编造，也不要再尝试抓取其它网页。", f"页面标题：{title or '未知标题'}", f"页面 URL：{url or '未知 URL'}", f"读取时间：{captured_at}", f"说明：{note}", "当前网页内容：", body, ] ) def _message_type(message: Any) -> str: if isinstance(message, dict): return str(message.get("type") or message.get("role") or "").lower() return str(getattr(message, "type", "") or getattr(message, "role", "") or "").lower() def _content_to_text(content: Any) -> str: if isinstance(content, str): return content.strip() if isinstance(content, list): parts: list[str] = [] for item in content: if isinstance(item, str): parts.append(item) elif isinstance(item, dict): if isinstance(item.get("text"), str): parts.append(item["text"]) elif item.get("type") == "text" and isinstance(item.get("content"), str): parts.append(item["content"]) return "\n".join(part.strip() for part in parts if part and part.strip()).strip() return "" def _latest_ai_answer(values: dict[str, Any]) -> str: messages = values.get("messages") if not isinstance(messages, list): return "" for message in reversed(messages): if _message_type(message) not in {"ai", "assistant"}: continue answer = _content_to_text(message.get("content") if isinstance(message, dict) else getattr(message, "content", None)) if answer: return answer return "" async def _sync_to_sandbox_if_needed(thread_id: str, virtual_path: str, file_path, data: bytes) -> None: sandbox_provider = get_sandbox_provider() if bool(getattr(sandbox_provider, "uses_thread_data_mounts", False)): return sandbox_id = sandbox_provider.acquire(thread_id) sandbox = sandbox_provider.get(sandbox_id) if sandbox is None: raise HTTPException(status_code=500, detail="Failed to acquire sandbox") _make_file_sandbox_writable(file_path) sandbox.update_file(virtual_path, data) @router.get("/status", response_model=BrowserContextStatusResponse) async def get_browser_context_status( config: AppConfig = Depends(get_config), ) -> BrowserContextStatusResponse: settings = _settings(config) return BrowserContextStatusResponse( enabled=settings.enabled, max_content_chars=settings.max_content_chars, max_title_chars=settings.max_title_chars, max_url_chars=settings.max_url_chars, task_result_timeout_seconds=settings.task_result_timeout_seconds, task_poll_interval_seconds=settings.task_poll_interval_seconds, task_lease_seconds=settings.task_lease_seconds, tool_result_max_chars=settings.tool_result_max_chars, ) @router.get("/tasks/next", response_model=BrowserContextTaskNextResponse) @require_permission("runs", "create") async def claim_browser_context_task( request: Request, worker_id: str | None = None, config: AppConfig = Depends(get_config), ) -> BrowserContextTaskNextResponse: settings = _require_enabled(config) effective_user_id = get_effective_user_id() effective_worker_id = worker_id or "chrome-extension" logger.info( "Browser task next poll user_id=%s worker_id=%s tasks_dir=%s client=%s", effective_user_id, effective_worker_id, tasks_dir(), getattr(request.client, "host", None), ) prune_browser_tasks() task = claim_next_browser_task( user_id=effective_user_id, worker_id=effective_worker_id, lease_seconds=settings.task_lease_seconds, ) if task is None: return BrowserContextTaskNextResponse(task=None) return BrowserContextTaskNextResponse(task=BrowserContextTaskPayload.model_validate(task)) @router.post("/tasks/{task_id}/result", response_model=BrowserContextTaskResultResponse) @require_permission("runs", "create") async def submit_browser_context_task_result( task_id: str, body: BrowserContextTaskResultRequest, request: Request, config: AppConfig = Depends(get_config), ) -> BrowserContextTaskResultResponse: settings = _require_enabled(config) error = _trim(body.error, 4000) content = body.content or "" is_act = (body.kind or "fetch") == "act" if not error and not is_act and not content.strip(): raise HTTPException(status_code=400, detail="Captured page content is empty") if not error and is_act and not isinstance(body.observation, dict): raise HTTPException(status_code=400, detail="Act task result requires an observation") task = complete_browser_task( task_id=task_id, user_id=get_effective_user_id(), title=_trim(body.title, settings.max_title_chars), url=_trim(body.url, settings.max_url_chars), content=content, content_type=body.content_type, error=error, metadata=body.metadata, observation=body.observation, ) if task is None: raise HTTPException(status_code=404, detail=f"Browser task {task_id} not found") result = task.get("result") or {} return BrowserContextTaskResultResponse( success=True, task_id=task_id, status=str(task.get("status") or "unknown"), truncated=bool(result.get("truncated")), ) @router.post("/capture", response_model=BrowserContextCaptureResponse) @require_permission("threads", "write") async def capture_browser_context( body: BrowserContextCaptureRequest, request: Request, config: AppConfig = Depends(get_config), ) -> BrowserContextCaptureResponse: settings = _require_enabled(config) raw_content = (body.content or "").strip() if not raw_content: raise HTTPException(status_code=400, detail="Captured page content is empty") title = _trim(body.title, settings.max_title_chars) url = _trim(body.url, settings.max_url_chars) truncated = len(raw_content) > settings.max_content_chars content = raw_content[: settings.max_content_chars] if truncated else raw_content requested_thread_id = (body.thread_id or "").strip() or None if requested_thread_id: thread_store = get_thread_store(request) allowed = await thread_store.check_access( requested_thread_id, get_effective_user_id(), require_existing=False, ) if not allowed: raise HTTPException(status_code=404, detail=f"Thread {requested_thread_id} not found") thread_response = await create_thread( ThreadCreateRequest( thread_id=requested_thread_id, metadata={ "source": "browser_context", "browser_context_url": url, "browser_context_title": title, }, ), request, ) thread_id = thread_response.thread_id captured_at = datetime.now(UTC).isoformat() markdown = _build_markdown( title=title, url=url, content=content, content_type=body.content_type, captured_at=captured_at, truncated=truncated, ) data = markdown.encode("utf-8") timestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S") prefix = _filename_part(settings.filename_prefix, "webpage", max_bytes=40) title_part = _filename_part(title, "page", max_bytes=120) filename = normalize_filename(f"{prefix}-{timestamp}-{title_part}.md") try: uploads_dir = ensure_uploads_dir(thread_id) file_path, fh = open_upload_file_no_symlink(uploads_dir, filename) with fh: fh.write(data) except UnsafeUploadPathError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: logger.exception("Failed to save browser context capture for thread %s", thread_id) raise HTTPException(status_code=500, detail="Failed to save captured page") from exc virtual_path = upload_virtual_path(filename) await _sync_to_sandbox_if_needed(thread_id, virtual_path, file_path, data) prompt = (body.prompt or "").strip() or DEFAULT_BROWSER_CONTEXT_PROMPT token = uuid4().hex route_base = settings.route_base.rstrip("/") or "/page/workspace" chat_path = f"{route_base}/chats/{thread_id}" query = urlencode( { "browser_context": token, "bc_filename": filename, "bc_size": str(len(data)), "bc_path": virtual_path, "bc_prompt": prompt, } ) chat_url = f"#{chat_path}?{query}" return BrowserContextCaptureResponse( success=True, thread_id=thread_id, file=BrowserContextFileInfo( filename=filename, size=len(data), path=virtual_path, virtual_path=virtual_path, artifact_url=upload_artifact_url(thread_id, filename), ), chat_path=chat_path, chat_url=chat_url, browser_context_token=token, prompt=prompt, truncated=truncated, message="Captured page saved as a zncm upload", ) @router.post("/ask", response_model=BrowserContextAskResponse) @require_permission("runs", "create") async def ask_browser_context( body: BrowserContextAskRequest, request: Request, config: AppConfig = Depends(get_config), ) -> BrowserContextAskResponse: settings = _require_enabled(config) raw_content = (body.content or "").strip() if not raw_content: raise HTTPException(status_code=400, detail="Captured page content is empty") question = (body.question or "").strip() if not question: raise HTTPException(status_code=400, detail="Question is empty") title = _trim(body.title, settings.max_title_chars) url = _trim(body.url, settings.max_url_chars) content_limit = max(1_000, min(settings.max_content_chars, settings.tool_result_max_chars)) truncated = len(raw_content) > content_limit content = raw_content[:content_limit] if truncated else raw_content requested_thread_id = (body.thread_id or "").strip() or None if requested_thread_id: thread_store = get_thread_store(request) allowed = await thread_store.check_access( requested_thread_id, get_effective_user_id(), require_existing=False, ) if not allowed: raise HTTPException(status_code=404, detail=f"Thread {requested_thread_id} not found") thread_response = await create_thread( ThreadCreateRequest( thread_id=requested_thread_id, assistant_id="lead_agent", metadata={ "source": "browser_context_page_assistant", "browser_context_url": url, "browser_context_title": title, }, ), request, ) thread_id = thread_response.thread_id captured_at = datetime.now(UTC).isoformat() is_follow_up = requested_thread_id is not None if is_follow_up: prompt_prefix = "\n\n".join( [ "你是 zncm 页面问答助手。请继续基于本会话里已经读取的网页内容回答用户问题。", "如果本会话里的网页内容不足以回答，请说明缺少哪些信息；不要编造，也不要再尝试抓取其它网页。", f"当前页面标题：{title or '未知标题'}", f"当前页面 URL：{url or '未知 URL'}", f"本轮提问时间：{captured_at}", ] ) else: prompt_prefix = _build_ask_prompt( title=title, url=url, content=content, content_type=body.content_type, captured_at=captured_at, truncated=truncated, ) run_body = RunCreateRequest( assistant_id="lead_agent", input={ "messages": [ { "type": "human", "content": question, "additional_kwargs": { "prompt_prefix": prompt_prefix, }, } ] }, metadata={ "source": "browser_context_page_assistant", "browser_context_url": url, "browser_context_title": title, "browser_context_truncated": truncated, }, stream_mode=["values"], on_disconnect="continue", multitask_strategy="reject", ) record = await start_run(run_body, thread_id, request) if record.task is not None: try: await record.task except asyncio.CancelledError: pass checkpointer = get_checkpointer(request) values: dict[str, Any] = {} try: checkpoint_tuple = await checkpointer.aget_tuple({"configurable": {"thread_id": thread_id}}) if checkpoint_tuple is not None: checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {} values = serialize_channel_values(checkpoint.get("channel_values", {}) or {}) except Exception: logger.exception("Failed to fetch browser context answer for run %s", record.run_id) answer = _latest_ai_answer(values) if not answer: answer = record.error or "zncm did not return an answer." return BrowserContextAskResponse( success=bool(answer), thread_id=thread_id, run_id=record.run_id, answer=answer, truncated=truncated, message="zncm page assistant answer completed", )