vega-index-client 0.1.0

Official Python SDK for VEGA INDEX API

vega-index-client

Official Python SDK for VEGA INDEX API — predictive viral inference engine pra música brasileira.

Install

# Production (PyPI — pendente)
pip install vega-index-client

# Direto do GitHub
pip install git+https://github.com/arthurvalle1/vegaindex.git#subdirectory=sdk

# Local dev
git clone https://github.com/arthurvalle1/vegaindex.git
pip install -e ./vegaindex/sdk

Async support:

pip install vega-index-client[async]   # adds httpx

Quick start

Demo público (sem API key)

from vega_index_client import Client

client = Client.public()
result = client.score_public(
    title="Envolver",
    artist="Anitta",
    genre="pop latino",
    language="pt-BR",
)
print(f"{result.track_id}: score={result.score} ({result.category})")
print(f"Horizons: 30d={result.horizon_30d:.0%} 90d={result.horizon_90d:.0%}")

Autenticado (API key)

from vega_index_client import Client

client = Client(api_key="vi_live_xxx")   # ou lu_live_xxx legacy

# Score single
result = client.score(
    title="Madrugada Sem Você",
    artist="Vega Romero",
    isrc="BRART2600142",
    genre="sertanejo",
    release_date="2026-06-12",
)
if result.is_high_potential():
    print(f"🚀 HIGH_POTENTIAL — {result.score}/100")
    for similar in result.similar_catalog_mert[:3]:
        print(f"  ↳ similar: {similar.artist} - {similar.title} (dist={similar.distance:.2f})")

Batch scoring

tracks = [
    {"title": "T1", "artist_primary": "A1"},
    {"title": "T2", "artist_primary": "A2"},
]
results = client.score_batch(tracks)
for r in results:
    print(r.track_id, r.score)

Histórico com pagination

page = client.history(limit=100, since="2026-01-01", category="HIGH_POTENTIAL")
while True:
    for job in page.jobs:
        print(f"{job.completed_at} {job.track_id} score={job.score}")
    if not page.has_more:
        break
    page = client.history(limit=100, cursor=page.next_cursor)

CSV export

client.history_csv("history.csv", since="2026-05-01")

Webhook management

# Cria — secret mostrado UMA VEZ
wh = client.create_webhook(
    url="https://yourapp.com/vega-hook",
    events=["high_potential"],
)
print(f"Secret: {wh.secret}")   # GUARDE PRA HMAC validation

# Lista (sem expor secret)
for w in client.list_webhooks():
    print(f"{w.webhook_id} {w.url} delivered={w.n_delivered} failed={w.n_failed}")

# Debug deliveries
deliveries = client.get_deliveries(wh.webhook_id)
for d in deliveries:
    print(f"{d.created_at} {d.event_type} status={d.status} attempts={d.attempts}")

# Delete (soft)
client.delete_webhook(wh.webhook_id)

PDF report

client.get_pdf("tr_abc123", save_to="report.pdf")

API key rotation

# Rotaciona com grace period 60min — key antiga válida nesse tempo
result = client.rotate_key(grace_period_minutes=60)
print(f"New key: {result['new_key']}")
# client.api_key já atualizado pra nova

Submit feedback (gold labels)

# Reportar outcome real pra fitar modelo
client.submit_feedback(
    track_id="tr_abc123",
    outcome="viralized",          # viralized / flopped / catalog_hit
    metric_type="streams_30d",
    metric_value=2_500_000,
    confidence="high",
)

Health check

print(client.health())                     # {"status": "ok"}
print(client.health(deep=True))            # 8 sub-checks (DB/ML/encoders/etc)

---

Async usage

import asyncio
from vega_index_client import AsyncClient

async def main():
    async with AsyncClient(api_key="vi_live_xxx") as client:
        result = await client.score("Envolver", "Anitta")
        print(result.score)

asyncio.run(main())

---

Exception handling

from vega_index_client import Client, AuthError, RateLimitedError

client = Client(api_key="invalid")
try:
    client.score("T", "A")
except AuthError:
    print("Bad key")
except RateLimitedError as e:
    print(f"Rate limited, retry in {e.retry_after}s")

Hierarquia:

VegaError
├── AuthError              (401)
├── QuotaExceededError     (402)
├── ValidationError        (400)
├── NotFoundError          (404)
├── RateLimitedError       (429) — inclui .retry_after
└── ServerError            (5xx)

---

API surface

Method	Endpoint
score(title, artist, ...)	POST /api/v1/score
score_batch(tracks)	POST /api/v1/score (with tracks list)
score_public(title, artist)	POST /api/v1/score/public
get_score(track_id)	GET /api/v1/score/{id}
get_pdf(track_id, save_to)	GET /api/v1/score/{id}/pdf
history(...)	GET /api/v1/score/history
history_csv(save_to, ...)	GET /api/v1/score/history.csv
search(query, ...)	GET /api/v1/search
create_webhook(url, events)	POST /api/v1/webhooks
list_webhooks()	GET /api/v1/webhooks
delete_webhook(id)	DELETE /api/v1/webhooks/{id}
get_deliveries(id)	GET /api/v1/webhooks/{id}/deliveries
rotate_key(grace_minutes)	POST /api/v1/keys/rotate
submit_feedback(...)	POST /api/v1/score/feedback
health(deep)	GET /healthz?deep=1

OpenAPI spec completo: https://www.vegaindex.app.br/api/v1/openapi.json

---

Configuração

Client(
    api_key="vi_live_xxx",
    base_url="https://www.vegaindex.app.br",   # default
    timeout=30.0,                                # default
)

Custom base_url útil pra:

• Local dev: http://localhost:5000
• Staging: https://staging.vegaindex.app.br

---

License

Proprietary. © 2026 ART Produções e Distribuição Ltda.

---

Suporte

• Email: [e-mail]
• Docs: https://www.vegaindex.app.br/docs
• OpenAPI: https://www.vegaindex.app.br/api/v1/openapi.json
• Swagger UI: https://www.vegaindex.app.br/api/v1/docs