perf(plugins): make the scraper plugins fast, and let the agent read their results

plugins/remote/host/.env.example

@@ -10,8 +10,14 @@ PORT=8000
 PLUGINS=google-search,web-crawl,youtube,tiktok,google-maps,amazon,walmart,indeed
 
 # Required when more than one copy runs, so a Stop reaches the copy with the call.
+# Google Search also shares its solved IPs there, so a copy adopts one instead of
+# paying for a captcha solve.
 REDIS_URL=redis://localhost:6379/0
 
+# off (or 0, false, no) holds no plugin's lifespan: nothing is warmed at startup,
+# such as a browser, and no shared pool is closed at shutdown. Unset, on.
+# PLUGIN_LIFESPANS=off
+
 # The license check's Keygen, under the backend's names and with its values.
 # The accounts endpoint; unset, Keygen Cloud's. Production's self-hosted Keygen CE is
 # reached inside the deployment:
@@ -43,3 +49,34 @@ SCRAPING_CAPTCHA_PROVIDER=capsolver
 # true runs a headed browser on the image's virtual display: TikTok's profile
 # feeds are withheld from a headless one.
 CRAWL_HEADED_XVFB_ENABLED=true
+
+# Speed settings, all optional: unset is what was measured. Most are kill switches,
+# where 0 brings back the path before the change; a number in brackets is the default.
+# Shared by the scraper plugins (proprietary/scraping/):
+# SCRAPING_HEDGE=0                no hedged second request anywhere
+# SCRAPING_MAX_HTTP=48            plain-HTTP requests in flight across plugins
+# SCRAPING_MAX_IDLE_SESSIONS=16   warm sessions kept between calls; 0 keeps none
+# SCRAPING_MAX_BROWSERS=3         pooled browsers, one per launch profile
+# SCRAPING_MAX_BROWSER_PAGES=6    pooled pages in all, 4 per plugin
+# SCRAPING_MAX_KEPT_CONTEXTS=4    contexts kept with their clearance; 0 keeps none
+# SCRAPING_BROWSER_IDLE_S=600     idle seconds before a pooled browser closes;
+# SCRAPING_DEFAULT_BROWSER_IDLE_S=1800  the same for the default launch profile
+# Per plugin:
+# GOOGLE_SEARCH_FAST_SETTLE=0, GOOGLE_SEARCH_SHRINK_DOM=0, GOOGLE_SEARCH_PARALLEL_PAGES=0,
+# GOOGLE_SEARCH_GOTO_PREFETCH=0, GOOGLE_SEARCH_GOTO_WAVE_SESSION=0,
+# GOOGLE_SEARCH_GOTO_AI_SOURCES=0, GOOGLE_SEARCH_SITE_RETRIES=0, GOOGLE_SEARCH_WARM_BROWSER=0
+# INDEED_KEPT_CONTEXT=0, INDEED_SPA_DETAILS=0, INDEED_TARGET_CONCURRENCY=1 (3);
+# INDEED_PIN_COUNTRY=0; INDEED_WARM_FIRST=1 and INDEED_ALWAYS_SOLVE=1 land on the home
+# page and run the Cloudflare solver on every page again
+# YOUTUBE_NATIVE_SUBTITLES=0, YOUTUBE_COMMENTS_NEXT_SEED=0, YOUTUBE_HEDGE=0;
+# YOUTUBE_KEEP_IDLE=1 keeps warm sessions between calls
+# GOOGLE_MAPS_SKIP_NID_MINT=0, GOOGLE_MAPS_KEEP_ALIVE=0, GOOGLE_MAPS_HEDGE_AFTER_S=0 (2.5);
+# GOOGLE_MAPS_KEEP_IDLE=1 keeps warm sessions between calls
+# AMAZON_HEDGE=0, AMAZON_NO_GOOGLE_REFERER=0, AMAZON_AOD_XHR=0, AMAZON_DETAIL_CONCURRENCY=8 (16)
+# WALMART_PARALLEL=0; WALMART_CANONICAL_URL=1 asks for the slug URL directly (its block
+# check failed, so it stays off)
+# TIKTOK_TRENDING_API=0, TIKTOK_COMMENTS_API=0, TIKTOK_BLOCK_MEDIA=0, TIKTOK_WARM_BROWSER=0,
+# TIKTOK_PAGE_HEDGE_S=0 (2), TIKTOK_BROWSER_CONCURRENCY=1 (2);
+# TIKTOK_BROWSER_WARMUP=1 and TIKTOK_PAGE_WARMUP=1 bring the msToken and ttwid warm-ups back
+# WEB_CRAWL_STATUS_ROUTING=0, WEB_CRAWL_HTML_CAP=0, WEB_CRAWL_RAW_STATIC=0,
+# WEB_CRAWL_BROWSER_POOL=0, WEB_CRAWL_CONCURRENCY=1 (6)

plugins/remote/host/__main__.py

@@ -2,6 +2,7 @@
 
 PLUGINS, a comma-separated list of ids, limits the container to those plugins.
 REDIS_URL passes a Stop to the other copies of the host.
+PLUGIN_LIFESPANS=off (or 0, false, no) holds no plugin's lifespan: no warm-ups.
 Logs are redacted before a plugin loads, so no credential reaches them.
 """
 
@@ -19,5 +20,7 @@
 app = host_app(
     plugins={plugin.strip() for plugin in only.split(",")} if only else None,
     broadcast=RedisBroadcast(redis_url) if redis_url else None,
+    lifespans=os.environ.get("PLUGIN_LIFESPANS", "on").strip().lower()
+    not in {"off", "0", "false", "no"},
 )
 uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))

plugins/remote/host/app.py

@@ -1,5 +1,6 @@
 """The plugin host: `/health`, `/cancel`, `/license`, and each plugin's MCP server at `/<id>/mcp`."""
 
+import logging
 from collections.abc import AsyncIterator, Collection
 from contextlib import AsyncExitStack, asynccontextmanager
 from pathlib import Path
@@ -17,25 +18,43 @@
 from host.cancellable_calls import CancellableCalls
 from host.license_route import license_route
 from host.mounted_plugins import MountedPlugin, mounted_plugins
+from host.post_only import PostOnly
 from host.refused_unless_allowed import RefusedUnlessAllowed
 
+logger = logging.getLogger(__name__)
+
 # plugins/remote/, where the host's folder sits beside the plugins.
 PLUGINS_ROOT = Path(__file__).resolve().parent.parent
+# How long one plugin's lifespan may take to close its pools at shutdown.
+CLOSE_SECONDS = 10.0
 
 
 def host_app(
     root: Path = PLUGINS_ROOT,
     plugins: Collection[str] | None = None,
     broadcast: CancelBroadcast | None = None,
+    lifespans: bool = True,
 ) -> Starlette:
-    """`plugins` limits the container to some; `broadcast` reaches the other copies."""
+    """`plugins` limits the container to some; `broadcast` reaches the other copies.
+
+    `lifespans=False` holds no plugin's `lifespan`: nothing is warmed at startup, and
+    no shared pool is closed at shutdown.
+    """
     mounted = mounted_plugins(root, plugins)
     calls = CallsInFlight()
     stops = broadcast or LocalBroadcast()
 
     @asynccontextmanager
     async def lifespan(_: Starlette) -> AsyncIterator[None]:
         async with AsyncExitStack() as stack:
+            # Entered first to close last, once the calls using their pools have ended.
+            held = await stack.enter_async_context(anyio.create_task_group())
+            kept = [
+                _Kept(plugin) for plugin in mounted if plugin.lifespan and lifespans
+            ]
+            for plugin in kept:
+                held.start_soon(plugin.run)
+            stack.push_async_callback(_close, kept)
             for plugin in mounted:
                 await stack.enter_async_context(plugin.server.session_manager.run())
             listening = await stack.enter_async_context(anyio.create_task_group())
@@ -61,7 +80,58 @@ def _served(plugin: MountedPlugin, calls: CallsInFlight) -> ASGIApp:
     )
     # A plugin without `access` is open to anyone. SurfSense offers no free plugin
     # yet; one will be limited per IP address here, as the plugins proposal says.
-    return app if plugin.access is None else RefusedUnlessAllowed(app, plugin.access)
+    if plugin.access is not None:
+        app = RefusedUnlessAllowed(app, plugin.access)
+    # Outermost, so a refused GET costs no license check.
+    return PostOnly(app)
+
+
+class _Kept:
+    """One plugin's lifespan, held open by a task of its own.
+
+    Startup waits for none of them, so a slow warm-up never delays `/health`, and
+    one that fails is logged while the plugin is still served.
+    """
+
+    def __init__(self, plugin: MountedPlugin) -> None:
+        self.plugin = plugin
+        self.scope = anyio.CancelScope()
+        self.entered = False
+        self.leave = anyio.Event()
+        self.left = anyio.Event()
+
+    async def run(self) -> None:
+        assert self.plugin.lifespan is not None
+        try:
+            with self.scope:
+                async with self.plugin.lifespan():
+                    self.entered = True
+                    await self.leave.wait()
+        except Exception:
+            logger.exception("[host] %s: its lifespan failed", self.plugin.id)
+        finally:
+            self.left.set()
+
+    async def close(self) -> None:
+        if not self.entered:
+            # Still starting: nothing to close yet.
+            self.scope.cancel()
+        self.leave.set()
+        with anyio.move_on_after(CLOSE_SECONDS) as waited:
+            await self.left.wait()
+        if waited.cancelled_caught:
+            logger.warning(
+                "[host] %s: its lifespan took over %ss to close",
+                self.plugin.id,
+                CLOSE_SECONDS,
+            )
+            self.scope.cancel()
+
+
+async def _close(kept: list[_Kept]) -> None:
+    """Last mounted first, one at a time, as an exit stack would."""
+    for plugin in reversed(kept):
+        await plugin.close()
 
 
 async def _health(_: Request) -> PlainTextResponse:

plugins/remote/host/mounted_plugins.py

@@ -7,6 +7,7 @@
 import importlib.util
 import sys
 from collections.abc import Callable, Collection
+from contextlib import AbstractAsyncContextManager
 from dataclasses import dataclass
 from pathlib import Path
 from types import ModuleType
@@ -30,11 +31,17 @@
 )
 
 
+# What a plugin keeps from startup to shutdown, such as warm browsers and clients.
+Lifespan = Callable[[], AbstractAsyncContextManager[object]]
+
+
 @dataclass(frozen=True)
 class MountedPlugin:
     id: str
     server: MCPServer
     access: Access | None
+    # The `lifespan` its `server.py` may define; the host holds it open while it serves.
+    lifespan: Lifespan | None = None
 
 
 def mounted_plugins(
@@ -70,7 +77,12 @@ def mounted_plugins(
 def _mounted(folder: Path) -> MountedPlugin:
     module = _loaded(folder / "server.py")
     build: Callable[[], MCPServer] = module.server
-    return MountedPlugin(plugin_id(folder), build(), getattr(module, "access", None))
+    return MountedPlugin(
+        plugin_id(folder),
+        build(),
+        getattr(module, "access", None),
+        getattr(module, "lifespan", None),
+    )
 
 
 def plugin_id(folder: Path) -> str:

plugins/remote/host/post_only.py

@@ -0,0 +1,57 @@
+"""A plugin's MCP endpoint takes POST only: no copy keeps a stream open for a client."""
+
+import json
+
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+# The SDK's own code for a request it does not take.
+INVALID_REQUEST = -32600
+_REFUSED = {"GET", "DELETE"}
+
+
+class PostOnly:
+    """GET and DELETE on `/mcp` are 405 with `Allow: POST`; anything else goes through.
+
+    Without sessions, the SDK answers the client's fire-and-forget GET with an
+    event stream that never carries a message, holding a connection, a task and
+    a ping every 15 s per plugin per agent. The client reads 405 as "no stream".
+    There is no session to DELETE.
+    """
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if (
+            scope["type"] == "http"
+            and scope["method"] in _REFUSED
+            and _route_path(scope).rstrip("/") == "/mcp"
+        ):
+            body = json.dumps(
+                {
+                    "jsonrpc": "2.0",
+                    "id": None,
+                    "error": {"code": INVALID_REQUEST, "message": "Method Not Allowed"},
+                }
+            ).encode()
+            await send(
+                {
+                    "type": "http.response.start",
+                    "status": 405,
+                    "headers": [
+                        (b"allow", b"POST"),
+                        (b"content-type", b"application/json"),
+                        (b"content-length", str(len(body)).encode()),
+                    ],
+                }
+            )
+            await send({"type": "http.response.body", "body": body})
+            return
+        await self.app(scope, receive, send)
+
+
+def _route_path(scope: Scope) -> str:
+    """The path under the plugin's mount, as Starlette's own routing reads it."""
+    path: str = scope["path"]
+    root_path: str = scope.get("root_path", "")
+    return path[len(root_path) :] if root_path and path.startswith(root_path) else path

plugins/remote/host/pyproject.toml

@@ -14,7 +14,12 @@ dependencies = [
     # Passes a Stop to the copy of the host running the call.
     "redis>=5.2",
     # The scraper plugins' fetchers and stealth browsers; `scrapling install` adds the browsers.
-    "scrapling[fetchers]>=0.4.11,<0.5",
+    # Exact: the shared browser pool reuses its private context and challenge helpers,
+    # and a contract test fails on an upgrade that moves them.
+    "scrapling[fetchers]==0.4.15",
+    # The shared fail-fast transport speaks curl_cffi directly and reads JSON with orjson.
+    "curl_cffi>=0.16.3",
+    "orjson>=3.13.0",
     # YouTube's subtitles.
     "youtube-transcript-api>=1.0.3",
     "fake-useragent>=2.2.0",

plugins/remote/host/tests/events.py

@@ -0,0 +1,3 @@
+"""What the shared-layer sample plugins did, in order, for a test to read."""
+
+EVENTS: list[str] = []

plugins/remote/host/tests/shared_layer_plugins/broken/server.py

@@ -0,0 +1,21 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from mcp.server.mcpserver import MCPServer
+
+
+@asynccontextmanager
+async def lifespan() -> AsyncIterator[None]:
+    raise RuntimeError("the warm-up failed")
+    yield
+
+
+def server() -> MCPServer:
+    broken = MCPServer("broken")
+
+    @broken.tool()
+    def search(query: str) -> str:
+        """Search."""
+        return f"found {query}"
+
+    return broken

plugins/remote/host/tests/shared_layer_plugins/first/server.py

@@ -0,0 +1,33 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+import anyio
+from mcp.server.mcpserver import MCPServer
+
+from host.tests.events import EVENTS
+
+
+@asynccontextmanager
+async def lifespan() -> AsyncIterator[None]:
+    EVENTS.append("first entered")
+    try:
+        yield
+    finally:
+        EVENTS.append("first left")
+
+
+def server() -> MCPServer:
+    first = MCPServer("first")
+
+    @first.tool()
+    async def wait() -> str:
+        """Runs until it is stopped."""
+        EVENTS.append("call started")
+        try:
+            await anyio.sleep(60)
+        except anyio.get_cancelled_exc_class():
+            EVENTS.append("call ended")
+            raise
+        return "done"
+
+    return first

plugins/remote/host/tests/shared_layer_plugins/guarded/server.py

@@ -0,0 +1,27 @@
+from mcp.server.mcpserver import MCPServer
+from starlette.datastructures import Headers
+
+from host.tests.events import EVENTS
+from mcp_server.access import Refusal
+
+
+class Guard:
+    """Refuses every request, noting each one it was asked about."""
+
+    async def refusal(self, headers: Headers) -> Refusal | None:
+        EVENTS.append("guard asked")
+        return Refusal(401, "missing_license")
+
+
+access = Guard()
+
+
+def server() -> MCPServer:
+    guarded = MCPServer("guarded")
+
+    @guarded.tool()
+    def search(query: str) -> str:
+        """Search."""
+        return f"found {query}"
+
+    return guarded