Naming and Trading Service

Intent

Enable dynamic discovery and selection of FHIR endpoints based on capability requirements and service characteristics, supporting federated healthcare networks.

Forces

Endpoint Capability Variation: FHIR servers implement different subsets of the specification, leading to compatibility issues.

Structure

The Naming and Trading Service pattern provides service discovery and capability-based endpoint selection for federated FHIR networks.

Key Components

NamingService

Provides endpoint lookup by name, identifier, or organization

TradingService

Matches capability requirements to available endpoints

ServiceRegistry

Central repository of registered endpoints and their metadata

CapabilityMatcher

Evaluates endpoint capabilities against request requirements

HealthMonitor

Tracks endpoint health and availability status

Behavior

Service Discovery Flow

The following sequence shows how a client discovers and selects an appropriate FHIR endpoint:

Discovery Process

Specify Requirements
Query Registry
Evaluate Capabilities
Apply Constraints
Rank Results
Return Selection

Implementation Considerations

Service Registration

Configuration schema for registering FHIR endpoints with the naming service. Includes capability declarations, SLA metadata, and health check settings.

Service Registration

# Service Registration Configuration
# Naming and Trading Service - Endpoint Registration

service_registration:
  endpoint_id: "acme-fhir-r4"
  organization:
    name: "ACME Healthcare"
    identifier: "urn:oid:2.16.840.1.113883.3.123"

  base_url: "https://fhir.acme-healthcare.org/r4"

  fhir_version: "4.0.1"

  capabilities:
    supported_resources:
      - Patient
      - Encounter
      - Observation
      - Condition
      - MedicationRequest
      - DiagnosticReport
      - DocumentReference

    supported_operations:
      - "$export"
      - "$everything"
      - "$validate"
      - "$match"

    supported_profiles:
      - "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"
      - "http://hl7.org/fhir/us/core/StructureDefinition/us-core-encounter"
      - "http://hl7.org/fhir/us/core/StructureDefinition/us-core-observation-lab"

    search_capabilities:
      include: true
      revinclude: true
      chained_parameters: true
      composite_parameters: true

  service_metadata:
    performance_tier: "premium"
    location: "us-east-1"
    availability_zone: "primary"
    sla:
      uptime: "99.9%"
      max_response_time: "500ms"
      support_level: "24x7"

    trust_framework:
      certification: "ONC-ACB"
      security_level: "high"
      data_classification: "PHI"

  health_check:
    endpoint: "/metadata"
    interval_seconds: 30
    timeout_seconds: 10
    failure_threshold: 3
    recovery_threshold: 2

# Multiple endpoint registration example
multi_endpoint_registration:
  - endpoint_id: "clinic-a"
    base_url: "https://fhir.clinic-a.org/r4"
    performance_tier: "standard"
    specialties: ["primary-care", "pediatrics"]

  - endpoint_id: "hospital-main"
    base_url: "https://fhir.hospital.org/r4"
    performance_tier: "premium"
    specialties: ["emergency", "inpatient", "surgery"]

  - endpoint_id: "lab-services"
    base_url: "https://fhir.lab-corp.org/r4"
    performance_tier: "standard"
    specialties: ["laboratory", "diagnostics"]

Capability Matching

Evaluates client capability requirements against registered endpoint capabilities to find suitable matches with quality scoring.

Capability Matching

from dataclasses import dataclass
from typing import List, Dict, Optional, Set
from enum import Enum

class MatchQuality(Enum):
    EXACT = "exact"
    COMPATIBLE = "compatible"
    PARTIAL = "partial"
    NONE = "none"

@dataclass
class CapabilityRequirement:
    """Represents a client's capability requirements"""
    resource_types: List[str]
    operations: List[str] = None
    profiles: List[str] = None
    search_params: Dict[str, List[str]] = None
    min_fhir_version: str = None

@dataclass
class CapabilityMatch:
    """Result of capability matching"""
    endpoint_id: str
    quality: MatchQuality
    score: float
    matched_resources: List[str]
    matched_operations: List[str]
    matched_profiles: List[str]
    missing_capabilities: List[str]

class CapabilityMatcher:
    """Matches capability requirements to endpoint capabilities"""

    def __init__(self):
        self.weight_resource = 1.0
        self.weight_operation = 0.8
        self.weight_profile = 0.6
        self.weight_search = 0.4

    def match(self, requirement: CapabilityRequirement, 
              capability_statement: dict) -> CapabilityMatch:
        """
        Evaluate how well an endpoint matches capability requirements.
        Returns a CapabilityMatch with quality assessment and score.
        """
        # Extract capabilities from CapabilityStatement
        endpoint_resources = self._extract_resources(capability_statement)
        endpoint_operations = self._extract_operations(capability_statement)
        endpoint_profiles = self._extract_profiles(capability_statement)

        # Match each requirement type
        resource_match = self._match_resources(
            requirement.resource_types, endpoint_resources
        )
        operation_match = self._match_operations(
            requirement.operations or [], endpoint_operations
        )
        profile_match = self._match_profiles(
            requirement.profiles or [], endpoint_profiles
        )

        # Calculate overall score
        score = self._calculate_score(
            resource_match, operation_match, profile_match
        )

        # Determine match quality
        quality = self._determine_quality(
            resource_match, operation_match, profile_match
        )

        # Identify missing capabilities
        missing = self._find_missing(
            requirement, endpoint_resources, 
            endpoint_operations, endpoint_profiles
        )

        return CapabilityMatch(
            endpoint_id=capability_statement.get('id', 'unknown'),
            quality=quality,
            score=score,
            matched_resources=resource_match['matched'],
            matched_operations=operation_match['matched'],
            matched_profiles=profile_match['matched'],
            missing_capabilities=missing
        )

    def _extract_resources(self, cap_statement: dict) -> Set[str]:
        """Extract supported resource types from CapabilityStatement"""
        resources = set()
        for rest in cap_statement.get('rest', []):
            for resource in rest.get('resource', []):
                resources.add(resource.get('type'))
        return resources

    def _extract_operations(self, cap_statement: dict) -> Set[str]:
        """Extract supported operations from CapabilityStatement"""
        operations = set()
        for rest in cap_statement.get('rest', []):
            # System-level operations
            for op in rest.get('operation', []):
                operations.add(op.get('name'))
            # Resource-level operations
            for resource in rest.get('resource', []):
                for op in resource.get('operation', []):
                    operations.add(op.get('name'))
        return operations

    def _extract_profiles(self, cap_statement: dict) -> Set[str]:
        """Extract supported profiles from CapabilityStatement"""
        profiles = set()
        for rest in cap_statement.get('rest', []):
            for resource in rest.get('resource', []):
                profiles.update(resource.get('supportedProfile', []))
        return profiles

    def _match_resources(self, required: List[str], 
                        available: Set[str]) -> dict:
        """Match required resources against available"""
        matched = [r for r in required if r in available]
        return {
            'matched': matched,
            'ratio': len(matched) / len(required) if required else 1.0
        }

    def _match_operations(self, required: List[str], 
                         available: Set[str]) -> dict:
        """Match required operations against available"""
        matched = [op for op in required if op in available]
        return {
            'matched': matched,
            'ratio': len(matched) / len(required) if required else 1.0
        }

    def _match_profiles(self, required: List[str], 
                       available: Set[str]) -> dict:
        """Match required profiles against available"""
        matched = [p for p in required if p in available]
        return {
            'matched': matched,
            'ratio': len(matched) / len(required) if required else 1.0
        }

    def _calculate_score(self, resource_match: dict, 
                        operation_match: dict, 
                        profile_match: dict) -> float:
        """Calculate weighted match score (0.0 to 1.0)"""
        total_weight = self.weight_resource + self.weight_operation + self.weight_profile

        score = (
            resource_match['ratio'] * self.weight_resource +
            operation_match['ratio'] * self.weight_operation +
            profile_match['ratio'] * self.weight_profile
        ) / total_weight

        return round(score, 3)

    def _determine_quality(self, resource_match: dict,
                          operation_match: dict,
                          profile_match: dict) -> MatchQuality:
        """Determine overall match quality"""
        if (resource_match['ratio'] == 1.0 and 
            operation_match['ratio'] == 1.0 and
            profile_match['ratio'] == 1.0):
            return MatchQuality.EXACT

        if resource_match['ratio'] == 1.0:
            if operation_match['ratio'] >= 0.5 or profile_match['ratio'] >= 0.5:
                return MatchQuality.COMPATIBLE
            return MatchQuality.PARTIAL

        if resource_match['ratio'] >= 0.5:
            return MatchQuality.PARTIAL

        return MatchQuality.NONE

    def _find_missing(self, requirement: CapabilityRequirement,
                     resources: Set[str], operations: Set[str],
                     profiles: Set[str]) -> List[str]:
        """Identify missing capabilities"""
        missing = []

        for r in requirement.resource_types:
            if r not in resources:
                missing.append(f"Resource: {r}")

        for op in (requirement.operations or []):
            if op not in operations:
                missing.append(f"Operation: {op}")

        for p in (requirement.profiles or []):
            if p not in profiles:
                missing.append(f"Profile: {p}")

        return missing


# Usage example
if __name__ == "__main__":
    matcher = CapabilityMatcher()

    # Define requirements
    requirements = CapabilityRequirement(
        resource_types=["Patient", "Observation", "Condition"],
        operations=["$export", "$everything"],
        profiles=["http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"]
    )

    # Sample capability statement
    cap_statement = {
        "id": "example-server",
        "rest": [{
            "resource": [
                {"type": "Patient", "supportedProfile": [
                    "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"
                ]},
                {"type": "Observation"},
                {"type": "Condition"}
            ],
            "operation": [
                {"name": "$export"},
                {"name": "$everything"}
            ]
        }]
    }

    result = matcher.match(requirements, cap_statement)
    print(f"Match Quality: {result.quality.value}")
    print(f"Score: {result.score}")
    print(f"Missing: {result.missing_capabilities}")

Health Monitoring

Continuous monitoring ensures endpoint availability: - Active Probing: Periodic capability statement requests - Passive Monitoring: Track success/failure rates of routed requests - Circuit Breaker: Temporarily remove unhealthy endpoints - Recovery Detection: Automatically restore recovered endpoints

Broker: Broker uses Naming/Trading to discover and select appropriate endpoints for routing
Capability Facade: Capability Facade provides simplified capability queries that Naming/Trading uses for matching
Security Strategy: Security Strategy validates credentials before allowing service discovery operations

Benefits

Dynamic Discovery: Endpoints can be added or removed without client changes
Capability Matching: Requests automatically route to capable servers
Load Balancing: Traffic distributed across multiple endpoints
Fault Tolerance: Automatic failover when endpoints become unavailable
Federated Networks: Supports multi-organization healthcare networks

Trade-offs

Registry Dependency: Central registry becomes critical infrastructure
Stale Data: Capability information may become outdated
Complexity: More complex than static endpoint configuration
Latency: Discovery adds overhead to initial requests

References

FHIR CapabilityStatement - Server capability declaration for service discovery
FHIR Endpoint Resource - Technical endpoint for healthcare services

Caching Strategy

Implement aggressive caching of capability information with background refresh. Most capability changes are infrequent, so cached data is usually valid.

Naming and Trading Service

Intent

Forces

Structure

Key Components

NamingService

TradingService

ServiceRegistry

CapabilityMatcher

HealthMonitor

Behavior

Service Discovery Flow

Discovery Process

Implementation Considerations

Service Registration

Capability Matching

Health Monitoring

Related Patterns

Benefits

Trade-offs

References