Listening for association requests

Assuming you have added your supported presentation contexts then you can start listening for association requests from peers with the AE.start_server() method:

from pynetdicom import AE
from pynetdicom.sop_class import Verification

ae = AE()
ae.add_supported_context(Verification)

# Listen for association requests on a IPv4 address, IPv6 is also supported
ae.start_server(("127.0.0.1", 11112))

The above is suitable as an implementation of the Verification Service Class, however other service classes will require that you implement and bind one or more of the intervention event handlers.

The association server can be started in both blocking (default) and non-blocking modes:

from pynetdicom import AE
from pynetdicom.sop_class import Verification

ae = AE()
ae.add_supported_context(Verification)

# Returns a ThreadedAssociationServer instance
server = ae.start_server(("127.0.0.1", 11112), block=False)

# Blocks
ae.start_server(("127.0.0.1", 11113), block=True)

The returned ThreadedAssociationServer instances can be stopped using shutdown() and all active associations can be stopped using AE.shutdown().

Specifying the AE Title

The AE title for each SCP can be set using the ae_title keyword parameter. If no value is set then the AE title of the parent AE will be used instead:

ae.start_server(("127.0.0.1", 11112), ae_title='STORE_SCP')

Specifying Presentation Contexts for each SCP

To support presentation contexts on a per-SCP basis you can use the contexts keyword parameter:

from pynetdicom import AE, build_context

ae = AE()
supported_cx = [build_context('1.2.840.10008.1.1')]
ae.start_server(("127.0.0.1", 11112), contexts=supported_cx)

Binding Event Handlers

If you want to bind handlers to any events within any Association instances generated by the SCP you can use the evt_handlers keyword parameter:

import logging

from pynetdicom import AE, evt, debug_logger
from pynetdicom.sop_class import Verification

debug_logger()
LOGGER = logging.getLogger('pynetdicom')

def handle_open(event):
    """Print the remote's (host, port) when connected."""
    LOGGER.info(f"Connected with remote at {event.address}")

def handle_accepted(event, arg1, arg2):
    """Demonstrate the use of the optional extra parameters"""
    LOGGER.info(f"Extra args: '{arg1}' and '{arg2}'")

# If a 2-tuple then only `event` parameter
# If a 3-tuple then the third value should be a list of objects to pass the handler
handlers = [
    (evt.EVT_CONN_OPEN, handle_open),
    (evt.EVT_ACCEPTED, handle_accepted, ['optional', 'parameters']),
]

ae = AE()
ae.add_supported_context(Verification)
ae.start_server(("127.0.0.1", 11112), evt_handlers=handlers)

Handlers can also be bound and unbound from events in an existing ThreadedAssociationServer, provided you run in non-blocking mode:

import logging
import time

from pynetdicom import AE, evt, debug_logger
from pynetdicom.sop_class import Verification

debug_logger()
LOGGER = logging.getLogger('pynetdicom')

def handle_open(event):
    """Print the remote's (host, port) when connected."""
    LOGGER.info(f"Connected with remote at {event.address}")

def handle_close(event):
    """Print the remote's (host, port) when disconnected."""
    LOGGER.info(f"Disconnected from remote at {event.address}")

handlers = [(evt.EVT_CONN_OPEN, handle_open)]

ae = AE()
ae.add_supported_context(Verification)
scp = ae.start_server(("127.0.0.1", 11112), block=False, evt_handlers=handlers)

time.sleep(20)

scp.unbind(evt.EVT_CONN_OPEN, handle_open)
scp.bind(evt.EVT_CONN_CLOSE, handle_close)

LOGGER.info("Bindings changed")

time.sleep(20)

scp.shutdown()

This will bind/unbind the handler from all currently running Association instances generated by the server as well as new Association instances generated in response to future association requests. Associations created using AE.associate() will be unaffected.

TLS

The client sockets generated by the association server can also be wrapped in TLS by supplying a ssl.SSLContext instance via the ssl_context keyword parameter:

import ssl

from pynetdicom import AE
from pynetdicom.sop_class import Verification

ae = AE()
ae.add_supported_context(Verification)

# Create the SSLContext, your requirements may vary
ssl_cx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_cx.verify_mode = ssl.CERT_REQUIRED
ssl_cx.load_cert_chain(certfile='server.crt', keyfile='server.key')
ssl_cx.load_verify_locations(cafile='client.crt')

server = ae.start_server(("127.0.0.1", 11112), block=False, ssl_context=ssl_cx)

Providing DIMSE services

If the association supports a service class that uses one or more of the DIMSE-C or -N services then a handler must be implemented and bound to the event corresponding to the service (excluding C-ECHO which has a default implementation that always returns a 0x0000 Success response):

DIMSE service	Event
C-ECHO	evt.EVT_C_ECHO
C-FIND	evt.EVT_C_FIND
C-GET	evt.EVT_C_GET
C-MOVE	evt.EVT_C_MOVE
C-STORE	evt.EVT_C_STORE
N-ACTION	evt.EVT_N_ACTION
N-CREATE	evt.EVT_N_CREATE
N-DELETE	evt.EVT_N_DELETE
N-EVENT-REPORT	evt.EVT_N_EVENT_REPORT
N-GET	evt.EVT_N_GET
N-SET	evt.EVT_N_SET

For instance, if your SCP is to support the Storage Service then you would implement and bind a handler for the evt.EVT_C_STORE event in manner similar to:

from pynetdicom import AE, evt
from pynetdicom.sop_class import CTImageStorage

ae = AE()
ae.add_supported_context(CTImageStorage)

def handle_store(event):
    """Handle evt.EVT_C_STORE"""
    # This is just a toy implementation that doesn't store anything and
    # always returns a Success response
    return 0x0000

handlers = [(evt.EVT_C_STORE, handle_store)]

# Listen for association requests
ae.start_server(("127.0.0.1", 11112), evt_handlers=handlers)

For more detailed information on implementing the DIMSE service provider handlers please see the handler implementation documentation and the examples corresponding to the service class you’re interested in.