stroblme
71628abcbc
The worker script is handed to the interpreter by path, so app/flow was sys.path[0] for every node: `import queue` got the engine's. It now drops its own directory before anything else imports, and runs with the deployment's credentials scrubbed out of its environment. Also: reload builds off the event loop, the pool wakes what is blocked on it when it stops, a refused metrics flush is kept for the next one rather than dropped, and the cascade events are paired through failures.
443 lines
16 KiB
Python
443 lines
16 KiB
Python
"""HTTP nodes: an inbound webhook trigger and an outbound request sender."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
import logging
|
|
import threading
|
|
from collections.abc import Iterable
|
|
from enum import Enum
|
|
from typing import TYPE_CHECKING, Any, Literal
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
import httpx
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from app.flow.messages import MessageSpec
|
|
from app.flow.nodes.base import Node
|
|
|
|
if TYPE_CHECKING:
|
|
from fastapi import FastAPI
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# One pooled client for every sender node: connections are the expensive part
|
|
# of an HTTP request, and a node that fires every second should keep its own.
|
|
_client: httpx.Client | None = None
|
|
_client_lock = threading.Lock()
|
|
|
|
|
|
def shared_client() -> httpx.Client:
|
|
"""The process-wide HTTP client, built on first use."""
|
|
global _client
|
|
with _client_lock:
|
|
if _client is None:
|
|
_client = httpx.Client()
|
|
return _client
|
|
|
|
|
|
def close_shared_client() -> None:
|
|
"""Release pooled connections at shutdown. Safe to call more than once."""
|
|
global _client
|
|
with _client_lock:
|
|
if _client is not None:
|
|
_client.close()
|
|
_client = None
|
|
|
|
|
|
def _first_catch_all(app: FastAPI) -> int:
|
|
"""Where a webhook has to go in to be reachable.
|
|
|
|
A mount at the root answers for every path, so anything registered after
|
|
it is dead. Returns that mount's index, or the end of the table.
|
|
"""
|
|
from starlette.routing import Mount
|
|
|
|
for index, route in enumerate(app.routes):
|
|
if isinstance(route, Mount) and route.path in ("", "/"):
|
|
return index
|
|
return len(app.routes)
|
|
|
|
|
|
class HttpNode(Node):
|
|
"""
|
|
HTTP node that can act as a trigger (receiver) or sender based on configuration.
|
|
|
|
This node integrates with FastAPI to either:
|
|
|
|
- **Trigger mode**: Receive incoming HTTP requests (GET/POST) and inject data
|
|
into the pipeline. Used when ``provides`` is specified but ``requires`` is empty.
|
|
- **Sender mode**: Make outgoing HTTP requests with pipeline data. Used when
|
|
``requires`` is specified.
|
|
|
|
:param url: The URL endpoint. For trigger mode, this is the route path
|
|
(e.g., "/sensors/temperature"). For sender mode, this is the full URL
|
|
to send requests to.
|
|
:type url: str
|
|
:param method: HTTP method - "GET" or "POST".
|
|
:type method: Literal["GET", "POST"]
|
|
:param requires: Messages required by this node (makes it a sender node).
|
|
:type requires: MessageSpec | list[MessageSpec] | None
|
|
:param provides: Messages provided by this node (makes it a trigger node).
|
|
:type provides: MessageSpec | list[MessageSpec] | None
|
|
:param params: Additional parameters for the node.
|
|
:type params: dict
|
|
:param name: Optional name for the node.
|
|
:type name: str | None
|
|
:param timeout: Request timeout in seconds (for sender mode).
|
|
:type timeout: float
|
|
:param headers: Additional HTTP headers.
|
|
:type headers: dict[str, str] | None
|
|
:param secret: Shared secret callers append to the webhook URL (trigger mode).
|
|
:type secret: str
|
|
|
|
:raises ValueError: If both ``requires`` and ``provides`` are empty, or if
|
|
the configuration is invalid.
|
|
|
|
:example:
|
|
Trigger node (receives POST requests):
|
|
|
|
>>> trigger = HttpNode(
|
|
... url="/api/sensors/temperature",
|
|
... method="POST",
|
|
... provides=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={},
|
|
... )
|
|
|
|
Sender node (makes POST requests):
|
|
|
|
>>> sender = HttpNode(
|
|
... url="https://api.example.com/data",
|
|
... method="POST",
|
|
... requires=[MessageSpec(name="temperature", dtype=DType.FLOAT)],
|
|
... params={},
|
|
... )
|
|
"""
|
|
|
|
class Mode(Enum):
|
|
"""Operating mode of the HTTP node."""
|
|
|
|
TRIGGER = "trigger" # Receives HTTP requests
|
|
SENDER = "sender" # Sends HTTP requests
|
|
|
|
# A repeated request is a repeated request, whatever the endpoint does
|
|
# with it.
|
|
idempotent = False
|
|
|
|
__slots__ = (
|
|
"url",
|
|
"method",
|
|
"mode",
|
|
"timeout",
|
|
"headers",
|
|
"secret",
|
|
"_route_registered",
|
|
)
|
|
|
|
class Params(BaseModel):
|
|
model_config = ConfigDict(extra="allow")
|
|
|
|
url: str
|
|
method: Literal["GET", "POST"] = "POST"
|
|
timeout: float = 30.0
|
|
headers: dict[str, str] = {}
|
|
secret: str = Field(
|
|
default="",
|
|
description=(
|
|
"Shared secret for a webhook, appended to its URL: "
|
|
"/hooks/<flow>/<url>/<secret>. Empty leaves the webhook open "
|
|
"to anyone."
|
|
),
|
|
json_schema_extra={"x-secret": True},
|
|
)
|
|
|
|
@classmethod
|
|
def instance_key(cls, params: dict[str, Any]) -> str | None:
|
|
# ponytail: a webhook's stored url is the path before the flow name is
|
|
# prefixed onto it, so two flows both receiving on "/tick" merge into
|
|
# one neuron. Key on mode as well if that ever misleads.
|
|
url = params.get("url")
|
|
if not url:
|
|
return None
|
|
# A URL may legally carry credentials, and this key becomes a group id
|
|
# in the brain graph — rendered into the response and into the DOM.
|
|
parts = urlsplit(str(url))
|
|
if parts.username or parts.password:
|
|
host = parts.netloc.rsplit("@", 1)[-1]
|
|
return urlunsplit(parts._replace(netloc=host))
|
|
return str(url)
|
|
|
|
def __init__(
|
|
self,
|
|
url: str | None = None,
|
|
method: Literal["GET", "POST"] | None = None,
|
|
requires: MessageSpec | Iterable[MessageSpec] = (),
|
|
provides: MessageSpec | Iterable[MessageSpec] = (),
|
|
params: dict[str, Any] | None = None,
|
|
name: str | None = None,
|
|
):
|
|
params = dict(params) if params else {}
|
|
if url is not None:
|
|
params["url"] = url
|
|
if method is not None:
|
|
params["method"] = method
|
|
cfg = self.Params.model_validate(params)
|
|
|
|
requires = Node._normalize_ports(requires)
|
|
provides = Node._normalize_ports(provides)
|
|
if not requires and not provides:
|
|
raise ValueError(
|
|
"An HTTP node needs either inputs (to send) or outputs (to receive)"
|
|
)
|
|
|
|
# Inputs mean this node sends data out; outputs mean it receives.
|
|
self.mode = HttpNode.Mode.SENDER if requires else HttpNode.Mode.TRIGGER
|
|
|
|
self.url = cfg.url
|
|
self.method = cfg.method.upper()
|
|
self.timeout = cfg.timeout
|
|
self.headers = cfg.headers
|
|
self.secret = cfg.secret
|
|
self._route_registered = False
|
|
|
|
# Set default name based on mode and URL
|
|
if name is None:
|
|
safe_url = self.url.replace("/", "_").replace(":", "").strip("_")
|
|
name = f"http_{self.mode.value}_{safe_url}"
|
|
|
|
# Initialize parent with appropriate function
|
|
# For trigger mode, f is a no-op since data is injected via inject()
|
|
# For sender mode, f handles the outgoing HTTP request
|
|
super().__init__(
|
|
f=(
|
|
self._noop_trigger
|
|
if self.mode == HttpNode.Mode.TRIGGER
|
|
else self._sender_handler
|
|
),
|
|
requires=requires,
|
|
provides=provides,
|
|
params=params,
|
|
name=name,
|
|
)
|
|
|
|
@staticmethod
|
|
def _noop_trigger(params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
|
|
"""
|
|
No-op function for trigger mode nodes.
|
|
|
|
Trigger mode nodes inject data via :meth:`inject`, not :meth:`__call__`.
|
|
This function exists only to satisfy the Node interface.
|
|
"""
|
|
return None
|
|
|
|
def _sender_handler(
|
|
self, params: dict[str, Any], **kwargs: Any
|
|
) -> dict[str, Any] | None:
|
|
"""
|
|
Send HTTP request with pipeline data (sender mode).
|
|
|
|
This method is called when upstream dependencies are satisfied.
|
|
It sends the required data via HTTP request.
|
|
|
|
Node functions already run on a worker thread, so the request is sent
|
|
synchronously over the shared pooled client: a node publishing every
|
|
second should reuse its connection, not open one per message.
|
|
|
|
:param params: Node parameters.
|
|
:type params: dict
|
|
:param kwargs: Pipeline data to send (from required messages).
|
|
:type kwargs: Any
|
|
:returns: Response data if the endpoint returns JSON, None otherwise.
|
|
:rtype: dict | None
|
|
"""
|
|
client = shared_client()
|
|
try:
|
|
if self.method == "GET":
|
|
response = client.get(
|
|
self.url,
|
|
params=kwargs,
|
|
headers=self.headers,
|
|
timeout=self.timeout,
|
|
)
|
|
else: # POST
|
|
response = client.post(
|
|
self.url,
|
|
json=kwargs,
|
|
headers=self.headers,
|
|
timeout=self.timeout,
|
|
)
|
|
|
|
response.raise_for_status()
|
|
|
|
try:
|
|
body = response.json()
|
|
except Exception:
|
|
return None
|
|
# Outputs are keyed by port, so only an object can be one. A bare
|
|
# scalar or list is a valid reply, just not something to publish.
|
|
return body if isinstance(body, dict) else None
|
|
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(
|
|
"HTTP error in node '%s': status=%s, url=%s",
|
|
self.name,
|
|
e.response.status_code,
|
|
self.url,
|
|
)
|
|
raise
|
|
except httpx.RequestError as e:
|
|
logger.error("Request error in node '%s': %s", self.name, e)
|
|
raise
|
|
|
|
async def start(self, app: FastAPI | None = None) -> None:
|
|
"""A webhook needs a route; a sender reaches out on its own."""
|
|
if self.mode is HttpNode.Mode.TRIGGER and app is not None:
|
|
self.register_route(app)
|
|
|
|
async def stop(self, app: FastAPI | None = None) -> None:
|
|
if self.mode is HttpNode.Mode.TRIGGER and app is not None:
|
|
self.unregister_route(app)
|
|
|
|
def register_route(self, app: FastAPI) -> None:
|
|
"""
|
|
Register this node's HTTP endpoint with a FastAPI application.
|
|
|
|
This method should only be called for trigger mode nodes. It creates
|
|
a route that, when called, triggers the node in the pipeline.
|
|
|
|
:param app: The FastAPI application instance.
|
|
:type app: FastAPI
|
|
:raises RuntimeError: If called on a sender mode node.
|
|
|
|
:example:
|
|
>>> from fastapi import FastAPI
|
|
>>> app = FastAPI()
|
|
>>> trigger_node = HttpNode(
|
|
... url="/sensors/data",
|
|
... method="POST",
|
|
... provides=[MessageSpec(name="value", dtype=DType.FLOAT)],
|
|
... params={},
|
|
... )
|
|
>>> trigger_node.register_route(app)
|
|
"""
|
|
if self.mode != HttpNode.Mode.TRIGGER:
|
|
raise RuntimeError("Can only register routes for trigger mode nodes")
|
|
|
|
if self._route_registered:
|
|
return
|
|
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
from starlette.routing import Route
|
|
|
|
async def handle_request(request: Request) -> JSONResponse:
|
|
"""
|
|
Handle incoming HTTP request and trigger the pipeline.
|
|
|
|
:param request: The incoming Starlette request.
|
|
:type request: Request
|
|
:returns: JSON response with trigger result.
|
|
:rtype: JSONResponse
|
|
"""
|
|
if self.secret and not hmac.compare_digest(
|
|
str(request.path_params.get("secret", "")).encode(),
|
|
self.secret.encode(),
|
|
):
|
|
# A wrong secret must look like no such hook at all.
|
|
return JSONResponse(content={"detail": "Not Found"}, status_code=404)
|
|
|
|
try:
|
|
# Parse request data
|
|
data: dict[str, Any]
|
|
if self.method == "GET":
|
|
data = dict(request.query_params)
|
|
else: # POST
|
|
content_type = request.headers.get("content-type", "")
|
|
if "application/json" in content_type:
|
|
data = await request.json()
|
|
elif "application/x-www-form-urlencoded" in content_type:
|
|
form = await request.form()
|
|
data = dict(form)
|
|
else:
|
|
data = await request.json() # Default to JSON
|
|
|
|
# Payload keys are port names; values arrive as text.
|
|
typed_data = {}
|
|
for spec in self.output_ports:
|
|
if spec.port in data:
|
|
typed_data[spec.port] = spec.coerce(data[spec.port])
|
|
|
|
# Inject data into the pipeline (trigger mode nodes inject via provides)
|
|
result = self.inject(typed_data)
|
|
|
|
return JSONResponse(
|
|
content={
|
|
"status": "triggered",
|
|
"node": self.name,
|
|
"data": typed_data,
|
|
"result": result if isinstance(result, dict) else None,
|
|
}
|
|
)
|
|
|
|
except ValueError as e:
|
|
return JSONResponse(
|
|
content={"error": str(e)},
|
|
status_code=400,
|
|
)
|
|
except Exception as e:
|
|
return JSONResponse(
|
|
content={"error": str(e)},
|
|
status_code=500,
|
|
)
|
|
|
|
# The secret is a path parameter, not part of the registered path, so
|
|
# neither the route table nor the logs below carry its value.
|
|
path = f"{self.url}/{{secret}}" if self.secret else self.url
|
|
|
|
# Create a Starlette Route and add it directly to the app's routes
|
|
route = Route(
|
|
path,
|
|
handle_request,
|
|
methods=[self.method],
|
|
name=self.id,
|
|
)
|
|
# Ahead of any catch-all mount. Starlette takes the first route that
|
|
# matches, and the MCP app is mounted at "/", so appending would put
|
|
# every webhook behind something that answers for every path.
|
|
app.routes.insert(_first_catch_all(app), route)
|
|
|
|
self._route_registered = True
|
|
logger.info(
|
|
"Registered %s route for node '%s': %s",
|
|
self.method,
|
|
self.name,
|
|
path,
|
|
)
|
|
|
|
def unregister_route(self, app: FastAPI) -> None:
|
|
"""
|
|
Unregister this node's HTTP endpoint from a FastAPI application.
|
|
|
|
:param app: The FastAPI application instance.
|
|
:type app: FastAPI
|
|
|
|
.. note::
|
|
FastAPI doesn't natively support route removal. This method
|
|
removes the route from the internal routes list, but the change
|
|
may not take effect until the application is restarted or
|
|
the OpenAPI schema is regenerated.
|
|
"""
|
|
if not self._route_registered:
|
|
return
|
|
|
|
# FastAPI doesn't have a clean way to remove routes
|
|
# We need to filter them out from the routes list
|
|
app.routes[:] = [
|
|
route
|
|
for route in app.routes
|
|
if not (hasattr(route, "name") and route.name == self.id)
|
|
]
|
|
|
|
self._route_registered = False
|
|
logger.info("Unregistered route for node '%s': %s", self.name, self.url)
|