mirror of
https://github.com/Alishahryar1/free-claude-code.git
synced 2026-04-28 03:20:01 +00:00
2fe15bd2cd
Add proxy support for providers based on [doc](https://www.python-httpx.org/advanced/proxies/): - Add per-provider proxy support (HTTP and SOCKS5) for all 4 providers: nvidia_nim, open_router, lmstudio, llamacpp - Each provider gets its own env var (NVIDIA_NIM_PROXY, OPENROUTER_PROXY, LMSTUDIO_PROXY, LLAMACPP_PROXY) for independent proxy configuration --------- Co-authored-by: Alishahryar1 <alishahryar2@gmail.com>
59 lines
1.7 KiB
Python
59 lines
1.7 KiB
Python
"""Base provider interface - extend this to implement your own provider."""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any
|
|
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class ProviderConfig(BaseModel):
|
|
"""Configuration for a provider.
|
|
|
|
Base fields apply to all providers. Provider-specific parameters
|
|
(e.g. NIM temperature, top_p) are passed by the provider constructor.
|
|
"""
|
|
|
|
api_key: str
|
|
base_url: str | None = None
|
|
rate_limit: int | None = None
|
|
rate_window: int = 60
|
|
max_concurrency: int = 5
|
|
http_read_timeout: float = 300.0
|
|
http_write_timeout: float = 10.0
|
|
http_connect_timeout: float = 2.0
|
|
enable_thinking: bool = True
|
|
proxy: str = ""
|
|
|
|
|
|
class BaseProvider(ABC):
|
|
"""Base class for all providers. Extend this to add your own."""
|
|
|
|
def __init__(self, config: ProviderConfig):
|
|
self._config = config
|
|
|
|
def _is_thinking_enabled(self, request: Any) -> bool:
|
|
"""Return whether thinking should be enabled for this request."""
|
|
thinking = getattr(request, "thinking", None)
|
|
request_enabled = (
|
|
thinking.enabled
|
|
if thinking is not None and hasattr(thinking, "enabled")
|
|
else True
|
|
)
|
|
return self._config.enable_thinking and request_enabled
|
|
|
|
@abstractmethod
|
|
async def cleanup(self) -> None:
|
|
"""Release any resources held by this provider."""
|
|
|
|
@abstractmethod
|
|
async def stream_response(
|
|
self,
|
|
request: Any,
|
|
input_tokens: int = 0,
|
|
*,
|
|
request_id: str | None = None,
|
|
) -> AsyncIterator[str]:
|
|
"""Stream response in Anthropic SSE format."""
|
|
if False:
|
|
yield "" # Required for ty/mypy to accept abstract async generator
|