import os import re import json import httpx from urllib.parse import urlparse from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.exceptions import RequestValidationError from pydantic import BaseModel from typing import List app = FastAPI() WEB_TOOLS = [ { "type": "function", "function": { "name": "web_fetch", "description": "Fetch a URL and return its content as text. Use this to browse the web, read articles, or view pages. If interact=True, also returns clickable links found on the page.", "parameters": { "type": "object", "properties": { "url": {"type": "string", "description": "The URL to fetch"}, "interact": {"type": "boolean", "default": False, "description": "Set to true to also extract clickable links/buttons"}, }, "required": ["url"], }, } }, { "type": "function", "function": { "name": "web_click", "description": "Click on a link or button by navigating to its URL. Use this to press buttons, follow links, or navigate webpages interactively.", "parameters": { "type": "object", "properties": { "url": {"type": "string", "description": "The URL/link to navigate to"}, }, "required": ["url"], }, } }, { "type": "function", "function": { "name": "web_search", "description": "Search the web using DuckDuckGo. Use this to find information, look up current events, or research topics.", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "The search query"}, }, "required": ["query"], }, } }, { "type": "function", "function": { "name": "web_type", "description": "Type into text fields and submit a form on a webpage. Use this to fill in textboxes, search bars, login forms, etc.", "parameters": { "type": "object", "properties": { "url": {"type": "string", "description": "The form action URL or page URL to submit to"}, "fields": {"type": "object", "description": "Dictionary of field names and values to fill in (e.g. {\"q\": \"hello\", \"username\": \"test\"})"}, "method": {"type": "string", "enum": ["POST", "GET"], "default": "POST", "description": "HTTP method"}, }, "required": ["url", "fields"], }, } }, ] def extract_links(html: str, base_url: str) -> list: links = [] for tag, attr in [("a", "href"), ("form", "action"), ("button", "data-href")]: pattern = f'<{tag}[^>]*{attr}=["\'](.*?)["\']' for m in re.finditer(pattern, html, re.IGNORECASE): href = m.group(1) if href and not href.startswith("#") and not href.startswith("javascript:"): if href.startswith("/"): parsed = urlparse(base_url) href = f"{parsed.scheme}://{parsed.netloc}{href}" links.append({"tag": tag, "url": href}) return links[:20] async def web_fetch(url: str, interact: bool = False) -> str: try: async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client: resp = await client.get(url, headers={"User-Agent": "Mozilla/5.0"}) html = resp.text text = html[:8000] result = f"URL: {url}\nStatus: {resp.status_code}\n\nContent:\n{text}" if interact: links = extract_links(html, url) if links: result += f"\n\nFound {len(links)} links/buttons:\n" + "\n".join(f" [{i}] <{l['tag']}> {l['url']}" for i, l in enumerate(links)) return result except Exception as e: return f"Error fetching {url}: {e}" async def web_click(url: str) -> str: return await web_fetch(url, interact=True) async def web_type(url: str, fields: dict, method: str = "POST") -> str: try: async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client: if method.upper() == "GET": resp = await client.get(url, params=fields, headers={"User-Agent": "Mozilla/5.0"}) else: resp = await client.post(url, data=fields, headers={"User-Agent": "Mozilla/5.0"}) html = resp.text[:6000] links = extract_links(html, str(resp.url)) result = f"URL: {resp.url}\nStatus: {resp.status_code}\n\nResponse:\n{html}" if links: result += "\n\nLinks found:\n" + "\n".join(f" [{i}] <{l['tag']}> {l['url']}" for i, l in enumerate(links[:10])) return result except Exception as e: return f"Error submitting form to {url}: {e}" async def web_search(query: str) -> str: try: url = f"https://html.duckduckgo.com/html/?q={httpx.utils.quote(query)}" async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client: resp = await client.get(url, headers={"User-Agent": "Mozilla/5.0"}) html = resp.text results = [] for m in re.finditer(r']*class="result__a"[^>]*href="(.*?)"[^>]*>(.*?)', html, re.DOTALL): href = m.group(1) title = re.sub(r'<[^>]+>', '', m.group(2)).strip() results.append(f"{title}\n {href}") if len(results) >= 5: break if not results: return f"Search results for '{query}':\n(no results found)" return f"Search results for '{query}':\n" + "\n".join(results) except Exception as e: return f"Error searching: {e}" @app.middleware("http") async def cache_body(request: Request, call_next): body = await request.body() request.state.body = body.decode(errors="replace") response = await call_next(request) print("=== REQUEST ===") print(f"Method: {request.method} {request.url}") print(f"Status: {response.status_code}") print(f"Body: {request.state.body}") if response.status_code >= 400: print(f"Headers: {dict(request.headers)}") return response @app.exception_handler(RequestValidationError) async def validation_handler(request: Request, exc: RequestValidationError): body = getattr(request.state, "body", await request.body()) print("=== VALIDATION ERROR ===") print(f"Method: {request.method} {request.url}") print(f"Headers: {dict(request.headers)}") print(f"Body: {body}") print(f"Errors: {exc.errors()}") return JSONResponse(status_code=422, content={"detail": exc.errors()}) INFERENCE_PORT_API_KEY = os.environ.get("INFERENCE_PORT_API_KEY", "") INFERENCE_PORT_URL = "https://api.inference.net/v1/chat/completions" INFERENCE_PORT_MODEL = "meta-llama/llama-3.1-8b-instruct/fp-8" FALLBACK_URL = "https://keylessai.thryx.workers.dev/v1/chat/completions" FALLBACK_MODEL = "gpt-4o" _history = [] MAX_HISTORY = 20 def compact_history(): global _history if len(_history) <= MAX_HISTORY: return keep = 10 old = _history[:-keep] recent = _history[-keep:] summary = " | ".join(f"{m['role']}: {m['content'][:150]}" for m in old) _history = [{"role": "user", "content": f"[Earlier: {summary}]"}] + recent def format_history(): return "\n".join(f"{m['role']}: {m['content']}" for m in _history) class Message(BaseModel): role: str content: str class ChatRequest(BaseModel): system: str messages: List[Message] def trim_to_last_n_pairs(messages: List[Message], n: int = 3) -> List[Message]: pairs = [] i = len(messages) - 1 while i >= 0 and len(pairs) < n * 2: pairs.insert(0, messages[i]) i -= 1 while pairs and pairs[0].role != "user": pairs.pop(0) return pairs @app.get("/debug") async def debug(): key = INFERENCE_PORT_API_KEY return { "key_set": bool(key), "key_preview": f"{key[:6]}...{key[-4:]}" if len(key) > 10 else f"[too short: {len(key)} chars]", "url": INFERENCE_PORT_URL, } async def _relay_logic(req: ChatRequest, trim: bool = True, include_system: bool = True, tools: list = None): if trim: msgs = trim_to_last_n_pairs(req.messages, n=3) else: msgs = req.messages messages = [] if include_system: messages.append({"role": "system", "content": req.system}) messages.extend({"role": m.role, "content": m.content} for m in msgs) providers = [ (INFERENCE_PORT_URL, INFERENCE_PORT_MODEL, INFERENCE_PORT_API_KEY), (FALLBACK_URL, FALLBACK_MODEL, ""), ] last_err = None for url, model, api_key in providers: if not api_key and "inference.net" in url: continue async with httpx.AsyncClient(timeout=60) as client: for attempt in range(5): payload = { "model": model, "messages": messages, "max_tokens": 4096, "temperature": 0.9, } if tools: payload["tools"] = tools headers = {"Content-Type": "application/json"} if api_key: headers["Authorization"] = f"Bearer {api_key}" resp = await client.post(url, headers=headers, json=payload) if resp.status_code == 429: last_err = resp.text break if resp.status_code != 200: last_err = resp.text break data = resp.json() msg = data["choices"][0]["message"] if msg.get("tool_calls"): messages.append({"role": "assistant", "content": msg.get("content") or "", "tool_calls": msg["tool_calls"]}) for tc in msg["tool_calls"]: fn = tc["function"] args = json.loads(fn["arguments"]) if fn["name"] == "web_fetch": result = await web_fetch(args["url"], args.get("interact", False)) elif fn["name"] == "web_click": result = await web_click(args["url"]) elif fn["name"] == "web_search": result = await web_search(args["query"]) elif fn["name"] == "web_type": result = await web_type(args["url"], args["fields"], args.get("method", "POST")) else: result = f"Unknown tool: {fn['name']}" messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result}) continue reply = (msg.get("content") or "").strip() return {"reply": reply, "provider": url} raise HTTPException(status_code=429 if last_err else 500, detail=last_err or "All providers failed") @app.post("/reset") async def reset(): _history.clear() return {"history": ""} @app.get("/history") async def get_history(): return {"history": format_history()} @app.post("/shortcuts") async def shortcuts(req: dict): messages_raw = req.get("messages") or req.get("message") or req.get("text") or "" if isinstance(messages_raw, str): messages_raw = [{"role": "user", "content": messages_raw}] for m in messages_raw: _history.append({"role": m["role"], "content": m["content"]}) compact_history() system = req.get("system", 'Respond to the following messages casually and concisely, with slight humor, in the style of a male american teenager: do not include quotation marks, and respond with lowercase capitalization, and use abbreviations commonly used by the american youth. Use from one word to 1 sentence and try not to repeat yourself too much with emojis or expressions like "bro"') chat_req = ChatRequest(system=system, messages=[Message(**m) for m in _history]) result = await _relay_logic(chat_req, trim=False, tools=WEB_TOOLS) reply = result["reply"] _history.append({"role": "assistant", "content": reply}) return {"reply": reply, "history": format_history()} @app.post("/relay") async def relay(req: ChatRequest): return await _relay_logic(req, trim=True) app.mount("/", StaticFiles(directory="static", html=True), name="static")