from __future__ import annotations
import inspect
import logging
import json
import asyncio
import uvicorn
import secrets
import os
import re
import shutil
import time
from collections import deque
from email.utils import formatdate
import os.path
import hashlib
import base64
from contextlib import asynccontextmanager
from urllib.parse import quote_plus
from fastapi import FastAPI, Response, Request, UploadFile, Form, Depends, Header
from fastapi.responses import (
StreamingResponse,
RedirectResponse,
HTMLResponse,
JSONResponse,
FileResponse,
)
from fastapi.exceptions import RequestValidationError
from fastapi.security import APIKeyHeader
from starlette.exceptions import HTTPException
from starlette.status import (
HTTP_200_OK,
HTTP_404_NOT_FOUND,
HTTP_401_UNAUTHORIZED,
HTTP_403_FORBIDDEN,
HTTP_429_TOO_MANY_REQUESTS,
HTTP_500_INTERNAL_SERVER_ERROR,
)
try:
from starlette.status import HTTP_422_UNPROCESSABLE_CONTENT
except ImportError:
HTTP_422_UNPROCESSABLE_CONTENT = 422
from starlette.staticfiles import NotModifiedResponse
from fastapi.encoders import jsonable_encoder
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials, HTTPBasic
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import FileResponse
from starlette.background import BackgroundTask
try:
from a2wsgi import WSGIMiddleware
has_a2wsgi = True
except ImportError:
has_a2wsgi = False
try:
from PIL import Image
has_pillow = True
except ImportError:
has_pillow = False
from types import SimpleNamespace
from typing import Union, Optional, List
try:
from typing import Annotated
except ImportError:
class Annotated:
pass
try:
from zendriver import util
has_nodriver = True
except ImportError:
has_nodriver = False
import g4f
import g4f.debug
from g4f.client import AsyncClient, ChatCompletion, ImagesResponse
from g4f.providers.response import BaseConversation, JsonConversation
from g4f.client.helper import filter_none
from g4f.config import DEFAULT_PORT, DEFAULT_TIMEOUT, DEFAULT_STREAM_TIMEOUT
from g4f.image import EXTENSIONS_MAP, is_data_an_media, process_image
from g4f.image.copy_images import get_media_dir, copy_media, get_source_url
from g4f.errors import (
ProviderNotFoundError,
ModelNotFoundError,
MissingAuthError,
NoValidHarFileError,
MissingRequirementsError,
RateLimitError,
)
from g4f.cookies import read_cookie_files, get_cookies_dir
from g4f.providers.types import ProviderType
from g4f.providers.response import AudioResponse
from g4f.providers.any_provider import AnyProvider
from g4f.providers.any_model_map import (
model_map,
vision_models,
image_models,
audio_models,
video_models,
)
from g4f.config import AppConfig
from g4f.client.factory import AbstractClientFactory
from g4f import Provider
from g4f.Provider import ProviderUtils
from g4f.gui import get_gui_app
from .stubs import (
ChatCompletionsConfig,
ImageGenerationConfig,
ResponsesConfig,
MessagesConfig,
ProviderResponseModel,
ModelResponseModel,
ErrorResponseModel,
ProviderResponseDetailModel,
FileResponseModel,
TranscriptionResponseModel,
AudioSpeechConfig,
)
from g4f import debug
try:
from g4f.gui.server.crypto import create_or_read_keys, decrypt_data, get_session_key
has_crypto = True
except ImportError:
has_crypto = False
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Request / response log store
# ---------------------------------------------------------------------------
_MAX_LOG_ENTRIES = 250
_MAX_BODY_LOG_SIZE = 1024 * 1024 # 1 MB
_SENSITIVE_HEADERS = {
"authorization",
"g4f-api-key",
"cookie",
"set-cookie",
"x-api-key",
}
_request_log: deque = deque(maxlen=_MAX_LOG_ENTRIES)
_log_id_counter: int = 0
def _sanitize_headers(headers: dict) -> dict:
return {
k: ("***" if k.lower() in _SENSITIVE_HEADERS else v) for k, v in headers.items()
}
def _try_parse_body(body_bytes: bytes, content_type: str):
if not body_bytes:
return None
if len(body_bytes) > _MAX_BODY_LOG_SIZE:
return f"<{len(body_bytes)} bytes – truncated>"
if "application/json" in content_type:
try:
return json.loads(body_bytes)
except Exception:
pass
try:
return body_bytes.decode("utf-8", errors="replace")
except Exception:
return f"<binary {len(body_bytes)} bytes>"
_LOGS_HTML = """<!doctype html>
<html lang="en">
<head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>g4f – Request Log</title>
<style>
*{box-sizing:border-box;margin:0;padding:0}
body{font-family:system-ui,-apple-system,sans-serif;background:#0d1117;color:#c9d1d9;min-height:100vh;font-size:14px}
a{color:#58a6ff}
.header{padding:14px 20px;border-bottom:1px solid #21262d;display:flex;align-items:center;gap:14px}
.header h1{font-size:15px;font-weight:600;color:#f0f6fc}
.header .sub{font-size:12px;color:#8b949e}
.toolbar{padding:10px 20px;border-bottom:1px solid #21262d;display:flex;gap:10px;align-items:center;flex-wrap:wrap}
.toolbar input[type=text]{flex:1;min-width:180px;background:#161b22;border:1px solid #30363d;color:#c9d1d9;padding:5px 10px;border-radius:6px;font-size:13px;outline:none}
.toolbar input[type=text]:focus{border-color:#58a6ff}
.toolbar label{display:flex;align-items:center;gap:5px;font-size:13px;color:#8b949e;cursor:pointer;user-select:none}
.btn{padding:5px 14px;border-radius:6px;border:1px solid #30363d;cursor:pointer;font-size:13px;background:#21262d;color:#c9d1d9}
.btn:hover{background:#30363d}
.btn-danger{border-color:#6e3435;background:#1c1214;color:#ffa198}
.btn-danger:hover{background:#6e3435}
.meta{margin-left:auto;font-size:12px;color:#6e7681}
.table-wrap{overflow-x:auto;padding:0 20px 40px}
table{width:100%;border-collapse:collapse;margin-top:14px;font-size:13px}
th{padding:6px 8px;text-align:left;color:#8b949e;font-weight:500;border-bottom:1px solid #21262d;white-space:nowrap}
td{padding:6px 8px;border-bottom:1px solid #161b22;white-space:nowrap;max-width:320px;overflow:hidden;text-overflow:ellipsis;vertical-align:middle}
tbody tr{cursor:pointer}
tbody tr:hover td{background:#161b22}
.GET{color:#3fb950}.POST{color:#58a6ff}.PUT{color:#e3b341}.DELETE{color:#f85149}.PATCH{color:#d2a8ff}
.s2{color:#3fb950}.s3{color:#58a6ff}.s4{color:#e3b341}.s5{color:#f85149}
.tag{display:inline-block;font-size:10px;padding:1px 6px;border-radius:10px;font-weight:500}
.tag-sse{background:#0d2044;color:#79c0ff}
.tag-body{background:#0d2820;color:#56d364}
.tag-empty{background:#1c2128;color:#6e7681}
.overlay{display:none;position:fixed;inset:0;background:rgba(0,0,0,.65);z-index:200;padding:24px;align-items:flex-start;justify-content:center;overflow-y:auto}
.overlay.active{display:flex}
.modal{background:#161b22;border:1px solid #30363d;border-radius:10px;width:100%;max-width:1040px;display:flex;flex-direction:column}
.modal-head{padding:14px 18px;border-bottom:1px solid #21262d;display:flex;justify-content:space-between;align-items:center;gap:10px}
.modal-head h2{font-size:13px;font-weight:600;color:#f0f6fc;overflow:hidden;text-overflow:ellipsis;white-space:nowrap;font-family:ui-monospace,monospace}
.modal-close{background:none;border:none;color:#6e7681;font-size:18px;cursor:pointer;line-height:1;padding:2px 6px;flex-shrink:0}
.modal-close:hover{color:#c9d1d9}
.modal-grid{display:grid;grid-template-columns:1fr 1fr 1fr}
.panel{padding:16px 18px;display:flex;flex-direction:column;gap:8px}
.panel:first-child{border-right:1px solid #21262d}
.panel:not(:last-child){border-right:1px solid #21262d}
.panel-title{font-size:11px;font-weight:600;text-transform:uppercase;letter-spacing:.06em;color:#6e7681}
.panel pre{background:#0d1117;border:1px solid #21262d;border-radius:6px;padding:12px;font-size:12px;line-height:1.5;overflow:auto;max-height:440px;white-space:pre-wrap;word-break:break-all;color:#c9d1d9;margin:0;font-family:ui-monospace,monospace}
@media(max-width:640px){.modal-grid{grid-template-columns:1fr}.panel:first-child{border-right:none;border-bottom:1px solid #21262d}.panel:not(:last-child){border-right:none;border-bottom:1px solid #21262d}}
</style>
</head>
<body>
<div class="header">
<h1>g4f Request Log</h1>
<span class="sub">last 500 entries · <a href="/v1">/v1 API</a></span>
</div>
<div class="toolbar">
<input type="text" id="q" placeholder="Filter by path, method, status, user…" oninput="render()">
<label><input type="checkbox" id="auto" checked onchange="toggleAuto()"> Auto‑refresh (3s)</label>
<button class="btn btn-danger" onclick="clearLogs()">Clear</button>
<span class="meta" id="meta"></span>
</div>
<div class="table-wrap"><table>
<thead><tr>
<th>#</th><th>Time (UTC)</th><th>Method</th><th>Path</th>
<th>Status</th><th>ms</th><th>User</th><th>Body</th>
</tr></thead>
<tbody id="tb"></tbody>
</table></div>
<div class="overlay" id="ov" onclick="overlayClick(event)">
<div class="modal" id="mod">
<div class="modal-head">
<h2 id="mtitle">–</h2>
<button class="modal-close" onclick="closeModal()">✕</button>
</div>
<div class="modal-grid">
<div class="panel"><div class="panel-title">Request</div><pre id="preq"></pre></div>
<div class="panel"><div class="panel-title">Response</div><pre id="pres"></pre></div>
<div class="panel"><div class="panel-title">Response Headers</div><pre id="presh"></pre></div>
</div>
</div>
</div>
<script>
'use strict';
var all=[], timer=null;
function sc(s){return s>=500?'s5':s>=400?'s4':s>=300?'s3':'s2';}
function btag(e){if(e.streaming&&e.response_body==null)return\'<span class="tag tag-sse">SSE…</span>\';if(e.response_body!=null)return\'<span class="tag tag-body">\'+(e.streaming?\'SSE\':\'body\')+\'</span>\';return\'<span class="tag tag-empty">–</span>\';}
function esc(s){return String(s??\'\'). replace(/&/g,\'&\').replace(/</g,\'<\').replace(/>/g,\'>\');}
function fmt(v){if(v==null)return\'(empty)\';if(typeof v===\'object\')return JSON.stringify(v,null,2);return String(v);}
async function load(){
try{var r=await fetch(\'/api/logs?limit=500\', { credentials: 'include' });if(!r.ok)return;var d=await r.json();all=d.entries||[];render();}catch(e){}
}
function render(){
var q=document.getElementById(\'q\').value.trim().toLowerCase();
var rows=q?all.filter(function(e){return(e.method+\' \'+e.path+\' \'+e.status+\' \'+(e.user||\'\')).toLowerCase().includes(q);}):all;
document.getElementById(\'meta\').textContent=rows.length+\' / \'+all.length+\' entries\';
document.getElementById(\'tb\').innerHTML=rows.map(function(e){
var t=(e.timestamp||\'\').replace(\'T\',\' \').replace(\'Z\',\'\');
var p=esc(e.path+(e.query?\'?\'+e.query:\'\'));
return\'<tr onclick="detail(\'+e.id+\')">\'+
\'<td style="color:#484f58">\'+e.id+\'</td>\'+
\'<td style="color:#6e7681;font-size:12px">\'+esc(t)+\'</td>\'+
\'<td class="\'+esc(e.method)+\'">\'+esc(e.method)+\'</td>\'+
\'<td title="\'+p+\'">\'+p+\'</td>\'+
\'<td class="\'+sc(e.status)+\'">\'+e.status+\'</td>\'+
\'<td style="color:#8b949e">\'+e.duration_ms+\'</td>\'+
\'<td style="color:#6e7681">\'+esc(e.user||\'\')+\'</td>\'+
\'<td>\'+btag(e)+\'</td>\'+
\'</tr>\';
}).join(\'\');
}
function detail(id){
var e=all.find(function(x){return x.id===id;});
if(!e)return;
document.getElementById(\'mtitle\').textContent=\'#\'+e.id+\' \'+e.method+\' \'+e.path+(e.query?\'?\'+e.query:\'\')+\' \u2192 \'+e.status+\' (\'+e.duration_ms+\'ms)\';
var req=\'\';
if(e.request_headers){req+=\'Headers:\\n\';for(var k in e.request_headers)req+=\' \'+k+\': \'+e.request_headers[k]+\'\\n\';}
if(e.request_body!=null)req+=\'\\nBody:\\n\'+fmt(e.request_body);
document.getElementById(\'preq\').textContent=req||\'(none)\';
document.getElementById(\'pres\').textContent=e.response_body!=null?fmt(e.response_body):(e.streaming?\'(streaming – collecting…)\':\'(empty)\');
var resh=\'\';
if(e.response_headers){resh+=\'Headers:\\n\';for(var k in e.response_headers)resh+=\' \'+k+\': \'+e.response_headers[k]+\'\\n\';}
document.getElementById(\'presh\').textContent=resh||\'(none)\';
document.getElementById(\'ov\').classList.add(\'active\');
}
function closeModal(){document.getElementById(\'ov\').classList.remove(\'active\');}
function overlayClick(ev){if(ev.target===document.getElementById(\'ov\'))closeModal();}
document.addEventListener(\'keydown\',function(e){if(e.key===\'Escape\')closeModal();});
async function clearLogs(){await fetch(\'/api/logs\',{method:\'DELETE\'});all=[];render();}
function toggleAuto(){clearInterval(timer);timer=null;if(document.getElementById(\'auto\').checked)timer=setInterval(load,3000);}
load();timer=setInterval(load,3000);
</script>
</body></html>"""
@asynccontextmanager
async def lifespan(app: FastAPI):
# Read cookie files if not ignored
if not AppConfig.ignore_cookie_files:
read_cookie_files()
else:
AppConfig.load_from_env()
yield
if has_nodriver:
for browser in util.get_registered_instances():
if browser.connection:
await browser.stop()
lock_file = os.path.join(get_cookies_dir(), ".browser_is_open")
if os.path.exists(lock_file):
try:
os.remove(lock_file)
except Exception as e:
debug.error(f"Failed to remove lock file {lock_file}:", e)
_LOG_SKIP_PREFIXES = ("/images/", "/media/", "/thumbnail/", "/dist/", "/.well-known/")
_LOG_SKIP_EXACT = {"/api/logs", "/logs", "/favicon.ico"}
def create_app():
app = FastAPI(lifespan=lifespan)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["*"],
)
@app.middleware("http")
async def log_requests(request: Request, call_next):
global _log_id_counter
path = request.url.path
if (
any(path.startswith(p) for p in _LOG_SKIP_PREFIXES)
or path in _LOG_SKIP_EXACT
):
return await call_next(request)
qs = f"?{request.url.query}" if request.url.query else ""
user = request.headers.get("x-user", "")
user_info = f" user={user}" if user else ""
logger.debug("→ %s %s%s%s", request.method, path, qs, user_info)
# Capture request body (Starlette caches after first read)
req_body_bytes = await request.body()
req_body = _try_parse_body(
req_body_bytes, request.headers.get("content-type", "")
)
start = time.monotonic()
response = await call_next(request)
duration_ms = round((time.monotonic() - start) * 1000)
resp_content_type = response.headers.get("content-type", "")
is_streaming = "text/event-stream" in resp_content_type
log_entry: dict = {}
resp_headers_log = _sanitize_headers(dict(response.headers))
if not is_streaming:
chunks: list[bytes] = []
async for chunk in response.body_iterator:
chunks.append(chunk)
resp_body_bytes = b"".join(chunks)
resp_body = _try_parse_body(resp_body_bytes, resp_content_type)
# Reconstruct response so it can still be sent to the client
resp_headers = {
k: v
for k, v in response.headers.items()
if k.lower() != "content-length"
}
response = Response(
content=resp_body_bytes,
status_code=response.status_code,
headers=resp_headers,
media_type=response.media_type,
)
else:
# Tee the streaming iterator: forward chunks to client AND accumulate for log
sse_chunks: list[bytes] = []
resp_body = None
orig_iterator = response.body_iterator
async def tee_iterator():
async for chunk in orig_iterator:
if isinstance(chunk, bytes):
sse_chunks.append(chunk)
else:
sse_chunks.append(chunk.encode("utf-8", errors="replace"))
yield chunk
# After iteration completes, parse and store the full SSE body
raw = b"".join(sse_chunks)
parsed = _try_parse_body(raw, "text/plain")
log_entry["response_body"] = parsed
response.body_iterator = tee_iterator()
level = logging.WARNING if response.status_code >= 400 else logging.INFO
logger.log(
level,
"%s %s%s → %d (%dms)%s",
request.method,
path,
qs,
response.status_code,
duration_ms,
user_info,
)
_log_id_counter += 1
log_entry.update(
{
"id": _log_id_counter,
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"method": request.method,
"path": path,
"query": request.url.query or None,
"status": response.status_code,
"duration_ms": duration_ms,
"user": user or None,
"streaming": is_streaming,
"request_headers": _sanitize_headers(dict(request.headers)),
"request_body": req_body,
"response_headers": resp_headers_log,
"response_body": resp_body, # None for SSE until iterator completes
}
)
_request_log.append(log_entry)
return response
api = Api(app)
api.register_routes()
api.register_authorization()
api.register_validation_exception_handler()
if AppConfig.gui:
if not has_a2wsgi:
raise MissingRequirementsError(
"a2wsgi is required for GUI. Install it with: pip install a2wsgi"
)
gui_app = WSGIMiddleware(
get_gui_app(AppConfig.demo, AppConfig.timeout, AppConfig.stream_timeout)
)
app.mount("/", gui_app)
if AppConfig.ignored_providers:
for provider in AppConfig.ignored_providers:
if provider in ProviderUtils.convert:
ProviderUtils.convert[provider].working = False
return app
def create_app_debug():
g4f.debug.logging = True
return create_app()
def create_app_with_gui_and_debug():
g4f.debug.logging = True
AppConfig.gui = True
return create_app()
def create_app_with_demo_and_debug():
g4f.debug.logging = True
AppConfig.gui = True
AppConfig.demo = True
return create_app()
class ErrorResponse(Response):
media_type = "application/json"
@classmethod
def from_exception(
cls,
exception: Exception,
config: Union[ChatCompletionsConfig, ImageGenerationConfig] = None,
status_code: int = HTTP_500_INTERNAL_SERVER_ERROR,
):
return cls(format_exception(exception, config), status_code)
@classmethod
def from_message(
cls,
message: str,
status_code: int = HTTP_500_INTERNAL_SERVER_ERROR,
headers: dict = None,
):
return cls(format_exception(message), status_code, headers=headers)
def render(self, content) -> bytes:
return str(content).encode(errors="ignore")
def update_headers(
request: Request, new_api_key: str = None, user: str = None
) -> Request:
new_headers = request.headers.mutablecopy()
if new_api_key:
new_headers["authorization"] = f"Bearer {new_api_key}"
else:
del new_headers["authorization"]
if user:
new_headers["x-user"] = user
request.scope["headers"] = new_headers.raw
delattr(request, "_headers")
return request
class Api:
def __init__(self, app: FastAPI) -> None:
self.app = app
self.client = AsyncClient()
self.get_g4f_api_key = APIKeyHeader(name="g4f-api-key")
self.conversations: dict[str, dict[str, BaseConversation]] = {}
security = HTTPBearer(auto_error=False)
basic_security = HTTPBasic()
async def get_username(self, request: Request) -> str:
credentials = await self.basic_security(request)
current_password_bytes = credentials.password.encode()
is_correct_password = secrets.compare_digest(
current_password_bytes, AppConfig.g4f_api_key.encode()
)
if not is_correct_password:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Basic"},
)
return credentials.username
def register_authorization(self):
if AppConfig.g4f_api_key:
print(
"Register authentication key:",
"".join(["*" for _ in range(len(AppConfig.g4f_api_key))]),
)
if has_crypto:
private_key, _ = create_or_read_keys()
session_key = get_session_key()
def _requires_api_key(path: str, demo: bool) -> bool:
"""Return ``True`` when *path* must present a G4F API key."""
return (
path.startswith("/v1/")
or (path.startswith("/api/") and path != "/api/logs")
or (path.startswith("/pa/") and not demo)
or (demo and path in ["/backend-api/v2/upload_cookies"])
)
@self.app.middleware("http")
async def authorization(request: Request, call_next):
user = None
if (
request.method != "OPTIONS"
and AppConfig.g4f_api_key is not None
or AppConfig.demo
):
update_authorization = False
try:
user_g4f_api_key = await self.get_g4f_api_key(request)
except HTTPException:
user_g4f_api_key = getattr(
await self.security(request), "credentials", None
)
update_authorization = True
if user_g4f_api_key:
user_g4f_api_key = user_g4f_api_key.split()
country = request.headers.get("Cf-Ipcountry", "")
if AppConfig.demo and user is None:
ip = request.headers.get("X-Forwarded-For", "")[:4].strip(":.")
user = request.headers.get("x-user", ip)
user = f"{country}:{user}" if country else user
if (
AppConfig.g4f_api_key is None
or not user_g4f_api_key
or not secrets.compare_digest(
AppConfig.g4f_api_key, user_g4f_api_key[0]
)
):
if has_crypto and user_g4f_api_key:
try:
expires, user = decrypt_data(
private_key, user_g4f_api_key[0]
).split(":", 1)
except Exception:
try:
data = json.loads(
decrypt_data(session_key, user_g4f_api_key[0])
)
debug.log(f"Decrypted G4F API key data: {data}")
expires = (
int(decrypt_data(private_key, data.pop("data")))
+ 86400
)
user = data.get("user", user)
if not user or "referrer" not in data:
raise ValueError("User not found")
except Exception:
return ErrorResponse.from_message(
f"Invalid G4F API key", HTTP_401_UNAUTHORIZED
)
user = f"{country}:{user}" if country else user
expires = int(expires) - int(time.time())
hours, remainder = divmod(expires, 3600)
minutes, seconds = divmod(remainder, 60)
if expires < 0:
debug.log(f"G4F API key expired for user '{user}'")
return ErrorResponse.from_message(
"G4F API key expired", HTTP_401_UNAUTHORIZED
)
count = 0
for char in user:
if char.isupper():
count += 1
if count >= 6:
debug.log(f"Invalid user name (screaming): '{user}'")
return ErrorResponse.from_message(
"Invalid user name (screaming)", HTTP_401_UNAUTHORIZED
)
debug.log(
f"User: '{user}' G4F API key expires in {hours}h {minutes}m {seconds}s"
)
elif user is None:
user = "admin"
path = request.url.path
if _requires_api_key(path, AppConfig.demo):
if request.method != "OPTIONS" and not path.endswith("/models"):
if not user_g4f_api_key:
return ErrorResponse.from_message(
"G4F API key required", HTTP_401_UNAUTHORIZED
)
if AppConfig.g4f_api_key is None and user is None:
return ErrorResponse.from_message(
"Invalid G4F API key", HTTP_403_FORBIDDEN
)
elif path == "/logs" or (
not AppConfig.demo
and not path.startswith("/images/")
and not path.startswith("/media/")
):
if user_g4f_api_key:
if user is None:
return ErrorResponse.from_message(
"Invalid G4F API key", HTTP_403_FORBIDDEN
)
elif (
path.startswith("/backend-api/")
or path.startswith("/chat/")
or path.startswith("/playground/")
or path in ["/logs"]
):
try:
new_user = await self.get_username(request)
if user is None:
user = new_user
except HTTPException as e:
return ErrorResponse.from_message(
e.detail, e.status_code, e.headers
)
if user_g4f_api_key and update_authorization:
new_api_key = user_g4f_api_key.pop()
if secrets.compare_digest(AppConfig.g4f_api_key, new_api_key):
new_api_key = None
else:
new_api_key = None
request = update_headers(request, new_api_key, user)
response = await call_next(request)
return response
def register_validation_exception_handler(self):
@self.app.exception_handler(RequestValidationError)
async def validation_exception_handler(
request: Request, exc: RequestValidationError
):
details = exc.errors()
modified_details = []
for error in details:
debug.log(
f"Validation error: {error['loc']} - {error['msg']} ({error['type']})"
)
modified_details.append(
{
"loc": error["loc"],
"message": error["msg"],
"type": error["type"],
}
)
return JSONResponse(
status_code=HTTP_422_UNPROCESSABLE_CONTENT,
content=jsonable_encoder({"detail": modified_details}),
)
def register_routes(self):
if not AppConfig.gui:
@self.app.get("/")
async def read_root():
return RedirectResponse("/v1", 302)
@self.app.get("/v1")
async def read_root_v1():
return HTMLResponse(
"g4f API: Go to "
'<a href="/v1/models">models</a>, '
'<a href="/v1/chat/completions">chat/completions</a>, '
'<a href="/v1/responses">responses</a> (OpenAI), '
'<a href="/v1/messages">messages</a> (Anthropic), or '
'<a href="/v1/media/generate">media/generate</a> <br><br>'
"Open Swagger UI at: "
'<a href="/docs">/docs</a>'
)
@self.app.get(
"/v1/models",
responses={
HTTP_200_OK: {"model": List[ModelResponseModel]},
},
)
async def models():
return {
"object": "list",
"data": [
{
"id": model,
"object": "model",
"created": 0,
"owned_by": "",
"image": isinstance(model, g4f.models.ImageModel),
"vision": isinstance(model, g4f.models.VisionModel),
"provider": False,
}
for model in AnyProvider.get_models()
]
+ [
{
"id": provider_name,
"object": "model",
"created": 0,
"owned_by": getattr(provider, "label", ""),
"image": bool(getattr(provider, "image_models", False)),
"vision": bool(getattr(provider, "vision_models", False)),
"provider": True,
}
for provider_name, provider in ProviderUtils.convert.items()
if provider.working
],
}
@self.app.get(
"/api/{provider:path}/models",
responses={
HTTP_200_OK: {"model": List[ModelResponseModel]},
},
)
async def models(
provider: str,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
if not hasattr(provider, "get_models"):
models = getattr(provider, "models", [])
elif credentials is not None and credentials.credentials != "secret":
models = provider.get_models(api_key=credentials.credentials)
else:
models = provider.get_models()
if inspect.isawaitable(models):
models = await models
return {
"object": "list",
"data": [
{
"id": model.get("id") if isinstance(model, dict) else model,
"object": "model",
"created": 0,
"owned_by": getattr(provider, "label", provider.__name__),
"image": (model.get("id") if isinstance(model, dict) else model)
in getattr(provider, "image_models", []),
"vision": (
model.get("id") if isinstance(model, dict) else model
)
in getattr(provider, "vision_models", []),
"audio": (model.get("id") if isinstance(model, dict) else model)
in getattr(provider, "audio_models", []),
"video": (model.get("id") if isinstance(model, dict) else model)
in getattr(provider, "video_models", []),
"type": "image"
if (model.get("id") if isinstance(model, dict) else model)
in getattr(provider, "image_models", [])
else "chat",
"count": getattr(provider, "models_count", {}).get(
model.get("id") if isinstance(model, dict) else model, 0
),
**(model if isinstance(model, dict) else {}),
}
for model in (
models.values() if isinstance(models, dict) else models
)
],
}
# quota endpoint mimics backend-api/v2/quota but exposed on public API
@self.app.get("/api/{provider:path}/quota")
async def provider_quota(
provider: str,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
if not hasattr(provider, "get_quota"):
return ErrorResponse.from_message(
"Provider doesn't support get_quota", HTTP_500_INTERNAL_SERVER_ERROR
)
try:
if credentials is not None and credentials.credentials != "secret":
usage = await provider.get_quota(api_key=credentials.credentials)
else:
usage = await provider.get_quota()
return usage
except MissingAuthError as e:
return ErrorResponse.from_message(
f"{type(e).__name__}: {e}", HTTP_401_UNAUTHORIZED
)
except Exception as e:
return ErrorResponse.from_message(
f"{type(e).__name__}: {e}", HTTP_500_INTERNAL_SERVER_ERROR
)
@self.app.get(
"/v1/models/{model_name}",
responses={
HTTP_200_OK: {"model": ModelResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
@self.app.post(
"/v1/models/{model_name}",
responses={
HTTP_200_OK: {"model": ModelResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
async def model_info(model_name: str) -> ModelResponseModel:
if model_name in g4f.models.ModelUtils.convert:
model_info = g4f.models.ModelUtils.convert[model_name]
return JSONResponse(
{
"id": model_name,
"object": "model",
"created": 0,
"owned_by": model_info.base_provider,
}
)
return ErrorResponse.from_message(
"The model does not exist.", HTTP_404_NOT_FOUND
)
responses = {
HTTP_200_OK: {"model": ChatCompletion},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_422_UNPROCESSABLE_CONTENT: {"model": ErrorResponseModel},
HTTP_429_TOO_MANY_REQUESTS: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.post("/v1/chat/completions", responses=responses)
@self.app.post("/api/{provider:path}/chat/completions", responses=responses)
@self.app.post(
"/api/{provider}/{conversation_id}/chat/completions", responses=responses
)
async def chat_completions(
config: ChatCompletionsConfig,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
provider: str = None,
conversation_id: str = None,
x_user: Annotated[str | None, Header()] = None,
):
if provider is not None:
config.provider = provider
if config.provider is None:
config.provider = AppConfig.provider
try:
provider = AbstractClientFactory.create_provider(None, config.provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
try:
if config.conversation_id is None:
config.conversation_id = conversation_id
if config.timeout is None:
config.timeout = AppConfig.timeout
if config.stream_timeout is None and config.stream:
config.stream_timeout = AppConfig.stream_timeout
if credentials is not None and credentials.credentials != "secret":
config.api_key = credentials.credentials
conversation = config.conversation
if conversation:
conversation = JsonConversation(**conversation)
elif config.conversation_id is not None and config.provider is not None:
if config.conversation_id in self.conversations:
if (
config.provider
in self.conversations[config.conversation_id]
):
conversation = self.conversations[config.conversation_id][
config.provider
]
if config.image is not None:
try:
is_data_an_media(config.image)
except ValueError as e:
return ErrorResponse.from_message(
f"The image you send must be a data URI. Example: data:image/jpeg;base64,...",
status_code=HTTP_422_UNPROCESSABLE_CONTENT,
)
if config.media is None:
config.media = config.images
if config.media is not None:
for image in config.media:
try:
is_data_an_media(image[0], image[1])
except ValueError as e:
example = json.dumps(
{
"media": [
["data:image/jpeg;base64,...", "filename.jpg"]
]
}
)
return ErrorResponse.from_message(
f"The media you send must be a data URIs. Example: {example}",
status_code=HTTP_422_UNPROCESSABLE_CONTENT,
)
# Create the completion response
response = self.client.chat.completions.create(
**filter_none(
**{
"model": AppConfig.model,
"provider": AppConfig.provider,
"proxy": AppConfig.proxy,
**(
config.model_dump(exclude_none=True)
if hasattr(config, "model_dump")
else config.dict(exclude_none=True)
),
**{
"provider": provider,
"conversation_id": None,
"conversation": conversation,
"user": x_user,
},
},
ignored=AppConfig.ignored_providers,
),
)
if not config.stream:
result = await response
return Response(
content=result.model_dump_json()
if hasattr(result, "model_dump_json")
else result.json(),
media_type="application/json",
headers=getattr(result, "_headers").get_dict()
if hasattr(result, "_headers")
else None,
)
first_chunk = await response.__anext__()
async def streaming():
yield f"data: {first_chunk.model_dump_json() if hasattr(first_chunk, 'model_dump_json') else first_chunk.json()}\n\n"
try:
async for chunk in response:
if isinstance(chunk, BaseConversation):
if (
config.conversation_id is not None
and config.provider is not None
):
if config.conversation_id not in self.conversations:
self.conversations[config.conversation_id] = {}
self.conversations[config.conversation_id][
config.provider
] = chunk
else:
yield f"data: {chunk.model_dump_json() if hasattr(chunk, 'model_dump_json') else chunk.json()}\n\n"
except GeneratorExit:
pass
except RateLimitError as e:
debug.error(e)
yield f"data: {format_exception(e, config)}\n\n"
except Exception as e:
logger.exception(e)
yield f"data: {format_exception(e, config)}\n\n"
yield "data: [DONE]\n\n"
headers = (
getattr(first_chunk, "_headers").get_dict()
if hasattr(first_chunk, "_headers")
else {}
)
headers = {
k.encode("latin-1", "ignore")
.decode("latin-1"): v.encode("latin-1", "ignore")
.decode("latin-1")
for k, v in headers.items()
}
return StreamingResponse(
streaming(), media_type="text/event-stream", headers=headers
)
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
except (MissingAuthError, NoValidHarFileError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
except RateLimitError as e:
return ErrorResponse.from_exception(
e, config, HTTP_429_TOO_MANY_REQUESTS
)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, config, HTTP_500_INTERNAL_SERVER_ERROR
)
# ------------------------------------------------------------------ #
# OpenAI Responses API (/v1/responses) #
# https://platform.openai.com/docs/api-reference/responses #
# ------------------------------------------------------------------ #
@self.app.post("/v1/responses", responses=responses)
@self.app.post("/api/{provider:path}/responses", responses=responses)
async def create_response(
config: ResponsesConfig,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
provider: str = None,
x_user: Annotated[str | None, Header()] = None,
):
if provider is not None:
config.provider = provider
if config.provider is None:
config.provider = AppConfig.provider
try:
provider = AbstractClientFactory.create_provider(None, config.provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
try:
if config.timeout is None:
config.timeout = AppConfig.timeout
if config.stream_timeout is None and config.stream:
config.stream_timeout = AppConfig.stream_timeout
if credentials is not None and credentials.credentials != "secret":
config.api_key = credentials.credentials
# Normalize `input` into a messages list.
messages = config.input
if isinstance(messages, str):
messages = [{"role": "user", "content": messages}]
if config.instructions:
messages = [
{"role": "system", "content": config.instructions},
*messages,
]
response = self.client.chat.completions.create(
**filter_none(
**{
"model": AppConfig.model,
"provider": AppConfig.provider,
"proxy": AppConfig.proxy,
**(
config.model_dump(exclude_none=True)
if hasattr(config, "model_dump")
else config.dict(exclude_none=True)
),
**{
"provider": provider,
"messages": messages,
"user": x_user,
},
},
ignored=AppConfig.ignored_providers,
),
)
if not config.stream:
result = await response
text = result.choices[0].message.content if result.choices else ""
usage = getattr(result, "usage", None)
if usage is not None and hasattr(usage, "model_dump"):
usage = usage.model_dump()
elif usage is not None and hasattr(usage, "dict"):
usage = usage.dict()
return JSONResponse(
{
"id": getattr(
result, "id", f"resp_{secrets.token_hex(12)}"
),
"object": "response",
"created_at": getattr(result, "created", int(time.time())),
"model": getattr(result, "model", config.model),
"provider": getattr(provider, "__name__", config.provider),
"output": [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": text}],
}
],
"output_text": text,
"usage": usage,
}
)
first_chunk = await response.__anext__()
async def responses_streaming():
yield f"data: {first_chunk.model_dump_json() if hasattr(first_chunk, 'model_dump_json') else first_chunk.json()}\n\n"
try:
async for chunk in response:
if isinstance(chunk, BaseConversation):
pass
else:
yield f"data: {chunk.model_dump_json() if hasattr(chunk, 'model_dump_json') else chunk.json()}\n\n"
except GeneratorExit:
pass
except RateLimitError as e:
debug.error(e)
yield f"data: {format_exception(e, config)}\n\n"
except Exception as e:
logger.exception(e)
yield f"data: {format_exception(e, config)}\n\n"
yield "data: [DONE]\n\n"
headers = (
getattr(first_chunk, "_headers").get_dict()
if hasattr(first_chunk, "_headers")
else {}
)
headers = {
k.encode("latin-1", "ignore")
.decode("latin-1"): v.encode("latin-1", "ignore")
.decode("latin-1")
for k, v in headers.items()
}
return StreamingResponse(
responses_streaming(),
media_type="text/event-stream",
headers=headers,
)
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
except (MissingAuthError, NoValidHarFileError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
except RateLimitError as e:
return ErrorResponse.from_exception(
e, config, HTTP_429_TOO_MANY_REQUESTS
)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, config, HTTP_500_INTERNAL_SERVER_ERROR
)
# ------------------------------------------------------------------ #
# Anthropic Messages API (/v1/messages) #
# https://docs.anthropic.com/en/api/messages #
# ------------------------------------------------------------------ #
@self.app.post("/v1/messages", responses=responses)
@self.app.post("/api/{provider:path}/messages", responses=responses)
async def create_message(
config: MessagesConfig,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
provider: str = None,
x_user: Annotated[str | None, Header()] = None,
):
if provider is not None:
config.provider = provider
if config.provider is None:
config.provider = AppConfig.provider
try:
provider = AbstractClientFactory.create_provider(None, config.provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
try:
if config.timeout is None:
config.timeout = AppConfig.timeout
if config.stream_timeout is None and config.stream:
config.stream_timeout = AppConfig.stream_timeout
if credentials is not None and credentials.credentials != "secret":
config.api_key = credentials.credentials
# Anthropic uses a top-level `system` field; fold it into messages.
messages = config.messages
if config.system:
system_content = config.system
if isinstance(system_content, list):
system_content = " ".join(
b.get("text", "") if isinstance(b, dict) else str(b)
for b in system_content
)
messages = [
{"role": "system", "content": system_content},
*messages,
]
response = self.client.chat.completions.create(
**filter_none(
**{
"model": AppConfig.model,
"provider": AppConfig.provider,
"proxy": AppConfig.proxy,
**(
config.model_dump(exclude_none=True)
if hasattr(config, "model_dump")
else config.dict(exclude_none=True)
),
**{
"provider": provider,
"messages": messages,
"user": x_user,
},
},
ignored=AppConfig.ignored_providers,
),
)
if not config.stream:
result = await response
text = result.choices[0].message.content if result.choices else ""
usage = getattr(result, "usage", None)
input_tokens = getattr(usage, "prompt_tokens", 0) or 0
output_tokens = getattr(usage, "completion_tokens", 0) or 0
return JSONResponse(
{
"id": getattr(result, "id", f"msg_{secrets.token_hex(12)}"),
"type": "message",
"role": "assistant",
"model": getattr(result, "model", config.model),
"provider": getattr(provider, "__name__", config.provider),
"content": [{"type": "text", "text": text}],
"stop_reason": getattr(
result.choices[0], "finish_reason", None
)
if result.choices
else None,
"stop_sequence": None,
"usage": {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
},
}
)
first_chunk = await response.__anext__()
async def messages_streaming():
yield f"data: {first_chunk.model_dump_json() if hasattr(first_chunk, 'model_dump_json') else first_chunk.json()}\n\n"
try:
async for chunk in response:
if isinstance(chunk, BaseConversation):
pass
else:
yield f"data: {chunk.model_dump_json() if hasattr(chunk, 'model_dump_json') else chunk.json()}\n\n"
except GeneratorExit:
pass
except RateLimitError as e:
debug.error(e)
yield f"data: {format_exception(e, config)}\n\n"
except Exception as e:
logger.exception(e)
yield f"data: {format_exception(e, config)}\n\n"
yield "data: [DONE]\n\n"
headers = (
getattr(first_chunk, "_headers").get_dict()
if hasattr(first_chunk, "_headers")
else {}
)
headers = {
k.encode("latin-1", "ignore")
.decode("latin-1"): v.encode("latin-1", "ignore")
.decode("latin-1")
for k, v in headers.items()
}
return StreamingResponse(
messages_streaming(),
media_type="text/event-stream",
headers=headers,
)
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
except (MissingAuthError, NoValidHarFileError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
except RateLimitError as e:
return ErrorResponse.from_exception(
e, config, HTTP_429_TOO_MANY_REQUESTS
)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, config, HTTP_500_INTERNAL_SERVER_ERROR
)
responses = {
HTTP_200_OK: {"model": ImagesResponse},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_429_TOO_MANY_REQUESTS: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.post("/v1/media/generate", responses=responses)
@self.app.post("/v1/images/generate", responses=responses)
@self.app.post("/v1/images/generations", responses=responses)
@self.app.post("/api/{provider:path}/images/generations", responses=responses)
async def generate_image(
request: Request,
config: ImageGenerationConfig,
provider: str = None,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
if provider is None:
provider = config.provider
if provider is None:
provider = AppConfig.provider
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
if (
config.api_key is None
and credentials is not None
and credentials.credentials != "secret"
):
config.api_key = credentials.credentials
try:
response = await self.client.images.generate(
**config.model_dump(exclude_none=True)
if hasattr(config, "model_dump")
else config.dict(exclude_none=True),
provider=provider,
)
for image in response.data:
if hasattr(image, "url") and image.url.startswith("/"):
image.url = f"{request.base_url}{image.url.lstrip('/')}"
return response
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
except MissingAuthError as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, config, HTTP_500_INTERNAL_SERVER_ERROR
)
@self.app.get(
"/v1/providers",
responses={
HTTP_200_OK: {"model": List[ProviderResponseModel]},
},
)
async def providers():
return [
{
"id": provider.__name__,
"object": "provider",
"created": 0,
"url": provider.url,
"label": getattr(provider, "label", None),
}
for provider in Provider.__providers__
if provider.working
]
@self.app.get(
"/v1/providers/{provider}",
responses={
HTTP_200_OK: {"model": ProviderResponseDetailModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
async def providers_info(provider: str):
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
return {
"id": provider.__name__,
"object": "provider",
"created": 0,
"url": provider.url,
"label": getattr(provider, "label", None),
"image_models": getattr(provider, "image_models", []) or [],
"vision_models": [
model
for model in [getattr(provider, "default_vision_model", None)]
if model
],
"params": [*provider.get_parameters()]
if hasattr(provider, "get_parameters")
else [],
}
# ------------------------------------------------------------------ #
# PA Provider routes #
# ------------------------------------------------------------------ #
@self.app.get(
"/pa/providers",
responses={
HTTP_200_OK: {},
},
)
async def pa_providers_list():
"""List all PA providers loaded from the workspace.
Filenames are never exposed; each provider is identified by a
stable opaque ID (SHA-256 of the path, first 8 hex chars).
"""
from g4f.mcp.pa_provider import get_pa_registry
return get_pa_registry().list_providers()
@self.app.get(
"/pa/providers/{provider_id}",
responses={
HTTP_200_OK: {},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
async def pa_providers_detail(provider_id: str):
"""Get details for a single PA provider by its opaque ID."""
from g4f.mcp.pa_provider import get_pa_registry
info = get_pa_registry().get_provider_info(provider_id)
if info is None:
return ErrorResponse.from_message(
f"PA provider '{provider_id}' not found", HTTP_404_NOT_FOUND
)
return info
# ------------------------------------------------------------------ #
# PA workspace static file serving (HTML/CSS/JS/images for browser) #
# ------------------------------------------------------------------ #
#: MIME types that are safe to serve for browser rendering.
#: Only these extensions are allowed; all others are refused with 403.
_WORKSPACE_SAFE_TYPES: dict[str, str] = {
"html": "text/html; charset=utf-8",
"htm": "text/html; charset=utf-8",
"css": "text/css; charset=utf-8",
"js": "application/javascript; charset=utf-8",
"mjs": "application/javascript; charset=utf-8",
"json": "application/json; charset=utf-8",
"txt": "text/plain; charset=utf-8",
"md": "text/markdown; charset=utf-8",
"svg": "image/svg+xml",
"png": "image/png",
"jpg": "image/jpeg",
"jpeg": "image/jpeg",
"gif": "image/gif",
"webp": "image/webp",
"ico": "image/x-icon",
"woff": "font/woff",
"woff2": "font/woff2",
"ttf": "font/ttf",
"otf": "font/otf",
"py": "text/plain; charset=utf-8",
}
@self.app.get(
"/pa/files/{file_path:path}",
responses={
HTTP_200_OK: {},
HTTP_403_FORBIDDEN: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
async def pa_serve_workspace_file(file_path: str, request: Request):
"""Securely serve a workspace file for browser rendering.
Only files within ``~/.g4f/workspace`` can be served. Path
traversal (``..``) is blocked. Only the MIME types listed in
``_WORKSPACE_SAFE_TYPES`` are served; all other extensions are
refused with **403 Forbidden** so that sensitive file types (e.g.
``.env``, ``.pa.py``, ``.py``) can never be read via this route.
HTML files are served with a ``Content-Security-Policy: sandbox``
directive (without ``allow-same-origin``), which forces the page
into a unique *null* browser origin. As a result the page cannot
access ``localStorage``, ``sessionStorage``, ``IndexedDB``, or
cookies belonging to the g4f server origin — the browser rejects
all such calls with a ``SecurityError``. The actual request
origin (``scheme://host``) is used in every source directive (e.g.
``default-src``) instead of ``'self'``, so that co-located CSS,
JS, images, and fonts still load correctly despite the document
having a null origin.
Non-HTML sub-resources (CSS, JS, images, fonts) are served without
the ``sandbox`` directive; they are leaf resources and do not run
in their own browsing context.
"""
from g4f.mcp.pa_provider import get_workspace_dir
workspace = get_workspace_dir()
# Normalise and check for traversal
try:
resolved = (workspace / file_path).resolve()
resolved.relative_to(workspace.resolve())
except (ValueError, Exception):
return ErrorResponse.from_message(
"Path traversal is not allowed", HTTP_403_FORBIDDEN
)
if not resolved.exists() or not resolved.is_file():
return ErrorResponse.from_message(
f"File not found: {file_path}", HTTP_404_NOT_FOUND
)
ext = resolved.suffix.lstrip(".").lower()
mime_type = _WORKSPACE_SAFE_TYPES.get(ext)
if mime_type is None:
return ErrorResponse.from_message(
f"File type '.{ext}' is not allowed for browser rendering",
HTTP_403_FORBIDDEN,
)
# Derive the actual request origin (scheme + authority) from the
# ASGI scope via request.url — this is set by the server
# infrastructure and is not controllable by the client (unlike the
# Host header, which can be spoofed to inject arbitrary values into
# the CSP). request.url.netloc includes the port when non-default.
request_origin = f"{request.url.scheme}://{request.url.netloc}"
is_html = ext in ("html", "htm")
if is_html:
# HTML documents are served with the CSP sandbox directive
# (without allow-same-origin). This forces the page into a
# unique null browsing-context origin so that it cannot access
# the g4f server's localStorage, sessionStorage, IndexedDB, or
# cookies. The page can still load sub-resources (CSS, JS,
# images) because they are referenced by the explicit
# request_origin in the source directives.
csp = (
"sandbox allow-scripts allow-forms allow-downloads allow-popups; "
f"default-src {request_origin} https://g4f.space; "
f"script-src {request_origin} 'unsafe-inline'; "
f"style-src {request_origin} 'unsafe-inline'; "
f"img-src {request_origin} data:; "
f"font-src {request_origin} data:; "
"connect-src 'none'; "
"object-src 'none'; "
"base-uri 'none';"
)
else:
# Non-HTML sub-resources (CSS, JS, images, fonts) don't need
# sandboxing — they are leaf assets without their own browsing
# context. Use the request origin for source directives.
csp = (
f"default-src {request_origin} https://g4f.space; "
f"script-src {request_origin} 'unsafe-inline'; "
f"style-src {request_origin} 'unsafe-inline'; "
f"img-src {request_origin} data:; "
f"font-src {request_origin} data:; "
"connect-src 'none'; "
"object-src 'none'; "
"base-uri 'none';"
)
headers = {
# Prevent the browser from sniffing a different content-type
"X-Content-Type-Options": "nosniff",
# Prevent this page from being framed by untrusted origins
"X-Frame-Options": "SAMEORIGIN",
# Basic XSS filter (belt-and-suspenders; CSP is more important)
"X-XSS-Protection": "1; mode=block",
"Content-Security-Policy": csp,
# Restrict powerful browser APIs that workspace pages don't need
"Permissions-Policy": (
"geolocation=(), camera=(), microphone=(), "
"payment=(), usb=(), fullscreen=()"
),
"Cache-Control": "no-store",
}
return FileResponse(
str(resolved),
media_type=mime_type,
headers=headers,
)
responses = {
HTTP_200_OK: {"model": TranscriptionResponseModel},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.post("/v1/audio/transcriptions", responses=responses)
@self.app.post(
"/api/{path_provider:path}/audio/transcriptions", responses=responses
)
@self.app.post("/api/markitdown", responses=responses)
async def convert(
file: UploadFile,
path_provider: Optional[str] = None,
model: Annotated[Optional[str], Form()] = None,
provider: Annotated[Optional[str], Form()] = None,
prompt: Annotated[Optional[str], Form()] = "Transcribe this audio",
):
if path_provider is not None:
provider = path_provider
if provider is None:
provider = "MarkItDown"
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
kwargs = {"modalities": ["text"]}
try:
response = await self.client.chat.completions.create(
messages=prompt,
model=model,
provider=provider,
media=[[file.file, file.filename]],
**kwargs,
)
return {
"text": response.choices[0].message.content,
"model": response.model,
"provider": response.provider,
}
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, None, HTTP_404_NOT_FOUND)
except MissingAuthError as e:
logger.exception(e)
return ErrorResponse.from_exception(e, None, HTTP_401_UNAUTHORIZED)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, None, HTTP_500_INTERNAL_SERVER_ERROR
)
responses = {
HTTP_200_OK: {"model": TranscriptionResponseModel},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.get("/markitdown/{url:path}", responses=responses)
async def convert_url(
request: Request,
url: str,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
"""Convert a URL to Markdown using MarkItDown.
The full URL (including scheme) is passed in the path, e.g.:
GET /markitdown/https://example.com/page
Query strings are preserved by reading them from the incoming
request and re-appending them to the target URL, e.g.:
GET /markitdown/https://example.com/page?foo=bar
"""
# FastAPI strips the query string from the {url:path} parameter,
# so re-attach it from the incoming request when present.
query_string = request.url.query
if query_string and "?" not in url:
url = f"{url}?{query_string}"
elif query_string:
# url already contains a '?', append remaining params with '&'
url = f"{url}&{query_string}"
if not url.startswith(("http://", "https://")):
return ErrorResponse.from_message(
f"Invalid URL: {url}. URL must start with http:// or https://",
HTTP_422_UNPROCESSABLE_CONTENT,
)
try:
from g4f.integration.markitdown import MarkItDown
md = MarkItDown()
result = md.convert_url(url)
text = result.text_content
if asyncio.iscoroutine(text):
text = await text
return JSONResponse(
{"text": text, "title": result.title, "url": url},
)
except ImportError as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, None, HTTP_500_INTERNAL_SERVER_ERROR
)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, None, HTTP_500_INTERNAL_SERVER_ERROR
)
responses = {
HTTP_200_OK: {"content": {"audio/*": {}}},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.post("/v1/audio/speech", responses=responses)
@self.app.post("/api/{provider:path}/audio/speech", responses=responses)
async def generate_speech(
config: AudioSpeechConfig,
provider: Optional[str] = None,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
api_key = None
if credentials is not None and credentials.credentials != "secret":
api_key = credentials.credentials
if provider is None:
provider = config.provider
if provider is None:
provider = AppConfig.media_provider
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
try:
audio = filter_none(
voice=config.voice,
format=config.response_format,
language=config.language,
)
response = await self.client.chat.completions.create(
messages=[
{
"role": "user",
"content": f"{config.instrcutions} Text: {config.input}",
}
],
model=config.model,
provider=provider,
prompt=config.input,
api_key=api_key,
download_media=config.download_media,
**filter_none(
audio=audio if audio else None,
),
)
if response.choices[0].message.audio is not None:
response = base64.b64decode(response.choices[0].message.audio.data)
return Response(
response,
media_type=f"audio/{config.response_format.replace('mp3', 'mpeg')}",
)
elif isinstance(response.choices[0].message.content, AudioResponse):
response = response.choices[0].message.content.data
response = response.replace("/media", get_media_dir())
def delete_file():
try:
os.remove(response)
except Exception as e:
logger.exception(e)
return FileResponse(
response, background=BackgroundTask(delete_file)
)
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, None, HTTP_404_NOT_FOUND)
except MissingAuthError as e:
logger.exception(e)
return ErrorResponse.from_exception(e, None, HTTP_401_UNAUTHORIZED)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, None, HTTP_500_INTERNAL_SERVER_ERROR
)
@self.app.post(
"/v1/upload_cookies",
responses={
HTTP_200_OK: {"model": List[FileResponseModel]},
},
)
def upload_cookies(
files: List[UploadFile],
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
response_data = []
if not AppConfig.ignore_cookie_files:
for file in files:
try:
if (
file
and file.filename.endswith(".json")
or file.filename.endswith(".har")
):
filename = os.path.basename(file.filename)
with open(
os.path.join(get_cookies_dir(), filename), "wb"
) as f:
shutil.copyfileobj(file.file, f)
response_data.append({"filename": filename})
finally:
file.file.close()
read_cookie_files()
return response_data
@self.app.get(
"/images/{filename}",
responses={
HTTP_200_OK: {"content": {"image/*": {}}},
HTTP_404_NOT_FOUND: {},
},
)
@self.app.get(
"/media/{filename}",
responses={
HTTP_200_OK: {"content": {"image/*": {}, "audio/*": {}}, "video/*": {}},
HTTP_404_NOT_FOUND: {},
},
)
async def get_media(filename, request: Request, thumbnail: bool = False):
def get_timestamp(str):
m = re.match("^[0-9]+", str)
if m:
return int(m.group(0))
else:
return 0
target = os.path.join(get_media_dir(), os.path.basename(filename))
if thumbnail and has_pillow:
thumbnail_dir = os.path.join(get_media_dir(), "thumbnails")
thumbnail = os.path.join(thumbnail_dir, filename)
if not os.path.isfile(target):
other_name = os.path.join(
get_media_dir(), os.path.basename(quote_plus(filename))
)
if os.path.isfile(other_name):
target = other_name
ext = os.path.splitext(filename)[1][1:]
mime_type = EXTENSIONS_MAP.get(ext)
stat_result = SimpleNamespace()
stat_result.st_size = 0
stat_result.st_mtime = get_timestamp(filename)
if thumbnail and has_pillow and os.path.isfile(thumbnail):
stat_result.st_size = os.stat(thumbnail).st_size
elif not thumbnail and os.path.isfile(target):
stat_result.st_size = os.stat(target).st_size
headers = {
"cache-control": "public, max-age=31536000",
"last-modified": formatdate(stat_result.st_mtime, usegmt=True),
"etag": f'"{hashlib.md5(filename.encode()).hexdigest()}"',
**(
{
"content-length": str(stat_result.st_size),
}
if stat_result.st_size
else {}
),
**(
{}
if thumbnail or mime_type is None
else {
"content-type": mime_type,
}
),
}
response = FileResponse(
target,
headers=headers,
filename=filename,
)
try:
if_none_match = request.headers["if-none-match"]
etag = response.headers["etag"]
if etag in [tag.strip(" W/") for tag in if_none_match.split(",")]:
return NotModifiedResponse(response.headers)
except KeyError:
pass
if not os.path.isfile(target) and mime_type is not None:
source_url = get_source_url(str(request.query_params))
ssl = None
if source_url is None:
backend_url = os.environ.get("G4F_BACKEND_URL")
if backend_url:
source_url = f"{backend_url}/media/{filename}"
ssl = False
if source_url is not None:
try:
await copy_media([source_url], target=target, ssl=ssl)
debug.log(f"File copied from {source_url}")
except Exception as e:
debug.error(f"Download failed: {source_url}")
debug.error(e)
return RedirectResponse(url=source_url)
if thumbnail and has_pillow:
try:
if not os.path.isfile(thumbnail):
image = Image.open(target)
os.makedirs(thumbnail_dir, exist_ok=True)
process_image(image, save=thumbnail)
debug.log(f"Thumbnail created: {thumbnail}")
except Exception as e:
logger.exception(e)
if thumbnail and os.path.isfile(thumbnail):
result = thumbnail
else:
result = target
if not os.path.isfile(result):
return ErrorResponse.from_message("File not found", HTTP_404_NOT_FOUND)
async def stream():
with open(result, "rb") as file:
while True:
chunk = file.read(65536)
if not chunk:
break
yield chunk
return StreamingResponse(stream(), headers=headers)
@self.app.get(
"/thumbnail/{filename}",
responses={
HTTP_200_OK: {"content": {"image/*": {}, "audio/*": {}}, "video/*": {}},
HTTP_404_NOT_FOUND: {},
},
)
async def get_media_thumbnail(filename: str, request: Request):
return await get_media(filename, request, True)
@self.app.get("/logs", response_class=HTMLResponse)
async def logs_inspector():
return HTMLResponse(_LOGS_HTML)
@self.app.get("/api/logs")
async def get_logs(limit: int = 500, offset: int = 0):
entries = list(_request_log)
total = len(entries)
start = max(0, total - limit - offset)
end = total - offset if offset < total else total
page = list(reversed(entries[start:end]))
return JSONResponse(
{"total": total, "entries": page}, headers={"Cache-Control": "no-store"}
)
@self.app.delete("/api/logs")
async def clear_logs():
_request_log.clear()
return JSONResponse({"status": "cleared"})
def format_exception(
e: Union[Exception, str],
config: Union[ChatCompletionsConfig, ImageGenerationConfig] = None,
image: bool = False,
) -> str:
provider = AppConfig.media_provider if image else AppConfig.provider
model = AppConfig.model
if config is not None:
if config.provider is not None:
provider = config.provider
if config.model is not None:
model = config.model
if isinstance(e, str):
message = e
else:
message = f"{e.__class__.__name__}: {e}"
return json.dumps(
{
"error": {"message": message},
**filter_none(
model=model, provider=getattr(provider, "__name__", provider)
),
}
)
def run_api(
host: str = "0.0.0.0",
port: int = None,
bind: str = None,
debug: bool = False,
use_colors: bool = None,
**kwargs,
) -> None:
print(
f"Starting server... [g4f v-{g4f.version.utils.current_version}]"
+ (" (debug)" if debug else "")
)
if use_colors is None:
use_colors = debug
if bind is not None:
host, port = bind.split(":")
if port is None:
port = DEFAULT_PORT
if AppConfig.demo and debug:
method = "create_app_with_demo_and_debug"
elif AppConfig.gui and debug:
method = "create_app_with_gui_and_debug"
else:
method = "create_app_debug" if debug else "create_app"
uvicorn.run(
f"g4f.api:{method}",
host=host,
port=int(port),
factory=True,
use_colors=use_colors,
**filter_none(**kwargs),
)
from __future__ import annotations
import inspect
import logging
import json
import asyncio
import uvicorn
import secrets
import os
import re
import shutil
import time
from collections import deque
from email.utils import formatdate
import os.path
import hashlib
import base64
from contextlib import asynccontextmanager
from urllib.parse import quote_plus
from fastapi import FastAPI, Response, Request, UploadFile, Form, Depends, Header
from fastapi.responses import (
StreamingResponse,
RedirectResponse,
HTMLResponse,
JSONResponse,
FileResponse,
)
from fastapi.exceptions import RequestValidationError
from fastapi.security import APIKeyHeader
from starlette.exceptions import HTTPException
from starlette.status import (
HTTP_200_OK,
HTTP_404_NOT_FOUND,
HTTP_401_UNAUTHORIZED,
HTTP_403_FORBIDDEN,
HTTP_429_TOO_MANY_REQUESTS,
HTTP_500_INTERNAL_SERVER_ERROR,
)
try:
from starlette.status import HTTP_422_UNPROCESSABLE_CONTENT
except ImportError:
HTTP_422_UNPROCESSABLE_CONTENT = 422
from starlette.staticfiles import NotModifiedResponse
from fastapi.encoders import jsonable_encoder
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials, HTTPBasic
from fastapi.middleware.cors import CORSMiddleware
from starlette.responses import FileResponse
from starlette.background import BackgroundTask
try:
from a2wsgi import WSGIMiddleware
has_a2wsgi = True
except ImportError:
has_a2wsgi = False
try:
from PIL import Image
has_pillow = True
except ImportError:
has_pillow = False
from types import SimpleNamespace
from typing import Union, Optional, List
try:
from typing import Annotated
except ImportError:
class Annotated:
pass
try:
from zendriver import util
has_nodriver = True
except ImportError:
has_nodriver = False
import g4f
import g4f.debug
from g4f.client import AsyncClient, ChatCompletion, ImagesResponse
from g4f.providers.response import BaseConversation, JsonConversation
from g4f.client.helper import filter_none
from g4f.config import DEFAULT_PORT, DEFAULT_TIMEOUT, DEFAULT_STREAM_TIMEOUT
from g4f.image import EXTENSIONS_MAP, is_data_an_media, process_image
from g4f.image.copy_images import get_media_dir, copy_media, get_source_url
from g4f.errors import (
ProviderNotFoundError,
ModelNotFoundError,
MissingAuthError,
NoValidHarFileError,
MissingRequirementsError,
RateLimitError,
)
from g4f.cookies import read_cookie_files, get_cookies_dir
from g4f.providers.types import ProviderType
from g4f.providers.response import AudioResponse
from g4f.providers.any_provider import AnyProvider
from g4f.providers.any_model_map import (
model_map,
vision_models,
image_models,
audio_models,
video_models,
)
from g4f.config import AppConfig
from g4f.client.factory import AbstractClientFactory
from g4f import Provider
from g4f.Provider import ProviderUtils
from g4f.gui import get_gui_app
from .stubs import (
ChatCompletionsConfig,
ImageGenerationConfig,
ResponsesConfig,
MessagesConfig,
ProviderResponseModel,
ModelResponseModel,
ErrorResponseModel,
ProviderResponseDetailModel,
FileResponseModel,
TranscriptionResponseModel,
AudioSpeechConfig,
)
from g4f import debug
try:
from g4f.gui.server.crypto import create_or_read_keys, decrypt_data, get_session_key
has_crypto = True
except ImportError:
has_crypto = False
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Request / response log store
# ---------------------------------------------------------------------------
_MAX_LOG_ENTRIES = 250
_MAX_BODY_LOG_SIZE = 1024 * 1024 # 1 MB
_SENSITIVE_HEADERS = {
"authorization",
"g4f-api-key",
"cookie",
"set-cookie",
"x-api-key",
}
_request_log: deque = deque(maxlen=_MAX_LOG_ENTRIES)
_log_id_counter: int = 0
def _sanitize_headers(headers: dict) -> dict:
return {
k: ("***" if k.lower() in _SENSITIVE_HEADERS else v) for k, v in headers.items()
}
def _try_parse_body(body_bytes: bytes, content_type: str):
if not body_bytes:
return None
if len(body_bytes) > _MAX_BODY_LOG_SIZE:
return f"<{len(body_bytes)} bytes – truncated>"
if "application/json" in content_type:
try:
return json.loads(body_bytes)
except Exception:
pass
try:
return body_bytes.decode("utf-8", errors="replace")
except Exception:
return f"<binary {len(body_bytes)} bytes>"
_LOGS_HTML = """<!doctype html>
<html lang="en">
<head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>g4f – Request Log</title>
<style>
*{box-sizing:border-box;margin:0;padding:0}
body{font-family:system-ui,-apple-system,sans-serif;background:#0d1117;color:#c9d1d9;min-height:100vh;font-size:14px}
a{color:#58a6ff}
.header{padding:14px 20px;border-bottom:1px solid #21262d;display:flex;align-items:center;gap:14px}
.header h1{font-size:15px;font-weight:600;color:#f0f6fc}
.header .sub{font-size:12px;color:#8b949e}
.toolbar{padding:10px 20px;border-bottom:1px solid #21262d;display:flex;gap:10px;align-items:center;flex-wrap:wrap}
.toolbar input[type=text]{flex:1;min-width:180px;background:#161b22;border:1px solid #30363d;color:#c9d1d9;padding:5px 10px;border-radius:6px;font-size:13px;outline:none}
.toolbar input[type=text]:focus{border-color:#58a6ff}
.toolbar label{display:flex;align-items:center;gap:5px;font-size:13px;color:#8b949e;cursor:pointer;user-select:none}
.btn{padding:5px 14px;border-radius:6px;border:1px solid #30363d;cursor:pointer;font-size:13px;background:#21262d;color:#c9d1d9}
.btn:hover{background:#30363d}
.btn-danger{border-color:#6e3435;background:#1c1214;color:#ffa198}
.btn-danger:hover{background:#6e3435}
.meta{margin-left:auto;font-size:12px;color:#6e7681}
.table-wrap{overflow-x:auto;padding:0 20px 40px}
table{width:100%;border-collapse:collapse;margin-top:14px;font-size:13px}
th{padding:6px 8px;text-align:left;color:#8b949e;font-weight:500;border-bottom:1px solid #21262d;white-space:nowrap}
td{padding:6px 8px;border-bottom:1px solid #161b22;white-space:nowrap;max-width:320px;overflow:hidden;text-overflow:ellipsis;vertical-align:middle}
tbody tr{cursor:pointer}
tbody tr:hover td{background:#161b22}
.GET{color:#3fb950}.POST{color:#58a6ff}.PUT{color:#e3b341}.DELETE{color:#f85149}.PATCH{color:#d2a8ff}
.s2{color:#3fb950}.s3{color:#58a6ff}.s4{color:#e3b341}.s5{color:#f85149}
.tag{display:inline-block;font-size:10px;padding:1px 6px;border-radius:10px;font-weight:500}
.tag-sse{background:#0d2044;color:#79c0ff}
.tag-body{background:#0d2820;color:#56d364}
.tag-empty{background:#1c2128;color:#6e7681}
.overlay{display:none;position:fixed;inset:0;background:rgba(0,0,0,.65);z-index:200;padding:24px;align-items:flex-start;justify-content:center;overflow-y:auto}
.overlay.active{display:flex}
.modal{background:#161b22;border:1px solid #30363d;border-radius:10px;width:100%;max-width:1040px;display:flex;flex-direction:column}
.modal-head{padding:14px 18px;border-bottom:1px solid #21262d;display:flex;justify-content:space-between;align-items:center;gap:10px}
.modal-head h2{font-size:13px;font-weight:600;color:#f0f6fc;overflow:hidden;text-overflow:ellipsis;white-space:nowrap;font-family:ui-monospace,monospace}
.modal-close{background:none;border:none;color:#6e7681;font-size:18px;cursor:pointer;line-height:1;padding:2px 6px;flex-shrink:0}
.modal-close:hover{color:#c9d1d9}
.modal-grid{display:grid;grid-template-columns:1fr 1fr 1fr}
.panel{padding:16px 18px;display:flex;flex-direction:column;gap:8px}
.panel:first-child{border-right:1px solid #21262d}
.panel:not(:last-child){border-right:1px solid #21262d}
.panel-title{font-size:11px;font-weight:600;text-transform:uppercase;letter-spacing:.06em;color:#6e7681}
.panel pre{background:#0d1117;border:1px solid #21262d;border-radius:6px;padding:12px;font-size:12px;line-height:1.5;overflow:auto;max-height:440px;white-space:pre-wrap;word-break:break-all;color:#c9d1d9;margin:0;font-family:ui-monospace,monospace}
@media(max-width:640px){.modal-grid{grid-template-columns:1fr}.panel:first-child{border-right:none;border-bottom:1px solid #21262d}.panel:not(:last-child){border-right:none;border-bottom:1px solid #21262d}}
</style>
</head>
<body>
<div class="header">
<h1>g4f Request Log</h1>
<span class="sub">last 500 entries · <a href="/v1">/v1 API</a></span>
</div>
<div class="toolbar">
<input type="text" id="q" placeholder="Filter by path, method, status, user…" oninput="render()">
<label><input type="checkbox" id="auto" checked onchange="toggleAuto()"> Auto‑refresh (3s)</label>
<button class="btn btn-danger" onclick="clearLogs()">Clear</button>
<span class="meta" id="meta"></span>
</div>
<div class="table-wrap"><table>
<thead><tr>
<th>#</th><th>Time (UTC)</th><th>Method</th><th>Path</th>
<th>Status</th><th>ms</th><th>User</th><th>Body</th>
</tr></thead>
<tbody id="tb"></tbody>
</table></div>
<div class="overlay" id="ov" onclick="overlayClick(event)">
<div class="modal" id="mod">
<div class="modal-head">
<h2 id="mtitle">–</h2>
<button class="modal-close" onclick="closeModal()">✕</button>
</div>
<div class="modal-grid">
<div class="panel"><div class="panel-title">Request</div><pre id="preq"></pre></div>
<div class="panel"><div class="panel-title">Response</div><pre id="pres"></pre></div>
<div class="panel"><div class="panel-title">Response Headers</div><pre id="presh"></pre></div>
</div>
</div>
</div>
<script>
'use strict';
var all=[], timer=null;
function sc(s){return s>=500?'s5':s>=400?'s4':s>=300?'s3':'s2';}
function btag(e){if(e.streaming&&e.response_body==null)return\'<span class="tag tag-sse">SSE…</span>\';if(e.response_body!=null)return\'<span class="tag tag-body">\'+(e.streaming?\'SSE\':\'body\')+\'</span>\';return\'<span class="tag tag-empty">–</span>\';}
function esc(s){return String(s??\'\'). replace(/&/g,\'&\').replace(/</g,\'<\').replace(/>/g,\'>\');}
function fmt(v){if(v==null)return\'(empty)\';if(typeof v===\'object\')return JSON.stringify(v,null,2);return String(v);}
async function load(){
try{var r=await fetch(\'/api/logs?limit=500\', { credentials: 'include' });if(!r.ok)return;var d=await r.json();all=d.entries||[];render();}catch(e){}
}
function render(){
var q=document.getElementById(\'q\').value.trim().toLowerCase();
var rows=q?all.filter(function(e){return(e.method+\' \'+e.path+\' \'+e.status+\' \'+(e.user||\'\')).toLowerCase().includes(q);}):all;
document.getElementById(\'meta\').textContent=rows.length+\' / \'+all.length+\' entries\';
document.getElementById(\'tb\').innerHTML=rows.map(function(e){
var t=(e.timestamp||\'\').replace(\'T\',\' \').replace(\'Z\',\'\');
var p=esc(e.path+(e.query?\'?\'+e.query:\'\'));
return\'<tr onclick="detail(\'+e.id+\')">\'+
\'<td style="color:#484f58">\'+e.id+\'</td>\'+
\'<td style="color:#6e7681;font-size:12px">\'+esc(t)+\'</td>\'+
\'<td class="\'+esc(e.method)+\'">\'+esc(e.method)+\'</td>\'+
\'<td title="\'+p+\'">\'+p+\'</td>\'+
\'<td class="\'+sc(e.status)+\'">\'+e.status+\'</td>\'+
\'<td style="color:#8b949e">\'+e.duration_ms+\'</td>\'+
\'<td style="color:#6e7681">\'+esc(e.user||\'\')+\'</td>\'+
\'<td>\'+btag(e)+\'</td>\'+
\'</tr>\';
}).join(\'\');
}
function detail(id){
var e=all.find(function(x){return x.id===id;});
if(!e)return;
document.getElementById(\'mtitle\').textContent=\'#\'+e.id+\' \'+e.method+\' \'+e.path+(e.query?\'?\'+e.query:\'\')+\' \u2192 \'+e.status+\' (\'+e.duration_ms+\'ms)\';
var req=\'\';
if(e.request_headers){req+=\'Headers:\\n\';for(var k in e.request_headers)req+=\' \'+k+\': \'+e.request_headers[k]+\'\\n\';}
if(e.request_body!=null)req+=\'\\nBody:\\n\'+fmt(e.request_body);
document.getElementById(\'preq\').textContent=req||\'(none)\';
document.getElementById(\'pres\').textContent=e.response_body!=null?fmt(e.response_body):(e.streaming?\'(streaming – collecting…)\':\'(empty)\');
var resh=\'\';
if(e.response_headers){resh+=\'Headers:\\n\';for(var k in e.response_headers)resh+=\' \'+k+\': \'+e.response_headers[k]+\'\\n\';}
document.getElementById(\'presh\').textContent=resh||\'(none)\';
document.getElementById(\'ov\').classList.add(\'active\');
}
function closeModal(){document.getElementById(\'ov\').classList.remove(\'active\');}
function overlayClick(ev){if(ev.target===document.getElementById(\'ov\'))closeModal();}
document.addEventListener(\'keydown\',function(e){if(e.key===\'Escape\')closeModal();});
async function clearLogs(){await fetch(\'/api/logs\',{method:\'DELETE\'});all=[];render();}
function toggleAuto(){clearInterval(timer);timer=null;if(document.getElementById(\'auto\').checked)timer=setInterval(load,3000);}
load();timer=setInterval(load,3000);
</script>
</body></html>"""
@asynccontextmanager
async def lifespan(app: FastAPI):
# Read cookie files if not ignored
if not AppConfig.ignore_cookie_files:
read_cookie_files()
else:
AppConfig.load_from_env()
yield
if has_nodriver:
for browser in util.get_registered_instances():
if browser.connection:
await browser.stop()
lock_file = os.path.join(get_cookies_dir(), ".browser_is_open")
if os.path.exists(lock_file):
try:
os.remove(lock_file)
except Exception as e:
debug.error(f"Failed to remove lock file {lock_file}:", e)
_LOG_SKIP_PREFIXES = ("/images/", "/media/", "/thumbnail/", "/dist/", "/.well-known/")
_LOG_SKIP_EXACT = {"/api/logs", "/logs", "/favicon.ico"}
def create_app():
app = FastAPI(lifespan=lifespan)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["*"],
)
@app.middleware("http")
async def log_requests(request: Request, call_next):
global _log_id_counter
path = request.url.path
if (
any(path.startswith(p) for p in _LOG_SKIP_PREFIXES)
or path in _LOG_SKIP_EXACT
):
return await call_next(request)
qs = f"?{request.url.query}" if request.url.query else ""
user = request.headers.get("x-user", "")
user_info = f" user={user}" if user else ""
logger.debug("→ %s %s%s%s", request.method, path, qs, user_info)
# Capture request body (Starlette caches after first read)
req_body_bytes = await request.body()
req_body = _try_parse_body(
req_body_bytes, request.headers.get("content-type", "")
)
start = time.monotonic()
response = await call_next(request)
duration_ms = round((time.monotonic() - start) * 1000)
resp_content_type = response.headers.get("content-type", "")
is_streaming = "text/event-stream" in resp_content_type
log_entry: dict = {}
resp_headers_log = _sanitize_headers(dict(response.headers))
if not is_streaming:
chunks: list[bytes] = []
async for chunk in response.body_iterator:
chunks.append(chunk)
resp_body_bytes = b"".join(chunks)
resp_body = _try_parse_body(resp_body_bytes, resp_content_type)
# Reconstruct response so it can still be sent to the client
resp_headers = {
k: v
for k, v in response.headers.items()
if k.lower() != "content-length"
}
response = Response(
content=resp_body_bytes,
status_code=response.status_code,
headers=resp_headers,
media_type=response.media_type,
)
else:
# Tee the streaming iterator: forward chunks to client AND accumulate for log
sse_chunks: list[bytes] = []
resp_body = None
orig_iterator = response.body_iterator
async def tee_iterator():
async for chunk in orig_iterator:
if isinstance(chunk, bytes):
sse_chunks.append(chunk)
else:
sse_chunks.append(chunk.encode("utf-8", errors="replace"))
yield chunk
# After iteration completes, parse and store the full SSE body
raw = b"".join(sse_chunks)
parsed = _try_parse_body(raw, "text/plain")
log_entry["response_body"] = parsed
response.body_iterator = tee_iterator()
level = logging.WARNING if response.status_code >= 400 else logging.INFO
logger.log(
level,
"%s %s%s → %d (%dms)%s",
request.method,
path,
qs,
response.status_code,
duration_ms,
user_info,
)
_log_id_counter += 1
log_entry.update(
{
"id": _log_id_counter,
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"method": request.method,
"path": path,
"query": request.url.query or None,
"status": response.status_code,
"duration_ms": duration_ms,
"user": user or None,
"streaming": is_streaming,
"request_headers": _sanitize_headers(dict(request.headers)),
"request_body": req_body,
"response_headers": resp_headers_log,
"response_body": resp_body, # None for SSE until iterator completes
}
)
_request_log.append(log_entry)
return response
api = Api(app)
api.register_routes()
api.register_authorization()
api.register_validation_exception_handler()
if AppConfig.gui:
if not has_a2wsgi:
raise MissingRequirementsError(
"a2wsgi is required for GUI. Install it with: pip install a2wsgi"
)
gui_app = WSGIMiddleware(
get_gui_app(AppConfig.demo, AppConfig.timeout, AppConfig.stream_timeout)
)
app.mount("/", gui_app)
if AppConfig.ignored_providers:
for provider in AppConfig.ignored_providers:
if provider in ProviderUtils.convert:
ProviderUtils.convert[provider].working = False
return app
def create_app_debug():
g4f.debug.logging = True
return create_app()
def create_app_with_gui_and_debug():
g4f.debug.logging = True
AppConfig.gui = True
return create_app()
def create_app_with_demo_and_debug():
g4f.debug.logging = True
AppConfig.gui = True
AppConfig.demo = True
return create_app()
class ErrorResponse(Response):
media_type = "application/json"
@classmethod
def from_exception(
cls,
exception: Exception,
config: Union[ChatCompletionsConfig, ImageGenerationConfig] = None,
status_code: int = HTTP_500_INTERNAL_SERVER_ERROR,
):
return cls(format_exception(exception, config), status_code)
@classmethod
def from_message(
cls,
message: str,
status_code: int = HTTP_500_INTERNAL_SERVER_ERROR,
headers: dict = None,
):
return cls(format_exception(message), status_code, headers=headers)
def render(self, content) -> bytes:
return str(content).encode(errors="ignore")
def update_headers(
request: Request, new_api_key: str = None, user: str = None
) -> Request:
new_headers = request.headers.mutablecopy()
if new_api_key:
new_headers["authorization"] = f"Bearer {new_api_key}"
else:
del new_headers["authorization"]
if user:
new_headers["x-user"] = user
request.scope["headers"] = new_headers.raw
delattr(request, "_headers")
return request
class Api:
def __init__(self, app: FastAPI) -> None:
self.app = app
self.client = AsyncClient()
self.get_g4f_api_key = APIKeyHeader(name="g4f-api-key")
self.conversations: dict[str, dict[str, BaseConversation]] = {}
security = HTTPBearer(auto_error=False)
basic_security = HTTPBasic()
async def get_username(self, request: Request) -> str:
credentials = await self.basic_security(request)
current_password_bytes = credentials.password.encode()
is_correct_password = secrets.compare_digest(
current_password_bytes, AppConfig.g4f_api_key.encode()
)
if not is_correct_password:
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Basic"},
)
return credentials.username
def register_authorization(self):
if AppConfig.g4f_api_key:
print(
"Register authentication key:",
"".join(["*" for _ in range(len(AppConfig.g4f_api_key))]),
)
if has_crypto:
private_key, _ = create_or_read_keys()
session_key = get_session_key()
def _requires_api_key(path: str, demo: bool) -> bool:
"""Return ``True`` when *path* must present a G4F API key."""
return (
path.startswith("/v1/")
or (path.startswith("/api/") and path != "/api/logs")
or (path.startswith("/pa/") and not demo)
or (demo and path in ["/backend-api/v2/upload_cookies"])
)
@self.app.middleware("http")
async def authorization(request: Request, call_next):
user = None
if (
request.method != "OPTIONS"
and AppConfig.g4f_api_key is not None
or AppConfig.demo
):
update_authorization = False
try:
user_g4f_api_key = await self.get_g4f_api_key(request)
except HTTPException:
user_g4f_api_key = getattr(
await self.security(request), "credentials", None
)
update_authorization = True
if user_g4f_api_key:
user_g4f_api_key = user_g4f_api_key.split()
country = request.headers.get("Cf-Ipcountry", "")
if AppConfig.demo and user is None:
ip = request.headers.get("X-Forwarded-For", "")[:4].strip(":.")
user = request.headers.get("x-user", ip)
user = f"{country}:{user}" if country else user
if (
AppConfig.g4f_api_key is None
or not user_g4f_api_key
or not secrets.compare_digest(
AppConfig.g4f_api_key, user_g4f_api_key[0]
)
):
if has_crypto and user_g4f_api_key:
try:
expires, user = decrypt_data(
private_key, user_g4f_api_key[0]
).split(":", 1)
except Exception:
try:
data = json.loads(
decrypt_data(session_key, user_g4f_api_key[0])
)
debug.log(f"Decrypted G4F API key data: {data}")
expires = (
int(decrypt_data(private_key, data.pop("data")))
+ 86400
)
user = data.get("user", user)
if not user or "referrer" not in data:
raise ValueError("User not found")
except Exception:
return ErrorResponse.from_message(
f"Invalid G4F API key", HTTP_401_UNAUTHORIZED
)
user = f"{country}:{user}" if country else user
expires = int(expires) - int(time.time())
hours, remainder = divmod(expires, 3600)
minutes, seconds = divmod(remainder, 60)
if expires < 0:
debug.log(f"G4F API key expired for user '{user}'")
return ErrorResponse.from_message(
"G4F API key expired", HTTP_401_UNAUTHORIZED
)
count = 0
for char in user:
if char.isupper():
count += 1
if count >= 6:
debug.log(f"Invalid user name (screaming): '{user}'")
return ErrorResponse.from_message(
"Invalid user name (screaming)", HTTP_401_UNAUTHORIZED
)
debug.log(
f"User: '{user}' G4F API key expires in {hours}h {minutes}m {seconds}s"
)
elif user is None:
user = "admin"
path = request.url.path
if _requires_api_key(path, AppConfig.demo):
if request.method != "OPTIONS" and not path.endswith("/models"):
if not user_g4f_api_key:
return ErrorResponse.from_message(
"G4F API key required", HTTP_401_UNAUTHORIZED
)
if AppConfig.g4f_api_key is None and user is None:
return ErrorResponse.from_message(
"Invalid G4F API key", HTTP_403_FORBIDDEN
)
elif path == "/logs" or (
not AppConfig.demo
and not path.startswith("/images/")
and not path.startswith("/media/")
):
if user_g4f_api_key:
if user is None:
return ErrorResponse.from_message(
"Invalid G4F API key", HTTP_403_FORBIDDEN
)
elif (
path.startswith("/backend-api/")
or path.startswith("/chat/")
or path.startswith("/playground/")
or path in ["/logs"]
):
try:
new_user = await self.get_username(request)
if user is None:
user = new_user
except HTTPException as e:
return ErrorResponse.from_message(
e.detail, e.status_code, e.headers
)
if user_g4f_api_key and update_authorization:
new_api_key = user_g4f_api_key.pop()
if secrets.compare_digest(AppConfig.g4f_api_key, new_api_key):
new_api_key = None
else:
new_api_key = None
request = update_headers(request, new_api_key, user)
response = await call_next(request)
return response
def register_validation_exception_handler(self):
@self.app.exception_handler(RequestValidationError)
async def validation_exception_handler(
request: Request, exc: RequestValidationError
):
details = exc.errors()
modified_details = []
for error in details:
debug.log(
f"Validation error: {error['loc']} - {error['msg']} ({error['type']})"
)
modified_details.append(
{
"loc": error["loc"],
"message": error["msg"],
"type": error["type"],
}
)
return JSONResponse(
status_code=HTTP_422_UNPROCESSABLE_CONTENT,
content=jsonable_encoder({"detail": modified_details}),
)
def register_routes(self):
if not AppConfig.gui:
@self.app.get("/")
async def read_root():
return RedirectResponse("/v1", 302)
@self.app.get("/v1")
async def read_root_v1():
return HTMLResponse(
"g4f API: Go to "
'<a href="/v1/models">models</a>, '
'<a href="/v1/chat/completions">chat/completions</a>, '
'<a href="/v1/responses">responses</a> (OpenAI), '
'<a href="/v1/messages">messages</a> (Anthropic), or '
'<a href="/v1/media/generate">media/generate</a> <br><br>'
"Open Swagger UI at: "
'<a href="/docs">/docs</a>'
)
@self.app.get(
"/v1/models",
responses={
HTTP_200_OK: {"model": List[ModelResponseModel]},
},
)
async def models():
return {
"object": "list",
"data": [
{
"id": model,
"object": "model",
"created": 0,
"owned_by": "",
"image": isinstance(model, g4f.models.ImageModel),
"vision": isinstance(model, g4f.models.VisionModel),
"provider": False,
}
for model in AnyProvider.get_models()
]
+ [
{
"id": provider_name,
"object": "model",
"created": 0,
"owned_by": getattr(provider, "label", ""),
"image": bool(getattr(provider, "image_models", False)),
"vision": bool(getattr(provider, "vision_models", False)),
"provider": True,
}
for provider_name, provider in ProviderUtils.convert.items()
if provider.working
],
}
@self.app.get(
"/api/{provider:path}/models",
responses={
HTTP_200_OK: {"model": List[ModelResponseModel]},
},
)
async def models(
provider: str,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
if not hasattr(provider, "get_models"):
models = getattr(provider, "models", [])
elif credentials is not None and credentials.credentials != "secret":
models = provider.get_models(api_key=credentials.credentials)
else:
models = provider.get_models()
if inspect.isawaitable(models):
models = await models
return {
"object": "list",
"data": [
{
"id": model.get("id") if isinstance(model, dict) else model,
"object": "model",
"created": 0,
"owned_by": getattr(provider, "label", provider.__name__),
"image": (model.get("id") if isinstance(model, dict) else model)
in getattr(provider, "image_models", []),
"vision": (
model.get("id") if isinstance(model, dict) else model
)
in getattr(provider, "vision_models", []),
"audio": (model.get("id") if isinstance(model, dict) else model)
in getattr(provider, "audio_models", []),
"video": (model.get("id") if isinstance(model, dict) else model)
in getattr(provider, "video_models", []),
"type": "image"
if (model.get("id") if isinstance(model, dict) else model)
in getattr(provider, "image_models", [])
else "chat",
"count": getattr(provider, "models_count", {}).get(
model.get("id") if isinstance(model, dict) else model, 0
),
**(model if isinstance(model, dict) else {}),
}
for model in (
models.values() if isinstance(models, dict) else models
)
],
}
# quota endpoint mimics backend-api/v2/quota but exposed on public API
@self.app.get("/api/{provider:path}/quota")
async def provider_quota(
provider: str,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
if not hasattr(provider, "get_quota"):
return ErrorResponse.from_message(
"Provider doesn't support get_quota", HTTP_500_INTERNAL_SERVER_ERROR
)
try:
if credentials is not None and credentials.credentials != "secret":
usage = await provider.get_quota(api_key=credentials.credentials)
else:
usage = await provider.get_quota()
return usage
except MissingAuthError as e:
return ErrorResponse.from_message(
f"{type(e).__name__}: {e}", HTTP_401_UNAUTHORIZED
)
except Exception as e:
return ErrorResponse.from_message(
f"{type(e).__name__}: {e}", HTTP_500_INTERNAL_SERVER_ERROR
)
@self.app.get(
"/v1/models/{model_name}",
responses={
HTTP_200_OK: {"model": ModelResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
@self.app.post(
"/v1/models/{model_name}",
responses={
HTTP_200_OK: {"model": ModelResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
async def model_info(model_name: str) -> ModelResponseModel:
if model_name in g4f.models.ModelUtils.convert:
model_info = g4f.models.ModelUtils.convert[model_name]
return JSONResponse(
{
"id": model_name,
"object": "model",
"created": 0,
"owned_by": model_info.base_provider,
}
)
return ErrorResponse.from_message(
"The model does not exist.", HTTP_404_NOT_FOUND
)
responses = {
HTTP_200_OK: {"model": ChatCompletion},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_422_UNPROCESSABLE_CONTENT: {"model": ErrorResponseModel},
HTTP_429_TOO_MANY_REQUESTS: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.post("/v1/chat/completions", responses=responses)
@self.app.post("/api/{provider:path}/chat/completions", responses=responses)
@self.app.post(
"/api/{provider}/{conversation_id}/chat/completions", responses=responses
)
async def chat_completions(
config: ChatCompletionsConfig,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
provider: str = None,
conversation_id: str = None,
x_user: Annotated[str | None, Header()] = None,
):
if provider is not None:
config.provider = provider
if config.provider is None:
config.provider = AppConfig.provider
try:
provider = AbstractClientFactory.create_provider(None, config.provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
try:
if config.conversation_id is None:
config.conversation_id = conversation_id
if config.timeout is None:
config.timeout = AppConfig.timeout
if config.stream_timeout is None and config.stream:
config.stream_timeout = AppConfig.stream_timeout
if credentials is not None and credentials.credentials != "secret":
config.api_key = credentials.credentials
conversation = config.conversation
if conversation:
conversation = JsonConversation(**conversation)
elif config.conversation_id is not None and config.provider is not None:
if config.conversation_id in self.conversations:
if (
config.provider
in self.conversations[config.conversation_id]
):
conversation = self.conversations[config.conversation_id][
config.provider
]
if config.image is not None:
try:
is_data_an_media(config.image)
except ValueError as e:
return ErrorResponse.from_message(
f"The image you send must be a data URI. Example: data:image/jpeg;base64,...",
status_code=HTTP_422_UNPROCESSABLE_CONTENT,
)
if config.media is None:
config.media = config.images
if config.media is not None:
for image in config.media:
try:
is_data_an_media(image[0], image[1])
except ValueError as e:
example = json.dumps(
{
"media": [
["data:image/jpeg;base64,...", "filename.jpg"]
]
}
)
return ErrorResponse.from_message(
f"The media you send must be a data URIs. Example: {example}",
status_code=HTTP_422_UNPROCESSABLE_CONTENT,
)
# Create the completion response
response = self.client.chat.completions.create(
**filter_none(
**{
"model": AppConfig.model,
"provider": AppConfig.provider,
"proxy": AppConfig.proxy,
**(
config.model_dump(exclude_none=True)
if hasattr(config, "model_dump")
else config.dict(exclude_none=True)
),
**{
"provider": provider,
"conversation_id": None,
"conversation": conversation,
"user": x_user,
},
},
ignored=AppConfig.ignored_providers,
),
)
if not config.stream:
result = await response
return Response(
content=result.model_dump_json()
if hasattr(result, "model_dump_json")
else result.json(),
media_type="application/json",
headers=getattr(result, "_headers").get_dict()
if hasattr(result, "_headers")
else None,
)
first_chunk = await response.__anext__()
async def streaming():
yield f"data: {first_chunk.model_dump_json() if hasattr(first_chunk, 'model_dump_json') else first_chunk.json()}\n\n"
try:
async for chunk in response:
if isinstance(chunk, BaseConversation):
if (
config.conversation_id is not None
and config.provider is not None
):
if config.conversation_id not in self.conversations:
self.conversations[config.conversation_id] = {}
self.conversations[config.conversation_id][
config.provider
] = chunk
else:
yield f"data: {chunk.model_dump_json() if hasattr(chunk, 'model_dump_json') else chunk.json()}\n\n"
except GeneratorExit:
pass
except RateLimitError as e:
debug.error(e)
yield f"data: {format_exception(e, config)}\n\n"
except Exception as e:
logger.exception(e)
yield f"data: {format_exception(e, config)}\n\n"
yield "data: [DONE]\n\n"
headers = (
getattr(first_chunk, "_headers").get_dict()
if hasattr(first_chunk, "_headers")
else {}
)
headers = {
k.encode("latin-1", "ignore")
.decode("latin-1"): v.encode("latin-1", "ignore")
.decode("latin-1")
for k, v in headers.items()
}
return StreamingResponse(
streaming(), media_type="text/event-stream", headers=headers
)
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
except (MissingAuthError, NoValidHarFileError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
except RateLimitError as e:
return ErrorResponse.from_exception(
e, config, HTTP_429_TOO_MANY_REQUESTS
)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, config, HTTP_500_INTERNAL_SERVER_ERROR
)
# ------------------------------------------------------------------ #
# OpenAI Responses API (/v1/responses) #
# https://platform.openai.com/docs/api-reference/responses #
# ------------------------------------------------------------------ #
@self.app.post("/v1/responses", responses=responses)
@self.app.post("/api/{provider:path}/responses", responses=responses)
async def create_response(
config: ResponsesConfig,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
provider: str = None,
x_user: Annotated[str | None, Header()] = None,
):
if provider is not None:
config.provider = provider
if config.provider is None:
config.provider = AppConfig.provider
try:
provider = AbstractClientFactory.create_provider(None, config.provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
try:
if config.timeout is None:
config.timeout = AppConfig.timeout
if config.stream_timeout is None and config.stream:
config.stream_timeout = AppConfig.stream_timeout
if credentials is not None and credentials.credentials != "secret":
config.api_key = credentials.credentials
# Normalize `input` into a messages list.
messages = config.input
if isinstance(messages, str):
messages = [{"role": "user", "content": messages}]
if config.instructions:
messages = [
{"role": "system", "content": config.instructions},
*messages,
]
response = self.client.chat.completions.create(
**filter_none(
**{
"model": AppConfig.model,
"provider": AppConfig.provider,
"proxy": AppConfig.proxy,
**(
config.model_dump(exclude_none=True)
if hasattr(config, "model_dump")
else config.dict(exclude_none=True)
),
**{
"provider": provider,
"messages": messages,
"user": x_user,
},
},
ignored=AppConfig.ignored_providers,
),
)
if not config.stream:
result = await response
text = result.choices[0].message.content if result.choices else ""
usage = getattr(result, "usage", None)
if usage is not None and hasattr(usage, "model_dump"):
usage = usage.model_dump()
elif usage is not None and hasattr(usage, "dict"):
usage = usage.dict()
return JSONResponse(
{
"id": getattr(
result, "id", f"resp_{secrets.token_hex(12)}"
),
"object": "response",
"created_at": getattr(result, "created", int(time.time())),
"model": getattr(result, "model", config.model),
"provider": getattr(provider, "__name__", config.provider),
"output": [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": text}],
}
],
"output_text": text,
"usage": usage,
}
)
first_chunk = await response.__anext__()
async def responses_streaming():
yield f"data: {first_chunk.model_dump_json() if hasattr(first_chunk, 'model_dump_json') else first_chunk.json()}\n\n"
try:
async for chunk in response:
if isinstance(chunk, BaseConversation):
pass
else:
yield f"data: {chunk.model_dump_json() if hasattr(chunk, 'model_dump_json') else chunk.json()}\n\n"
except GeneratorExit:
pass
except RateLimitError as e:
debug.error(e)
yield f"data: {format_exception(e, config)}\n\n"
except Exception as e:
logger.exception(e)
yield f"data: {format_exception(e, config)}\n\n"
yield "data: [DONE]\n\n"
headers = (
getattr(first_chunk, "_headers").get_dict()
if hasattr(first_chunk, "_headers")
else {}
)
headers = {
k.encode("latin-1", "ignore")
.decode("latin-1"): v.encode("latin-1", "ignore")
.decode("latin-1")
for k, v in headers.items()
}
return StreamingResponse(
responses_streaming(),
media_type="text/event-stream",
headers=headers,
)
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
except (MissingAuthError, NoValidHarFileError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
except RateLimitError as e:
return ErrorResponse.from_exception(
e, config, HTTP_429_TOO_MANY_REQUESTS
)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, config, HTTP_500_INTERNAL_SERVER_ERROR
)
# ------------------------------------------------------------------ #
# Anthropic Messages API (/v1/messages) #
# https://docs.anthropic.com/en/api/messages #
# ------------------------------------------------------------------ #
@self.app.post("/v1/messages", responses=responses)
@self.app.post("/api/{provider:path}/messages", responses=responses)
async def create_message(
config: MessagesConfig,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
provider: str = None,
x_user: Annotated[str | None, Header()] = None,
):
if provider is not None:
config.provider = provider
if config.provider is None:
config.provider = AppConfig.provider
try:
provider = AbstractClientFactory.create_provider(None, config.provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
try:
if config.timeout is None:
config.timeout = AppConfig.timeout
if config.stream_timeout is None and config.stream:
config.stream_timeout = AppConfig.stream_timeout
if credentials is not None and credentials.credentials != "secret":
config.api_key = credentials.credentials
# Anthropic uses a top-level `system` field; fold it into messages.
messages = config.messages
if config.system:
system_content = config.system
if isinstance(system_content, list):
system_content = " ".join(
b.get("text", "") if isinstance(b, dict) else str(b)
for b in system_content
)
messages = [
{"role": "system", "content": system_content},
*messages,
]
response = self.client.chat.completions.create(
**filter_none(
**{
"model": AppConfig.model,
"provider": AppConfig.provider,
"proxy": AppConfig.proxy,
**(
config.model_dump(exclude_none=True)
if hasattr(config, "model_dump")
else config.dict(exclude_none=True)
),
**{
"provider": provider,
"messages": messages,
"user": x_user,
},
},
ignored=AppConfig.ignored_providers,
),
)
if not config.stream:
result = await response
text = result.choices[0].message.content if result.choices else ""
usage = getattr(result, "usage", None)
input_tokens = getattr(usage, "prompt_tokens", 0) or 0
output_tokens = getattr(usage, "completion_tokens", 0) or 0
return JSONResponse(
{
"id": getattr(result, "id", f"msg_{secrets.token_hex(12)}"),
"type": "message",
"role": "assistant",
"model": getattr(result, "model", config.model),
"provider": getattr(provider, "__name__", config.provider),
"content": [{"type": "text", "text": text}],
"stop_reason": getattr(
result.choices[0], "finish_reason", None
)
if result.choices
else None,
"stop_sequence": None,
"usage": {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
},
}
)
first_chunk = await response.__anext__()
async def messages_streaming():
yield f"data: {first_chunk.model_dump_json() if hasattr(first_chunk, 'model_dump_json') else first_chunk.json()}\n\n"
try:
async for chunk in response:
if isinstance(chunk, BaseConversation):
pass
else:
yield f"data: {chunk.model_dump_json() if hasattr(chunk, 'model_dump_json') else chunk.json()}\n\n"
except GeneratorExit:
pass
except RateLimitError as e:
debug.error(e)
yield f"data: {format_exception(e, config)}\n\n"
except Exception as e:
logger.exception(e)
yield f"data: {format_exception(e, config)}\n\n"
yield "data: [DONE]\n\n"
headers = (
getattr(first_chunk, "_headers").get_dict()
if hasattr(first_chunk, "_headers")
else {}
)
headers = {
k.encode("latin-1", "ignore")
.decode("latin-1"): v.encode("latin-1", "ignore")
.decode("latin-1")
for k, v in headers.items()
}
return StreamingResponse(
messages_streaming(),
media_type="text/event-stream",
headers=headers,
)
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
except (MissingAuthError, NoValidHarFileError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
except RateLimitError as e:
return ErrorResponse.from_exception(
e, config, HTTP_429_TOO_MANY_REQUESTS
)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, config, HTTP_500_INTERNAL_SERVER_ERROR
)
responses = {
HTTP_200_OK: {"model": ImagesResponse},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_429_TOO_MANY_REQUESTS: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.post("/v1/media/generate", responses=responses)
@self.app.post("/v1/images/generate", responses=responses)
@self.app.post("/v1/images/generations", responses=responses)
@self.app.post("/api/{provider:path}/images/generations", responses=responses)
async def generate_image(
request: Request,
config: ImageGenerationConfig,
provider: str = None,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
if provider is None:
provider = config.provider
if provider is None:
provider = AppConfig.provider
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
if (
config.api_key is None
and credentials is not None
and credentials.credentials != "secret"
):
config.api_key = credentials.credentials
try:
response = await self.client.images.generate(
**config.model_dump(exclude_none=True)
if hasattr(config, "model_dump")
else config.dict(exclude_none=True),
provider=provider,
)
for image in response.data:
if hasattr(image, "url") and image.url.startswith("/"):
image.url = f"{request.base_url}{image.url.lstrip('/')}"
return response
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_404_NOT_FOUND)
except MissingAuthError as e:
logger.exception(e)
return ErrorResponse.from_exception(e, config, HTTP_401_UNAUTHORIZED)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, config, HTTP_500_INTERNAL_SERVER_ERROR
)
@self.app.get(
"/v1/providers",
responses={
HTTP_200_OK: {"model": List[ProviderResponseModel]},
},
)
async def providers():
return [
{
"id": provider.__name__,
"object": "provider",
"created": 0,
"url": provider.url,
"label": getattr(provider, "label", None),
}
for provider in Provider.__providers__
if provider.working
]
@self.app.get(
"/v1/providers/{provider}",
responses={
HTTP_200_OK: {"model": ProviderResponseDetailModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
async def providers_info(provider: str):
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
return {
"id": provider.__name__,
"object": "provider",
"created": 0,
"url": provider.url,
"label": getattr(provider, "label", None),
"image_models": getattr(provider, "image_models", []) or [],
"vision_models": [
model
for model in [getattr(provider, "default_vision_model", None)]
if model
],
"params": [*provider.get_parameters()]
if hasattr(provider, "get_parameters")
else [],
}
# ------------------------------------------------------------------ #
# PA Provider routes #
# ------------------------------------------------------------------ #
@self.app.get(
"/pa/providers",
responses={
HTTP_200_OK: {},
},
)
async def pa_providers_list():
"""List all PA providers loaded from the workspace.
Filenames are never exposed; each provider is identified by a
stable opaque ID (SHA-256 of the path, first 8 hex chars).
"""
from g4f.mcp.pa_provider import get_pa_registry
return get_pa_registry().list_providers()
@self.app.get(
"/pa/providers/{provider_id}",
responses={
HTTP_200_OK: {},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
async def pa_providers_detail(provider_id: str):
"""Get details for a single PA provider by its opaque ID."""
from g4f.mcp.pa_provider import get_pa_registry
info = get_pa_registry().get_provider_info(provider_id)
if info is None:
return ErrorResponse.from_message(
f"PA provider '{provider_id}' not found", HTTP_404_NOT_FOUND
)
return info
# ------------------------------------------------------------------ #
# PA workspace static file serving (HTML/CSS/JS/images for browser) #
# ------------------------------------------------------------------ #
#: MIME types that are safe to serve for browser rendering.
#: Only these extensions are allowed; all others are refused with 403.
_WORKSPACE_SAFE_TYPES: dict[str, str] = {
"html": "text/html; charset=utf-8",
"htm": "text/html; charset=utf-8",
"css": "text/css; charset=utf-8",
"js": "application/javascript; charset=utf-8",
"mjs": "application/javascript; charset=utf-8",
"json": "application/json; charset=utf-8",
"txt": "text/plain; charset=utf-8",
"md": "text/markdown; charset=utf-8",
"svg": "image/svg+xml",
"png": "image/png",
"jpg": "image/jpeg",
"jpeg": "image/jpeg",
"gif": "image/gif",
"webp": "image/webp",
"ico": "image/x-icon",
"woff": "font/woff",
"woff2": "font/woff2",
"ttf": "font/ttf",
"otf": "font/otf",
"py": "text/plain; charset=utf-8",
}
@self.app.get(
"/pa/files/{file_path:path}",
responses={
HTTP_200_OK: {},
HTTP_403_FORBIDDEN: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
},
)
async def pa_serve_workspace_file(file_path: str, request: Request):
"""Securely serve a workspace file for browser rendering.
Only files within ``~/.g4f/workspace`` can be served. Path
traversal (``..``) is blocked. Only the MIME types listed in
``_WORKSPACE_SAFE_TYPES`` are served; all other extensions are
refused with **403 Forbidden** so that sensitive file types (e.g.
``.env``, ``.pa.py``, ``.py``) can never be read via this route.
HTML files are served with a ``Content-Security-Policy: sandbox``
directive (without ``allow-same-origin``), which forces the page
into a unique *null* browser origin. As a result the page cannot
access ``localStorage``, ``sessionStorage``, ``IndexedDB``, or
cookies belonging to the g4f server origin — the browser rejects
all such calls with a ``SecurityError``. The actual request
origin (``scheme://host``) is used in every source directive (e.g.
``default-src``) instead of ``'self'``, so that co-located CSS,
JS, images, and fonts still load correctly despite the document
having a null origin.
Non-HTML sub-resources (CSS, JS, images, fonts) are served without
the ``sandbox`` directive; they are leaf resources and do not run
in their own browsing context.
"""
from g4f.mcp.pa_provider import get_workspace_dir
workspace = get_workspace_dir()
# Normalise and check for traversal
try:
resolved = (workspace / file_path).resolve()
resolved.relative_to(workspace.resolve())
except (ValueError, Exception):
return ErrorResponse.from_message(
"Path traversal is not allowed", HTTP_403_FORBIDDEN
)
if not resolved.exists() or not resolved.is_file():
return ErrorResponse.from_message(
f"File not found: {file_path}", HTTP_404_NOT_FOUND
)
ext = resolved.suffix.lstrip(".").lower()
mime_type = _WORKSPACE_SAFE_TYPES.get(ext)
if mime_type is None:
return ErrorResponse.from_message(
f"File type '.{ext}' is not allowed for browser rendering",
HTTP_403_FORBIDDEN,
)
# Derive the actual request origin (scheme + authority) from the
# ASGI scope via request.url — this is set by the server
# infrastructure and is not controllable by the client (unlike the
# Host header, which can be spoofed to inject arbitrary values into
# the CSP). request.url.netloc includes the port when non-default.
request_origin = f"{request.url.scheme}://{request.url.netloc}"
is_html = ext in ("html", "htm")
if is_html:
# HTML documents are served with the CSP sandbox directive
# (without allow-same-origin). This forces the page into a
# unique null browsing-context origin so that it cannot access
# the g4f server's localStorage, sessionStorage, IndexedDB, or
# cookies. The page can still load sub-resources (CSS, JS,
# images) because they are referenced by the explicit
# request_origin in the source directives.
csp = (
"sandbox allow-scripts allow-forms allow-downloads allow-popups; "
f"default-src {request_origin} https://g4f.space; "
f"script-src {request_origin} 'unsafe-inline'; "
f"style-src {request_origin} 'unsafe-inline'; "
f"img-src {request_origin} data:; "
f"font-src {request_origin} data:; "
"connect-src 'none'; "
"object-src 'none'; "
"base-uri 'none';"
)
else:
# Non-HTML sub-resources (CSS, JS, images, fonts) don't need
# sandboxing — they are leaf assets without their own browsing
# context. Use the request origin for source directives.
csp = (
f"default-src {request_origin} https://g4f.space; "
f"script-src {request_origin} 'unsafe-inline'; "
f"style-src {request_origin} 'unsafe-inline'; "
f"img-src {request_origin} data:; "
f"font-src {request_origin} data:; "
"connect-src 'none'; "
"object-src 'none'; "
"base-uri 'none';"
)
headers = {
# Prevent the browser from sniffing a different content-type
"X-Content-Type-Options": "nosniff",
# Prevent this page from being framed by untrusted origins
"X-Frame-Options": "SAMEORIGIN",
# Basic XSS filter (belt-and-suspenders; CSP is more important)
"X-XSS-Protection": "1; mode=block",
"Content-Security-Policy": csp,
# Restrict powerful browser APIs that workspace pages don't need
"Permissions-Policy": (
"geolocation=(), camera=(), microphone=(), "
"payment=(), usb=(), fullscreen=()"
),
"Cache-Control": "no-store",
}
return FileResponse(
str(resolved),
media_type=mime_type,
headers=headers,
)
responses = {
HTTP_200_OK: {"model": TranscriptionResponseModel},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.post("/v1/audio/transcriptions", responses=responses)
@self.app.post(
"/api/{path_provider:path}/audio/transcriptions", responses=responses
)
@self.app.post("/api/markitdown", responses=responses)
async def convert(
file: UploadFile,
path_provider: Optional[str] = None,
model: Annotated[Optional[str], Form()] = None,
provider: Annotated[Optional[str], Form()] = None,
prompt: Annotated[Optional[str], Form()] = "Transcribe this audio",
):
if path_provider is not None:
provider = path_provider
if provider is None:
provider = "MarkItDown"
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
kwargs = {"modalities": ["text"]}
try:
response = await self.client.chat.completions.create(
messages=prompt,
model=model,
provider=provider,
media=[[file.file, file.filename]],
**kwargs,
)
return {
"text": response.choices[0].message.content,
"model": response.model,
"provider": response.provider,
}
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, None, HTTP_404_NOT_FOUND)
except MissingAuthError as e:
logger.exception(e)
return ErrorResponse.from_exception(e, None, HTTP_401_UNAUTHORIZED)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, None, HTTP_500_INTERNAL_SERVER_ERROR
)
responses = {
HTTP_200_OK: {"model": TranscriptionResponseModel},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.get("/markitdown/{url:path}", responses=responses)
async def convert_url(
request: Request,
url: str,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
"""Convert a URL to Markdown using MarkItDown.
The full URL (including scheme) is passed in the path, e.g.:
GET /markitdown/https://example.com/page
Query strings are preserved by reading them from the incoming
request and re-appending them to the target URL, e.g.:
GET /markitdown/https://example.com/page?foo=bar
"""
# FastAPI strips the query string from the {url:path} parameter,
# so re-attach it from the incoming request when present.
query_string = request.url.query
if query_string and "?" not in url:
url = f"{url}?{query_string}"
elif query_string:
# url already contains a '?', append remaining params with '&'
url = f"{url}&{query_string}"
if not url.startswith(("http://", "https://")):
return ErrorResponse.from_message(
f"Invalid URL: {url}. URL must start with http:// or https://",
HTTP_422_UNPROCESSABLE_CONTENT,
)
try:
from g4f.integration.markitdown import MarkItDown
md = MarkItDown()
result = md.convert_url(url)
text = result.text_content
if asyncio.iscoroutine(text):
text = await text
return JSONResponse(
{"text": text, "title": result.title, "url": url},
)
except ImportError as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, None, HTTP_500_INTERNAL_SERVER_ERROR
)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, None, HTTP_500_INTERNAL_SERVER_ERROR
)
responses = {
HTTP_200_OK: {"content": {"audio/*": {}}},
HTTP_401_UNAUTHORIZED: {"model": ErrorResponseModel},
HTTP_404_NOT_FOUND: {"model": ErrorResponseModel},
HTTP_500_INTERNAL_SERVER_ERROR: {"model": ErrorResponseModel},
}
@self.app.post("/v1/audio/speech", responses=responses)
@self.app.post("/api/{provider:path}/audio/speech", responses=responses)
async def generate_speech(
config: AudioSpeechConfig,
provider: Optional[str] = None,
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
api_key = None
if credentials is not None and credentials.credentials != "secret":
api_key = credentials.credentials
if provider is None:
provider = config.provider
if provider is None:
provider = AppConfig.media_provider
try:
provider = AbstractClientFactory.create_provider(None, provider)
except ProviderNotFoundError as e:
return ErrorResponse.from_message(str(e), 404)
try:
audio = filter_none(
voice=config.voice,
format=config.response_format,
language=config.language,
)
response = await self.client.chat.completions.create(
messages=[
{
"role": "user",
"content": f"{config.instrcutions} Text: {config.input}",
}
],
model=config.model,
provider=provider,
prompt=config.input,
api_key=api_key,
download_media=config.download_media,
**filter_none(
audio=audio if audio else None,
),
)
if response.choices[0].message.audio is not None:
response = base64.b64decode(response.choices[0].message.audio.data)
return Response(
response,
media_type=f"audio/{config.response_format.replace('mp3', 'mpeg')}",
)
elif isinstance(response.choices[0].message.content, AudioResponse):
response = response.choices[0].message.content.data
response = response.replace("/media", get_media_dir())
def delete_file():
try:
os.remove(response)
except Exception as e:
logger.exception(e)
return FileResponse(
response, background=BackgroundTask(delete_file)
)
except (ModelNotFoundError, ProviderNotFoundError) as e:
logger.exception(e)
return ErrorResponse.from_exception(e, None, HTTP_404_NOT_FOUND)
except MissingAuthError as e:
logger.exception(e)
return ErrorResponse.from_exception(e, None, HTTP_401_UNAUTHORIZED)
except Exception as e:
logger.exception(e)
return ErrorResponse.from_exception(
e, None, HTTP_500_INTERNAL_SERVER_ERROR
)
@self.app.post(
"/v1/upload_cookies",
responses={
HTTP_200_OK: {"model": List[FileResponseModel]},
},
)
def upload_cookies(
files: List[UploadFile],
credentials: Annotated[
HTTPAuthorizationCredentials, Depends(Api.security)
] = None,
):
response_data = []
if not AppConfig.ignore_cookie_files:
for file in files:
try:
if (
file
and file.filename.endswith(".json")
or file.filename.endswith(".har")
):
filename = os.path.basename(file.filename)
with open(
os.path.join(get_cookies_dir(), filename), "wb"
) as f:
shutil.copyfileobj(file.file, f)
response_data.append({"filename": filename})
finally:
file.file.close()
read_cookie_files()
return response_data
@self.app.get(
"/images/{filename}",
responses={
HTTP_200_OK: {"content": {"image/*": {}}},
HTTP_404_NOT_FOUND: {},
},
)
@self.app.get(
"/media/{filename}",
responses={
HTTP_200_OK: {"content": {"image/*": {}, "audio/*": {}}, "video/*": {}},
HTTP_404_NOT_FOUND: {},
},
)
async def get_media(filename, request: Request, thumbnail: bool = False):
def get_timestamp(str):
m = re.match("^[0-9]+", str)
if m:
return int(m.group(0))
else:
return 0
target = os.path.join(get_media_dir(), os.path.basename(filename))
if thumbnail and has_pillow:
thumbnail_dir = os.path.join(get_media_dir(), "thumbnails")
thumbnail = os.path.join(thumbnail_dir, filename)
if not os.path.isfile(target):
other_name = os.path.join(
get_media_dir(), os.path.basename(quote_plus(filename))
)
if os.path.isfile(other_name):
target = other_name
ext = os.path.splitext(filename)[1][1:]
mime_type = EXTENSIONS_MAP.get(ext)
stat_result = SimpleNamespace()
stat_result.st_size = 0
stat_result.st_mtime = get_timestamp(filename)
if thumbnail and has_pillow and os.path.isfile(thumbnail):
stat_result.st_size = os.stat(thumbnail).st_size
elif not thumbnail and os.path.isfile(target):
stat_result.st_size = os.stat(target).st_size
headers = {
"cache-control": "public, max-age=31536000",
"last-modified": formatdate(stat_result.st_mtime, usegmt=True),
"etag": f'"{hashlib.md5(filename.encode()).hexdigest()}"',
**(
{
"content-length": str(stat_result.st_size),
}
if stat_result.st_size
else {}
),
**(
{}
if thumbnail or mime_type is None
else {
"content-type": mime_type,
}
),
}
response = FileResponse(
target,
headers=headers,
filename=filename,
)
try:
if_none_match = request.headers["if-none-match"]
etag = response.headers["etag"]
if etag in [tag.strip(" W/") for tag in if_none_match.split(",")]:
return NotModifiedResponse(response.headers)
except KeyError:
pass
if not os.path.isfile(target) and mime_type is not None:
source_url = get_source_url(str(request.query_params))
ssl = None
if source_url is None:
backend_url = os.environ.get("G4F_BACKEND_URL")
if backend_url:
source_url = f"{backend_url}/media/{filename}"
ssl = False
if source_url is not None:
try:
await copy_media([source_url], target=target, ssl=ssl)
debug.log(f"File copied from {source_url}")
except Exception as e:
debug.error(f"Download failed: {source_url}")
debug.error(e)
return RedirectResponse(url=source_url)
if thumbnail and has_pillow:
try:
if not os.path.isfile(thumbnail):
image = Image.open(target)
os.makedirs(thumbnail_dir, exist_ok=True)
process_image(image, save=thumbnail)
debug.log(f"Thumbnail created: {thumbnail}")
except Exception as e:
logger.exception(e)
if thumbnail and os.path.isfile(thumbnail):
result = thumbnail
else:
result = target
if not os.path.isfile(result):
return ErrorResponse.from_message("File not found", HTTP_404_NOT_FOUND)
async def stream():
with open(result, "rb") as file:
while True:
chunk = file.read(65536)
if not chunk:
break
yield chunk
return StreamingResponse(stream(), headers=headers)
@self.app.get(
"/thumbnail/{filename}",
responses={
HTTP_200_OK: {"content": {"image/*": {}, "audio/*": {}}, "video/*": {}},
HTTP_404_NOT_FOUND: {},
},
)
async def get_media_thumbnail(filename: str, request: Request):
return await get_media(filename, request, True)
@self.app.get("/logs", response_class=HTMLResponse)
async def logs_inspector():
return HTMLResponse(_LOGS_HTML)
@self.app.get("/api/logs")
async def get_logs(limit: int = 500, offset: int = 0):
entries = list(_request_log)
total = len(entries)
start = max(0, total - limit - offset)
end = total - offset if offset < total else total
page = list(reversed(entries[start:end]))
return JSONResponse(
{"total": total, "entries": page}, headers={"Cache-Control": "no-store"}
)
@self.app.delete("/api/logs")
async def clear_logs():
_request_log.clear()
return JSONResponse({"status": "cleared"})
def format_exception(
e: Union[Exception, str],
config: Union[ChatCompletionsConfig, ImageGenerationConfig] = None,
image: bool = False,
) -> str:
provider = AppConfig.media_provider if image else AppConfig.provider
model = AppConfig.model
if config is not None:
if config.provider is not None:
provider = config.provider
if config.model is not None:
model = config.model
if isinstance(e, str):
message = e
else:
message = f"{e.__class__.__name__}: {e}"
return json.dumps(
{
"error": {"message": message},
**filter_none(
model=model, provider=getattr(provider, "__name__", provider)
),
}
)
def run_api(
host: str = "0.0.0.0",
port: int = None,
bind: str = None,
debug: bool = False,
use_colors: bool = None,
**kwargs,
) -> None:
print(
f"Starting server... [g4f v-{g4f.version.utils.current_version}]"
+ (" (debug)" if debug else "")
)
if use_colors is None:
use_colors = debug
if bind is not None:
host, port = bind.split(":")
if port is None:
port = DEFAULT_PORT
if AppConfig.demo and debug:
method = "create_app_with_demo_and_debug"
elif AppConfig.gui and debug:
method = "create_app_with_gui_and_debug"
else:
method = "create_app_debug" if debug else "create_app"
uvicorn.run(
f"g4f.api:{method}",
host=host,
port=int(port),
factory=True,
use_colors=use_colors,
**filter_none(**kwargs),
)
