# src/pages/builder/templates/agent-python/adapter.py """ASGI entry point — uvicorn runs this as the application.""" from __future__ import annotations import asyncio import hashlib import importlib import json import os import re import sys from typing import Any, Callable from urllib.parse import parse_qs import httpx # --- Observability bootstrap (must run before user code imports) --- # Reads entry names from AGENT_OBSERVABILITY_ENTRIES env var (set by main.py # or the dev runner). Idempotent — safe to call multiple times. _obs_entries_raw = os.environ.get("AGENT_OBSERVABILITY_ENTRIES", "") if _obs_entries_raw: try: from _platform._observability import setup as _obs_setup _obs_setup(json.loads(_obs_entries_raw)) except Exception as _obs_err: print(f"[observability] setup failed: {_obs_err}", file=sys.stderr, flush=True) from .store import InMemoryStore, BlobBackedStore from .runtime import AgentRuntime, RouteEntry from .context import EoContext, StreamResponse # --- Geo parsing (aligns with Node agent-node.ts parseEoConnectingGeo) --- def _parse_eo(headers: dict) -> EoContext: """Parse eo-connecting-geo header into structured EoContext. Format: key=value key="quoted value" Aligns with Node Agent parseEoConnectingGeo logic. """ import urllib.parse as _up geo_str = "" if isinstance(headers, dict): geo_str = _up.unquote(headers.get("eo-connecting-geo", "")) if not geo_str: return EoContext(geo={}, client_ip=str(headers.get("eo-connecting-ip", "")) if isinstance(headers, dict) else "") result = {} matches = __import__("re").findall(r'[a-z_]+="[^"]*"|[a-z_]+=[A-Za-z0-9.-]+', geo_str) for m in matches: key, val = m.split("=", 1) result[key] = val.strip('"') geo = {} if "asn" in result: geo["asn"] = result["asn"] if "nation_name" in result: geo["countryName"] = result["nation_name"] if "region_code" in result: geo["regionCode"] = result["region_code"] geo["countryCodeAlpha2"] = result["region_code"].split("-")[0] if "nation_numeric" in result: geo["countryCodeNumeric"] = result["nation_numeric"] if "region_name" in result: geo["regionName"] = result["region_name"] if "city_name" in result: geo["cityName"] = result["city_name"] if "latitude" in result: geo["latitude"] = result["latitude"] if "longitude" in result: geo["longitude"] = result["longitude"] if "network_operator" in result: geo["cisp"] = result["network_operator"] client_ip = str(headers.get("eo-connecting-ip", "")) if isinstance(headers, dict) else "" return EoContext(geo=geo, client_ip=client_ip) # --- Platform response headers (align with Node agent + python-function) --- def _build_platform_headers(request_headers: dict, status_code: int) -> list[tuple[bytes, bytes]]: """Build platform response headers matching Node agent behavior. Injects: - Functions-Request-Id: from incoming x-scf-request-id header (or _LAMBDA_RUNTIME_REQUEST_ID env) - eo-pages-inner-scf-status: actual response status code - eo-pages-inner-status-intercept: 'true' if status >= 500 else 'false' For agent-python (long-running ASGI), the request-id comes from the incoming request header x-scf-request-id (same as Node agent), not from _LAMBDA_RUNTIME_REQUEST_ID env var (which is per-invocation in SCF). We check both for compatibility. """ request_id = ( request_headers.get("x-scf-request-id", "") or os.environ.get("_LAMBDA_RUNTIME_REQUEST_ID", "") ) result: list[tuple[bytes, bytes]] = [] if request_id: result.append((b"functions-request-id", request_id.encode("latin1"))) result.append((b"eo-pages-inner-scf-status", str(status_code).encode("latin1"))) result.append(( b"eo-pages-inner-status-intercept", b"true" if status_code >= 500 else b"false", )) return result # LangGraph 通过抛出这些异常实现 graph 暂停 / 控制流跳转，并非业务错误。 # interrupt() 命中时会抛出 GraphInterrupt（Exception 子类），冒泡出 handler # generator 后会落进流式迭代的 except 分支——这属于「正常暂停等待人工」， # 不能当 streaming error 报错或把 root span 标 ERROR。 _INTERRUPT_EXCEPTION_NAMES = frozenset({ "GraphInterrupt", "NodeInterrupt", "ParentCommand", }) def _is_interrupt_exception(err: BaseException) -> bool: """按类继承链（MRO）匹配 LangGraph 控制流异常名。 走 MRO 是为了让 GraphInterrupt 的子类也能识别，同时足够严格、 不会误伤名字里恰好含 "interrupt" 的业务异常。 """ for cls in type(err).__mro__: if cls.__name__ in _INTERRUPT_EXCEPTION_NAMES: return True return False # --- Proxy transport for credential requests --- _PROXY_UUID = "{{PAGES_PROXY_UUID}}" _PROXY_HOST = "{{PAGES_PROXY_HOST}}" def _is_proxy_enabled() -> bool: """Check if proxy placeholders have been replaced with real values.""" return ( not _PROXY_UUID.startswith("{{") and not _PROXY_HOST.startswith("{{") and bool(_PROXY_UUID) and bool(_PROXY_HOST) ) class _ProxyTransport(httpx.AsyncBaseTransport): """Rewrites requests to go through the EdgeOne proxy with signing headers. Signature algorithm matches Node fetch-proxy.js: sign = md5(timestamp + '-' + pathname + '-' + uuid) """ def __init__(self) -> None: self._inner = httpx.AsyncHTTPTransport() async def handle_async_request(self, request: httpx.Request) -> httpx.Response: original_url = request.url pathname = original_url.raw_path.decode("ascii").split("?")[0] timestamp = str(int(__import__("time").time() * 1000)) sign = hashlib.md5( f"{timestamp}-{pathname}-{_PROXY_UUID}".encode() ).hexdigest() # Rewrite URL to proxy host, preserving path and query proxy_url = original_url.copy_with( host=_PROXY_HOST.replace("https://", "").replace("http://", ""), scheme="https", ) # Add signing headers request.headers["oe-host"] = str(original_url.host) request.headers["oe-timestamp"] = timestamp request.headers["oe-sign"] = sign # Build new request targeting proxy proxy_request = httpx.Request( method=request.method, url=proxy_url, headers=request.headers, content=request.content, ) return await self._inner.handle_async_request(proxy_request) async def aclose(self) -> None: await self._inner.aclose() def _build_proxy_client() -> httpx.AsyncClient | None: """Build an httpx.AsyncClient with proxy transport, or None if proxy is not configured.""" if not _is_proxy_enabled(): return None return httpx.AsyncClient(transport=_ProxyTransport(), timeout=30.0) # --- Cold start --- try: _route_table_raw = json.loads(os.environ.get("AGENT_ROUTE_TABLE", "{}")) except (json.JSONDecodeError, ValueError): print("[agent-python] ERROR: Invalid AGENT_ROUTE_TABLE, using empty", file=sys.stderr, flush=True) _route_table_raw = {} try: _timeout = int(os.environ.get("AGENT_TIMEOUT", "300")) except ValueError: print("[agent-python] WARNING: Invalid AGENT_TIMEOUT, using 300", file=sys.stderr, flush=True) _timeout = 300 # --- Dev mode: inject credential from environment variable --- # When user runs `edgeone pages link`, CLI injects PAGES_BLOB_DEPLOY_CREDENTIAL # into the Python process env. If present and valid (not placeholder), inject it # into pages-blob SDK so it can connect to COS without build-time replacement. _dev_credential = os.environ.get("PAGES_BLOB_DEPLOY_CREDENTIAL", "") if _dev_credential and not (_dev_credential.startswith("{{") and _dev_credential.endswith("}}")): try: from pages_blob import set_environment_context from pages_blob.types import PagesBlobContext set_environment_context(PagesBlobContext(deploy_credential=_dev_credential)) except ImportError: pass # --- Store resolver (matches Node agent-node.ts pattern) --- def _to_agent_store_name(route_path: str) -> str: """Convert route path to blob store name. Mirrors Node toAgentStoreName().""" normalized = re.sub(r"^/+|/+$", "", str(route_path or "")) parts = [p for p in normalized.split("/") if p] joined = "-".join(parts) joined = re.sub(r"[^a-zA-Z0-9_-]+", "-", joined) joined = re.sub(r"-+", "-", joined) joined = re.sub(r"^[-_]+|[-_]+$", "", joined) joined = joined.lower() suffix = joined or "index" return "agent-" + suffix def _create_store_resolver() -> Callable[[str], Any]: """Build store resolver: blob-backed in production, in-memory for dev.""" cache: dict[str, Any] = {} # 0. Local persist mode（dev 专用）：PAGES_BLOB_LOCAL_PERSIST=1 时 # 绕开 pages-blob SDK，用本地文件做持久化后端。 # 细节见 local_blob_store.py 顶部注释。 from .local_blob_store import LocalFileBlobStore, get_local_base_dir _local_base = get_local_base_dir() if _local_base is not None: print( f"[agent-python] Local persist mode: storing blobs under {_local_base}", file=sys.stderr, flush=True, ) # Try pages-blob _get_store = None try: from pages_blob import get_store as _get_store_fn _get_store = _get_store_fn print("[agent-python] pages-blob available, using BlobBackedStore", file=sys.stderr, flush=True) except Exception as _blob_import_err: print(f"[agent-python] pages-blob not available: {type(_blob_import_err).__name__}: {_blob_import_err}", file=sys.stderr, flush=True) print("[agent-python] using InMemoryStore", file=sys.stderr, flush=True) # Build proxy client for credential requests (only in prod when placeholders are replaced) _proxy_client = _build_proxy_client() if _proxy_client is not None: print("[agent-python] Proxy enabled for credential requests", file=sys.stderr, flush=True) else: print("[agent-python] Proxy not configured, using direct connections", file=sys.stderr, flush=True) def resolver(route_path: str) -> Any: cache_key = str(route_path or "/") if cache_key in cache: return cache[cache_key] store_name = _to_agent_store_name(cache_key) # Local persist 优先——和 blob 走一样的 BlobBackedStore 包装（envelope/TTL # 语义完全一致），只是底层换成文件。 if _local_base is not None: local_blob = LocalFileBlobStore(store_name, _local_base) store = BlobBackedStore(local_blob) cache[cache_key] = store return store if _get_store is not None: try: _project_id = os.environ.get("PAGES_PROJECT_ID", "") or None blob_store = _get_store(store_name, project_id=_project_id, token=_dev_credential or None, consistency="strong", http_client=_proxy_client) if (_dev_credential and _project_id) else _get_store(store_name, consistency="strong", http_client=_proxy_client) store = BlobBackedStore(blob_store) except Exception as e: print(f"[agent-python] WARNING: Failed to create blob store for {cache_key}: {e}, falling back to InMemoryStore", file=sys.stderr, flush=True) store = InMemoryStore() else: store = InMemoryStore() cache[cache_key] = store return store return resolver # --- Store blob (dedicated namespace for ctx.store) --- def _get_store_blob_name() -> str: """Compute the store blob namespace name. prod: memory-{projectId} dev: memory-{projectId}-dev Missing projectId: dev → memory-local-dev, prod → memory-agent (with warning). projectId resolution order (mirrors Node getMemoryProjectId): 0. BUILDER_PROJECT_ID from runtime_config (build-time injected) 1. PAGES_PROJECT_ID env var 2. EDGEONE_PROJECT_ID env var 3. ProjectId env var (injected by CLI dev server) """ # Build-time injected project ID (aligns with agent-node.ts BUILDER_PROJECT_ID) _builder_project_id = "" try: from _platform.runtime_config import BUILDER_PROJECT_ID as _bpid _builder_project_id = _bpid except (ImportError, AttributeError, Exception): pass project_id = ( _builder_project_id or os.environ.get("PAGES_PROJECT_ID", "").strip() or os.environ.get("EDGEONE_PROJECT_ID", "").strip() or os.environ.get("ProjectId", "").strip() ) # Determine mode from runtime_config or AGENT_MODE env try: from _platform.runtime_config import AGENT_MODE is_dev = AGENT_MODE == "dev" except (ImportError, Exception): env_val = os.environ.get("PAGES_ENV", "").strip().lower() is_dev = env_val in ("dev", "development", "local", "") if project_id: if is_dev: return f"memory-{project_id}-dev" return f"memory-{project_id}" else: if is_dev: return "memory-local-dev" print( "[agent-python] WARNING: PAGES_PROJECT_ID not set in production, " "using fallback store name 'memory-agent'", file=sys.stderr, flush=True, ) return "memory-agent" class _InMemoryBlobAdapter: """In-memory adapter matching raw blob store duck-type interface. Used as conversation store fallback when pages-blob is unavailable. Implements: get(key, type=), set(key, value), delete(key), list(prefix=). """ def __init__(self) -> None: self._data: dict[str, str] = {} async def get(self, key: str, *, type: str = "text", **_kwargs) -> Any: value = self._data.get(key) if value is None: return None if type == "json": return json.loads(value) return value async def set(self, key: str, value: Any, **_kwargs) -> None: if isinstance(value, str): self._data[key] = value elif isinstance(value, (bytes, bytearray)): self._data[key] = bytes(value).decode("utf-8") else: self._data[key] = json.dumps(value, ensure_ascii=False) async def delete(self, key: str) -> None: self._data.pop(key, None) async def list(self, *, prefix: str | None = None, **_kwargs) -> Any: from dataclasses import dataclass, field as _field @dataclass class _BlobInfo: key: str @dataclass class _ListResult: blobs: list = _field(default_factory=list) keys = sorted(k for k in self._data if prefix is None or k.startswith(prefix)) return _ListResult(blobs=[_BlobInfo(key=k) for k in keys]) def _create_store_blob() -> Any: """Create the dedicated blob store for ctx.store. Returns the raw blob store instance (not wrapped in BlobBackedStore — ConversationMemory manages its own data format). """ store_name = _get_store_blob_name() from .local_blob_store import LocalFileBlobStore, get_local_base_dir local_base = get_local_base_dir() if local_base is not None: print( f"[agent-python] Conversation store: local persist → {local_base / store_name}", file=sys.stderr, flush=True, ) return LocalFileBlobStore(store_name, local_base) try: from pages_blob import get_store as _get_store_fn proxy_client = _build_proxy_client() _mem_project_id = os.environ.get("PAGES_PROJECT_ID", "") or None _mem_token = os.environ.get("PAGES_BLOB_DEPLOY_CREDENTIAL", "") or None if _mem_token and _mem_project_id: blob_store = _get_store_fn(store_name, project_id=_mem_project_id, token=_mem_token, consistency="strong", http_client=proxy_client) else: blob_store = _get_store_fn(store_name, consistency="strong", http_client=proxy_client) print( f"[agent-python] Conversation store: blob-backed → {store_name}", file=sys.stderr, flush=True, ) return blob_store except ImportError: print( "[agent-python] Conversation store: in-memory (data will be lost on restart)\n" "[agent-python] → To persist locally: set PAGES_BLOB_LOCAL_PERSIST=1 in .env\n" "[agent-python] → To use cloud storage: run `edgeone pages link`", file=sys.stderr, flush=True, ) return _InMemoryBlobAdapter() except Exception as e: print( f"[agent-python] WARNING: Failed to create conversation store: {e}\n" "[agent-python] Conversation store: in-memory (data will be lost on restart)\n" "[agent-python] → To persist locally: set PAGES_BLOB_LOCAL_PERSIST=1 in .env\n" "[agent-python] → To use cloud storage: run `edgeone pages link`", file=sys.stderr, flush=True, ) return _InMemoryBlobAdapter() _store_resolver = _create_store_resolver() _store_blob = _create_store_blob() _registry: dict[str, RouteEntry] = {} # 加载失败的路由：route_path → 失败原因。404 响应时回显，避免静默 404 难以排查。 _failed_routes: dict[str, str] = {} for route_path, info in _route_table_raw.items(): module_path = info["module"] is_index = info.get("isIndex", False) try: mod = importlib.import_module(module_path) handler = getattr(mod, "handler", None) if handler is None: reason = f"module '{module_path}' has no 'handler' function" print(f"[agent-python] WARNING: {reason}, skipping", file=sys.stderr, flush=True) _failed_routes[route_path] = reason continue _registry[route_path] = RouteEntry(handler=handler, is_index=is_index, module_path=module_path) except Exception as e: reason = f"failed to import '{module_path}': {e}" print(f"[agent-python] ERROR loading {module_path}: {e}", file=sys.stderr, flush=True) _failed_routes[route_path] = reason _runtime = AgentRuntime( registry=_registry, store_resolver=_store_resolver, timeout=_timeout, store_blob=_store_blob, failed_routes=_failed_routes, ) print(f"[agent-python] Loaded {len(_registry)} routes, timeout={_timeout}s", file=sys.stderr, flush=True) if _failed_routes: print( f"[agent-python] WARNING: {len(_failed_routes)} route(s) failed to load: " f"{', '.join(_failed_routes.keys())}", file=sys.stderr, flush=True, ) # --- ASGI app --- async def app(scope: dict, receive, send) -> None: if scope["type"] == "lifespan": msg = await receive() if msg["type"] == "lifespan.startup": await send({"type": "lifespan.startup.complete"}) msg = await receive() if msg["type"] == "lifespan.shutdown": # Flush telemetry before process exits try: from _platform._observability.telemetry import get_telemetry _tel = get_telemetry() if _tel: _tel.shutdown() except (ImportError, Exception): pass await send({"type": "lifespan.shutdown.complete"}) return if scope["type"] != "http": return # Read body body_parts = [] while True: msg = await receive() body_parts.append(msg.get("body", b"")) if not msg.get("more_body", False): break raw_body = b"".join(body_parts) try: parsed = json.loads(raw_body) if raw_body else {} body = parsed if isinstance(parsed, dict) else {} except (json.JSONDecodeError, ValueError) as e: # JSON 解析失败直接返回 400，避免静默降级导致路由逻辑误判 # 注意：此时 headers dict 尚未构建，需要从 scope 提取原始 headers 来获取 request-id _early_headers = { k.decode("utf-8") if isinstance(k, bytes) else k: v.decode("utf-8") if isinstance(v, bytes) else v for k, v in scope.get("headers", []) } error_body = json.dumps({ "code": "INVALID_REQUEST_BODY", "message": f"Request body is not valid JSON: {e}", }).encode("utf-8") _err_resp_headers: list[tuple[bytes, bytes]] = [ (b"content-type", b"application/json"), (b"content-length", str(len(error_body)).encode()), ] _err_resp_headers.extend(_build_platform_headers(_early_headers, 400)) await send({ "type": "http.response.start", "status": 400, "headers": _err_resp_headers, }) await send({"type": "http.response.body", "body": error_body}) return path = scope.get("path", "/") method = scope.get("method", "POST") # Parse query string: ?foo=bar&x=1 → {"foo": "bar", "x": "1"} # 多值参数只取最后一个（和 Node req.query 行为一致）； # 需要多值的场景业务侧可直接 parse scope["query_string"]。 raw_qs = scope.get("query_string", b"") if isinstance(raw_qs, bytes): raw_qs = raw_qs.decode("utf-8", errors="replace") query = {k: v[-1] for k, v in parse_qs(raw_qs, keep_blank_values=True).items()} if raw_qs else {} headers = { k.decode("utf-8") if isinstance(k, bytes) else k: v.decode("utf-8") if isinstance(v, bytes) else v for k, v in scope.get("headers", []) } # Log Makers request path (matches python-function format) # 部署环境下通过 eo-pages-host header 拼接完整 URL # 对齐 agent-node: eo-pages-host 缺失时回退到 host header _pages_host = headers.get("eo-pages-host", "") or headers.get("host", "") if _pages_host: _pages_proto = headers.get("x-forwarded-proto", "https") _pages_full_path = f"{_pages_proto}://{_pages_host}{path}" print(f"Makers request path: {_pages_full_path}", file=sys.stderr, flush=True) # Dispatch to runtime. 同时起一个后台 task 监听 http.disconnect， # 客户端主动断开（curl Ctrl+C、浏览器关标签、网关超时踢掉）时： # 1. 从 header 解析 conversation_id； # 2. 调 runtime.request_cancel(conv_id) —— 业务 handler 下次检查 # is_cancelled 就能优雅返回，没检查的也会在下一个 await 点抛 # CancelledError，被 runtime 的 except CancelledError 捕获。 # 注意：不能简单 `dispatch_task.cancel()`，因为 runtime.handle 本身 # 也有 finally 清理（active_tasks pop 等），直接 cancel 整个 handle # 会让这些清理被 CancelledError 打断、留下幽灵 entry。 # # 流式响应特殊处理：dispatch_task 完成（generator 拿到了）后 watcher 还 # 必须继续工作，因为真正的耗时在后续 async for 迭代 chunk 阶段；client # 在 chunk 流期间断开时，watcher 要能继续触发 cancel。所以 watcher 的 # 关闭点延后到「流式 chunk 全部发完 / 或非流式响应已 send」之后。 _conv_id = headers.get("makers-conversation-id") or headers.get("conversation-id") or "" # --- Observability: per-request root span --- _obs_root_span = None _obs_token = None _obs_conv_token = None try: # 必须在创建 root span 之前写入 conversation_id：AgentContextPropagator # 在 span 的 on_start 那一刻读取 ContextVar 来注入 agent.conversation_id。 # 若放在 start_request_span 之后，root span 的 on_start 已经跑完，会读到 # None，导致 root span 漏掉 agent.conversation_id（只有子 span 有）。 try: from _platform._observability import set_agent_conversation_id as _obs_set_conv _obs_conv_token = _obs_set_conv(_conv_id or None) except (ImportError, Exception): pass from _platform._observability import start_request_span as _obs_start_span # agent.conversation_id 与 agent.run_id 由 AgentContextPropagator 在 onStart # 时统一注入到所有 span（含 root），这里不再手动放进 attributes 以避免重复。 _obs_root_span, _obs_ctx = _obs_start_span( f"agent.request:{path}", {"http.method": method, "http.route": path}, ) from opentelemetry.context import attach as _obs_attach _obs_token = _obs_attach(_obs_ctx) # Make context available to CrewAI's background thread try: from _platform._observability import set_request_context set_request_context(_obs_ctx) except (ImportError, Exception): pass except (ImportError, Exception): pass dispatch_task = asyncio.create_task( _runtime.handle(path, method, headers, query, body, eo=_parse_eo(headers)) ) # 用 Event 让 watcher 区分「流被业务自己 cancel 了」和「client 真的断开」： # 流式 generator 内部 await asyncio.sleep 时如果业务调了 task.cancel， # 会抛 CancelledError；但那不是 client 断开，不应该再 print "client # disconnected" 误导排查。 _disconnect_seen = asyncio.Event() async def _watch_disconnect() -> None: try: while True: msg = await receive() if msg.get("type") == "http.disconnect": _disconnect_seen.set() if _conv_id and _runtime.request_cancel(_conv_id): print( f"[agent-python] Client disconnected, cancelling run conv={_conv_id}", file=sys.stderr, flush=True, ) return except asyncio.CancelledError: # 正常完成时 watcher 会被 adapter finally cancel 掉，吞掉即可。 return _watcher = asyncio.create_task(_watch_disconnect()) result = None try: result = await dispatch_task except BaseException: # dispatch 本身挂了的话先把 watcher 关了再 raise，避免 watcher 卡 receive。 _watcher.cancel() raise # 注意：这里不再立即 cancel watcher —— 流式分支需要它在 chunk 迭代时继续 # 监听 disconnect。非流式分支会在下面 send 完后统一 cancel。 # Log Makers response status (matches python-function format) print(f"Makers response status: {result.status}", file=sys.stderr, flush=True) # Send response response_headers = [(k.encode(), str(v).encode()) for k, v in result.headers.items()] # Inject platform headers (Functions-Request-Id, eo-pages-inner-scf-status, eo-pages-inner-status-intercept) response_headers.extend(_build_platform_headers(headers, result.status)) # ───────── 流式分支 ───────── # runtime 已决议好 status / headers / content-type，body 是 StreamResponse； # 这里只负责把 chunk 一帧帧 send 到 ASGI 通道。 # # 关键点： # 1. http.response.start 只能 send 一次（ASGI 协议要求），所以一旦 send # 过去就不能再改 status —— generator 中途异常只能 close 连接。 # 2. 每个 chunk send 时 more_body=True，最后再 send 一个空 body # more_body=False 标识结束；不发结尾帧的话 uvicorn 会一直挂连接。 # 3. async for 期间要监听 watcher 触发的 cancel —— is_cancelled / 任务被 # cancel 时 generator 自然抛 CancelledError 我们走 except 关流就行； # 不要尝试在 stream 里再 send 错误响应（headers 已发出去了）。 # 4. 最终（无论成功失败）都要调 result._cleanup() 清掉 active_runs， # 否则同 conversation_id 会一直被 409 拒绝。 if result.is_streaming: stream: StreamResponse = result.body # type: ignore[assignment] await send({ "type": "http.response.start", "status": result.status, "headers": response_headers, }) # 流式响应的取消机制有别于非流式： # - 非流式时，handler 本身的 task 在 await 点能被 task.cancel() 打断； # - 流式时，handler 调用立刻返回 generator 对象（async generator # 函数体并不会进入），handler 的 task 已经 done，再 cancel 它没用。 # 真正在跑的是 adapter 这里的 `async for chunk in stream.body`。 # 所以要在 async for 阶段把「下一帧」和「中断事件」做成 race， # 谁先到处理谁。中断事件包括： # 1. client disconnect（_disconnect_seen） # 2. /stop 或 abort_active_run 设置的 cancel signal chunks_sent = 0 agen = stream.body.__aiter__() # 获取 runtime 附着的 cancel signal（/stop、abort_active_run 会 set 它） _cancel_signal: asyncio.Event | None = getattr(result, "_cancel_signal", None) try: while True: next_chunk_task = asyncio.create_task(agen.__anext__()) disconnect_task = asyncio.create_task(_disconnect_seen.wait()) race_set: set[asyncio.Task] = {next_chunk_task, disconnect_task} # 如果有 cancel signal，加入 race；这样 /stop 和 # abort_active_run 也能立即中断流式迭代。 cancel_signal_task: asyncio.Task | None = None if _cancel_signal is not None and not _cancel_signal.is_set(): cancel_signal_task = asyncio.create_task(_cancel_signal.wait()) race_set.add(cancel_signal_task) done, pending = await asyncio.wait( race_set, return_when=asyncio.FIRST_COMPLETED, ) # 取消未完成的那个 —— 注意 next_chunk_task 被 cancel 时， # generator 内部 await 点会抛 CancelledError，业务侧的 finally # （文件关闭、连接释放等）能正常跑。 for p in pending: p.cancel() # 等 cancel 真正生效，避免 task 泄漏告警 try: await p except (asyncio.CancelledError, StopAsyncIteration, Exception): pass # 检测中断：client disconnect 或 cancel signal _interrupted = False _interrupt_reason = "" if disconnect_task in done: _interrupted = True _interrupt_reason = "client disconnected" elif cancel_signal_task is not None and cancel_signal_task in done: _interrupted = True _interrupt_reason = "cancel signal received (/stop or abort_active_run)" if _interrupted: print( f"[agent-python] Stream interrupted ({_interrupt_reason}) after {chunks_sent} chunks for {path}", file=sys.stderr, flush=True, ) # 主动 aclose generator，让业务侧 try/finally 能跑收尾。 # 这会传播 GeneratorExit 到上游（如 LLM streaming），释放连接。 try: await agen.aclose() except Exception: pass break # next_chunk_task 完成：可能是正常 chunk、StopAsyncIteration、或异常。 try: chunk = next_chunk_task.result() except StopAsyncIteration: break except asyncio.CancelledError: # 被我们自己 cancel 的（理论上不会走到 —— pending 集合排除了 done）； # 兜底处理。 break if chunk is None: continue payload = _encode_stream_chunk(chunk) if not payload: continue await send({ "type": "http.response.body", "body": payload, "more_body": True, }) chunks_sent += 1 except asyncio.CancelledError: # ASGI server 主动 cancel 整个 app coroutine（极少见，比如服务关停）。 # 已发的 chunk 不会回滚，直接关连接即可。注意要 re-raise 让上层 # task 状态收敛。 print( f"[agent-python] Streaming cancelled (server shutdown) after {chunks_sent} chunks for {path}", file=sys.stderr, flush=True, ) try: await send({"type": "http.response.body", "body": b"", "more_body": False}) except Exception: pass raise except Exception as e: if _is_interrupt_exception(e): # LangGraph interrupt()：graph 暂停等待人工输入，属正常控制流， # 不是 streaming error。静默收尾：不打 error 日志、不把 root span # 标 ERROR，正常发结尾空帧关闭流。前端通过 __interrupt__ 事件 # 单独获取暂停草稿。 print( f"[agent-python] Stream paused (interrupt) after {chunks_sent} chunks for {path}", file=sys.stderr, flush=True, ) if _obs_root_span: try: _obs_root_span.set_attribute("agent.interrupt", True) except Exception: pass try: await send({"type": "http.response.body", "body": b"", "more_body": False}) except Exception: pass else: # generator 内部业务异常：headers 已发，没法再改 status，只能记日志 # 并以空 body 收尾。客户端看到的会是「收到部分数据后流被截断」—— # 业务侧应该自行在最后一个 chunk 包含「success/error」标识。 print( f"[agent-python] Streaming error after {chunks_sent} chunks for {path}: {e}", file=sys.stderr, flush=True, ) if _obs_root_span: try: from opentelemetry.trace.status import Status as _OStatus, StatusCode as _OSC _obs_root_span.set_status(_OStatus(_OSC.ERROR, str(e))) except (ImportError, Exception): pass try: await send({"type": "http.response.body", "body": b"", "more_body": False}) except Exception: pass else: # 正常结束（包括 disconnect 后的优雅 break）：发结尾帧 more_body=False。 try: await send({"type": "http.response.body", "body": b"", "more_body": False}) except Exception: # client 已断的话 send 会 raise，吞掉就行。 pass print( f"[agent-python] Streaming complete: {path} chunks={chunks_sent}", file=sys.stderr, flush=True, ) finally: _watcher.cancel() cleanup = getattr(result, "_cleanup", None) if callable(cleanup): try: cleanup() except Exception as cleanup_err: print( f"[agent-python] Cleanup error for {path}: {cleanup_err}", file=sys.stderr, flush=True, ) # --- Observability: end request root span (streaming) --- if _obs_root_span: try: from _platform._observability import end_request_span as _obs_end_span _obs_end_span(_obs_root_span) except (ImportError, Exception): try: _obs_root_span.end() except Exception: pass if _obs_token: try: from opentelemetry.context import detach as _obs_detach _obs_detach(_obs_token) except (ImportError, Exception): pass # Reset agent.conversation_id ContextVar (used by AgentContextPropagator) if _obs_conv_token is not None: try: from _platform._observability import reset_agent_conversation_id as _obs_reset_conv _obs_reset_conv(_obs_conv_token) except (ImportError, Exception): pass # Clear CrewAI request context try: from _platform._observability import clear_request_context clear_request_context() except (ImportError, Exception): pass return # ───────── 非流式分支（旧行为，原样保留） ───────── _watcher.cancel() if result.body is None: response_headers.append((b"content-length", b"0")) await send({"type": "http.response.start", "status": result.status, "headers": response_headers}) await send({"type": "http.response.body", "body": b""}) elif isinstance(result.body, str): encoded = result.body.encode("utf-8") if not any(k == b"content-type" for k, _ in response_headers): response_headers.append((b"content-type", b"text/plain; charset=utf-8")) response_headers.append((b"content-length", str(len(encoded)).encode())) await send({"type": "http.response.start", "status": result.status, "headers": response_headers}) await send({"type": "http.response.body", "body": encoded}) else: encoded = json.dumps(result.body, ensure_ascii=False).encode("utf-8") response_headers.append((b"content-type", b"application/json; charset=utf-8")) response_headers.append((b"content-length", str(len(encoded)).encode())) await send({"type": "http.response.start", "status": result.status, "headers": response_headers}) await send({"type": "http.response.body", "body": encoded}) # --- Observability: end request root span --- if _obs_root_span: try: from _platform._observability import end_request_span as _obs_end_span if result and result.status >= 400: from opentelemetry.trace.status import Status as _OStatus, StatusCode as _OSC _obs_root_span.set_status(_OStatus(_OSC.ERROR)) _obs_end_span(_obs_root_span) except (ImportError, Exception): try: _obs_root_span.end() except Exception: pass if _obs_token: try: from opentelemetry.context import detach as _obs_detach _obs_detach(_obs_token) except (ImportError, Exception): pass # Reset agent.conversation_id ContextVar (used by AgentContextPropagator) if _obs_conv_token is not None: try: from _platform._observability import reset_agent_conversation_id as _obs_reset_conv _obs_reset_conv(_obs_conv_token) except (ImportError, Exception): pass # Clear CrewAI request context try: from _platform._observability import clear_request_context clear_request_context() except (ImportError, Exception): pass def _encode_stream_chunk(chunk: Any) -> bytes: """把 generator 产出的 chunk 归一成 bytes，方便 ASGI send。 支持： - bytes / bytearray / memoryview → 原样（zero-copy 转 bytes） - str → utf-8 编码 - dict / list / 其他 JSON 兼容 → json.dumps + '\\n'（NDJSON 风格） 返回 b"" 表示「空 chunk，跳过 send」（adapter 上层会过滤）。 设计取舍：dict 自动 NDJSON 化是给「不显式包 SSE 的最简流」一个合理默认， 比如 handler 写 `yield {"token": "x"}` 就能产出 `{"token":"x"}\\n`。如果 业务要严格 SSE，应该用 sse(...) 显式构造。 """ if chunk is None: return b"" if isinstance(chunk, (bytes, bytearray, memoryview)): return bytes(chunk) if isinstance(chunk, str): return chunk.encode("utf-8") # dict / list / 其他可 JSON 序列化的对象 → NDJSON 帧（一行一个 JSON）。 # 加 \n 是 NDJSON 规范要求；客户端用 readline 切帧会很方便。 try: return (json.dumps(chunk, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8") except (TypeError, ValueError): # 兜底：repr 转字符串，加换行。这种情况通常是用户传了一个不可序列化 # 的对象（如 Future、Lock），日志里能看到原始 chunk 的 repr 帮 debug。 return (repr(chunk) + "\n").encode("utf-8")