SDKの概要

from vonage_cloud_runtime.vcr import VCR
vcr = VCR()

from vonage_cloud_runtime.vcr import VCR
vcr = VCR()

session = vcr.createSession()           # default 7-day TTL
session = vcr.createSession(3600)       # 1-hour TTL
session = vcr.createSessionWithId('user-123')
session = vcr.getSessionFromRequest(req)
session = vcr.getGlobalSession()

from vonage_cloud_runtime.vcr import VCR
from vonage_cloud_runtime.providers.state.state import State
from vonage_cloud_runtime.providers.queue.queue import Queue
from vonage_cloud_runtime.providers.scheduler.scheduler import Scheduler
from vonage_cloud_runtime.providers.messages.messages import Messages
from vonage_cloud_runtime.providers.voice.voice import Voice
from vonage_cloud_runtime.providers.assets.assets import Assets

vcr = VCR()
session = vcr.createSession()

# For shared state across replicas, use getInstanceState() — not State(session)
state     = vcr.getInstanceState()
queue     = Queue(session)
scheduler = Scheduler(session)
messages  = Messages(session)
voice     = Voice(session)
assets    = Assets(session)

import os
from fastapi import FastAPI, Request
from vonage_cloud_runtime.vcr import VCR
from vonage_cloud_runtime.providers.voice.voice import Voice
from vonage_cloud_runtime.providers.messages.messages import Messages
from vonage_cloud_runtime.providers.state.state import State

app = FastAPI()
vcr = VCR()

# Use getGlobalSession() for registering instance-wide subscriptions.
# Do NOT use vcr.createSession() here — it creates a random session that
# cannot be shared across replicas or requests.
global_session = vcr.getGlobalSession()
voice = Voice(global_session)
messages = Messages(global_session)

@app.on_event("startup")
async def startup():
    await voice.onCall("onCall")
    await messages.onMessage("onMessage",
        {"type": "sms", "number": os.environ.get("VONAGE_NUMBER")},
        {"type": "sms", "number": None}
    )

@app.post("/onCall")
async def on_call(request: Request):
    body = await request.json()
    # Scope session to the conversation using its unique ID from the callback
    session = vcr.createSessionWithId(body.get("conversation_uuid"))
    state = State(session)
    await state.set("status", "answered")
    return [{"action": "talk", "text": "Hello from VCR!"}]

@app.post("/onMessage")
async def on_message(request: Request):
    body = await request.json()
    # Scope session to the sender — deterministic, same across replicas
    session = vcr.createSessionWithId(body.get("from"))
    state = State(session)
    await state.set("lastMessage", body.get("text"))
    return {"status": "ok"}

@app.get("/_/health")
async def health():
    return {"status": "ok"}

if __name__ == "__main__":
    import uvicorn
    port = int(os.environ.get("VCR_PORT", 8080))
    uvicorn.run(app, host="0.0.0.0", port=port)