scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-14 09:03:16 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 214

Merge f456e23d0b into b4a8bc715f

Browse source
This commit is contained in:
udemex10 2026-11-05 07:15:26 +00:00 • committed by GitHub
commit 868d4a18ea
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 37 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
litellm/integrations/generic_api/generic_api_compatible_callbacks.json
View file

@ -34,6 +34,15 @@
},
"environment_variables": ["QUALIFIRE_API_KEY", "QUALIFIRE_WEBHOOK_URL"]
},
"veqen": {
"event_types": ["llm_api_success", "llm_api_failure"],
"endpoint": "https://veqen-app.com/v1/integrations/litellm",
"headers": {
"Content-Type": "application/json",
"Authorization": "Bearer {{environment_variables.VEQEN_API_KEY}}"
},
"environment_variables": ["VEQEN_API_KEY"]
},
"ramp": {
"event_types": ["llm_api_success"],
"endpoint": "https://api.ramp.com/developer/v1/ai-usage/litellm",

28
tests/unit/integrations/generic_api/test_generic_api_callback.py
View file

@ -731,6 +731,34 @@ class _InProcessLogSink:
return httpx.Response(200)
@pytest.mark.asyncio
async def test_veqen_delivers_success_and_failure_as_authenticated_json(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("VEQEN_API_KEY", "synthetic-ingest-key")
sink: Final = _InProcessLogSink()
handler: Final = AsyncHTTPHandler(transport=httpx.MockTransport(sink.respond))
before_tasks: Final = frozenset(asyncio.all_tasks())
logger: Final = GenericAPILogger(
callback_name="veqen", async_httpx_client=handler, batch_size=1000, flush_interval=3600
)
owned_tasks: Final = frozenset(asyncio.all_tasks()) - before_tasks
try:
await logger.async_log_success_event({"standard_logging_object": {"id": "success"}}, None, None, None)
await logger.async_log_failure_event({"standard_logging_object": {"id": "failure"}}, None, None, None)
await logger.flush_queue()
assert tuple(_received_log_ids(sink.requests, "json_array")) == ("success", "failure")
assert len(sink.requests) == 1
request: Final = sink.requests[0]
assert request.method == "POST"
assert str(request.url) == "https://veqen-app.com/v1/integrations/litellm"
assert request.headers["authorization"] == "Bearer synthetic-ingest-key"
assert request.headers["content-type"] == "application/json"
finally:
for task in owned_tasks:
task.cancel()
await asyncio.gather(*owned_tasks, return_exceptions=True)
await handler.close()
@pytest.mark.asyncio
@pytest.mark.parametrize("log_format", ("json_array", "ndjson", "single"))
async def test_generic_api_keeps_failure_arriving_during_success_batch(log_format: LOG_FORMAT_TYPES) -> None: