Gemini2API 服务端实现，Python抛砖引玉版本

___ · 发表于昨天 18:28

立刻注册账号，享受更清爽的界面！

您需要登录才可以下载或查看，没有账号？注册

×

https://github.com/HanaokaYuzu/Gemini-API
然后花了一点时间做了一个API，可以对接 Cherry Studio
➡️代码

import asyncio
import json
from datetime import datetime, timezone
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import List, Optional
import time
import uuid
import logging
from gemini_webapi import GeminiClient, set_log_level
from gemini_webapi.constants import Model
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
set_log_level("INFO")
app = FastAPI(title="Gemini API FastAPI Server")
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Global client
gemini_client = None
# Authentication credentials
SECURE_1PSID = ""
SECURE_1PSIDTS = ""
# Pydantic models for API requests and responses
class Message(BaseModel):
role: str
content: str
name: Optional[str] = None
class ChatCompletionRequest(BaseModel):
model: str
messages: List[Message]
temperature: Optional[float] = 0.7
top_p: Optional[float] = 1.0
n: Optional[int] = 1
stream: Optional[bool] = False
max_tokens: Optional[int] = None
presence_penalty: Optional[float] = 0
frequency_penalty: Optional[float] = 0
user: Optional[str] = None
class Choice(BaseModel):
index: int
message: Message
finish_reason: str
class Usage(BaseModel):
prompt_tokens: int
completion_tokens: int
total_tokens: int
class ChatCompletionResponse(BaseModel):
id: str
object: str = "chat.completion"
created: int
model: str
choices: List[Choice]
usage: Usage
class ModelData(BaseModel):
id: str
object: str = "model"
created: int
owned_by: str = "google"
class ModelList(BaseModel):
object: str = "list"
data: List[ModelData]
# Simple error handler middleware
@app.middleware("http")
async def error_handling(request: Request, call_next):
try:
return await call_next(request)
except Exception as e:
logger.error(f"Request failed: {str(e)}")
return JSONResponse(
status_code=500,
content={ "error": { "message": str(e), "type": "internal_server_error" } }
)
# Get list of available models
@app.get("/v1/models")
async def list_models():
"""返回 gemini_webapi 中声明的模型列表"""
now = int(datetime.now(tz=timezone.utc).timestamp())
data = [
{
"id": m.model_name, # 如 "gemini-2.0-flash"
"object": "model",
"created": now,
"owned_by": "google-gemini-web"
}
for m in Model
]
print(data)
return {"object": "list", "data": data}
# Helper to convert between Gemini and OpenAI model names
def map_model_name(openai_model_name: str) -> Model:
"""根据模型名称字符串查找匹配的 Model 枚举值"""
# 打印所有可用模型以便调试
all_models = [m.model_name if hasattr(m, "model_name") else str(m) for m in Model]
logger.info(f"Available models: {all_models}")
# 首先尝试直接查找匹配的模型名称
for m in Model:
model_name = m.model_name if hasattr(m, "model_name") else str(m)
if openai_model_name.lower() in model_name.lower():
return m
# 如果找不到匹配项，使用默认映射
model_keywords = {
"gemini-pro": ["pro", "2.0"],
"gemini-pro-vision": ["vision", "pro"],
"gemini-flash": ["flash", "2.0"],
"gemini-1.5-pro": ["1.5", "pro"],
"gemini-1.5-flash": ["1.5", "flash"],
}
# 根据关键词匹配
keywords = model_keywords.get(openai_model_name, ["pro"]) # 默认使用pro模型
for m in Model:
model_name = m.model_name if hasattr(m, "model_name") else str(m)
if all(kw.lower() in model_name.lower() for kw in keywords):
return m
# 如果还是找不到，返回第一个模型
return next(iter(Model))
# Prepare conversation history from OpenAI messages format
def prepare_conversation(messages: List[Message]) -> str:
conversation = ""
for msg in messages:
if msg.role == "system":
conversation += f"System: {msg.content}\n\n"
elif msg.role == "user":
conversation += f"Human: {msg.content}\n\n"
elif msg.role == "assistant":
conversation += f"Assistant: {msg.content}\n\n"
# Add a final prompt for the assistant to respond to
conversation += "Assistant: "
return conversation
# Dependency to get the initialized Gemini client
async def get_gemini_client():
global gemini_client
if gemini_client is None:
try:
gemini_client = GeminiClient(SECURE_1PSID, SECURE_1PSIDTS)
await gemini_client.init(timeout=30)
except Exception as e:
logger.error(f"Failed to initialize Gemini client: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"Failed to initialize Gemini client: {str(e)}"
)
return gemini_client
@app.post("/v1/chat/completions")
async def create_chat_completion(request: ChatCompletionRequest):
try:
# 确保客户端已初始化
global gemini_client
if gemini_client is None:
gemini_client = GeminiClient(SECURE_1PSID, SECURE_1PSIDTS)
await gemini_client.init(timeout=30)
logger.info("Gemini client initialized successfully")
# 转换消息为对话格式
conversation = prepare_conversation(request.messages)
logger.info(f"Prepared conversation: {conversation}")
# 获取适当的模型
model = map_model_name(request.model)
logger.info(f"Using model: {model}")
# 生成响应
logger.info("Sending request to Gemini...")
response = await gemini_client.generate_content(conversation, model=model)
# 提取文本响应
reply_text = ""
if hasattr(response, "text"):
reply_text = response.text
else:
reply_text = str(response)
logger.info(f"Response: {reply_text}")
if not reply_text or reply_text.strip() == "":
logger.warning("Empty response received from Gemini")
reply_text = "服务器返回了空响应。请检查 Gemini API 凭据是否有效。"
# 创建响应对象
completion_id = f"chatcmpl-{uuid.uuid4()}"
created_time = int(time.time())
# 检查客户端是否请求流式响应
if request.stream:
# 实现流式响应
async def generate_stream():
# 创建 SSE 格式的流式响应
# 先发送开始事件
data = {
"id": completion_id,
"object": "chat.completion.chunk",
"created": created_time,
"model": request.model,
"choices": [
{
"index": 0,
"delta": {
"role": "assistant"
},
"finish_reason": None
}
]
}
yield f"data: {json.dumps(data)}\n\n"
# 模拟流式输出 - 将文本按字符分割发送
for char in reply_text:
data = {
"id": completion_id,
"object": "chat.completion.chunk",
"created": created_time,
"model": request.model,
"choices": [
{
"index": 0,
"delta": {
"content": char
},
"finish_reason": None
}
]
}
yield f"data: {json.dumps(data)}\n\n"
# 可选：添加短暂延迟以模拟真实的流式输出
await asyncio.sleep(0.01)
# 发送结束事件
data = {
"id": completion_id,
"object": "chat.completion.chunk",
"created": created_time,
"model": request.model,
"choices": [
{
"index": 0,
"delta": { },
"finish_reason": "stop"
}
]
}
yield f"data: {json.dumps(data)}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
generate_stream(),
media_type="text/event-stream"
)
else:
# 非流式响应（原来的逻辑）
result = {
"id": completion_id,
"object": "chat.completion",
"created": created_time,
"model": request.model,
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": reply_text
},
"finish_reason": "stop"
}
],
"usage": {
"prompt_tokens": len(conversation.split()),
"completion_tokens": len(reply_text.split()),
"total_tokens": len(conversation.split()) + len(reply_text.split())
}
}
logger.info(f"Returning response: {result}")
return result
except Exception as e:
logger.error(f"Error generating completion: {str(e)}", exc_info=True)
raise HTTPException(
status_code=500,
detail=f"Error generating completion: {str(e)}"
)
@app.get("/")
async def root():
return { "status": "online", "message": "Gemini API FastAPI Server is running" }
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, log_level="info")

复制代码

0. 填入Secure_1PSID 和Secure_1PSIDTS （登录下 Gemini 去 Application - Cookie 里面找）

Secure_1PSID = "COOKIE VALUE HERE"
Secure_1PSIDTS = "COOKIE VALUE HERE"

复制代码

1.安装一下依赖

uv add fastapi uvicorn gemini-webapi

复制代码

或者pip也可以

pip install fastapi uvicorn gemini-webapi

复制代码

2. 激活一下环境

source venv/bin/activate

复制代码

3. 启动

uvicorn main:app --reload --host 127.0.0.1 --port 8000

复制代码

tips: 没有任何密码，仅作为抛砖引玉

这个问题一般是 IP 不太行或者请求太频繁（后者等待一段时间即可），见issue：https://github.com/HanaokaYuzu/Gemini-API/issues/72

"message": "500 {"detail":"Error generating completion: Failed to initialize client. SECURE_1PSIDTS could get expired frequently,

复制代码

Gemini2API 服务端实现，Python抛砖引玉版本

立刻注册账号，享受更清爽的界面！

浏览过的版块

相关网站

站内导航

商业推广