diff --git a/litellm/a2a_protocol/litellm_completion_bridge/handler.py b/litellm/a2a_protocol/litellm_completion_bridge/handler.py index 67ffcf4f8f..52e471ff70 100644 --- a/litellm/a2a_protocol/litellm_completion_bridge/handler.py +++ b/litellm/a2a_protocol/litellm_completion_bridge/handler.py @@ -20,9 +20,20 @@ from litellm.a2a_protocol.litellm_completion_bridge.transformation import ( ) from litellm.a2a_protocol.providers.config_manager import A2AProviderConfigManager +# litellm_params key carrying the authenticated principal (hashed virtual key) so +# A2A provider configs can scope provider-side state (e.g. LangFlow session memory) +# per key instead of trusting the client-supplied A2A contextId. +A2A_USER_API_KEY_HASH_PARAM = "litellm_a2a_user_api_key_hash" + # Agent metadata fields stored in litellm_params that are not valid litellm.acompletion() kwargs _AGENT_ONLY_PARAMS = frozenset( - {"is_public", "agent_name", "agent_id", "agent_card_params"} + { + "is_public", + "agent_name", + "agent_id", + "agent_card_params", + A2A_USER_API_KEY_HASH_PARAM, + } ) @@ -37,6 +48,8 @@ class A2ACompletionBridgeHandler: params: Dict[str, Any], litellm_params: Dict[str, Any], api_base: Optional[str] = None, + *, + _skip_a2a_provider_routing: bool = False, ) -> Dict[str, Any]: """ Handle non-streaming A2A request via litellm.acompletion. @@ -50,25 +63,24 @@ class A2ACompletionBridgeHandler: Returns: A2A SendMessageResponse dict """ - # Get provider config for custom_llm_provider custom_llm_provider = litellm_params.get("custom_llm_provider") - a2a_provider_config = A2AProviderConfigManager.get_provider_config( - custom_llm_provider=custom_llm_provider, - model=litellm_params.get("model"), - ) - - # If provider config exists, use it - if a2a_provider_config is not None: - verbose_logger.info(f"A2A: Using provider config for {custom_llm_provider}") - - response_data = await a2a_provider_config.handle_non_streaming( - request_id=request_id, - params=params, - api_base=api_base, - litellm_params=litellm_params, + if not _skip_a2a_provider_routing: + a2a_provider_config = A2AProviderConfigManager.get_provider_config( + custom_llm_provider=custom_llm_provider, + model=litellm_params.get("model"), ) - return response_data + if a2a_provider_config is not None: + verbose_logger.info( + f"A2A: Using provider config for {custom_llm_provider}" + ) + + return await a2a_provider_config.handle_non_streaming( + request_id=request_id, + params=params, + api_base=api_base, + litellm_params=litellm_params, + ) # Extract message from params message = params.get("message", {}) @@ -137,6 +149,8 @@ class A2ACompletionBridgeHandler: params: Dict[str, Any], litellm_params: Dict[str, Any], api_base: Optional[str] = None, + *, + _skip_a2a_provider_routing: bool = False, ) -> AsyncIterator[Dict[str, Any]]: """ Handle streaming A2A request via litellm.acompletion with stream=True. @@ -156,28 +170,27 @@ class A2ACompletionBridgeHandler: Yields: A2A streaming response events """ - # Get provider config for custom_llm_provider custom_llm_provider = litellm_params.get("custom_llm_provider") - a2a_provider_config = A2AProviderConfigManager.get_provider_config( - custom_llm_provider=custom_llm_provider, - model=litellm_params.get("model"), - ) - - # If provider config exists, use it - if a2a_provider_config is not None: - verbose_logger.info( - f"A2A: Using provider config for {custom_llm_provider} (streaming)" + if not _skip_a2a_provider_routing: + a2a_provider_config = A2AProviderConfigManager.get_provider_config( + custom_llm_provider=custom_llm_provider, + model=litellm_params.get("model"), ) - async for chunk in a2a_provider_config.handle_streaming( - request_id=request_id, - params=params, - api_base=api_base, - litellm_params=litellm_params, - ): - yield chunk + if a2a_provider_config is not None: + verbose_logger.info( + f"A2A: Using provider config for {custom_llm_provider} (streaming)" + ) - return + async for chunk in a2a_provider_config.handle_streaming( + request_id=request_id, + params=params, + api_base=api_base, + litellm_params=litellm_params, + ): + yield chunk + + return # Extract message from params message = params.get("message", {}) diff --git a/litellm/a2a_protocol/providers/config_manager.py b/litellm/a2a_protocol/providers/config_manager.py index ecb8f66bde..a421afec18 100644 --- a/litellm/a2a_protocol/providers/config_manager.py +++ b/litellm/a2a_protocol/providers/config_manager.py @@ -48,6 +48,11 @@ class A2AProviderConfigManager: return BedrockAgentCoreA2AConfig() + if custom_llm_provider == "langflow": + from litellm.a2a_protocol.providers.langflow.config import LangFlowA2AConfig + + return LangFlowA2AConfig() + if custom_llm_provider == "watsonx_orchestrate": from litellm.a2a_protocol.providers.watsonx_orchestrate.config import ( WatsonxOrchestrateA2AConfig, diff --git a/litellm/a2a_protocol/providers/langflow/__init__.py b/litellm/a2a_protocol/providers/langflow/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/litellm/a2a_protocol/providers/langflow/config.py b/litellm/a2a_protocol/providers/langflow/config.py new file mode 100644 index 0000000000..9302c38126 --- /dev/null +++ b/litellm/a2a_protocol/providers/langflow/config.py @@ -0,0 +1,62 @@ +from typing import Any, AsyncIterator, Dict, Optional + +from litellm.a2a_protocol.litellm_completion_bridge.handler import ( + A2A_USER_API_KEY_HASH_PARAM, + A2ACompletionBridgeHandler, +) +from litellm.a2a_protocol.providers.base import BaseA2AProviderConfig +from litellm.llms.langflow.a2a import merge_a2a_session_into_litellm_params + + +class LangFlowA2AConfig(BaseA2AProviderConfig): + """A2A bridge for LangFlow: scopes contextId to the authenticated key as the + LangFlow session_id, then uses completion.""" + + async def handle_non_streaming( + self, + request_id: str, + params: Dict[str, Any], + api_base: Optional[str] = None, + **kwargs, + ) -> Dict[str, Any]: + litellm_params = kwargs.get("litellm_params") + if not litellm_params: + raise ValueError( + "litellm_params is required for LangFlowA2AConfig " + "(must contain custom_llm_provider and model)" + ) + litellm_params = merge_a2a_session_into_litellm_params( + litellm_params, params, litellm_params.get(A2A_USER_API_KEY_HASH_PARAM) + ) + return await A2ACompletionBridgeHandler.handle_non_streaming( + request_id=request_id, + params=params, + litellm_params=litellm_params, + api_base=api_base, + _skip_a2a_provider_routing=True, + ) + + async def handle_streaming( + self, + request_id: str, + params: Dict[str, Any], + api_base: Optional[str] = None, + **kwargs, + ) -> AsyncIterator[Dict[str, Any]]: + litellm_params = kwargs.get("litellm_params") + if not litellm_params: + raise ValueError( + "litellm_params is required for LangFlowA2AConfig " + "(must contain custom_llm_provider and model)" + ) + litellm_params = merge_a2a_session_into_litellm_params( + litellm_params, params, litellm_params.get(A2A_USER_API_KEY_HASH_PARAM) + ) + async for chunk in A2ACompletionBridgeHandler.handle_streaming( + request_id=request_id, + params=params, + litellm_params=litellm_params, + api_base=api_base, + _skip_a2a_provider_routing=True, + ): + yield chunk diff --git a/litellm/llms/langflow/__init__.py b/litellm/llms/langflow/__init__.py new file mode 100644 index 0000000000..d1270fc91f --- /dev/null +++ b/litellm/llms/langflow/__init__.py @@ -0,0 +1 @@ +"""LangFlow LLM provider for LiteLLM.""" diff --git a/litellm/llms/langflow/a2a.py b/litellm/llms/langflow/a2a.py new file mode 100644 index 0000000000..dbe3e02401 --- /dev/null +++ b/litellm/llms/langflow/a2a.py @@ -0,0 +1,37 @@ +import hashlib +from typing import Any, Dict, Optional + + +def get_session_id_from_a2a_params(params: Dict[str, Any]) -> Optional[str]: + message = params.get("message", {}) + if isinstance(message, dict): + return message.get("contextId") + return getattr(message, "contextId", None) + + +def scope_session_to_principal(session_id: str, principal: Optional[str]) -> str: + """ + Bind a client-supplied A2A contextId to the authenticated principal. + + Without this, two distinct keys authorized for the same LangFlow agent could + set the same contextId and read/append to each other's LangFlow memory. The + principal is hashed (it is already a hashed token) so the raw value is never + sent to the LangFlow backend, while the original contextId is kept as a + suffix for operator-side correlation. + """ + if not principal: + return session_id + principal_prefix = hashlib.sha256(principal.encode("utf-8")).hexdigest()[:16] + return f"{principal_prefix}-{session_id}" + + +def merge_a2a_session_into_litellm_params( + litellm_params: Dict[str, Any], + params: Dict[str, Any], + principal: Optional[str] = None, +) -> Dict[str, Any]: + merged = dict(litellm_params) + session_id = get_session_id_from_a2a_params(params) + if session_id and "session_id" not in merged: + merged["session_id"] = scope_session_to_principal(session_id, principal) + return merged diff --git a/litellm/llms/langflow/chat/__init__.py b/litellm/llms/langflow/chat/__init__.py new file mode 100644 index 0000000000..286b12e31f --- /dev/null +++ b/litellm/llms/langflow/chat/__init__.py @@ -0,0 +1 @@ +"""LangFlow chat transformation.""" diff --git a/litellm/llms/langflow/chat/transformation.py b/litellm/llms/langflow/chat/transformation.py new file mode 100644 index 0000000000..f898163ad0 --- /dev/null +++ b/litellm/llms/langflow/chat/transformation.py @@ -0,0 +1,327 @@ +"""LangFlow run API: POST {api_base}/api/v1/run/{flow_id}""" + +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union +from urllib.parse import quote + +import httpx + +from litellm._logging import verbose_logger +from litellm.litellm_core_utils.prompt_templates.common_utils import ( + convert_content_list_to_str, +) +from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException +from litellm.types.llms.openai import AllMessageValues +from litellm.types.utils import Choices, Message, ModelResponse, Usage + +if TYPE_CHECKING: + from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj + from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler + from litellm.utils import CustomStreamWrapper + + LiteLLMLoggingObj = _LiteLLMLoggingObj +else: + LiteLLMLoggingObj = Any + HTTPHandler = Any + AsyncHTTPHandler = Any + CustomStreamWrapper = Any + + +class LangFlowError(BaseLLMException): + """Exception class for LangFlow API errors.""" + + pass + + +class LangFlowConfig(BaseConfig): + """ + Configuration for the LangFlow API. + + LangFlow is a visual, low-code platform for building AI agents and pipelines. + Each flow has a unique flow_id and is invoked via a simple HTTP endpoint. + """ + + def __init__(self, **kwargs): + super().__init__(**kwargs) + + def _get_openai_compatible_provider_info( + self, + api_base: Optional[str], + api_key: Optional[str], + ) -> Tuple[Optional[str], Optional[str]]: + from litellm.secret_managers.main import get_secret_str + + api_base = ( + api_base or get_secret_str("LANGFLOW_API_BASE") or "http://localhost:7860" + ) + api_key = api_key or get_secret_str("LANGFLOW_API_KEY") + return api_base, api_key + + def get_supported_openai_params(self, model: str) -> List[str]: + return ["stream"] + + def map_openai_params( + self, + non_default_params: dict, + optional_params: dict, + model: str, + drop_params: bool, + ) -> dict: + return optional_params + + def _get_flow_id(self, model: str, optional_params: dict) -> str: + """ + Extract flow_id from the authorized model name only. + + Model format: "langflow/{flow_id}". Request kwargs must not override + flow_id (would allow calling another flow with the same API key). + """ + if optional_params.get("flow_id") is not None: + raise LangFlowError( + status_code=400, + message=( + "flow_id cannot be set via request parameters; " + "use model langflow/{flow_id}" + ), + ) + + flow_id = (model.split("/", 1)[1] if "/" in model else model).strip() + if not flow_id: + raise LangFlowError( + status_code=400, + message="flow_id is required; use model langflow/{flow_id}", + ) + return flow_id + + def get_complete_url( + self, + api_base: Optional[str], + api_key: Optional[str], + model: str, + optional_params: dict, + litellm_params: dict, + stream: Optional[bool] = None, + ) -> str: + if api_base is None: + raise ValueError( + "api_base is required for LangFlow. Set it via LANGFLOW_API_BASE env var or api_base parameter." + ) + + api_base = api_base.rstrip("/") + flow_id = quote(self._get_flow_id(model, optional_params), safe="") + return f"{api_base}/api/v1/run/{flow_id}" + + def _get_last_user_message(self, messages: List[AllMessageValues]) -> str: + """Extract the text of the last user message to use as input_value.""" + for msg in reversed(messages): + if msg.get("role") == "user": + content = msg.get("content", "") + if isinstance(content, list): + content = convert_content_list_to_str(msg) + if not isinstance(content, str): + content = str(content) + return content + + # Fallback: use last message regardless of role + if messages: + content = messages[-1].get("content", "") + if isinstance(content, list): + content = convert_content_list_to_str(messages[-1]) + if not isinstance(content, str): + content = str(content) + return content + + return "" + + def _reject_caller_tweaks(self, params: dict) -> None: + if params.get("tweaks") is not None: + raise LangFlowError( + status_code=400, + message=( + "tweaks cannot be set via request parameters; they would " + "override the operator-configured LangFlow flow components" + ), + ) + + def transform_request( + self, + model: str, + messages: List[AllMessageValues], + optional_params: dict, + litellm_params: dict, + headers: dict, + ) -> dict: + """ + Transform the request to LangFlow format. + + LangFlow request format: + { + "input_value": "", + "input_type": "chat", + "output_type": "chat", + "session_id": "" + } + """ + self._reject_caller_tweaks(optional_params) + + input_value = self._get_last_user_message(messages) + + payload: Dict[str, Any] = { + "input_value": input_value, + "input_type": optional_params.get("input_type", "chat"), + "output_type": optional_params.get("output_type", "chat"), + } + + session_id = optional_params.get("session_id") + if session_id: + payload["session_id"] = session_id + + verbose_logger.debug(f"LangFlow request payload: {payload}") + return payload + + def _extract_content_from_response(self, response_json: dict) -> Optional[str]: + """ + Extract the assistant text from a LangFlow run response. + + Expected structure: + {"outputs": [{"outputs": [{"results": {"message": {"text": "..."}}}]}]} + + Returns None when no message text is present so the caller can surface an + explicit error instead of forwarding a raw JSON blob as the answer. + """ + outputs = response_json.get("outputs", []) + if not (isinstance(outputs, list) and outputs): + return None + + first_output = outputs[0] + if not isinstance(first_output, dict): + return None + + inner_outputs = first_output.get("outputs", []) + if not (isinstance(inner_outputs, list) and inner_outputs): + return None + + first_inner = inner_outputs[0] + if not isinstance(first_inner, dict): + return None + + results = first_inner.get("results", {}) + if isinstance(results, dict): + message = results.get("message", {}) + if isinstance(message, dict) and message.get("text"): + return message["text"] + + outputs_dict = first_inner.get("outputs", {}) + if isinstance(outputs_dict, dict): + for val in outputs_dict.values(): + if isinstance(val, dict): + msg = val.get("message", {}) + if isinstance(msg, dict) and msg.get("text"): + return msg["text"] + + return None + + def transform_response( + self, + model: str, + raw_response: httpx.Response, + model_response: ModelResponse, + logging_obj: LiteLLMLoggingObj, + request_data: dict, + messages: List[AllMessageValues], + optional_params: dict, + litellm_params: dict, + encoding: Any, + api_key: Optional[str] = None, + json_mode: Optional[bool] = None, + ) -> ModelResponse: + try: + response_json = raw_response.json() + except Exception as e: + raise LangFlowError( + message=f"LangFlow returned a non-JSON response: {e}", + status_code=raw_response.status_code, + ) + + verbose_logger.debug(f"LangFlow response: {response_json}") + + content = self._extract_content_from_response(response_json) + if content is None: + raise LangFlowError( + message=( + "Could not extract a message from the LangFlow response; " + "ensure the flow ends in a Chat Output component" + ), + status_code=500, + ) + + message = Message(content=content, role="assistant") + choice = Choices(finish_reason="stop", index=0, message=message) + + model_response.choices = [choice] + model_response.model = model + + try: + from litellm.utils import token_counter + + prompt_tokens = token_counter(model=model, messages=messages) + completion_tokens = token_counter( + model=model, text=content, count_response_tokens=True + ) + usage = Usage( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=prompt_tokens + completion_tokens, + ) + setattr(model_response, "usage", usage) + except Exception as e: + verbose_logger.warning(f"Failed to calculate token usage: {e}") + + return model_response + + def sign_request( + self, + headers: dict, + optional_params: dict, + request_data: dict, + api_base: str, + api_key: Optional[str] = None, + model: Optional[str] = None, + stream: Optional[bool] = None, + fake_stream: Optional[bool] = None, + ) -> Tuple[dict, Optional[bytes]]: + self._reject_caller_tweaks(request_data) + return headers, None + + def validate_environment( + self, + headers: dict, + model: str, + messages: List[AllMessageValues], + optional_params: dict, + litellm_params: dict, + api_key: Optional[str] = None, + api_base: Optional[str] = None, + ) -> dict: + headers["Content-Type"] = "application/json" + + if api_key: + headers["x-api-key"] = api_key + + return headers + + def get_error_class( + self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers] + ) -> BaseLLMException: + return LangFlowError(status_code=status_code, message=error_message) + + @property + def supports_stream_param_in_request_body(self) -> bool: + return False + + def should_fake_stream( + self, + model: Optional[str], + stream: Optional[bool], + custom_llm_provider: Optional[str] = None, + ) -> bool: + return stream is True diff --git a/litellm/main.py b/litellm/main.py index 09c70998cf..3ef094042e 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -4503,6 +4503,39 @@ def completion( # type: ignore # noqa: PLR0915 client=client, ) + elif custom_llm_provider == "langflow": + # LangFlow - Visual AI Agent Platform + from litellm.llms.langflow.chat.transformation import LangFlowConfig + + ( + api_base, + api_key, + ) = LangFlowConfig()._get_openai_compatible_provider_info( + api_base=api_base or litellm.api_base, + api_key=api_key or litellm.api_key, + ) + + headers = headers or litellm.headers + + response = base_llm_http_handler.completion( + model=model, + stream=stream, + messages=messages, + acompletion=acompletion, + api_base=api_base, + model_response=model_response, + optional_params=optional_params, + litellm_params=litellm_params, + shared_session=shared_session, + custom_llm_provider=custom_llm_provider, + timeout=timeout, + headers=headers, + encoding=_get_encoding(), + api_key=api_key, + logging_obj=logging, + client=client, + ) + else: raise LiteLLMUnknownProvider( model=model, custom_llm_provider=custom_llm_provider diff --git a/litellm/proxy/agent_endpoints/a2a_endpoints.py b/litellm/proxy/agent_endpoints/a2a_endpoints.py index 993d30e381..7b56155982 100644 --- a/litellm/proxy/agent_endpoints/a2a_endpoints.py +++ b/litellm/proxy/agent_endpoints/a2a_endpoints.py @@ -389,6 +389,19 @@ async def invoke_agent_a2a( # noqa: PLR0915 litellm_params = agent.litellm_params or {} custom_llm_provider = litellm_params.get("custom_llm_provider") + # Hand the authenticated key hash to the completion bridge so provider + # configs can scope provider-side session state per key (e.g. LangFlow + # session memory) instead of trusting the client-supplied A2A contextId. + if custom_llm_provider and user_api_key_dict.api_key: + from litellm.a2a_protocol.litellm_completion_bridge.handler import ( + A2A_USER_API_KEY_HASH_PARAM, + ) + + litellm_params = { + **litellm_params, + A2A_USER_API_KEY_HASH_PARAM: user_api_key_dict.api_key, + } + # URL is required unless using completion bridge with a provider that derives endpoint from model # (e.g., bedrock/agentcore derives endpoint from ARN in model string) if not agent_url and not custom_llm_provider: diff --git a/litellm/proxy/public_endpoints/agent_create_fields.json b/litellm/proxy/public_endpoints/agent_create_fields.json index e58bd97cce..36484cc106 100644 --- a/litellm/proxy/public_endpoints/agent_create_fields.json +++ b/litellm/proxy/public_endpoints/agent_create_fields.json @@ -7,6 +7,48 @@ "credential_fields": [], "litellm_params_template": {} }, + { + "agent_type": "langflow", + "agent_type_display_name": "LangFlow", + "description": "Connect to LangFlow AI agents via the LangFlow Platform API", + "logo_url": "/ui/assets/logos/langflow.svg", + "model_template": "langflow/{flow_id}", + "credential_fields": [ + { + "key": "flow_id", + "label": "Flow ID", + "placeholder": "your-flow-id", + "tooltip": "The Flow ID from your LangFlow deployment (found in the flow URL or settings)", + "required": true, + "field_type": "text", + "default_value": null, + "include_in_litellm_params": false + }, + { + "key": "api_base", + "label": "LangFlow API Base", + "placeholder": "http://localhost:7860", + "tooltip": "The base URL for your LangFlow server (e.g., http://localhost:7860 or your deployed LangFlow URL)", + "required": true, + "field_type": "text", + "default_value": "http://localhost:7860", + "include_in_litellm_params": true + }, + { + "key": "api_key", + "label": "LangFlow API Key", + "placeholder": null, + "tooltip": "API key for authenticating with your LangFlow server (x-api-key header)", + "required": false, + "field_type": "password", + "default_value": null, + "include_in_litellm_params": true + } + ], + "litellm_params_template": { + "custom_llm_provider": "langflow" + } + }, { "agent_type": "langgraph", "agent_type_display_name": "LangGraph", diff --git a/litellm/types/utils.py b/litellm/types/utils.py index d3c2c8c18f..63c2513aed 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -3364,6 +3364,7 @@ class LlmProviders(str, Enum): AMAZON_NOVA = "amazon_nova" A2A_AGENT = "a2a_agent" LANGGRAPH = "langgraph" + LANGFLOW = "langflow" MINIMAX = "minimax" SYNTHETIC = "synthetic" APERTIS = "apertis" diff --git a/litellm/utils.py b/litellm/utils.py index 68982ea2b3..6188206148 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -8394,6 +8394,10 @@ class ProviderConfigManager: lambda: ProviderConfigManager._get_langgraph_config(), False, ), + LlmProviders.LANGFLOW: ( + lambda: ProviderConfigManager._get_langflow_config(), + False, + ), } @staticmethod @@ -8465,6 +8469,13 @@ class ProviderConfigManager: return LangGraphConfig() + @staticmethod + def _get_langflow_config() -> BaseConfig: + """Get LangFlow config.""" + from litellm.llms.langflow.chat.transformation import LangFlowConfig + + return LangFlowConfig() + @staticmethod def get_provider_chat_config( # noqa: PLR0915 model: str, diff --git a/provider_endpoints_support.json b/provider_endpoints_support.json index 1e8357a813..3a01541060 100644 --- a/provider_endpoints_support.json +++ b/provider_endpoints_support.json @@ -2430,6 +2430,24 @@ "interactions": true } }, + "langflow": { + "display_name": "LangFlow (`langflow`)", + "url": "https://docs.litellm.ai/docs/providers/langflow", + "endpoints": { + "chat_completions": true, + "messages": false, + "responses": false, + "embeddings": false, + "image_generations": false, + "audio_transcriptions": false, + "audio_speech": false, + "moderations": false, + "batches": false, + "rerank": false, + "a2a": true, + "interactions": false + } + }, "vertex_ai/agent_engine": { "display_name": "Vertex AI Agent Engine (`vertex_ai/agent_engine`)", "url": "https://docs.litellm.ai/docs/providers/vertex_ai_agent_engine", diff --git a/tests/test_litellm/llms/langflow/chat/test_langflow_chat_transformation.py b/tests/test_litellm/llms/langflow/chat/test_langflow_chat_transformation.py new file mode 100644 index 0000000000..c03919a065 --- /dev/null +++ b/tests/test_litellm/llms/langflow/chat/test_langflow_chat_transformation.py @@ -0,0 +1,398 @@ +from unittest.mock import MagicMock, patch + +import httpx +import pytest + +from litellm.llms.langflow.chat.transformation import LangFlowConfig, LangFlowError +from litellm.types.utils import LlmProviders, ModelResponse +from litellm.utils import ProviderConfigManager + + +def test_flow_id_cannot_be_overridden_via_optional_params(): + config = LangFlowConfig() + url = config.get_complete_url( + api_base="http://localhost:7860", + api_key=None, + model="langflow/authorized-flow", + optional_params={}, + litellm_params={}, + stream=False, + ) + assert url.endswith("/api/v1/run/authorized-flow") + + with pytest.raises(LangFlowError): + config.get_complete_url( + api_base="http://localhost:7860", + api_key=None, + model="langflow/authorized-flow", + optional_params={"flow_id": "malicious-flow"}, + litellm_params={}, + stream=False, + ) + + +def test_langflow_config_get_complete_url(): + config = LangFlowConfig() + url = config.get_complete_url( + api_base="http://localhost:7860", + api_key=None, + model="langflow/my-flow-id", + optional_params={}, + litellm_params={}, + stream=False, + ) + assert url == "http://localhost:7860/api/v1/run/my-flow-id" + + +def test_langflow_config_get_complete_url_requires_api_base(): + config = LangFlowConfig() + with pytest.raises(ValueError): + config.get_complete_url( + api_base=None, + api_key=None, + model="langflow/my-flow-id", + optional_params={}, + litellm_params={}, + stream=False, + ) + + +def test_langflow_config_flow_id_is_path_segment_encoded(): + config = LangFlowConfig() + url = config.get_complete_url( + api_base="http://localhost:7860", + api_key=None, + model="langflow/../../secret?x=1", + optional_params={}, + litellm_params={}, + stream=False, + ) + assert url == "http://localhost:7860/api/v1/run/..%2F..%2Fsecret%3Fx%3D1" + assert "/api/v1/run/" in url + assert url.rsplit("/api/v1/run/", 1)[1] not in ("..", "../..") + + +@pytest.mark.parametrize("model", ["langflow/", "langflow/ "]) +def test_langflow_config_rejects_empty_flow_id(model): + config = LangFlowConfig() + with pytest.raises(LangFlowError): + config.get_complete_url( + api_base="http://localhost:7860", + api_key=None, + model=model, + optional_params={}, + litellm_params={}, + stream=False, + ) + + +def test_langflow_config_strips_flow_id_whitespace(): + config = LangFlowConfig() + url = config.get_complete_url( + api_base="http://localhost:7860", + api_key=None, + model="langflow/ my-flow-id ", + optional_params={}, + litellm_params={}, + stream=False, + ) + assert url == "http://localhost:7860/api/v1/run/my-flow-id" + + +def test_langflow_config_transform_request_includes_session_id(): + config = LangFlowConfig() + request = config.transform_request( + model="langflow/my-flow-id", + messages=[{"role": "user", "content": "hello"}], + optional_params={"session_id": "sess-abc"}, + litellm_params={}, + headers={}, + ) + + assert request["input_value"] == "hello" + assert request["input_type"] == "chat" + assert request["output_type"] == "chat" + assert request["session_id"] == "sess-abc" + + +def test_langflow_config_transform_request_uses_last_user_message(): + config = LangFlowConfig() + request = config.transform_request( + model="langflow/my-flow-id", + messages=[ + {"role": "system", "content": "be helpful"}, + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "ok"}, + {"role": "user", "content": [{"type": "text", "text": "second"}]}, + ], + optional_params={}, + litellm_params={}, + headers={}, + ) + + assert request["input_value"] == "second" + assert "session_id" not in request + + +def test_langflow_config_transform_request_falls_back_to_last_message(): + config = LangFlowConfig() + request = config.transform_request( + model="langflow/my-flow-id", + messages=[{"role": "assistant", "content": "only assistant"}], + optional_params={}, + litellm_params={}, + headers={}, + ) + + assert request["input_value"] == "only assistant" + + +def test_langflow_config_transform_request_empty_messages(): + config = LangFlowConfig() + request = config.transform_request( + model="langflow/my-flow-id", + messages=[], + optional_params={}, + litellm_params={}, + headers={}, + ) + + assert request["input_value"] == "" + + +def test_langflow_config_rejects_tweaks_from_request_params(): + config = LangFlowConfig() + with pytest.raises(LangFlowError): + config.transform_request( + model="langflow/my-flow-id", + messages=[{"role": "user", "content": "hi"}], + optional_params={"tweaks": {"HttpComponent": {"url": "http://attacker"}}}, + litellm_params={}, + headers={}, + ) + + +def test_langflow_config_rejects_tweaks_from_request_body(): + config = LangFlowConfig() + with pytest.raises(LangFlowError): + config.sign_request( + headers={}, + optional_params={}, + request_data={ + "input_value": "hi", + "tweaks": {"HttpComponent": {"url": "http://attacker"}}, + }, + api_base="http://localhost:7860", + ) + + +def test_langflow_config_sign_request_passes_through_without_tweaks(): + config = LangFlowConfig() + headers, body = config.sign_request( + headers={"x-api-key": "secret"}, + optional_params={}, + request_data={"input_value": "hi"}, + api_base="http://localhost:7860", + ) + assert headers == {"x-api-key": "secret"} + assert body is None + + +def test_langflow_config_validate_environment_sets_api_key_header(): + config = LangFlowConfig() + headers = config.validate_environment( + headers={}, + model="langflow/my-flow-id", + messages=[{"role": "user", "content": "hi"}], + optional_params={}, + litellm_params={}, + api_key="secret", + ) + assert headers["Content-Type"] == "application/json" + assert headers["x-api-key"] == "secret" + + +def test_langflow_extra_body_cannot_inject_tweaks_into_run_payload(): + import json + + import litellm + from litellm.llms.custom_httpx.http_handler import HTTPHandler + + posted_bodies = [] + + def fake_post(*args, **kwargs): + body = kwargs.get("data") + posted_bodies.append(json.loads(body) if isinstance(body, str) else body) + resp = MagicMock(spec=httpx.Response) + resp.status_code = 200 + resp.json.return_value = { + "outputs": [{"outputs": [{"results": {"message": {"text": "hi"}}}]}] + } + resp.headers = {} + resp.text = "{}" + return resp + + with patch.object(HTTPHandler, "post", side_effect=fake_post): + with pytest.raises(Exception): + litellm.completion( + model="langflow/my-flow", + messages=[{"role": "user", "content": "hello"}], + api_base="http://example.com", + api_key="sk-test", + extra_body={"tweaks": {"HttpComponent": {"url": "http://attacker"}}}, + ) + + assert all("tweaks" not in (body or {}) for body in posted_bodies) + + +def test_langflow_config_extract_response(): + config = LangFlowConfig() + content = config._extract_content_from_response( + { + "session_id": "sess-abc", + "outputs": [ + { + "outputs": [ + { + "results": { + "message": {"text": "Hello from LangFlow"}, + } + } + ] + } + ], + } + ) + assert content == "Hello from LangFlow" + + +def test_langflow_config_extract_response_from_outputs_dict(): + config = LangFlowConfig() + content = config._extract_content_from_response( + { + "outputs": [ + { + "outputs": [ + { + "results": {}, + "outputs": { + "message": {"message": {"text": "via outputs dict"}} + }, + } + ] + } + ], + } + ) + assert content == "via outputs dict" + + +def test_langflow_extract_response_returns_none_when_no_message(): + config = LangFlowConfig() + assert config._extract_content_from_response({"outputs": []}) is None + assert config._extract_content_from_response({"detail": "flow failed"}) is None + assert config._extract_content_from_response({"outputs": ["not-a-dict"]}) is None + assert ( + config._extract_content_from_response({"outputs": [{"outputs": ["bad"]}]}) + is None + ) + assert ( + config._extract_content_from_response( + {"outputs": [{"outputs": [{"results": {"message": {"text": ""}}}]}]} + ) + is None + ) + + +def test_langflow_transform_response_builds_model_response_with_usage(): + config = LangFlowConfig() + raw_response = httpx.Response( + status_code=200, + json={ + "session_id": "sess-abc", + "outputs": [ + {"outputs": [{"results": {"message": {"text": "Hello from LangFlow"}}}]} + ], + }, + ) + + result = config.transform_response( + model="langflow/my-flow-id", + raw_response=raw_response, + model_response=ModelResponse(), + logging_obj=None, + request_data={}, + messages=[{"role": "user", "content": "hi"}], + optional_params={}, + litellm_params={}, + encoding=None, + ) + + assert result.choices[0].message.content == "Hello from LangFlow" + assert result.choices[0].finish_reason == "stop" + assert result.model == "langflow/my-flow-id" + assert result.usage.completion_tokens > 0 + assert result.usage.total_tokens == ( + result.usage.prompt_tokens + result.usage.completion_tokens + ) + + +def test_langflow_transform_response_raises_on_unparseable_body(): + config = LangFlowConfig() + raw_response = httpx.Response(status_code=200, json={"detail": "flow failed"}) + + with pytest.raises(LangFlowError): + config.transform_response( + model="langflow/my-flow-id", + raw_response=raw_response, + model_response=ModelResponse(), + logging_obj=None, + request_data={}, + messages=[{"role": "user", "content": "hi"}], + optional_params={}, + litellm_params={}, + encoding=None, + ) + + +def test_langflow_transform_response_raises_on_non_json_body(): + config = LangFlowConfig() + raw_response = httpx.Response( + status_code=200, content=b"not json", headers={"content-type": "text/plain"} + ) + + with pytest.raises(LangFlowError): + config.transform_response( + model="langflow/my-flow-id", + raw_response=raw_response, + model_response=ModelResponse(), + logging_obj=None, + request_data={}, + messages=[{"role": "user", "content": "hi"}], + optional_params={}, + litellm_params={}, + encoding=None, + ) + + +def test_langflow_config_get_error_class(): + config = LangFlowConfig() + err = config.get_error_class(error_message="boom", status_code=503, headers={}) + assert isinstance(err, LangFlowError) + assert err.status_code == 503 + + +def test_langflow_config_stream_behavior_flags(): + config = LangFlowConfig() + assert config.supports_stream_param_in_request_body is False + assert config.should_fake_stream(model="langflow/x", stream=True) is True + assert config.should_fake_stream(model="langflow/x", stream=False) is False + + +def test_langflow_provider_config_registered(): + cfg = ProviderConfigManager.get_provider_chat_config( + model="langflow/flow-1", + provider=LlmProviders.LANGFLOW, + ) + assert cfg is not None + assert cfg.__class__.__name__ == "LangFlowConfig" diff --git a/tests/test_litellm/llms/langflow/test_langflow_a2a.py b/tests/test_litellm/llms/langflow/test_langflow_a2a.py new file mode 100644 index 0000000000..c49ec8d87c --- /dev/null +++ b/tests/test_litellm/llms/langflow/test_langflow_a2a.py @@ -0,0 +1,159 @@ +from unittest.mock import AsyncMock, patch + +import pytest + +from litellm.a2a_protocol.litellm_completion_bridge.handler import ( + A2A_USER_API_KEY_HASH_PARAM, +) +from litellm.a2a_protocol.providers.config_manager import A2AProviderConfigManager +from litellm.llms.langflow.a2a import merge_a2a_session_into_litellm_params + + +def test_merge_a2a_session_into_litellm_params(): + merged = merge_a2a_session_into_litellm_params( + {"custom_llm_provider": "langflow", "model": "langflow/flow-1"}, + {"message": {"contextId": "shared-session-99"}}, + ) + assert merged["session_id"] == "shared-session-99" + + +def test_merge_a2a_session_is_scoped_per_principal(): + """The LangFlow session must be bound to the authenticated key so two + distinct keys cannot share memory by reusing the same A2A contextId, while + the same key keeps a stable session across turns.""" + base = {"custom_llm_provider": "langflow", "model": "langflow/flow-1"} + params = {"message": {"contextId": "ctx-1"}} + + key_a = merge_a2a_session_into_litellm_params(base, params, "hash-a")["session_id"] + key_a_again = merge_a2a_session_into_litellm_params(base, params, "hash-a")[ + "session_id" + ] + key_b = merge_a2a_session_into_litellm_params(base, params, "hash-b")["session_id"] + + assert key_a == key_a_again, "same key + contextId must stay on one session" + assert key_a != key_b, "different keys must not collide on the same contextId" + assert key_a != "ctx-1", "raw client contextId must not be used verbatim" + assert key_a.endswith("-ctx-1"), "original contextId kept for correlation" + assert "hash-a" not in key_a, "raw principal must not be sent to LangFlow" + + +def test_merge_a2a_session_without_context_id_is_noop(): + merged = merge_a2a_session_into_litellm_params( + {"custom_llm_provider": "langflow", "model": "langflow/flow-1"}, + {"message": {"role": "user"}}, + ) + assert "session_id" not in merged + + +def test_langflow_a2a_provider_config_registered(): + cfg = A2AProviderConfigManager.get_provider_config( + custom_llm_provider="langflow", + model="langflow/flow-1", + ) + assert cfg is not None + assert cfg.__class__.__name__ == "LangFlowA2AConfig" + + +@pytest.mark.asyncio +async def test_langflow_a2a_config_passes_session_id_to_completion(): + from litellm.a2a_protocol.providers.langflow.config import LangFlowA2AConfig + + mock_response = type( + "R", + (), + { + "choices": [ + type( + "C", + (), + {"message": type("M", (), {"content": "ok"})()}, + )() + ] + }, + )() + + with patch("litellm.acompletion", new_callable=AsyncMock) as mock_acompletion: + mock_acompletion.return_value = mock_response + + await LangFlowA2AConfig().handle_non_streaming( + request_id="req-1", + params={ + "message": { + "role": "user", + "parts": [{"kind": "text", "text": "hi"}], + "contextId": "shared-session-99", + } + }, + litellm_params={ + "custom_llm_provider": "langflow", + "model": "langflow/flow-1", + "api_base": "http://localhost:7860", + }, + api_base="http://localhost:7860", + ) + + assert ( + mock_acompletion.call_args.kwargs.get("session_id") == "shared-session-99" + ) + + +@pytest.mark.asyncio +async def test_langflow_a2a_config_scopes_session_by_authenticated_key(): + from litellm.a2a_protocol.providers.langflow.config import LangFlowA2AConfig + + mock_response = type( + "R", + (), + {"choices": [type("C", (), {"message": type("M", (), {"content": "ok"})()})()]}, + )() + + with patch("litellm.acompletion", new_callable=AsyncMock) as mock_acompletion: + mock_acompletion.return_value = mock_response + + await LangFlowA2AConfig().handle_non_streaming( + request_id="req-1", + params={ + "message": { + "role": "user", + "parts": [{"kind": "text", "text": "hi"}], + "contextId": "ctx-1", + } + }, + litellm_params={ + "custom_llm_provider": "langflow", + "model": "langflow/flow-1", + "api_base": "http://localhost:7860", + A2A_USER_API_KEY_HASH_PARAM: "hashed-key-1", + }, + api_base="http://localhost:7860", + ) + + forwarded = mock_acompletion.call_args.kwargs + assert forwarded.get("session_id") != "ctx-1" + assert forwarded.get("session_id").endswith("-ctx-1") + assert ( + A2A_USER_API_KEY_HASH_PARAM not in forwarded + ), "internal principal param must not leak to the LLM call" + + +@pytest.mark.asyncio +async def test_langflow_a2a_config_requires_litellm_params_non_streaming(): + from litellm.a2a_protocol.providers.langflow.config import LangFlowA2AConfig + + with pytest.raises(ValueError, match="litellm_params is required"): + await LangFlowA2AConfig().handle_non_streaming( + request_id="req-1", + params={"message": {"contextId": "shared-session-99"}}, + ) + + +@pytest.mark.asyncio +async def test_langflow_a2a_config_requires_litellm_params_streaming(): + from litellm.a2a_protocol.providers.langflow.config import LangFlowA2AConfig + + with pytest.raises(ValueError, match="litellm_params is required"): + async for _ in LangFlowA2AConfig().handle_streaming( + request_id="req-1", + params={"message": {"contextId": "shared-session-99"}}, + ): + pass diff --git a/tests/test_litellm/proxy/agent_endpoints/test_a2a_endpoints.py b/tests/test_litellm/proxy/agent_endpoints/test_a2a_endpoints.py index 268e6d2dc1..a32f2eadb9 100644 --- a/tests/test_litellm/proxy/agent_endpoints/test_a2a_endpoints.py +++ b/tests/test_litellm/proxy/agent_endpoints/test_a2a_endpoints.py @@ -246,3 +246,105 @@ async def test_invoke_agent_a2a_handles_none_agent_card_params(): assert body["jsonrpc"] == "2.0" assert body["error"]["code"] == -32000 assert "no URL configured" in body["error"]["message"] + + +@pytest.mark.asyncio +async def test_invoke_agent_a2a_injects_authenticated_key_hash_for_bridge(): + """Completion-bridge agents must receive the authenticated key hash in + litellm_params so provider configs (e.g. LangFlow) can scope provider-side + session memory per key. Regression for cross-key A2A session bleed.""" + from litellm.a2a_protocol.litellm_completion_bridge.handler import ( + A2A_USER_API_KEY_HASH_PARAM, + ) + from litellm.proxy._types import UserAPIKeyAuth + + captured = {} + + async def mock_add_litellm_data(data, **kwargs): + data["proxy_server_request"] = { + "url": "http://localhost:4000/a2a/lf-agent", + "method": "POST", + "headers": {}, + "body": {}, + } + data.setdefault("metadata", {}) + return data + + async def capture_asend_message(**kwargs): + captured.update(kwargs) + resp = MagicMock() + resp.model_dump.return_value = {"jsonrpc": "2.0", "id": "test-id", "result": {}} + return resp + + mock_agent = MagicMock() + mock_agent.agent_id = "lf-agent" + mock_agent.agent_name = "lf-agent" + # No URL: the bridge derives the endpoint from the LangFlow agent config. + mock_agent.agent_card_params = {"name": "LF Agent"} + mock_agent.litellm_params = { + "custom_llm_provider": "langflow", + "model": "langflow/flow-1", + } + mock_agent.static_headers = None + mock_agent.extra_headers = None + + mock_request = MagicMock() + mock_request.headers = {} + mock_request.json = AsyncMock( + return_value={ + "jsonrpc": "2.0", + "id": "test-id", + "method": "message/send", + "params": { + "message": { + "role": "user", + "parts": [{"kind": "text", "text": "hi"}], + "messageId": "msg-1", + "contextId": "ctx-1", + } + }, + } + ) + + mock_user_api_key_dict = UserAPIKeyAuth( + api_key="sk-hashed-123", + user_id="test-user", + team_id="test-team", + ) + + with ( + patch( + "litellm.proxy.agent_endpoints.a2a_endpoints._get_agent", + return_value=mock_agent, + ), + patch( + "litellm.proxy.common_request_processing.add_litellm_data_to_request", + side_effect=mock_add_litellm_data, + ), + patch( + "litellm.proxy.agent_endpoints.auth.agent_permission_handler.AgentRequestHandler.is_agent_allowed", + new=AsyncMock(return_value=True), + ), + patch( + "litellm.a2a_protocol.asend_message", + new=AsyncMock(side_effect=capture_asend_message), + ), + patch("litellm.proxy.proxy_server.general_settings", {}), + patch("litellm.proxy.proxy_server.proxy_config", MagicMock()), + patch("litellm.proxy.proxy_server.version", "1.0.0"), + patch("litellm.a2a_protocol.main.A2A_SDK_AVAILABLE", True), + patch.dict(sys.modules, {"a2a": MagicMock(), "a2a.types": MagicMock()}), + ): + from litellm.proxy.agent_endpoints.a2a_endpoints import invoke_agent_a2a + + await invoke_agent_a2a( + agent_id="lf-agent", + request=mock_request, + fastapi_response=MagicMock(), + user_api_key_dict=mock_user_api_key_dict, + ) + + assert ( + captured.get("litellm_params", {}).get(A2A_USER_API_KEY_HASH_PARAM) + == mock_user_api_key_dict.api_key + ), "authenticated key hash was not forwarded to the completion bridge" diff --git a/ui/litellm-dashboard/public/assets/logos/langflow.svg b/ui/litellm-dashboard/public/assets/logos/langflow.svg new file mode 100644 index 0000000000..1c7b36c4dd --- /dev/null +++ b/ui/litellm-dashboard/public/assets/logos/langflow.svg @@ -0,0 +1,5 @@ + + + + + diff --git a/ui/litellm-dashboard/src/components/agents/agent_type_utils.ts b/ui/litellm-dashboard/src/components/agents/agent_type_utils.ts index fd04aa4c26..8a78a7fabb 100644 --- a/ui/litellm-dashboard/src/components/agents/agent_type_utils.ts +++ b/ui/litellm-dashboard/src/components/agents/agent_type_utils.ts @@ -10,11 +10,13 @@ export const detectAgentType = (agent: Agent): string => { const customProvider = agent.litellm_params?.custom_llm_provider; // Check by custom_llm_provider first + if (customProvider === "langflow") return "langflow"; if (customProvider === "langgraph") return "langgraph"; if (customProvider === "azure_ai") return "azure_ai_foundry"; if (customProvider === "bedrock") return "bedrock_agentcore"; // Check by model prefix + if (model.startsWith("langflow/")) return "langflow"; if (model.startsWith("langgraph/")) return "langgraph"; if (model.startsWith("azure_ai/agents/")) return "azure_ai_foundry"; if (model.startsWith("bedrock/agentcore/")) return "bedrock_agentcore";