#!/usr/bin/env python3
"""
gemini-web2api + Zep Cloud Memory — Professional Agent Edition
===============================================================
نسخة احترافية كاملة مع:
- ✅ Zep Cloud: ذاكرة دائمة عبر كل الجلسات (حقائق + ملخص + بحث دلالي)
- ✅ Agent Loop حقيقي (ReAct: Reason → Act → Observe → Repeat)
- ✅ Tool Calling موثوق مع JSON parser متعدد المستويات
- ✅ Streaming حقيقي مع Tool Calls
- ✅ SQLite Cache + Rate Limiting
- ✅ Retry Logic + Fallback للنموذج الافتراضي
- ✅ Gradio UI محسّن
المتطلبات:
pip install fastapi uvicorn aiosqlite httpx gradio zep-cloud pydantic
متغيرات البيئة:
ZEP_API_KEY - مفتاح Zep Cloud (مطلوب للذاكرة الدائمة)
API_KEYS - مفاتيح API مفصولة بفاصلة (افتراضي: sk-gemini)
GEMINI_COOKIE - كوكي Gemini (اختياري، يُحسّن الجودة)
HTTPS_PROXY - بروكسي (اختياري)
GEMINI_BL - Gemini BL header (اختياري)
"""
import asyncio
import json
import time
import hashlib
import re
import os
import ssl
import sys
import uuid
import argparse
import urllib.parse
import urllib.request
from typing import Optional, List, Dict, Any, Tuple, Callable
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import aiosqlite
import gradio as gr
try:
import httpx
HAS_HTTPX = True
except ImportError:
HAS_HTTPX = False
try:
from zep_cloud.client import AsyncZep
from zep_cloud.types import Message as ZepMessage
HAS_ZEP = True
except ImportError:
HAS_ZEP = False
__version__ = "3.0.0-zep"
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Configuration ────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def load_config() -> dict:
defaults = {
# ─── Server
"port": 7860,
"host": "0.0.0.0",
"log_requests": True,
# ─── Gemini
"gemini_bl": "boq_assistant-bard-web-server_20260525.09_p0",
"default_model": "gemini-3.5-flash",
"cookie_file": None,
"proxy": None,
"retry_attempts": 3,
"retry_delay_sec": 2,
"request_timeout_sec": 180,
# ─── API
"api_keys": ["sk-gemini"],
"rate_limit_per_minute": 20,
# ─── Cache
"cache_ttl_seconds": 3600,
"db_path": "cache.db",
"cleanup_interval_seconds": 300,
# ─── Agent Loop
"agent_max_iterations": 10,
"agent_max_tokens_per_step": 4096,
# ─── Zep Cloud Memory
"zep_api_key": os.environ.get("ZEP_API_KEY", ""),
"zep_memory_type": "perpetual", # perpetual | summary_retriever
"zep_min_fact_rating": 0.3, # حد أدنى لجودة الحقائق
"zep_recent_messages": 20, # عدد الرسائل الأخيرة في السياق
"zep_search_limit": 5, # نتائج البحث الدلالي
"zep_save_system_msgs": False, # هل تحفظ رسائل النظام؟
}
# تحميل ملف الإعدادات إذا وُجد
for p in ["./config.json", os.path.expanduser("~/.config/gemini-web2api/config.json")]:
if os.path.exists(p):
try:
with open(p) as f:
defaults.update(json.load(f))
print(f"[Config] Loaded: {p}")
except Exception as e:
print(f"[Config] Error: {e}")
break
# متغيرات البيئة تُلغي الإعدادات
if os.environ.get("API_KEYS"):
defaults["api_keys"] = [k.strip() for k in os.environ["API_KEYS"].split(",") if k.strip()]
if os.environ.get("GEMINI_COOKIE"):
defaults["_cookie_inline"] = os.environ["GEMINI_COOKIE"]
if os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY"):
defaults["proxy"] = os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY")
if os.environ.get("GEMINI_BL"):
defaults["gemini_bl"] = os.environ["GEMINI_BL"]
if os.environ.get("ZEP_API_KEY"):
defaults["zep_api_key"] = os.environ["ZEP_API_KEY"]
return defaults
CONFIG = load_config()
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Models ───────────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
MODELS = {
"gemini-3.7-flash": {
"mode": 1, "think": 4,
"desc": "Fast general-purpose model (~12k chars)",
},
"gemini-3.6-flash": {
"mode": 1, "think": 4,
"desc": "All-around model (Gemini 3.6 Flash)",
},
"gemini-3.5-flash": {
"mode": 1, "think": 4,
"desc": "Standard flash model",
},
"gemini-3.5-flash-thinking": {
"mode": 2, "think": 0,
"desc": "Deep thinking mode (~20k chars)",
},
"gemini-3.1-pro": {
"mode": 3, "think": 4,
"desc": "Pro model",
},
"gemini-auto": {
"mode": 4, "think": 4,
"desc": "Auto model selection",
},
"gemini-3.5-flash-thinking-lite": {
"mode": 5, "think": 0,
"desc": "Dynamic thinking (~15k chars)",
},
"gemini-flash-lite": {
"mode": 6, "think": 4,
"desc": "Lightweight fast model (~10k chars)",
},
}
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Pydantic Schemas ─────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
class ChatMessage(BaseModel):
role: str
content: Any
tool_calls: Optional[List[dict]] = None
tool_call_id: Optional[str] = None
name: Optional[str] = None
class ChatCompletionRequest(BaseModel):
model: str = "gemini-3.5-flash"
messages: List[ChatMessage]
temperature: Optional[float] = 0.7
max_tokens: Optional[int] = 32000
stream: Optional[bool] = False
tools: Optional[List[dict]] = None
tool_choice: Optional[Any] = "auto"
user: Optional[str] = "default"
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Logging ──────────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def log(msg: str, level: str = "INFO"):
if CONFIG.get("log_requests"):
icons = {
"INFO": "ℹ️", "AGENT": "🤖", "TOOL": "🔧",
"ERROR": "❌", "WARN": "⚠️", "MEM": "🧠",
}
icon = icons.get(level, "•")
sys.stderr.write(f"[{time.strftime('%H:%M:%S')}] {icon} {msg}\n")
sys.stderr.flush()
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Services (Cache + Rate Limit) ────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
@dataclass
class RateLimitEntry:
timestamps: List[float] = field(default_factory=list)
last_cleanup: float = field(default_factory=time.time)
class Services:
def __init__(self):
self._rate_limits: Dict[str, RateLimitEntry] = {}
self._lock = asyncio.Lock()
self._cleanup_task: Optional[asyncio.Task] = None
async def init_db(self):
async with aiosqlite.connect(CONFIG["db_path"]) as db:
await db.execute("""
CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
val TEXT,
exp REAL
)
""")
await db.execute("CREATE INDEX IF NOT EXISTS idx_exp ON cache(exp)")
await db.commit()
async def start_cleanup(self):
async def loop():
while True:
await asyncio.sleep(CONFIG["cleanup_interval_seconds"])
await self._cleanup()
self._cleanup_task = asyncio.create_task(loop())
async def _cleanup(self):
now = time.time()
cutoff = now - 3600
async with self._lock:
expired = [ip for ip, e in self._rate_limits.items() if e.last_cleanup < cutoff]
for ip in expired:
del self._rate_limits[ip]
try:
async with aiosqlite.connect(CONFIG["db_path"]) as db:
await db.execute("DELETE FROM cache WHERE exp < ?", (now,))
await db.commit()
except Exception:
pass
async def check_rate_limit(self, client_id: str) -> Tuple[bool, int]:
now = time.time()
cutoff = now - 60
async with self._lock:
if client_id not in self._rate_limits:
self._rate_limits[client_id] = RateLimitEntry()
entry = self._rate_limits[client_id]
entry.timestamps = [t for t in entry.timestamps if t > cutoff]
entry.last_cleanup = now
limit = CONFIG["rate_limit_per_minute"]
if len(entry.timestamps) >= limit:
wait = 60 - (now - min(entry.timestamps))
return False, int(wait) + 1
entry.timestamps.append(now)
return True, 0
async def get_cache(self, key: str) -> Optional[str]:
try:
async with aiosqlite.connect(CONFIG["db_path"]) as db:
async with db.execute(
"SELECT val FROM cache WHERE key=? AND exp>?", (key, time.time())
) as cur:
row = await cur.fetchone()
return row[0] if row else None
except Exception:
return None
async def set_cache(self, key: str, value: str):
try:
async with aiosqlite.connect(CONFIG["db_path"]) as db:
await db.execute(
"INSERT OR REPLACE INTO cache VALUES (?, ?, ?)",
(key, value, time.time() + CONFIG["cache_ttl_seconds"])
)
await db.commit()
except Exception:
pass
services = Services()
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Zep Cloud Memory Manager ─────────────────────────────────────════════════
# ═══════════════════════════════════════════════════════════════════════════════
class ZepMemoryManager:
"""
مدير الذاكرة الدائمة عبر Zep Cloud.
الميزات:
─────────
• كل user_id → جلسة (session) ثابتة لا تُمسح بين الدردشات
• يسترجع: الحقائق المستخرجة + ملخص المحادثة + آخر N رسالة
• يبحث دلالياً في كامل تاريخ المحادثة بناءً على السؤال الحالي
• يحفظ كل رسالة (user + assistant + tool) تلقائياً بعد كل رد
• يتعامل مع تغيير الموضوع: كل رسالة مرتبطة بـ timestamp وسياق
"""
def __init__(self):
self.client: Optional[AsyncZep] = None
self._enabled = False
self._init_error: str = ""
async def init(self):
"""تهيئة عميل Zep Cloud."""
if not HAS_ZEP:
self._init_error = "zep-cloud not installed. Run: pip install zep-cloud"
log(self._init_error, "WARN")
return
api_key = CONFIG.get("zep_api_key", "").strip()
if not api_key:
self._init_error = "ZEP_API_KEY not set — persistent memory disabled"
log(self._init_error, "WARN")
return
try:
self.client = AsyncZep(api_key=api_key)
# اختبار الاتصال
await self.client.user.list(page_number=1, page_size=1)
self._enabled = True
log("✅ Zep Cloud memory connected and ready", "MEM")
except Exception as e:
self._init_error = f"Zep Cloud connection failed: {e}"
log(self._init_error, "ERROR")
@property
def enabled(self) -> bool:
return self._enabled and self.client is not None
# ─── Session Management ────────────────────────────────────────────────────
def _session_id(self, user_id: str) -> str:
"""
session_id ثابت ومبني على user_id.
لا يتغير بين الجلسات → هذا هو مفتاح الاستمرارية.
"""
safe = re.sub(r'[^a-zA-Z0-9_-]', '_', user_id)
return f"agent_{safe}"
async def ensure_user_session(self, user_id: str) -> bool:
"""
ينشئ المستخدم والجلسة في Zep إن لم تكونا موجودَين.
يعيد True عند النجاح.
"""
if not self.enabled:
return False
try:
session_id = self._session_id(user_id)
# إنشاء المستخدم (يتجاهل الخطأ إذا كان موجوداً)
try:
await self.client.user.add(
user_id=user_id,
metadata={
"source": "gemini-web2api",
"created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
}
)
log(f"Created Zep user: {user_id}", "MEM")
except Exception:
pass # المستخدم موجود بالفعل
# إنشاء الجلسة (يتجاهل الخطأ إذا كانت موجودة)
try:
await self.client.memory.add_session(
session_id=session_id,
user_id=user_id,
metadata={
"agent_version": __version__,
"model": CONFIG.get("default_model"),
}
)
log(f"Created Zep session: {session_id}", "MEM")
except Exception:
pass # الجلسة موجودة بالفعل
return True
except Exception as e:
log(f"ensure_user_session error: {e}", "WARN")
return False
# ─── Memory Retrieval ──────────────────────────────────────────────────────
async def get_full_context(self, user_id: str, current_query: str = "") -> str:
"""
يسترجع السياق الكامل من Zep ويُعيده كنص منسَّق.
يشمل:
1. الحقائق المستخرجة تلقائياً (مثل: اسم المستخدم، تفضيلاته، مهامه)
2. ملخص تلقائي لكل المحادثات السابقة
3. آخر N رسالة للسياق الفوري
4. نتائج بحث دلالي إذا كان هناك استعلام
هذا يضمن:
- تذكر تغييرات الموضوع (كل شيء مُدوَّن)
- الحفاظ على السياق في المحادثات الطويلة جداً
- استرجاع المعلومات ذات الصلة حتى من محادثات قديمة
"""
if not self.enabled:
return ""
session_id = self._session_id(user_id)
parts = []
try:
# ─── 1. الذاكرة الأساسية (حقائق + ملخص + رسائل أخيرة)
memory = await self.client.memory.get(
session_id=session_id,
min_rating=CONFIG.get("zep_min_fact_rating", 0.3),
)
# الحقائق المستخرجة تلقائياً من كل المحادثات
if hasattr(memory, 'facts') and memory.facts:
facts_lines = []
for f in memory.facts:
fact_text = f if isinstance(f, str) else getattr(f, 'fact', str(f))
if fact_text:
facts_lines.append(f" • {fact_text}")
if facts_lines:
parts.append(
"### KNOWN FACTS (extracted automatically from all past conversations)\n"
+ "\n".join(facts_lines)
)
# الملخص التلقائي للمحادثات السابقة
if hasattr(memory, 'summary') and memory.summary:
summary_text = (
memory.summary if isinstance(memory.summary, str)
else getattr(memory.summary, 'content', '')
)
if summary_text and summary_text.strip():
parts.append(
"### CONVERSATION HISTORY SUMMARY\n"
+ summary_text.strip()
)
# آخر الرسائل للسياق الفوري
if hasattr(memory, 'messages') and memory.messages:
n = CONFIG.get("zep_recent_messages", 20)
recent = memory.messages[-n:] if len(memory.messages) > n else memory.messages
if recent:
lines = []
for m in recent:
role = getattr(m, 'role_type', getattr(m, 'role', 'user'))
content = getattr(m, 'content', '') or ''
if content.strip():
lines.append(f" [{role.upper()}]: {content.strip()}")
if lines:
parts.append(
f"### RECENT MESSAGES (last {len(recent)})\n"
+ "\n".join(lines)
)
except Exception as e:
log(f"get_memory error: {e}", "WARN")
# ─── 2. البحث الدلالي بناءً على الاستعلام الحالي
if current_query and current_query.strip():
try:
search_results = await self.client.memory.search_sessions(
text=current_query,
user_id=user_id,
limit=CONFIG.get("zep_search_limit", 5),
)
if search_results and hasattr(search_results, 'results') and search_results.results:
relevant_lines = []
for r in search_results.results:
msg = getattr(r, 'message', None)
if msg:
content = getattr(msg, 'content', '') or ''
role = getattr(msg, 'role_type', 'unknown')
score = getattr(r, 'score', 0)
if content.strip() and score > 0.5:
relevant_lines.append(
f" [{role.upper()}] (relevance: {score:.2f}): {content.strip()}"
)
if relevant_lines:
parts.append(
"### SEMANTICALLY RELEVANT PAST CONTEXT (for current query)\n"
+ "\n".join(relevant_lines)
)
except Exception as e:
log(f"semantic search error: {e}", "WARN")
if not parts:
return ""
# ─── بناء السياق النهائي
context = (
"╔══════════════════════════════════════════════════════════╗\n"
"║ PERSISTENT MEMORY — CROSS-SESSION CONTEXT ║\n"
"║ Use this to maintain full continuity across all chats ║\n"
"╚══════════════════════════════════════════════════════════╝\n\n"
+ "\n\n".join(parts)
+ "\n\n[END OF PERSISTENT MEMORY]"
)
log(f"Retrieved memory context ({len(context)} chars) for user={user_id}", "MEM")
return context
# ─── Memory Saving ─────────────────────────────────────────────────────────
async def save_turn(self, user_id: str, messages: List[dict], assistant_reply: str = ""):
"""
يحفظ دوران المحادثة (رسائل المستخدم + رد المساعد) في Zep.
يُستدعى بعد كل رد ناجح.
Zep يستخرج الحقائق ويُحدّث الملخص تلقائياً في الخلفية.
"""
if not self.enabled:
return
session_id = self._session_id(user_id)
zep_msgs = []
try:
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content") or ""
# تجاهل رسائل النظام إذا كانت الإعدادات تمنعها
if role == "system" and not CONFIG.get("zep_save_system_msgs"):
continue
# تحويل content القائمة إلى نص
if isinstance(content, list):
content = " ".join(
c.get("text", "") for c in content
if isinstance(c, dict) and c.get("type") in ("text", "input_text")
)
content = (content or "").strip()
if not content:
continue
# تحويل tool calls إلى نص قابل للقراءة
tool_calls_data = msg.get("tool_calls")
if tool_calls_data and role == "assistant":
tc_summary = json.dumps(
[{"tool": tc["function"]["name"], "args": tc["function"].get("arguments", "{}")}
for tc in tool_calls_data],
ensure_ascii=False
)
content = f"{content}\n[Tool calls: {tc_summary}]" if content else f"[Tool calls: {tc_summary}]"
role_map = {
"user": "user",
"assistant": "assistant",
"tool": "tool",
"system": "system",
}
zep_role = role_map.get(role, "user")
zep_msgs.append(
ZepMessage(
role_type=zep_role,
role=role,
content=content,
)
)
# إضافة رد المساعد إذا لم يكن في القائمة
if assistant_reply and assistant_reply.strip():
already_saved = any(
m.get("role") == "assistant" and (m.get("content") or "") == assistant_reply
for m in messages
)
if not already_saved:
zep_msgs.append(
ZepMessage(
role_type="assistant",
role="assistant",
content=assistant_reply.strip(),
)
)
if zep_msgs:
await self.client.memory.add(
session_id=session_id,
messages=zep_msgs,
)
log(f"💾 Saved {len(zep_msgs)} messages to Zep (user={user_id})", "MEM")
except Exception as e:
log(f"save_turn error: {e}", "WARN")
async def add_fact(self, user_id: str, fact: str):
"""
إضافة حقيقة صريحة يذكرها المستخدم (للأشياء المهمة جداً).
Zep يستخرج الحقائق تلقائياً أيضاً، لكن هذا للتأكيد.
"""
if not self.enabled:
return
try:
session_id = self._session_id(user_id)
await self.client.memory.add(
session_id=session_id,
messages=[
ZepMessage(
role_type="system",
role="system",
content=f"[IMPORTANT FACT]: {fact}",
)
]
)
log(f"Added explicit fact for user={user_id}: {fact[:50]}...", "MEM")
except Exception as e:
log(f"add_fact error: {e}", "WARN")
def status(self) -> dict:
return {
"enabled": self.enabled,
"provider": "Zep Cloud",
"error": self._init_error if not self.enabled else None,
}
# المثيل العام
zep_memory = ZepMemoryManager()
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Cookie & Headers ─────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def load_cookie() -> Tuple[str, Optional[str]]:
inline = CONFIG.get("_cookie_inline", "")
if inline:
if inline.startswith("{"):
try:
data = json.loads(inline)
return data.get("cookie", ""), data.get("sapisid") or None
except Exception:
pass
pairs = dict(p.split("=", 1) for p in inline.split("; ") if "=" in p)
return inline, pairs.get("SAPISID") or None
cookie_file = CONFIG.get("cookie_file")
if not cookie_file or not os.path.exists(cookie_file):
return "", None
try:
with open(cookie_file) as f:
content = f.read().strip()
if content.startswith("{"):
data = json.loads(content)
return data.get("cookie", ""), data.get("sapisid") or None
pairs = dict(p.split("=", 1) for p in content.split("; ") if "=" in p)
return content, pairs.get("SAPISID") or None
except Exception as e:
log(f"Cookie load error: {e}", "ERROR")
return "", None
def make_sapisidhash(sapisid: str) -> str:
ts = int(time.time())
h = hashlib.sha1(f"{ts} {sapisid} https://gemini.google.com".encode()).hexdigest()
return f"SAPISIDHASH {ts}_{h}"
def _build_headers() -> dict:
headers = {
"Content-Type": "application/x-www-form-urlencoded",
"Origin": "https://gemini.google.com",
"Referer": "https://gemini.google.com/app",
"X-Same-Domain": "1",
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
),
}
cookie_str, sapisid = load_cookie()
if cookie_str:
headers["Cookie"] = cookie_str
if sapisid:
headers["Authorization"] = make_sapisidhash(sapisid)
return headers
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Gemini Protocol ──────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def _build_gemini_payload(prompt: str, model_id: int, think_mode: int) -> Tuple[bytes, str]:
inner = [None] * 80
inner[0] = [prompt, 0, None, None, None, None, 0]
inner[1] = ["en"]
inner[2] = ["", "", "", None, None, None, None, None, None, ""]
inner[6] = [0]
inner[7] = 1
inner[10] = 1
inner[11] = 0
inner[17] = [[think_mode]]
inner[18] = 0
inner[27] = 1
inner[30] = [4]
inner[41] = [2]
inner[53] = 0
inner[59] = str(uuid.uuid4())
inner[61] = []
inner[68] = 1
inner[79] = model_id
outer = [None, json.dumps(inner)]
body = urllib.parse.urlencode({"f.req": json.dumps(outer)}).encode()
reqid = int(time.time()) % 1000000
url = (
"https://gemini.google.com/_/BardChatUi/data/"
"assistant.lamda.BardFrontendService/StreamGenerate"
f"?bl={CONFIG['gemini_bl']}&hl=en&_reqid={reqid}&rt=c"
)
return body, url
def gemini_stream_generate(prompt: str, model_id: int, think_mode: int) -> str:
body, url = _build_gemini_payload(prompt, model_id, think_mode)
headers = _build_headers()
proxy = CONFIG.get("proxy")
last_err = None
for attempt in range(CONFIG["retry_attempts"]):
try:
req = urllib.request.Request(url, data=body, headers=headers, method="POST")
ctx = ssl.create_default_context()
if proxy:
opener = urllib.request.build_opener(
urllib.request.ProxyHandler({"http": proxy, "https": proxy}),
urllib.request.HTTPSHandler(context=ctx)
)
resp = opener.open(req, timeout=CONFIG["request_timeout_sec"])
else:
resp = urllib.request.urlopen(req, context=ctx, timeout=CONFIG["request_timeout_sec"])
return resp.read().decode("utf-8", errors="replace")
except Exception as e:
last_err = e
if attempt < CONFIG["retry_attempts"] - 1:
log(f"Retry {attempt+1}/{CONFIG['retry_attempts']}: {e}", "WARN")
time.sleep(CONFIG["retry_delay_sec"])
raise last_err
def clean_gemini_text(text: str) -> str:
text = re.sub(
r'```(?:python|javascript|text)\?code_(?:reference|stdout)&code_event_index=\d+\n.*?```\n?',
'', text, flags=re.DOTALL
)
return text.strip()
def extract_response_text(raw: str) -> str:
texts = []
for line in raw.split("\n"):
if '"wrb.fr"' not in line or len(line) < 200:
continue
try:
arr = json.loads(line)
inner_str = arr[0][2]
if not inner_str or len(inner_str) < 50:
continue
inner = json.loads(inner_str)
if isinstance(inner, list) and len(inner) > 4 and inner[4]:
for part in inner[4]:
if isinstance(part, list) and len(part) > 1 and part[1]:
if isinstance(part[1], list):
for t in part[1]:
if isinstance(t, str) and len(t) > 0:
texts.append(t)
except (json.JSONDecodeError, IndexError, TypeError):
pass
text = ""
for t in reversed(texts):
if t.strip():
text = t
break
return clean_gemini_text(text)
def gemini_stream_iter(prompt: str, model_id: int, think_mode: int):
"""Streaming حقيقي يُعيد النص تدريجياً."""
body_bytes, url = _build_gemini_payload(prompt, model_id, think_mode)
body_str = body_bytes.decode()
headers = _build_headers()
proxy = CONFIG.get("proxy")
if not HAS_HTTPX:
raw = gemini_stream_generate(prompt, model_id, think_mode)
text = extract_response_text(raw)
if text:
yield text
return
prev_text = ""
transport = httpx.HTTPTransport(proxy=proxy) if proxy else None
with httpx.Client(transport=transport, timeout=CONFIG["request_timeout_sec"], verify=True) as client:
with client.stream("POST", url, content=body_str, headers=headers) as resp:
buf = ""
for chunk in resp.iter_text():
buf += chunk
while "\n" in buf:
line, buf = buf.split("\n", 1)
if '"wrb.fr"' not in line or len(line) < 200:
continue
try:
arr = json.loads(line)
inner_str = arr[0][2]
if not inner_str or len(inner_str) < 50:
continue
inner2 = json.loads(inner_str)
if isinstance(inner2, list) and len(inner2) > 4 and inner2[4]:
for part in inner2[4]:
if (isinstance(part, list) and len(part) > 1
and part[1] and isinstance(part[1], list)):
for t in part[1]:
if isinstance(t, str) and len(t) > len(prev_text):
delta = t[len(prev_text):]
delta = clean_gemini_text(delta)
if delta:
yield delta
prev_text = t
except (json.JSONDecodeError, IndexError, TypeError):
pass
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Tool Call Parser ─────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def parse_tool_calls(text: str) -> Tuple[str, Optional[List[dict]]]:
"""
محلّل Tool Calls متعدد المستويات (4 أنماط).
"""
tool_calls = []
# النمط 1: ```tool_call {...}```
p1 = r'```tool_call\s*\n(.*?)\n?```'
for m in re.finditer(p1, text, re.DOTALL):
tc = _try_parse_tc(m.group(1).strip())
if tc:
tool_calls.append(tc)
# النمط 2: ```json {...}``` (يحتوي على name + arguments)
if not tool_calls:
p2 = r'```json\s*\n(.*?)\n?```'
for m in re.finditer(p2, text, re.DOTALL):
tc = _try_parse_tc(m.group(1).strip())
if tc:
tool_calls.append(tc)
# النمط 3: ...
if not tool_calls:
p3 = r'(.*?)'
for m in re.finditer(p3, text, re.DOTALL):
tc = _try_parse_tc(m.group(1).strip())
if tc:
tool_calls.append(tc)
# النمط 4: JSON مباشر
if not tool_calls:
jp = r'\{[^{}]*"name"\s*:\s*"[^"]+[^{}]*"arguments"\s*:[^{}]*\}'
for m in re.finditer(jp, text, re.DOTALL):
tc = _try_parse_tc(m.group(0))
if tc:
tool_calls.append(tc)
if not tool_calls:
return text, None
# تنظيف النص
clean = re.sub(p1, '', text, flags=re.DOTALL)
clean = re.sub(r'```json\s*\n.*?\n?```', '', clean, flags=re.DOTALL)
clean = re.sub(r'.*?', '', clean, flags=re.DOTALL)
return clean.strip(), tool_calls
def _try_parse_tc(raw: str) -> Optional[dict]:
try:
data = json.loads(raw)
except json.JSONDecodeError:
try:
data = json.loads(_fix_json(raw))
except json.JSONDecodeError:
return None
if not isinstance(data, dict):
return None
name = data.get("name") or data.get("function") or data.get("tool")
if not name:
return None
args = data.get("arguments") or data.get("parameters") or data.get("args") or {}
if isinstance(args, str):
try:
args = json.loads(args)
except Exception:
args = {"value": args}
return {
"id": f"call_{uuid.uuid4().hex[:8]}",
"type": "function",
"function": {
"name": name,
"arguments": json.dumps(args, ensure_ascii=False),
},
}
def _fix_json(s: str) -> str:
s = re.sub(r',\s*([}\]])', r'\1', s)
s = re.sub(r'(\w+)\s*:', r'"\1":', s)
s = re.sub(r'""(\w+)""', r'"\1"', s)
return s
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Agent System Prompt ──────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
AGENT_SYSTEM_PROMPT = """You are a professional AI assistant with persistent memory and access to tools.
## MEMORY INSTRUCTIONS
- The [PERSISTENT MEMORY] block above contains everything you remember about this user from ALL previous conversations.
- Use this memory to maintain full continuity even when topics change suddenly.
- If the user references something from a past conversation, look it up in the memory block first.
- Facts, preferences, tasks, and details from previous chats are all available to you.
## TOOL USAGE
When you need to call a tool, use this EXACT format:
```tool_call
{"name": "tool_name_here", "arguments": {"param1": "value1", "param2": "value2"}}
```
### Rules:
1. Always use tools when they are needed — do not guess information a tool can provide.
2. Call tools one at a time and wait for the result before calling the next.
3. Incorporate tool results into your final answer.
4. Never fabricate tool results.
5. If a tool fails, explain what happened and try an alternative approach.
## AVAILABLE TOOLS
{tool_definitions}
## RESPONSE STYLE
- Be thorough and professional in task execution.
- Acknowledge when you remember something from a previous conversation.
- If topics change mid-conversation, handle the new topic fully while retaining all past context.
- Respond in the same language the user uses.
"""
def build_tool_definitions_text(tools: List[dict]) -> str:
if not tools:
return "No tools currently available."
lines = []
for i, tool in enumerate(tools, 1):
fn = tool.get("function", tool) if tool.get("type") == "function" else tool
name = fn.get("name", f"tool_{i}")
desc = fn.get("description", "No description")
params = fn.get("parameters", {})
lines.append(f"### {i}. `{name}`")
lines.append(f"**Description:** {desc}")
if params and params.get("properties"):
lines.append("**Parameters:**")
for pname, pinfo in params["properties"].items():
req = pname in params.get("required", [])
ptype = pinfo.get("type", "any")
pdesc = pinfo.get("description", "")
lines.append(f" - `{pname}` ({ptype}){'*' if req else ''}: {pdesc}")
lines.append("")
return "\n".join(lines)
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Prompt Builder ───────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def messages_to_prompt(
messages: list,
tools: list = None,
memory_context: str = "",
) -> str:
"""
يبني prompt كامل بالترتيب:
1. ذاكرة Zep (السياق الدائم عبر الجلسات)
2. System prompt الأدوات (إذا كانت هناك أدوات)
3. رسائل المحادثة الحالية
"""
parts = []
# ─── 1. الذاكرة الدائمة أولاً (الأعلى أولوية)
if memory_context and memory_context.strip():
parts.append(memory_context)
# ─── 2. System prompt للأدوات
if tools:
tool_defs = build_tool_definitions_text(tools)
sys_prompt = AGENT_SYSTEM_PROMPT.replace("{tool_definitions}", tool_defs)
parts.append(f"[SYSTEM — AGENT INSTRUCTIONS]\n{sys_prompt}")
else:
parts.append(
"[SYSTEM]\nYou are a professional AI assistant with persistent memory. "
"Use the memory context above to maintain continuity across all conversations."
)
# ─── 3. معالجة رسائل المحادثة
for msg in messages:
role = msg.get("role", "user") if isinstance(msg, dict) else msg.role
content = msg.get("content", "") if isinstance(msg, dict) else (msg.content or "")
tc_data = msg.get("tool_calls") if isinstance(msg, dict) else getattr(msg, "tool_calls", None)
# تحويل content متعدد الأجزاء
if isinstance(content, list):
content = " ".join(
c.get("text", "") for c in content
if isinstance(c, dict) and c.get("type") in ("text", "input_text")
)
content = (content or "").strip()
if role == "system":
# دمج system prompt المستخدم مع ما سبق
if content:
parts.append(f"[ADDITIONAL SYSTEM INSTRUCTIONS]\n{content}")
elif role == "user":
if content:
parts.append(f"[USER]: {content}")
elif role == "assistant":
if tc_data:
tc_str = "\n".join(
f"```tool_call\n{json.dumps({'name': tc['function']['name'], 'arguments': json.loads(tc['function'].get('arguments', '{}'))}, ensure_ascii=False)}\n```"
for tc in tc_data
)
assistant_part = ""
if content:
assistant_part = f"[ASSISTANT]: {content}\n"
assistant_part += f"[ASSISTANT called tools]:\n{tc_str}"
parts.append(assistant_part)
elif content:
parts.append(f"[ASSISTANT]: {content}")
elif role == "tool":
tool_name = msg.get("name", "") if isinstance(msg, dict) else getattr(msg, "name", "")
tool_id = msg.get("tool_call_id", "") if isinstance(msg, dict) else getattr(msg, "tool_call_id", "")
label = tool_name or tool_id or "unknown_tool"
if content:
parts.append(
f"[TOOL RESULT — {label}]:\n```json\n{content}\n```\n"
f"→ Continue based on this result."
)
return "\n\n".join(p for p in parts if p and p.strip())
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Agent Loop ───────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def run_agent_sync(
messages: List[dict],
model_id: int,
think_mode: int,
tools: Optional[List[dict]],
memory_context: str = "",
max_iterations: int = None,
) -> Tuple[str, Optional[List[dict]], List[dict]]:
"""
Agent Loop (ReAct):
Reason → Act (tool call) → Observe (result) → Reason → ...
يعيد (final_text, tool_calls_if_any, updated_messages)
ملاحظة: في وضع non-executor (الافتراضي)، يُعيد tool calls للـ client
ليُنفّذها ويُرسل النتائج في الطلب التالي. هذا هو السلوك الصحيح
لبروتوكول OpenAI Tool Calling.
"""
if max_iterations is None:
max_iterations = CONFIG["agent_max_iterations"]
current_messages = list(messages)
for iteration in range(1, max_iterations + 1):
log(f"Agent iteration {iteration}/{max_iterations}", "AGENT")
# بناء الـ prompt مع الذاكرة
prompt = messages_to_prompt(current_messages, tools, memory_context)
try:
raw = gemini_stream_generate(prompt, model_id, think_mode)
text = extract_response_text(raw)
except Exception as e:
log(f"Gemini call error (iter {iteration}): {e}", "ERROR")
return f"[Error communicating with model: {e}]", None, current_messages
if not text:
log("Empty response from Gemini", "WARN")
if iteration == 1:
return "[No response from model]", None, current_messages
break
# تحليل tool calls
clean_text, tool_calls = parse_tool_calls(text)
if tool_calls and tools:
names = [tc["function"]["name"] for tc in tool_calls]
log(f"Tool calls detected: {names}", "TOOL")
# أضف رد المساعد للسياق
current_messages.append({
"role": "assistant",
"content": clean_text or None,
"tool_calls": tool_calls,
})
# أعد tool calls للـ client (OpenAI protocol)
return clean_text or "", tool_calls, current_messages
else:
# رد نهائي
log(f"✅ Final answer at iteration {iteration}", "AGENT")
return clean_text or text, None, current_messages
log(f"Max iterations ({max_iterations}) reached", "WARN")
return text or "[Max iterations reached]", None, current_messages
def run_agent_with_executor_sync(
messages: List[dict],
model_id: int,
think_mode: int,
tools: Optional[List[dict]],
tool_executor: Optional[Callable[[str, dict], str]],
memory_context: str = "",
max_iterations: int = None,
) -> Tuple[str, Optional[List[dict]], List[dict]]:
"""
Agent Loop الكامل مع تنفيذ الأدوات داخلياً.
يستخدم عندما يكون tool_executor متاحاً (نادر في هذا السيناريو).
"""
if max_iterations is None:
max_iterations = CONFIG["agent_max_iterations"]
current_messages = list(messages)
for iteration in range(1, max_iterations + 1):
log(f"[Full Agent] Iteration {iteration}/{max_iterations}", "AGENT")
prompt = messages_to_prompt(current_messages, tools, memory_context)
try:
raw = gemini_stream_generate(prompt, model_id, think_mode)
text = extract_response_text(raw)
except Exception as e:
log(f"Gemini error: {e}", "ERROR")
return f"Error: {e}", None, current_messages
if not text:
break
clean_text, tool_calls = parse_tool_calls(text)
if not tool_calls or not tools:
log(f"✅ Final answer at iteration {iteration}", "AGENT")
return clean_text or text, None, current_messages
current_messages.append({
"role": "assistant",
"content": clean_text or None,
"tool_calls": tool_calls,
})
for tc in tool_calls:
fn_name = tc["function"]["name"]
try:
fn_args = json.loads(tc["function"]["arguments"])
except Exception:
fn_args = {}
log(f"Executing: {fn_name}({fn_args})", "TOOL")
if tool_executor:
try:
result = tool_executor(fn_name, fn_args)
result_str = result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)
except Exception as e:
result_str = json.dumps({"error": str(e)})
log(f"Tool error: {e}", "ERROR")
else:
result_str = json.dumps({"status": "no_executor", "tool": fn_name})
current_messages.append({
"role": "tool",
"tool_call_id": tc["id"],
"name": fn_name,
"content": result_str,
})
return text or "", None, current_messages
def call_gemini_sync(
prompt: str,
model_id: int,
think_mode: int,
tools: list = None,
) -> Tuple[str, Optional[list]]:
"""استدعاء بسيط بدون agent loop."""
raw = gemini_stream_generate(prompt, model_id, think_mode)
text = extract_response_text(raw)
tool_calls = None
if tools and text:
text, tool_calls = parse_tool_calls(text)
return text or "", tool_calls
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Model Resolution ─────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def resolve_model(model_name: str) -> Tuple[Optional[str], Optional[int], Optional[int], Optional[str]]:
think_override = None
if "@think=" in model_name:
model_name, think_str = model_name.rsplit("@think=", 1)
try:
think_override = int(think_str)
except ValueError:
pass
cfg = MODELS.get(model_name)
if not cfg:
default = CONFIG.get("default_model", "gemini-3.5-flash")
cfg = MODELS.get(default)
if cfg:
log(f"Unknown model '{model_name}', falling back to '{default}'", "WARN")
model_name = default
else:
return None, None, None, f"Unknown model: {model_name}. Available: {list(MODELS.keys())}"
return model_name, cfg["mode"], (think_override if think_override is not None else cfg["think"]), None
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Helpers ──────────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def extract_last_user_query(messages: List[dict]) -> str:
"""يستخرج آخر رسالة للمستخدم كنص للبحث الدلالي."""
for msg in reversed(messages):
if msg.get("role") == "user":
content = msg.get("content", "")
if isinstance(content, list):
content = " ".join(
c.get("text", "") for c in content
if isinstance(c, dict) and c.get("type") in ("text", "input_text")
)
return (content or "").strip()
return ""
def _build_completion_response(
cid: str, created: int, model: str, text: str,
tool_calls: Optional[list], prompt: str = ""
) -> dict:
msg: dict = {"role": "assistant", "content": text or None}
if tool_calls:
msg["tool_calls"] = tool_calls
finish = "tool_calls" if tool_calls else "stop"
return {
"id": cid,
"object": "chat.completion",
"created": created,
"model": model,
"choices": [{"index": 0, "message": msg, "finish_reason": finish}],
"usage": {
"prompt_tokens": len(prompt) // 4,
"completion_tokens": len(text) // 4 if text else 0,
"total_tokens": (len(prompt) + len(text)) // 4 if text else len(prompt) // 4,
},
}
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Authentication ───────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
security = HTTPBearer(auto_error=False)
async def verify_api_key(
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
) -> str:
api_keys = CONFIG.get("api_keys", [])
if not api_keys:
return "anonymous"
if not credentials:
raise HTTPException(status_code=401, detail="Missing Authorization header")
token = credentials.credentials
if token not in api_keys:
raise HTTPException(status_code=401, detail="Invalid API key")
return token
# ═══════════════════════════════════════════════════════════════════════════════
# ─── FastAPI App ──────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
@asynccontextmanager
async def lifespan(app: FastAPI):
await services.init_db()
await services.start_cleanup()
await zep_memory.init() # ← تهيئة Zep Cloud
log(f"gemini-web2api v{__version__} started", "INFO")
log(f"Memory: {zep_memory.status()}", "MEM")
yield
if services._cleanup_task:
services._cleanup_task.cancel()
app = FastAPI(
title="Gemini Web2API + Zep Memory (Agent Edition)",
version=__version__,
description="OpenAI-compatible Gemini proxy with persistent Zep Cloud memory and full Agent Loop",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ═══════════════════════════════════════════════════════════════════════════════
# ─── /v1/models ───────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
@app.get("/v1/models")
async def list_models():
return {
"object": "list",
"data": [
{
"id": name,
"object": "model",
"created": 1700000000,
"owned_by": "google",
"description": cfg["desc"],
"capabilities": {
"tool_use": True,
"streaming": True,
"agent_loop": True,
"persistent_memory": zep_memory.enabled,
},
}
for name, cfg in MODELS.items()
],
}
@app.get("/v1beta/models")
async def google_models_list():
return {
"models": [
{
"name": f"models/{name}",
"displayName": name,
"description": cfg["desc"],
"supportedGenerationMethods": ["generateContent", "streamGenerateContent"],
}
for name, cfg in MODELS.items()
]
}
# ═══════════════════════════════════════════════════════════════════════════════
# ─── /v1/chat/completions ─────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
@app.post("/v1/chat/completions")
async def chat_completions(
body: ChatCompletionRequest,
request: Request,
_: str = Depends(verify_api_key),
):
# ─── Rate Limiting
client_ip = request.client.host
allowed, wait_time = await services.check_rate_limit(client_ip)
if not allowed:
raise HTTPException(
status_code=429,
detail=f"Rate limit exceeded. Retry in {wait_time}s.",
headers={"Retry-After": str(wait_time)},
)
# ─── Model Resolution
model_name, model_id, think_mode, err = resolve_model(
body.model or CONFIG["default_model"]
)
if err:
raise HTTPException(status_code=400, detail=err)
tools = body.tools
messages = [m.dict(exclude_none=True) for m in body.messages]
# user_id: نستخدمه كمفتاح للذاكرة — ثابت لكل مستخدم
user_id = body.user or client_ip or "default"
# ═══════════════════════════════════════════════════════
# ✅ ZEP: تجهيز الجلسة واسترجاع الذاكرة الكاملة
# ═══════════════════════════════════════════════════════
await zep_memory.ensure_user_session(user_id)
last_query = extract_last_user_query(messages)
# استرجاع الذاكرة والبحث الدلالي بالتوازي
memory_context = await zep_memory.get_full_context(user_id, last_query)
# ═══════════════════════════════════════════════════════
cid = f"chatcmpl-{uuid.uuid4().hex[:12]}"
created = int(time.time())
# Cache key (بدون ذاكرة لأن الذاكرة تتغير)
cache_key = hashlib.md5(
f"{model_name}:{user_id}:{last_query[:200]}:{bool(tools)}".encode()
).hexdigest()
# ─── Streaming ───────────────────────────────────────────
if body.stream:
async def stream_gen():
full_text = ""
final_tool_calls = None
try:
if tools:
loop = asyncio.get_event_loop()
text, tc, updated_msgs = await loop.run_in_executor(
None, run_agent_sync,
messages, model_id, think_mode, tools,
memory_context, CONFIG["agent_max_iterations"]
)
full_text = text or ""
final_tool_calls = tc
if tc:
# إرسال tool calls chunk
chunk = {
"id": cid, "object": "chat.completion.chunk",
"created": created, "model": model_name,
"choices": [{
"index": 0,
"delta": {"role": "assistant", "content": full_text or None, "tool_calls": tc},
"finish_reason": "tool_calls",
}],
}
yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
else:
if full_text:
chunk = {
"id": cid, "object": "chat.completion.chunk",
"created": created, "model": model_name,
"choices": [{
"index": 0,
"delta": {"role": "assistant", "content": full_text},
"finish_reason": None,
}],
}
yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
else:
# Streaming بدون أدوات
prompt = messages_to_prompt(messages, memory_context=memory_context)
for delta in gemini_stream_iter(prompt, model_id, think_mode):
full_text += delta
chunk = {
"id": cid, "object": "chat.completion.chunk",
"created": created, "model": model_name,
"choices": [{
"index": 0,
"delta": {"content": delta},
"finish_reason": None,
}],
}
yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
# Final chunk
finish = "tool_calls" if final_tool_calls else "stop"
yield f"data: {json.dumps({'id': cid, 'object': 'chat.completion.chunk', 'created': created, 'model': model_name, 'choices': [{'index': 0, 'delta': {}, 'finish_reason': finish}]})}\n\n"
yield "data: [DONE]\n\n"
# ✅ ZEP: حفظ المحادثة بعد الانتهاء
if full_text and not final_tool_calls:
save_msgs = [m for m in messages if m.get("role") in ("user", "assistant", "tool")]
await zep_memory.save_turn(user_id, save_msgs, full_text)
except Exception as e:
log(f"Stream error: {e}", "ERROR")
yield f"data: {json.dumps({'error': {'message': str(e), 'type': 'server_error'}})}\n\n"
return StreamingResponse(stream_gen(), media_type="text/event-stream")
# ─── Non-streaming ────────────────────────────────────────
# Cache (بدون أدوات فقط)
if not tools:
cached = await services.get_cache(cache_key)
if cached:
return JSONResponse(content=json.loads(cached))
try:
loop = asyncio.get_event_loop()
if tools:
text, tool_calls, updated_msgs = await loop.run_in_executor(
None, run_agent_sync,
messages, model_id, think_mode, tools,
memory_context, CONFIG["agent_max_iterations"]
)
else:
prompt = messages_to_prompt(messages, memory_context=memory_context)
text, tool_calls = await loop.run_in_executor(
None, call_gemini_sync,
prompt, model_id, think_mode, None
)
except Exception as e:
log(f"Error: {e}", "ERROR")
raise HTTPException(status_code=502, detail=f"Upstream error: {e}")
# ✅ ZEP: حفظ المحادثة
if text and not tool_calls:
save_msgs = [m for m in messages if m.get("role") in ("user", "assistant", "tool")]
await zep_memory.save_turn(user_id, save_msgs, text)
prompt_text = messages_to_prompt(messages)
resp_data = _build_completion_response(cid, created, model_name, text, tool_calls, prompt_text)
if not tools and text:
await services.set_cache(cache_key, json.dumps(resp_data))
return JSONResponse(content=resp_data)
# ═══════════════════════════════════════════════════════════════════════════════
# ─── /v1/responses (Codex CLI / Responses API) ────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
@app.post("/v1/responses")
async def responses_api(
request: Request,
_: str = Depends(verify_api_key),
):
req = await request.json()
model_name, model_id, think_mode, err = resolve_model(
req.get("model", CONFIG["default_model"])
)
if err:
raise HTTPException(status_code=400, detail=err)
input_items = req.get("input", [])
tools = req.get("tools")
messages = []
user_id = req.get("user", "default")
if req.get("instructions"):
messages.append({"role": "system", "content": req["instructions"]})
if isinstance(input_items, str):
messages.append({"role": "user", "content": input_items})
elif isinstance(input_items, list):
for item in input_items:
if isinstance(item, str):
messages.append({"role": "user", "content": item})
elif isinstance(item, dict):
if item.get("type") == "function_call_output":
messages.append({
"role": "tool",
"tool_call_id": item.get("call_id", ""),
"name": item.get("name", ""),
"content": item.get("output", ""),
})
elif item.get("role") == "assistant":
cp = item.get("content", [])
text_acc, tc_list = "", []
if isinstance(cp, list):
for c in cp:
if isinstance(c, dict):
if c.get("type") == "output_text":
text_acc += c.get("text", "")
elif c.get("type") == "function_call":
tc_list.append(c)
elif isinstance(cp, str):
text_acc = cp
m = {"role": "assistant", "content": text_acc or None}
if tc_list:
m["tool_calls"] = [
{
"id": tc.get("call_id", f"call_{i}"),
"type": "function",
"function": {
"name": tc.get("name", ""),
"arguments": tc.get("arguments", "{}"),
}
}
for i, tc in enumerate(tc_list)
]
messages.append(m)
else:
role = item.get("role", "user")
content = item.get("content", "")
if isinstance(content, list):
content = " ".join(
c.get("text", "") for c in content
if c.get("type") in ("text", "input_text")
)
messages.append({"role": role, "content": content})
if tools:
tools = [
{
"type": "function",
"function": {
"name": t["name"],
"description": t.get("description", ""),
"parameters": t.get("parameters", {}),
}
}
if t.get("type") == "function" and "function" not in t else t
for t in tools
]
if not any(m.get("content") for m in messages):
raise HTTPException(status_code=400, detail="Empty input")
# ✅ Zep Memory
await zep_memory.ensure_user_session(user_id)
last_query = extract_last_user_query(messages)
memory_context = await zep_memory.get_full_context(user_id, last_query)
loop = asyncio.get_event_loop()
try:
if tools:
text, tool_calls, _ = await loop.run_in_executor(
None, run_agent_sync,
messages, model_id, think_mode, tools,
memory_context, CONFIG["agent_max_iterations"]
)
else:
prompt = messages_to_prompt(messages, memory_context=memory_context)
text, tool_calls = await loop.run_in_executor(
None, call_gemini_sync, prompt, model_id, think_mode, None
)
except Exception as e:
raise HTTPException(status_code=502, detail=f"Upstream error: {e}")
# ✅ حفظ في Zep
if text and not tool_calls:
save_msgs = [m for m in messages if m.get("role") in ("user", "assistant", "tool")]
await zep_memory.save_turn(user_id, save_msgs, text)
rid = f"resp_{uuid.uuid4().hex[:16]}"
mid = f"msg_{uuid.uuid4().hex[:12]}"
output = []
if tool_calls:
for tc in tool_calls:
output.append({
"type": "function_call",
"id": tc["id"],
"call_id": tc["id"],
"name": tc["function"]["name"],
"arguments": tc["function"]["arguments"],
"status": "completed",
})
if text or not tool_calls:
output.append({
"type": "message",
"id": mid,
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": text or "", "annotations": []}],
})
prompt_text = messages_to_prompt(messages)
if req.get("stream"):
async def responses_stream():
yield f"event: response.created\ndata: {json.dumps({'type': 'response.created', 'response': {'id': rid, 'object': 'response', 'status': 'in_progress', 'model': model_name, 'output': []}})}\n\n"
for item in output:
if item["type"] == "function_call":
ev = {
"type": "response.function_call_arguments.done",
"item_id": item["id"], "call_id": item["call_id"],
"name": item["name"], "arguments": item["arguments"],
}
yield f"event: response.function_call_arguments.done\ndata: {json.dumps(ev)}\n\n"
elif item["type"] == "message":
for ci, cp in enumerate(item["content"]):
ev = {
"type": "response.output_text.done",
"item_id": item["id"], "content_index": ci, "text": cp["text"],
}
yield f"event: response.output_text.done\ndata: {json.dumps(ev)}\n\n"
resp_obj = {"id": rid, "object": "response", "status": "completed", "model": model_name, "output": output}
yield f"event: response.completed\ndata: {json.dumps({'type': 'response.completed', 'response': resp_obj})}\n\n"
return StreamingResponse(responses_stream(), media_type="text/event-stream")
return JSONResponse({
"id": rid, "object": "response", "created_at": int(time.time()),
"status": "completed", "model": model_name, "output": output,
"usage": {
"input_tokens": len(prompt_text) // 4,
"output_tokens": len(text) // 4 if text else 0,
"total_tokens": (len(prompt_text) + len(text)) // 4 if text else len(prompt_text) // 4,
},
})
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Google Native API /v1beta ────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def _google_contents_to_messages(req: dict) -> list:
messages = []
sys_inst = req.get("systemInstruction")
if sys_inst:
sys_text = " ".join(p.get("text", "") for p in sys_inst.get("parts", []) if p.get("text"))
if sys_text:
messages.append({"role": "system", "content": sys_text})
for content in req.get("contents", []):
role = "assistant" if content.get("role") == "model" else "user"
text = " ".join(p.get("text", "") for p in content.get("parts", []) if p.get("text"))
messages.append({"role": role, "content": text})
return messages
@app.post("/v1beta/models/{model_path:path}")
async def google_generate(model_path: str, request: Request):
req = await request.json()
model_name_raw = model_path.split(":")[0]
is_stream = ":streamGenerateContent" in model_path
model_name, model_id, think_mode, err = resolve_model(model_name_raw)
if err:
raise HTTPException(status_code=400, detail=err)
messages = _google_contents_to_messages(req)
if not messages:
raise HTTPException(status_code=400, detail="Empty content")
# Zep للـ Google API أيضاً
user_id = req.get("user", "google_api")
await zep_memory.ensure_user_session(user_id)
last_query = extract_last_user_query(messages)
memory_context = await zep_memory.get_full_context(user_id, last_query)
prompt = messages_to_prompt(messages, memory_context=memory_context)
loop = asyncio.get_event_loop()
try:
text, _ = await loop.run_in_executor(
None, call_gemini_sync, prompt, model_id, think_mode, None
)
except Exception as e:
raise HTTPException(status_code=502, detail=f"Upstream error: {e}")
if text:
await zep_memory.save_turn(user_id, messages, text)
candidate = {
"content": {"parts": [{"text": text or ""}], "role": "model"},
"finishReason": "STOP",
"index": 0,
}
response_obj = {
"candidates": [candidate],
"usageMetadata": {
"promptTokenCount": len(prompt) // 4,
"candidatesTokenCount": len(text) // 4 if text else 0,
"totalTokenCount": (len(prompt) + len(text)) // 4 if text else len(prompt) // 4,
},
"modelVersion": model_name,
}
if is_stream:
async def google_stream():
yield f"data: {json.dumps(response_obj)}\n\n"
return StreamingResponse(google_stream(), media_type="text/event-stream")
return JSONResponse(response_obj)
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Health & Root ────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
@app.get("/")
async def root():
return {
"service": "gemini-web2api",
"version": __version__,
"models": list(MODELS.keys()),
"memory": zep_memory.status(),
"features": {
"persistent_memory": zep_memory.enabled,
"agent_loop": True,
"tool_calling": True,
"streaming": True,
"cache": True,
"rate_limiting": True,
},
"endpoints": [
"/v1/chat/completions",
"/v1/models",
"/v1/responses",
"/v1beta/models",
"/health",
"/memory/status",
],
}
@app.get("/health")
async def health():
return {
"status": "healthy",
"timestamp": time.time(),
"version": __version__,
"memory_enabled": zep_memory.enabled,
}
@app.get("/memory/status")
async def memory_status():
"""حالة الذاكرة الدائمة."""
return {
"provider": "Zep Cloud",
"enabled": zep_memory.enabled,
"config": {
"memory_type": CONFIG.get("zep_memory_type"),
"recent_messages": CONFIG.get("zep_recent_messages"),
"search_limit": CONFIG.get("zep_search_limit"),
"min_fact_rating": CONFIG.get("zep_min_fact_rating"),
},
"error": zep_memory._init_error if not zep_memory.enabled else None,
}
@app.post("/memory/add-fact")
async def add_fact_endpoint(
request: Request,
_: str = Depends(verify_api_key),
):
"""إضافة حقيقة صريحة لذاكرة مستخدم محدد."""
body = await request.json()
user_id = body.get("user_id", "default")
fact = body.get("fact", "")
if not fact:
raise HTTPException(status_code=400, detail="fact is required")
await zep_memory.ensure_user_session(user_id)
await zep_memory.add_fact(user_id, fact)
return {"status": "ok", "fact": fact, "user_id": user_id}
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Gradio UI ────────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
def build_gradio_interface():
async def chat_handler(
message: str,
history: list,
model_choice: str,
user_id_box: str,
request: gr.Request,
):
if not message or not message.strip():
yield history
return
model_name, model_id, think_mode, err = resolve_model(model_choice)
if err:
history.append((message, f"❌ {err}"))
yield history
return
uid = (user_id_box or "").strip() or (request.client.host if request.client else "default")
# تجهيز الجلسة واسترجاع الذاكرة
await zep_memory.ensure_user_session(uid)
memory_context = await zep_memory.get_full_context(uid, message)
history.append((message, "⏳ ..."))
yield history
full_response = ""
try:
msgs = [{"role": "user", "content": message}]
prompt = messages_to_prompt(msgs, memory_context=memory_context)
for chunk in gemini_stream_iter(prompt, model_id, think_mode):
full_response += chunk
history[-1] = (message, full_response)
yield history
except Exception as e:
full_response = f"❌ Error: {e}"
history[-1] = (message, full_response)
yield history
return
# حفظ في Zep
if full_response:
await zep_memory.save_turn(
uid,
[{"role": "user", "content": message}],
full_response,
)
log(f"Saved chat turn to Zep for user={uid}", "MEM")
mem_status = "✅ Zep Cloud متصل" if HAS_ZEP else "❌ zep-cloud غير مثبّت"
api_key_display = CONFIG.get("api_keys", ["sk-gemini"])[0]
with gr.Blocks(
title="Gemini Agent + Zep Memory",
theme=gr.themes.Soft(),
) as demo:
gr.Markdown(f"""
# 🤖 Gemini Professional Agent + 🧠 Zep Persistent Memory
### v{__version__} — ذاكرة دائمة عبر كل الجلسات
| الخاصية | القيمة |
|---------|--------|
| **Memory** | {mem_status} |
| **Agent Loop** | ✅ حتى {CONFIG['agent_max_iterations']} iterations |
| **API Key** | `{api_key_display}` |
| **Base URL** | `/v1` |
""")
with gr.Row():
with gr.Column(scale=4):
chatbot = gr.Chatbot(
height=520,
label="المحادثة — تُحفظ تلقائياً في Zep",
show_copy_button=True,
)
with gr.Column(scale=1):
gr.Markdown("""
### 🧠 الذاكرة الدائمة
يتذكر الوكيل عبر **كل الجلسات**:
- الحقائق المستخرجة تلقائياً
- ملخص كامل للمحادثات
- آخر 20 رسالة بالتفصيل
- بحث دلالي في كل التاريخ
### ✅ المميزات
- يتذكر تغييرات الموضوع
- يحفظ تفاصيل المهام
- Agent Loop احترافي
- Tool Calling موثوق
- Streaming حقيقي
""")
with gr.Row():
user_id_box = gr.Textbox(
label="🆔 User ID (اختياري — للتمييز بين المستخدمين)",
placeholder="مثال: user_123 أو اتركه فارغاً",
scale=2,
)
model_dd = gr.Dropdown(
choices=list(MODELS.keys()),
value=CONFIG["default_model"],
label="🤖 النموذج",
scale=2,
)
with gr.Row():
msg_box = gr.Textbox(
label="رسالتك",
placeholder="اكتب هنا... يمكنك تغيير الموضوع في أي وقت، الوكيل يتذكر كل شيء ✨",
lines=3,
scale=5,
)
send_btn = gr.Button("إرسال ▶", variant="primary", scale=1)
with gr.Row():
clear_btn = gr.Button("🗑️ مسح العرض (لا يمسح الذاكرة)")
# ربط الأحداث
msg_box.submit(
chat_handler,
[msg_box, chatbot, model_dd, user_id_box],
[chatbot],
).then(lambda: "", None, [msg_box])
send_btn.click(
chat_handler,
[msg_box, chatbot, model_dd, user_id_box],
[chatbot],
).then(lambda: "", None, [msg_box])
clear_btn.click(lambda: [], None, [chatbot])
return demo
demo = build_gradio_interface()
app = gr.mount_gradio_app(app, demo, path="/chat")
# ═══════════════════════════════════════════════════════════════════════════════
# ─── Entry Point ──────────────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════════
if __name__ == "__main__":
import uvicorn
parser = argparse.ArgumentParser(description="Gemini Web2API + Zep Memory Agent")
parser.add_argument("--port", type=int, default=CONFIG["port"])
parser.add_argument("--host", type=str, default=CONFIG["host"])
parser.add_argument("--cookie-file", type=str, default=None)
parser.add_argument("--proxy", type=str, default=None)
parser.add_argument("--max-iterations", type=int, default=CONFIG["agent_max_iterations"])
parser.add_argument("--zep-key", type=str, default=None, help="Zep Cloud API Key")
parser.add_argument("--version", action="version", version=f"v{__version__}")
args = parser.parse_args()
if args.cookie_file:
CONFIG["cookie_file"] = args.cookie_file
if args.proxy:
CONFIG["proxy"] = args.proxy
if args.zep_key:
CONFIG["zep_api_key"] = args.zep_key
CONFIG["agent_max_iterations"] = args.max_iterations
print(f"\n{'═'*65}")
print(f" gemini-web2api v{__version__} — Professional Agent + Zep Memory")
print(f"{'═'*65}")
print(f" API: http://localhost:{args.port}/v1")
print(f" Chat UI: http://localhost:{args.port}/chat")
print(f" Health: http://localhost:{args.port}/health")
print(f" Memory: http://localhost:{args.port}/memory/status")
print(f" Models: {', '.join(MODELS.keys())}")
print(f" Agent Loop: max {CONFIG['agent_max_iterations']} iterations")
print(f" Zep Memory: {'✅ Key configured' if CONFIG.get('zep_api_key') else '❌ ZEP_API_KEY not set'}")
print(f" Cookie: {'✅' if CONFIG.get('cookie_file') or CONFIG.get('_cookie_inline') else '❌ none (anonymous)'}")
print(f" Proxy: {CONFIG.get('proxy') or '❌ none'}")
print(f"{'═'*65}\n")
uvicorn.run(
app,
host=args.host,
port=args.port,
loop="asyncio",
timeout_keep_alive=120,
)