| 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'<a[^>]*class="result__a"[^>]*href="(.*?)"[^>]*>(.*?)</a>', 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") |
|
|
