Unofficial asynchronous Python client library for the Rivian API. Provides GraphQL-based access to vehicle state, commands, and charging data for Rivian vehicles (R1T/R1S).
- Authentication: Username/password login with OTP support
- Vehicle State: Real-time vehicle data via WebSocket subscriptions
- Vehicle Commands: Remote control (lock/unlock, climate, charging, etc.)
- Phone Enrollment: BLE pairing for both Gen 1 and Gen 2 vehicles
- Parallax Protocol: Cloud-based vehicle commands and data retrieval (analysis complete, implementation planned)
- Charging: Live charging session data, wallbox management, and smart charging schedules (v2.1+)
- User Management: Access user info, shared drivers, and vehicle images
- Location Sharing: Send destinations to vehicle navigation (v2.1+)
- Trip Planning: Multi-stop trip planning with charging stops (v2.1+)
- Trailer Management: Manage trailer profiles for R1T (v2.1+)
- Advanced Key Management: CCC/WCC2 digital key support (v2.1+)
- Type-Safe: Built with gql DSL for GraphQL queries with comprehensive schema
pip install rivian-python-clientpip install rivian-python-client[ble]from rivian import Rivian
async with Rivian() as client:
# Simple login (returns OTP token if MFA is enabled)
otp_token = await client.login("user@example.com", "password")
# If OTP required, validate it
if otp_token:
await client.login_with_otp("user@example.com", "123456", otp_token)# Get user info with vehicles (v2.0+ returns dict directly)
user_info = await client.get_user_information()
print(f"User ID: {user_info['id']}")
for vehicle in user_info['vehicles']:
print(f"Vehicle: {vehicle['name']} ({vehicle['vin']})")
# Include enrolled phones
user_info = await client.get_user_information(include_phones=True)
for phone in user_info.get('enrolledPhones', []):
print(f"Phone ID: {phone['vas']['vasPhoneId']}")# Get drivers and keys for a vehicle
drivers = await client.get_drivers_and_keys(vehicle_id)
print(f"VIN: {drivers['vin']}")
for user in drivers['invitedUsers']:
print(f"User: {user.get('email')} - Devices: {len(user.get('devices', []))}")
# Get vehicle images
images = await client.get_vehicle_images(extension="png", resolution="@2x")
for img in images['getVehicleMobileImages']:
print(f"Image: {img['url']}")# Get registered wallboxes
wallboxes = await client.get_registered_wallboxes()
for wallbox in wallboxes:
print(f"Wallbox: {wallbox['name']} - Status: {wallbox['chargingStatus']}")from rivian.const import VEHICLE_STATE_PROPERTIES
# Subscribe to vehicle state updates
async for update in client.subscribe_for_vehicle_updates(
vehicle_id=vehicle_id,
properties=VEHICLE_STATE_PROPERTIES
):
print(f"Battery: {update['batteryLevel']['value']}%")
print(f"Range: {update['distanceToEmpty']['value']} miles")from rivian import VehicleCommand
# Unlock vehicle
command_id = await client.send_vehicle_command(
vehicle_id=vehicle_id,
command=VehicleCommand.UNLOCK_ALL_CLOSURES
)
# Monitor command status via WebSocket
async for state in client.subscribe_for_command_state(command_id):
print(f"Command state: {state['state']}")
if state['state'] in ['COMPLETE', 'FAILED']:
breakParallax is Rivian's cloud-based protocol for remote vehicle commands and data retrieval. Unlike BLE commands which require proximity, Parallax operates through Rivian's cloud infrastructure and works from anywhere with internet connectivity.
Query Methods (Read-Only):
get_charging_session_live_data()- Real-time charging metricsget_climate_hold_status()- Current cabin climate stateget_ota_status()- Software update statusget_trip_progress()- Navigation progress
Control Methods (Write):
set_climate_hold()- Configure cabin climateset_charging_schedule()- Set charging time windows
# Get live charging session data
data = await client.get_charging_session_live_data(vehicle_id)
if data['success']:
# Payload contains Base64-encoded protobuf
print(f"Charging payload: {data['payload']}")# Enable climate hold at 22°C for 2 hours
result = await client.set_climate_hold(
vehicle_id=vehicle_id,
enabled=True,
temp_celsius=22.0,
duration_minutes=120
)
print(f"Climate set: {result['success']}")
# Check climate status
status = await client.get_climate_hold_status(vehicle_id)
print(f"Climate active: {status['success']}")# Charge only between 10 PM and 6 AM
result = await client.set_charging_schedule(
vehicle_id=vehicle_id,
start_hour=22,
start_minute=0,
end_hour=6,
end_minute=0
)
print(f"Schedule set: {result['success']}")from rivian import ParallaxCommand, RVMType
# Build custom command
cmd = ParallaxCommand(RVMType.HALLOWEEN_SETTINGS, b"")
result = await client.send_parallax_command(vehicle_id, cmd)Phase 1 (6 implemented): Charging data, Climate, OTA, Trip progress Phase 2 (12 pending): Energy analytics, Geofence, GearGuard, Wheels, Ventilation, Passive entry, Halloween
For complete protocol details: PARALLAX_PROTOCOL.md
# Get referral code
referral = await client.get_referral_code()
print(f"Share this code: {referral['code']}")
print(f"Or share this URL: {referral['url']}")
# Get vehicle invitations
invites = await client.get_invitations_by_user()
for invite in invites:
print(f"Invited to {invite['vehicleModel']} by {invite['invitedByFirstName']}")# Get service appointments
appointments = await client.get_service_appointments(vehicle_id)
for appt in appointments:
print(f"{appt['serviceType']} on {appt['scheduledTime']} - {appt['status']}")
# Get active service requests
requests = await client.get_active_service_requests(vehicle_id)
for req in requests:
print(f"{req['category']}: {req['description']} [{req['status']}]")
# Get users with access to vehicle
users = await client.get_vehicle_provisioned_users(vehicle_id)
for user in users:
print(f"{user['firstName']} {user['lastName']} - {', '.join(user['roles'])}")# Register multiple notification tokens
tokens = [
{"token": "ios_token_1", "platform": "ios", "deviceId": "device1"},
{"token": "android_token_1", "platform": "android", "deviceId": "device2"}
]
result = await client.register_notification_tokens(tokens)
print(f"Registered {len(result['registeredTokens'])} tokens")
# Register single push notification
await client.register_push_notification_token("my_token", "ios", vehicle_id)
# Register live notification token
await client.register_live_notification_token(vehicle_id, "live_token")# Get chat session
session = await client.get_chat_session(vehicle_id)
print(f"Chat session {session['sessionId']}: {session['status']}")# Get charging schedules
schedules = await client.get_charging_schedules(vehicle_id)
for schedule in schedules['schedules']:
print(f"{schedule['name']}: {schedule['departureTime']} on {schedule['days']}")
# Create/update departure schedule
schedule = await client.update_departure_schedule(
vehicle_id=vehicle_id,
schedule={
"name": "Weekday Commute",
"enabled": True,
"days": ["MONDAY", "TUESDAY", "WEDNESDAY", "THURSDAY", "FRIDAY"],
"departureTime": "08:00",
"cabinPreconditioning": True,
"cabinPreconditioningTemp": 21.0,
"targetSOC": 80
}
)
# Enable/disable smart charging
await client.enroll_in_smart_charging(vehicle_id)
await client.unenroll_from_smart_charging(vehicle_id)# Share GPS coordinates
result = await client.share_location_to_vehicle(
vehicle_id=vehicle_id,
latitude=37.7749,
longitude=-122.4194
)
# Share Google Place
result = await client.share_place_id_to_vehicle(
vehicle_id=vehicle_id,
place_id="ChIJN1t_tDeuEmsRUsoyG83frY4"
)# Plan multi-stop trip with charging
trip = await client.plan_trip_with_multi_stop(
vehicle_id=vehicle_id,
waypoints=[
{"latitude": 37.7749, "longitude": -122.4194, "name": "San Francisco"},
{"latitude": 34.0522, "longitude": -118.2437, "name": "Los Angeles"}
],
options={
"avoidTolls": False,
"targetArrivalSOC": 20
}
)
print(f"Trip: {trip['totalDistance']} miles, {trip['totalDuration']} minutes")
for stop in trip['chargingStops']:
print(f"Charging stop: {stop['location']['name']}")# Get trailer profiles
trailers = await client.get_trailer_profiles(vehicle_id)
for trailer in trailers:
print(f"{trailer['name']}: {trailer['length']}m, pinned={trailer['pinnedToGear']}")
# Pin/unpin trailer to gear
await client.update_pin_to_gear(vehicle_id, trailer_id, pinned=True)# Subscribe to Gear Guard config updates
def on_config_update(config):
print(f"Gear Guard: {config['videoMode']}, storage={config['storageRemaining']}%")
unsubscribe = await client.subscribe_for_gear_guard_config(vehicle_id, on_config_update)Query methods now return dict/list[dict] directly instead of ClientResponse. No need to call .json().
Before (v1.x):
response = await client.get_user_information()
data = await response.json()
user = data["data"]["currentUser"]
vehicles = user["vehicles"]After (v2.0):
user = await client.get_user_information()
# Returns dict directly - no .json() needed
vehicles = user["vehicles"]Affected Methods:
get_user_information()- Returnsdictget_drivers_and_keys(vehicle_id)- Returnsdictget_registered_wallboxes()- Returnslist[dict]get_vehicle_images()- Returnsdict[str, list[dict]]
The new login() and login_with_otp() methods provide a simpler authentication flow.
Before (v1.x):
await client.create_csrf_token()
await client.authenticate(username, password)
if client._otp_needed:
await client.validate_otp(username, otp_code)After (v2.0 - Recommended):
otp_token = await client.login(username, password)
if otp_token:
await client.login_with_otp(username, otp_code, otp_token)Note: Legacy auth methods (create_csrf_token(), authenticate(), validate_otp()) still work but are deprecated.
Methods that still return ClientResponse (unchanged):
get_vehicle_state(vin)- Use WebSocket subscriptions for real-time dataget_vehicle_command_state(command_id)- Usesubscribe_for_command_state()insteadget_live_charging_session(vin)- Usesubscribe_for_charging_session()insteadget_vehicle_ota_update_details(vehicle_id)
Poetry is used for dependency management.
curl -sSL https://install.python-poetry.org | python3 -Install project dependencies into the poetry virtual environment and setup pre-commit hooks:
poetry install
pre-commit installpoetry install --extras blepoetry run pytest# Specific test file
poetry run pytest tests/rivian_test.py
# Specific test function
poetry run pytest tests/rivian_test.py::test_authentication# Run ruff linter with auto-fix
poetry run ruff check --fix
# Run ruff formatter
poetry run ruff format
# Run type checking
poetry run mypy src/rivianPre-commit hooks automatically run ruff (linter with --fix) and ruff-format on staged files.
- Python 3.9+
- Full support for Python 3.9 - 3.13
- Uses conditional imports for version-specific features
MIT License - See LICENSE file for details
This is an unofficial client library and is not affiliated with, endorsed by, or connected to Rivian Automotive, LLC. Use at your own risk.
Contributions are welcome! Please open an issue or pull request for bug fixes, features, or documentation improvements.
See CHANGELOG.md for detailed version history and migration guides.