ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(agents): add LangFlow agent provider with A2A session bridging (#28963)

Browse Source 
* 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 <cursoragent@cursor.com>

* docs(providers): document langflow in provider_endpoints_support.json

Co-authored-by: Cursor <cursoragent@cursor.com>

* 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 <cursoragent@cursor.com>

* 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 <cursoragent@cursor.com>

* 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 <cursoragent@cursor.com>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
Sameer Kankute 2026-06-03 03:15:56 +05:30 committed by GitHub
parent c1602587c1
commit ae7ac72331
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 1265 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
litellm/a2a_protocol/litellm_completion_bridge/handler.py
View File

@ -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", {})

5
litellm/a2a_protocol/providers/config_manager.py
View File

@ -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,

0
litellm/a2a_protocol/providers/langflow/__init__.py Normal file
View File

62
litellm/a2a_protocol/providers/langflow/config.py Normal file
View File

@ -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

1
litellm/llms/langflow/__init__.py Normal file
View File

@ -0,0 +1 @@
"""LangFlow LLM provider for LiteLLM."""

37
litellm/llms/langflow/a2a.py Normal file
View File

@ -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

1
litellm/llms/langflow/chat/__init__.py Normal file
View File

@ -0,0 +1 @@
"""LangFlow chat transformation."""

327
litellm/llms/langflow/chat/transformation.py Normal file
View File

@ -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": "<last user message>",
"input_type": "chat",
"output_type": "chat",
"session_id": "<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

33
litellm/main.py
View File

@ -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

13
litellm/proxy/agent_endpoints/a2a_endpoints.py
View File

@ -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:

42
litellm/proxy/public_endpoints/agent_create_fields.json
View File

@ -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",

1
litellm/types/utils.py
View File

@ -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"

11
litellm/utils.py
View File

@ -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,

18
provider_endpoints_support.json
View File

@ -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",

398
tests/test_litellm/llms/langflow/chat/test_langflow_chat_transformation.py Normal file
View File

@ -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"

159
tests/test_litellm/llms/langflow/test_langflow_a2a.py Normal file
View File

@ -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

102
tests/test_litellm/proxy/agent_endpoints/test_a2a_endpoints.py
View File

@ -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"

5
ui/litellm-dashboard/public/assets/logos/langflow.svg Normal file
View File

@ -0,0 +1,5 @@
<svg width="470" height="470" viewBox="0 0 470 470" fill="none" xmlns="http://www.w3.org/2000/svg">
<path d="M342.604 243.34H389.75C398.998 243.34 406.489 250.831 406.489 260.079V287.892C406.489 297.14 398.998 304.631 389.75 304.631H348.629C344.186 304.631 339.928 306.4 336.787 309.54L266.225 380.091C263.084 383.232 258.827 385 254.383 385H220.463C211.39 385 203.956 377.765 203.724 368.691L202.991 340.297C202.747 330.886 210.308 323.115 219.73 323.115H248.927C253.371 323.115 257.629 321.347 260.769 318.206L330.739 248.237C333.879 245.097 338.137 243.328 342.58 243.328L342.604 243.34Z" fill="black"/>
<path d="M202.619 85H249.765C259.013 85 266.504 92.4913 266.504 101.739V129.552C266.504 138.8 259.013 146.291 249.765 146.291H208.644C204.201 146.291 199.943 148.06 196.802 151.2L126.24 221.763C123.099 224.904 118.842 226.672 114.398 226.672H80.4777C71.4044 226.672 63.9712 219.436 63.7386 210.363L63.0058 181.968C62.7615 172.558 70.3226 164.799 79.7449 164.799H108.942C113.386 164.799 117.643 163.031 120.784 159.89L190.753 89.9205C193.894 86.7798 198.152 85.0116 202.595 85.0116L202.619 85Z" fill="black"/>
<path d="M342.603 120.829H389.75C398.997 120.829 406.489 128.32 406.489 137.568V165.381C406.489 174.629 398.997 182.12 389.75 182.12H348.629C344.185 182.12 339.928 183.888 336.787 187.029L266.225 257.591C263.084 260.732 258.826 262.5 254.383 262.5H213.169C208.853 262.5 204.701 264.164 201.583 267.153L122.366 343.067C119.248 346.056 115.096 347.72 110.78 347.72H81.9083C72.6605 347.72 65.1692 340.217 65.1692 330.981V302.4C65.1692 293.152 72.6605 285.661 81.9083 285.661H110.571C115.014 285.661 119.272 283.892 122.413 280.752L197.64 205.525C200.78 202.384 205.038 200.616 209.481 200.616H248.927C253.371 200.616 257.628 198.848 260.769 195.707L330.738 125.738C333.879 122.597 338.136 120.829 342.58 120.829H342.603Z" fill="black"/>
</svg>
Side by Side

After

Width:  |  Height:  |  Size: 1.8 KiB

2
ui/litellm-dashboard/src/components/agents/agent_type_utils.ts
View File

@ -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";