# SPDX-FileCopyrightText: Copyright (c) 2026 Abhinav Kalvacherla # SPDX-License-Identifier: Apache-2.0 """ Generic HTTP executor for client-supplied function/API definitions. A function definition (supplied per-session, see session_config.py) is a dict: { "name": "get_weather", # required — tool name "description": "Current weather for a lat/lon.", "endpoint": "https://api.open-meteo.com/v1/forecast", # required — may contain {placeholders} "method": "GET", # GET|POST|PUT|PATCH|DELETE (default GET) "param_location": "query", # query|body|path (default: query for GET, body otherwise) "param_locations": {"id": "path"}, # optional per-parameter override "headers": {"X-Api-Key": "..."}, # optional static request headers "static_params": {"current_weather": true}, # optional server-merged constants "parameters": { ...JSON schema... } # OpenAI tool parameters schema } `build_openai_tools` turns the {name, description, parameters} parts into the OpenAI `tools` array given to the LLM. `execute_function` takes the LLM-extracted args, builds the HTTP request per the spec, and returns the response text. Security: the server makes requests to client-supplied URLs (an SSRF surface). `_enforce_allowlist` restricts hosts and blocks private/loopback/link-local IPs. """ from __future__ import annotations import ipaddress import logging import os import socket from urllib.parse import urlparse import httpx logger = logging.getLogger(__name__) MAX_RESPONSE_CHARS = 1500 HTTP_TIMEOUT = 4.0 # below rag_timeout; the retriever's wait_for is the hard ceiling def build_openai_tools(functions: list[dict]) -> list[dict]: """Map declarative function defs → OpenAI `tools` array for the LLM.""" tools = [] for f in functions: tools.append({ "type": "function", "function": { "name": f["name"], "description": f.get("description", ""), "parameters": f.get("parameters") or {"type": "object", "properties": {}}, }, }) return tools def _default_location(spec: dict) -> str: return "query" if str(spec.get("method", "GET")).upper() == "GET" else "body" def _split_args(spec: dict, args: dict) -> tuple[str, dict, dict | None]: """Route merged args into URL path / query / body per the spec.""" loc = spec.get("param_location") or _default_location(spec) per = spec.get("param_locations") or {} merged = {**(spec.get("static_params") or {}), **(args or {})} url = spec["endpoint"] query: dict = {} body: dict = {} for k, v in merged.items(): placeholder = "{%s}" % k where = per.get(k, "path" if placeholder in url else loc) if where == "path": url = url.replace(placeholder, str(v)) elif where == "body": body[k] = v else: query[k] = v return url, query, (body or None) def _enforce_allowlist(url: str, cfg) -> None: """Reject disallowed hosts and any host resolving to a non-public IP (SSRF guard).""" parsed = urlparse(url) if parsed.scheme not in ("http", "https"): raise PermissionError(f"scheme {parsed.scheme!r} not allowed") host = parsed.hostname or "" server_allow = {h for h in (os.environ.get("MOSHI_TOOL_ALLOWED_HOSTS") or "").split(",") if h} sess_allow = set(getattr(cfg, "allowed_hosts", None) or []) if server_allow and sess_allow: allow = server_allow & sess_allow else: allow = server_allow or sess_allow if allow and host not in allow: raise PermissionError(f"host {host!r} not in allowlist") # Block private / loopback / link-local / reserved targets regardless of allowlist. try: infos = socket.getaddrinfo(host, None) except socket.gaierror as exc: raise PermissionError(f"cannot resolve host {host!r}: {exc}") from exc for info in infos: ip = ipaddress.ip_address(info[4][0]) if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast: raise PermissionError(f"host {host!r} resolves to non-public IP {ip}") async def execute_function(spec: dict, args: dict, cfg) -> str: """Build and make the HTTP call described by `spec` with `args`; return response text.""" _enforce_allowlist(spec["endpoint"], cfg) method = str(spec.get("method", "GET")).upper() url, query, body = _split_args(spec, args) headers = spec.get("headers") or {} logger.info("[http] %s %s params=%s body=%s", method, url, query, body) async with httpx.AsyncClient(timeout=HTTP_TIMEOUT, follow_redirects=False) as client: resp = await client.request(method, url, params=query or None, json=body, headers=headers) resp.raise_for_status() text = resp.text return text[:MAX_RESPONSE_CHARS]