File size: 5,013 Bytes
ee34fec 11654f3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | # 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]
|