update GigaChat access_token logic

This commit is contained in:
klankerr 2026-02-27 15:16:33 +03:00 committed by Kirill Basyul
parent 61acb373ca
commit 20bb81ad3f
5 changed files with 98 additions and 7 deletions

View file

@ -91,7 +91,7 @@ def get_access_token(
auth_url = auth_url or _get_auth_url()
# Check cache
cache_key = f"gigachat_token:{credentials[:16]}"
cache_key = f"gigachat_token:{credentials[:16]}:{scope}:{auth_url}"
cached = _token_cache.get_cache(cache_key)
if cached:
token, expires_at = cached
@ -128,7 +128,8 @@ async def get_access_token_async(
auth_url = auth_url or _get_auth_url()
# Check cache
cache_key = f"gigachat_token:{credentials[:16]}"
cache_key = f"gigachat_token:{credentials[:16]}:{scope}:{auth_url}"
cached = _token_cache.get_cache(cache_key)
if cached:
token, expires_at = cached

View file

@ -83,6 +83,8 @@ class GigaChatConfig(BaseConfig):
# Instance variables for current request context
self._current_credentials: Optional[str] = None
self._current_api_base: Optional[str] = None
self._current_scope: Optional[str] = None
self._current_auth_url: Optional[str] = None
def get_complete_url(
self,
@ -106,6 +108,8 @@ class GigaChatConfig(BaseConfig):
litellm_params: dict,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
scope: Optional[str] = None,
auth_url: Optional[str] = None,
) -> dict:
"""
Set up headers with OAuth token.
@ -116,11 +120,32 @@ class GigaChatConfig(BaseConfig):
or get_secret_str("GIGACHAT_CREDENTIALS")
or get_secret_str("GIGACHAT_API_KEY")
)
access_token = get_access_token(credentials=credentials)
scope = (
scope
or optional_params.get("scope")
or litellm_params.get("scope")
or get_secret_str("GIGACHAT_SCOPE")
)
auth_url = (
auth_url
or optional_params.get("auth_url")
or litellm_params.get("auth_url")
or get_secret_str("GIGACHAT_AUTH_URL")
)
access_token = get_access_token(
credentials=credentials,
scope=scope,
auth_url=auth_url,
)
# Store credentials for image uploads
self._current_credentials = credentials
self._current_api_base = api_base
self._current_scope = scope
self._current_auth_url = auth_url
headers["Authorization"] = f"Bearer {access_token}"
headers["Content-Type"] = "application/json"
@ -273,6 +298,8 @@ class GigaChatConfig(BaseConfig):
image_url=image_url,
credentials=self._current_credentials,
api_base=self._current_api_base,
scope=self._current_scope,
auth_url=self._current_auth_url,
)
except Exception as e:
verbose_logger.error(f"Failed to upload image: {e}")

View file

@ -12,6 +12,7 @@ import httpx
from litellm import LlmProviders
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.secret_managers.main import get_secret_str
from litellm.llms.base_llm.embedding.transformation import BaseEmbeddingConfig
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.types.llms.openai import AllEmbeddingInputValues, AllMessageValues
@ -189,12 +190,38 @@ class GigaChatEmbeddingConfig(BaseEmbeddingConfig):
litellm_params: dict,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
scope: Optional[str] = None,
auth_url: Optional[str] = None,
) -> dict:
"""
Set up headers with OAuth token for GigaChat.
"""
# Get access token via OAuth
access_token = get_access_token(api_key)
# Get access token
credentials = (
api_key
or get_secret_str("GIGACHAT_CREDENTIALS")
or get_secret_str("GIGACHAT_API_KEY")
)
scope = (
scope
or optional_params.get("scope")
or litellm_params.get("scope")
or get_secret_str("GIGACHAT_SCOPE")
)
auth_url = (
auth_url
or optional_params.get("auth_url")
or litellm_params.get("auth_url")
or get_secret_str("GIGACHAT_AUTH_URL")
)
access_token = get_access_token(
credentials=credentials,
scope=scope,
auth_url=auth_url,
)
default_headers = {
"Content-Type": "application/json",

View file

@ -82,6 +82,8 @@ def upload_file_sync(
image_url: str,
credentials: Optional[str] = None,
api_base: Optional[str] = None,
scope: Optional[str] = None,
auth_url: Optional[str] = None,
) -> Optional[str]:
"""
Upload file to GigaChat and return file_id (sync).
@ -114,7 +116,11 @@ def upload_file_sync(
filename = f"{uuid.uuid4()}.{ext}"
# Get access token
access_token = get_access_token(credentials)
access_token = get_access_token(
credentials=credentials,
scope=scope,
auth_url=auth_url,
)
# Upload to GigaChat
base_url = api_base or GIGACHAT_BASE_URL
@ -147,6 +153,8 @@ async def upload_file_async(
image_url: str,
credentials: Optional[str] = None,
api_base: Optional[str] = None,
scope: Optional[str] = None,
auth_url: Optional[str] = None,
) -> Optional[str]:
"""
Upload file to GigaChat and return file_id (async).
@ -179,7 +187,11 @@ async def upload_file_async(
filename = f"{uuid.uuid4()}.{ext}"
# Get access token
access_token = await get_access_token_async(credentials)
access_token = await get_access_token_async(
credentials=credentials,
scope=scope,
auth_url=auth_url,
)
# Upload to GigaChat
base_url = api_base or GIGACHAT_BASE_URL

View file

@ -1157,6 +1157,30 @@
"options": null,
"default_value": "https://gigachat.devices.sberbank.ru/api/v1"
},
{
"key": "auth_url",
"label": "GigaChat Auth URL",
"placeholder": "https://ngw.devices.sberbank.ru:9443/api/v2/oauth",
"tooltip": null,
"required": false,
"field_type": "text",
"options": null,
"default_value": "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"
},
{
"key": "scope",
"label": "GigaChat Scope",
"placeholder": null,
"tooltip": null,
"required": false,
"field_type": "select",
"options": [
"GIGACHAT_API_PERS",
"GIGACHAT_API_B2B",
"GIGACHAT_API_CORP"
],
"default_value": "GIGACHAT_API_PERS"
},
{
"key": "api_key",
"label": "GigaChat API Key",