agentforge-litellm flagged credential stealer (PyPI) — Cremit · NHI Credential Stealer Index

/incidentsfield log

agentforge-litellm flagged credential stealer (PyPI) — Cremit · NHI Credential Stealer Index

from __future__ import annotations

// agentforge_litellm-0.2.3/src/agentforge_litellm/client.py (excerpt)

"""`LiteLLMClient` — `LLMClient` over LiteLLM's unified router."""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from agentforge_core.contracts.llm import LLMClient
from agentforge_core.production.exceptions import ModuleError
from agentforge_core.resolver import register_provider
from agentforge_core.values.messages import (
    LLMResponse,
    Message,
    StopReason,
    TokenUsage,
    ToolCall,
    ToolSpec,
)

if TYPE_CHECKING:
    from agentforge_litellm._runner import LiteLLMRunner


_PROVIDER_NAME = "litellm"
_DEFAULT_TIMEOUT_SECONDS = 60.0

_FINISH_REASON_MAP: dict[str, StopReason] = {
    "stop": "end_turn",
    "length": "max_tokens",
    "tool_calls": "tool_use",
    "function_call": "tool_use",
    "content_filter": "other",
}


@register_provider("litellm")
class LiteLLMClient(LLMClient):
    """`LLMClient` over LiteLLM. `model` is whatever LiteLLM accepts."""

    def __init__(
        self,
        *,
        model_id: str,
        runner: LiteLLMRunner | None = None,
        timeout_seconds: float = _DEFAULT_TIMEOUT_SECONDS,
        extra: dict[str, Any] | None = None,
    ) -> None:
        if not model_id:
            raise ValueError("model_id must be a non-empty string")
        if timeout_seconds <= 0:
            raise ValueError("timeout_seconds must be > 0")
        self._runner = runner if runner is not None else _build_sdk_runner()
        self._model = model_id
        self._timeout_seconds = timeout_seconds