From ae7ac72331ef21d990fece0cb8cd66ab1d594af2 Mon Sep 17 00:00:00 2001 From: Sameer Kankute Date: Wed, 3 Jun 2026 03:15:56 +0530 Subject: [PATCH] feat(agents): add LangFlow agent provider with A2A session bridging (#28963) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(agents): add LangFlow agent provider with A2A session bridging Register LangFlow as a completion provider and agent type (UI + /api/v1/run), and map A2A contextId to LangFlow session_id for multi-turn conversations. Co-authored-by: Cursor * docs(providers): document langflow in provider_endpoints_support.json Co-authored-by: Cursor * fix(agents): address Greptile review for LangFlow integration Move A2A contextId→session_id mapping into LangFlow A2A provider config, add langflow.svg logo, remove live integration test, use model for token count. Co-authored-by: Cursor * fix(langflow): prevent flow_id override via request optional_params Derive flow_id only from the authorized model name and reject flow_id kwargs so callers cannot invoke a different LangFlow run endpoint. Co-authored-by: Cursor * refactor(langflow): remove redundant flow_id branch in _get_flow_id * fix(langflow): surface an error when the run response has no extractable message Previously the response parser returned the raw JSON blob as the assistant message when it could not find message text, silently presenting an unparseable payload as a valid answer. It now returns None and the caller raises a LangFlowError so the failure is visible to the client. * fix(langflow): URL-encode flow_id path segment to prevent path injection flow_id is taken from the model suffix and interpolated into /api/v1/run/{flow_id}. Without path-segment encoding a model such as langflow/../../x (or one containing ?) could move the request off the run endpoint to another path on the configured LangFlow server using the operator x-api-key. Encode the segment with quote(safe="") so it always stays a single path segment. * fix(langflow): reject empty flow_id from model name * fix(langflow): return stripped flow_id so validation matches URL path * fix(langflow): reject caller-supplied tweaks to prevent flow component override * fix(langflow): reject caller-supplied tweaks injected via extra_body The transform_request guard only inspected optional_params, but extra_body is popped before transform_request runs and merged into the request body afterward, letting a caller reintroduce tweaks and override the operator-configured LangFlow flow components. Validate the final request body in sign_request so tweaks cannot reach LangFlow through extra_body. * test(langflow): move provider tests into mirrored coverage path The langflow tests lived under tests/llm_translation/, whose CircleCI job runs without --cov and uploads nothing to Codecov, so none of the new langflow code counted toward patch coverage (codecov/patch reported 9.78% of the diff hit against a 70.83% target). Relocate them to tests/test_litellm/llms/langflow/, which the GitHub Actions provider job runs with --cov=./litellm and uploads, and add regression tests for the previously untested happy paths (transform_response building the ModelResponse with usage, non-JSON body handling, last-user message extraction, outputs-dict response shape, sign_request pass-through, error class and stream flags). Patch coverage on the diff is now ~88%. * fix(langflow): require litellm_params in A2A config instead of silent empty fallback * fix(langflow): scope A2A session_id to the authenticated key The LangFlow A2A bridge used the LangFlow session_id verbatim from the client-controlled A2A contextId, so two distinct virtual keys authorized for the same agent could read or append to each other's LangFlow conversation memory by reusing a contextId. Hand the authenticated key hash to the completion bridge through litellm_params and namespace the forwarded session_id with it. The same key keeps a stable session across turns, while different keys can no longer collide on a shared contextId. The principal is hashed before it is embedded in the session_id, so the stored token is never sent to the LangFlow backend; the original contextId is preserved as a suffix for operator-side correlation. * fix(langflow): wire authenticated key hash through A2A bridge and tests Define A2A_USER_API_KEY_HASH_PARAM in the completion bridge handler, strip it before litellm.acompletion, inject the authenticated key hash at the proxy A2A endpoint, and add regression tests for per-key LangFlow session scoping. --------- Co-authored-by: Cursor Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com> --- .../litellm_completion_bridge/handler.py | 83 ++-- .../a2a_protocol/providers/config_manager.py | 5 + .../providers/langflow/__init__.py | 0 .../a2a_protocol/providers/langflow/config.py | 62 +++ litellm/llms/langflow/__init__.py | 1 + litellm/llms/langflow/a2a.py | 37 ++ litellm/llms/langflow/chat/__init__.py | 1 + litellm/llms/langflow/chat/transformation.py | 327 ++++++++++++++ litellm/main.py | 33 ++ .../proxy/agent_endpoints/a2a_endpoints.py | 13 + .../public_endpoints/agent_create_fields.json | 42 ++ litellm/types/utils.py | 1 + litellm/utils.py | 11 + provider_endpoints_support.json | 18 + .../chat/test_langflow_chat_transformation.py | 398 ++++++++++++++++++ .../llms/langflow/test_langflow_a2a.py | 159 +++++++ .../agent_endpoints/test_a2a_endpoints.py | 102 +++++ .../public/assets/logos/langflow.svg | 5 + .../src/components/agents/agent_type_utils.ts | 2 + 19 files changed, 1265 insertions(+), 35 deletions(-) create mode 100644 litellm/a2a_protocol/providers/langflow/__init__.py create mode 100644 litellm/a2a_protocol/providers/langflow/config.py create mode 100644 litellm/llms/langflow/__init__.py create mode 100644 litellm/llms/langflow/a2a.py create mode 100644 litellm/llms/langflow/chat/__init__.py create mode 100644 litellm/llms/langflow/chat/transformation.py create mode 100644 tests/test_litellm/llms/langflow/chat/test_langflow_chat_transformation.py create mode 100644 tests/test_litellm/llms/langflow/test_langflow_a2a.py create mode 100644 ui/litellm-dashboard/public/assets/logos/langflow.svg 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";