mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
test(proxy): pin the max_parallel_requests slot release when a streamed /v1/messages MCP auto-execute ends (#47788)
* test(proxy): cover the max_parallel_requests slot a streamed /v1/messages MCP auto-execute keeps A streamed /v1/messages request that auto-executes MCP tools replays its answer as a stream after its model rounds already logged success. The stream path frees the request's max_parallel_requests slot only on a client disconnect, so after message_stop the slot stays taken and the key's next request answers 429 The test streams such a replay to the end and expects the slot to be free. It fails until the fix lands * fix(proxy): free the max_parallel_requests slot when a replayed /v1/messages MCP stream completes A streamed /v1/messages request that auto-executes MCP tools runs its model rounds without streaming and replays the final answer as a stream. The first round claims the request's only success log, and its limiter callback keeps the slot because the tool call reused it afterwards. The stream path released the slot only on a client disconnect, so after message_stop the key stayed at its max_parallel_requests limit and its next request answered 429 Stream cleanup now releases the stashed slot when the stream completed and its logging object already claimed the success log, which only happens when the stream replays a response that was logged before it started. A real stream still releases through its own success callback * test(mcp): free the key's only parallel slot after a streamed auto-execute ends Run a streamed MCP auto-execute on a key with max_parallel_requests 1 on every LLM surface, then send the next request on that key and expect 200. On main both /v1/messages surfaces answer 429 here, while /v1/responses and /v1/chat/completions already pass * test(proxy): keep only the MCP stream regression tests, since #48628 already frees the slot at stream end * test(proxy): require the MCP stream regression test's response to stream and reach its end event
This commit is contained in:
parent
d8a43a7cd3
commit
94e4ebd63d
2 changed files with 121 additions and 0 deletions
|
|
@ -702,6 +702,36 @@ def test_streaming_responses_opens_the_gateway_tool_call_empty_and_streams_its_a
|
|||
assert len(rig.upstream_tools()) == 2
|
||||
|
||||
|
||||
STREAM_END_EVENT: Final[Mapping[Surface, str]] = {
|
||||
"chat": "data: [DONE]",
|
||||
"responses": "response.completed",
|
||||
"messages": "message_stop",
|
||||
"messages_bridge": "message_stop",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("surface", SURFACES)
|
||||
def test_streamed_auto_execute_gives_back_the_keys_only_parallel_slot_once_the_stream_ends(
|
||||
gateway: Gateway, surface: Surface
|
||||
) -> None:
|
||||
with _rig(gateway, surface) as rig:
|
||||
key: Final = rig.scenario.key(object_permission={"mcp_servers": [rig.server_id]}, max_parallel_requests=1)
|
||||
streamed: Final = rig.send(key, [AUTO], stream=True)
|
||||
assert streamed.status_code == 200, streamed.text
|
||||
assert streamed.headers["content-type"].startswith("text/event-stream"), streamed.headers
|
||||
assert STREAM_END_EVENT[surface] in streamed.text, streamed.text
|
||||
assert ANSWER in streamed.text, streamed.text
|
||||
assert [call["body"]["params"]["name"] for call in _peer_add_calls(rig.peer)] == ["add"]
|
||||
follow_up: Final = eventually(
|
||||
lambda: rig.send(key, [], prompt=f"{DIRECT} {rig.alias}"),
|
||||
lambda response: response.status_code == 200,
|
||||
seconds=20,
|
||||
return_last_on_timeout=True,
|
||||
)
|
||||
assert follow_up.status_code == 200, follow_up.text
|
||||
assert rig.final_text(follow_up.json()) == ANSWER
|
||||
|
||||
|
||||
def test_streaming_chat_through_a_toolset_gateway_url_serves_a_team_key_without_its_own_grant(
|
||||
gateway: Gateway,
|
||||
) -> None:
|
||||
|
|
|
|||
|
|
@ -4332,6 +4332,97 @@ async def test_first_round_logged_after_tool_call_reuse_releases_only_at_request
|
|||
assert await in_flight() == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_completed_mcp_replay_stream_frees_slot_its_logged_first_round_kept() -> None:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.llms.anthropic.messages.native.fake_stream_iterator import (
|
||||
FakeAnthropicMessagesStreamIterator,
|
||||
)
|
||||
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
|
||||
|
||||
reused_at: Final = datetime(2026, 1, 5, 12, 0, 0)
|
||||
api_key: Final = hash_token("sk-mcp-replayed-stream")
|
||||
local_cache: Final = DualCache()
|
||||
handler: Final = _PROXY_MaxParallelRequestsHandler(
|
||||
internal_usage_cache=InternalUsageCache(local_cache),
|
||||
time_provider=lambda: reused_at,
|
||||
)
|
||||
proxy_logging_obj: Final = ProxyLogging(user_api_key_cache=DualCache())
|
||||
proxy_logging_obj.proxy_hook_mapping["parallel_request_limiter"] = handler
|
||||
user_api_key_dict: Final = UserAPIKeyAuth(api_key=api_key, max_parallel_requests=1)
|
||||
parallel_key: Final = f"{{api_key:{api_key}}}:max_parallel_requests"
|
||||
owner_call_id: Final = "mcp-replayed-stream-owner"
|
||||
messages: Final = [{"role": "user", "content": "add 2 and 3"}]
|
||||
|
||||
async def in_flight() -> int:
|
||||
return handler._gauge_in_flight_from_cache_value(
|
||||
await local_cache.async_get_cache(key=parallel_key)
|
||||
)
|
||||
|
||||
async def pre_call(data: dict[str, object], call_type: str) -> None:
|
||||
await handler.async_pre_call_hook(
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
cache=local_cache,
|
||||
data=data,
|
||||
call_type=call_type,
|
||||
)
|
||||
|
||||
await pre_call(
|
||||
{"model": "claude-haiku-5-5", "messages": messages, "stream": True, "litellm_call_id": owner_call_id},
|
||||
CallTypes.anthropic_messages.value,
|
||||
)
|
||||
await pre_call(
|
||||
{"model": "MCP: add", "messages": [{"role": "user", "content": "Tool: add"}], "mcp_server_name": "math"},
|
||||
CallTypes.call_mcp_tool.value,
|
||||
)
|
||||
logging_obj: Final = LiteLLMLoggingObj(
|
||||
model="claude-haiku-5-5",
|
||||
messages=messages,
|
||||
stream=True,
|
||||
call_type=CallTypes.anthropic_messages.value,
|
||||
start_time=None,
|
||||
litellm_call_id=owner_call_id,
|
||||
function_id="mcp-replayed-stream",
|
||||
)
|
||||
assert logging_obj.claim_async_success_log()
|
||||
await handler.async_log_success_event(
|
||||
kwargs={"litellm_call_id": owner_call_id, "stream": False},
|
||||
response_obj=None,
|
||||
start_time=None,
|
||||
end_time=reused_at - timedelta(seconds=1),
|
||||
)
|
||||
assert await in_flight() == 1
|
||||
|
||||
replay: Final = FakeAnthropicMessagesStreamIterator(
|
||||
response={
|
||||
"id": "msg_replayed",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": "claude-haiku-5-5",
|
||||
"content": [{"type": "text", "text": "2 + 3 = 5"}],
|
||||
"stop_reason": "end_turn",
|
||||
"usage": {"input_tokens": 20, "output_tokens": 8},
|
||||
}
|
||||
)
|
||||
with _override_litellm_callbacks([]):
|
||||
streamed: Final = [
|
||||
chunk
|
||||
async for chunk in ProxyBaseLLMRequestProcessing.async_sse_data_generator(
|
||||
response=replay,
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
request_data={
|
||||
"model": "claude-haiku-5-5",
|
||||
"litellm_call_id": owner_call_id,
|
||||
"litellm_logging_obj": logging_obj,
|
||||
},
|
||||
proxy_logging_obj=proxy_logging_obj,
|
||||
)
|
||||
]
|
||||
|
||||
assert b"event: message_stop" in streamed[-1]
|
||||
assert await in_flight() == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mcp_tool_call_reusing_request_slot_keeps_request_max_parallel_headers() -> None:
|
||||
api_key: Final = hash_token("sk-mcp-reuse-headers")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue