Microsoft Azure Media Services Client Library for Python - A management library for Azure Media Services that provides programmatic access to media processing and streaming capabilities in the cloud.
83
Quality
Pending
Does it follow best practices?
Impact
83%
1.09xAverage score across 10 eval scenarios
Comprehensive live streaming capabilities for real-time content delivery including live event management, input configuration, preview settings, and output publishing. Supports RTMP and fragmented MP4 ingest with adaptive bitrate encoding and global distribution through Azure's content delivery network.
Create and manage live streaming events with complete lifecycle control and long-running operation support.
def list(resource_group_name: str, account_name: str) -> Iterable[LiveEvent]:
"""
List all live events in a media service account.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
Returns:
Iterable of LiveEvent objects with configuration and status
"""
def get(
resource_group_name: str,
account_name: str,
live_event_name: str
) -> LiveEvent:
"""
Get a specific live event with complete configuration details.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
Returns:
LiveEvent object with current state and configuration
"""
def begin_create(
resource_group_name: str,
account_name: str,
live_event_name: str,
parameters: LiveEvent
) -> LROPoller[LiveEvent]:
"""
Create a new live event with specified configuration.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name for the new live event (str)
- parameters: Live event configuration (LiveEvent)
Returns:
LROPoller for tracking the creation operation
"""
def begin_update(
resource_group_name: str,
account_name: str,
live_event_name: str,
parameters: LiveEvent
) -> LROPoller[LiveEvent]:
"""
Update an existing live event configuration.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
- parameters: Updated live event configuration (LiveEvent)
Returns:
LROPoller for tracking the update operation
"""
def begin_delete(
resource_group_name: str,
account_name: str,
live_event_name: str
) -> LROPoller[None]:
"""
Delete a live event.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
Returns:
LROPoller for tracking the deletion operation
"""Control live event state transitions with resource allocation and streaming control.
def begin_allocate(
resource_group_name: str,
account_name: str,
live_event_name: str
) -> LROPoller[None]:
"""
Allocate resources for a live event.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
Returns:
LROPoller for tracking the allocation operation
"""
def begin_start(
resource_group_name: str,
account_name: str,
live_event_name: str
) -> LROPoller[None]:
"""
Start a live event to begin accepting ingest streams.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
Returns:
LROPoller for tracking the start operation
"""
def begin_stop(
resource_group_name: str,
account_name: str,
live_event_name: str,
parameters: LiveEventActionInput
) -> LROPoller[None]:
"""
Stop a live event and optionally remove streaming endpoints.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
- parameters: Stop action configuration (LiveEventActionInput)
Returns:
LROPoller for tracking the stop operation
"""
def begin_reset(
resource_group_name: str,
account_name: str,
live_event_name: str
) -> LROPoller[None]:
"""
Reset a live event, clearing any cached state.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
Returns:
LROPoller for tracking the reset operation
"""Manage live outputs for archiving live streams to assets with automatic recording capabilities.
def list(
resource_group_name: str,
account_name: str,
live_event_name: str
) -> Iterable[LiveOutput]:
"""
List all live outputs for a live event.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
Returns:
Iterable of LiveOutput objects with archiving configuration
"""
def get(
resource_group_name: str,
account_name: str,
live_event_name: str,
live_output_name: str
) -> LiveOutput:
"""
Get a specific live output with configuration details.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
- live_output_name: Name of the live output (str)
Returns:
LiveOutput object with current state and configuration
"""
def begin_create(
resource_group_name: str,
account_name: str,
live_event_name: str,
live_output_name: str,
parameters: LiveOutput
) -> LROPoller[LiveOutput]:
"""
Create a live output for archiving live streams.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
- live_output_name: Name for the new live output (str)
- parameters: Live output configuration (LiveOutput)
Returns:
LROPoller for tracking the creation operation
"""
def begin_delete(
resource_group_name: str,
account_name: str,
live_event_name: str,
live_output_name: str
) -> LROPoller[None]:
"""
Delete a live output.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
- live_output_name: Name of the live output (str)
Returns:
LROPoller for tracking the deletion operation
"""Monitor long-running live streaming operations with status tracking and result retrieval.
def async_operation(
resource_group_name: str,
account_name: str,
operation_id: str
) -> AsyncOperationResult:
"""
Get the result of an asynchronous live streaming operation.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- operation_id: Identifier of the async operation (str)
Returns:
AsyncOperationResult with operation status and details
"""
def operation_location(
resource_group_name: str,
account_name: str,
live_event_name: str,
operation_id: str
) -> LiveEvent:
"""
Get live event operation result by location.
Parameters:
- resource_group_name: Name of the resource group (str)
- account_name: Name of the media service account (str)
- live_event_name: Name of the live event (str)
- operation_id: Identifier of the operation (str)
Returns:
LiveEvent object with updated state
"""class LiveEvent:
"""Live streaming event configuration and state."""
name: str
description: str
location: str
resource_state: str # LiveEventResourceState enum
input: LiveEventInput
preview: LiveEventPreview
encoding: LiveEventEncoding
hostname_prefix: str
use_static_hostname: bool
stream_options: List[str] # StreamOptionsFlag enum
created: str
last_modified: str
provisioning_state: str
class LiveEventInput:
"""Live event input configuration."""
streaming_protocol: str # LiveEventInputProtocol enum (FragmentedMP4, RTMP)
access_control: LiveEventInputAccessControl
key_frame_interval_duration: str
access_token: str
endpoints: List[LiveEventEndpoint]
class LiveEventInputAccessControl:
"""Access control for live event input."""
ip: IPAccessControl
class IPAccessControl:
"""IP-based access control configuration."""
allow: List[IPRange]
class IPRange:
"""IP address range for access control."""
name: str
address: str
subnet_prefix_length: int
class LiveEventEndpoint:
"""Live event streaming endpoint."""
protocol: str
url: str
class LiveEventPreview:
"""Live event preview configuration."""
endpoints: List[LiveEventEndpoint]
access_control: LiveEventPreviewAccessControl
preview_locator: str
streaming_policy_name: str
alternative_media_id: str
class LiveEventPreviewAccessControl:
"""Access control for live event preview."""
ip: IPAccessControl
class LiveEventEncoding:
"""Live event encoding configuration."""
encoding_type: str # LiveEventEncodingType enum
preset_name: str
stretch_mode: str # StretchMode enum
key_frame_interval: str
class LiveOutput:
"""Live output for archiving live streams."""
name: str
description: str
asset_name: str
archive_window_length: str
manifest_name: str
hls: Hls
output_snap_time: int
created: str
last_modified: str
provisioning_state: str
resource_state: str # LiveOutputResourceState enum
class Hls:
"""HLS-specific live output configuration."""
fragments_per_ts_segment: int
class LiveEventActionInput:
"""Input parameters for live event actions."""
remove_outputs_on_stop: bool
class AsyncOperationResult:
"""Result of an asynchronous operation."""
name: str
status: str # AsyncOperationStatus enum
properties: dict
error: dictfrom azure.mgmt.media import AzureMediaServices
from azure.mgmt.media.models import (
LiveEvent, LiveEventInput, LiveEventPreview, LiveEventEncoding,
LiveEventInputProtocol, LiveEventEncodingType, IPAccessControl, IPRange
)
from azure.identity import DefaultAzureCredential
client = AzureMediaServices(
credential=DefaultAzureCredential(),
subscription_id="your-subscription-id"
)
# Configure input access control (allow all IPs - use restrictive rules in production)
ip_range = IPRange(
name="AllowAll",
address="0.0.0.0",
subnet_prefix_length=0
)
input_access_control = LiveEventInputAccessControl(
ip=IPAccessControl(allow=[ip_range])
)
# Configure live event input
live_input = LiveEventInput(
streaming_protocol=LiveEventInputProtocol.RTMP,
access_control=input_access_control,
key_frame_interval_duration="PT2S"
)
# Configure preview
preview = LiveEventPreview(
access_control=LiveEventPreviewAccessControl(
ip=IPAccessControl(allow=[ip_range])
)
)
# Configure encoding
encoding = LiveEventEncoding(
encoding_type=LiveEventEncodingType.STANDARD,
preset_name="Default720p"
)
# Create live event
live_event = LiveEvent(
location="East US",
description="Sample live streaming event",
input=live_input,
preview=preview,
encoding=encoding,
use_static_hostname=True
)
# Start creation
create_operation = client.live_events.begin_create(
resource_group_name="my-resource-group",
account_name="my-media-service",
live_event_name="sample-live-event",
parameters=live_event
)
# Wait for creation to complete
created_event = create_operation.result()
print(f"Live event created: {created_event.name}")
# Start the live event
start_operation = client.live_events.begin_start(
resource_group_name="my-resource-group",
account_name="my-media-service",
live_event_name="sample-live-event"
)
start_operation.result()
print("Live event started and ready for ingest")
# Get updated live event with ingest URLs
running_event = client.live_events.get(
resource_group_name="my-resource-group",
account_name="my-media-service",
live_event_name="sample-live-event"
)
print(f"Live event state: {running_event.resource_state}")
print("Ingest endpoints:")
for endpoint in running_event.input.endpoints:
print(f" {endpoint.protocol}: {endpoint.url}")from azure.mgmt.media.models import LiveOutput, Hls
# Create asset for recording
from azure.mgmt.media.models import Asset
recording_asset = Asset(
description="Live stream recording"
)
client.assets.create_or_update(
resource_group_name="my-resource-group",
account_name="my-media-service",
asset_name="live-recording-asset",
parameters=recording_asset
)
# Create live output
live_output = LiveOutput(
description="Archive live stream",
asset_name="live-recording-asset",
archive_window_length="PT1H", # 1 hour archive window
manifest_name="output",
hls=Hls(fragments_per_ts_segment=5)
)
output_operation = client.live_outputs.begin_create(
resource_group_name="my-resource-group",
account_name="my-media-service",
live_event_name="sample-live-event",
live_output_name="archive-output",
parameters=live_output
)
created_output = output_operation.result()
print(f"Live output created: {created_output.name}")
print(f"Recording to asset: {created_output.asset_name}")from azure.mgmt.media.models import LiveEventActionInput
import time
# Monitor live event status
while True:
event_status = client.live_events.get(
resource_group_name="my-resource-group",
account_name="my-media-service",
live_event_name="sample-live-event"
)
print(f"Live event state: {event_status.resource_state}")
if event_status.resource_state == "Running":
print("Live event is running successfully")
break
elif event_status.resource_state in ["Error", "Stopped"]:
print("Live event failed or stopped")
break
time.sleep(10)
# Later: Stop the live event
stop_input = LiveEventActionInput(remove_outputs_on_stop=False)
stop_operation = client.live_events.begin_stop(
resource_group_name="my-resource-group",
account_name="my-media-service",
live_event_name="sample-live-event",
parameters=stop_input
)
stop_operation.result()
print("Live event stopped")Install with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-mediadocs
evals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10