Skip to content

Latest commit

 

History

History
2340 lines (1879 loc) · 82.3 KB

CARRIER_INTEGRATION_GUIDE.md

File metadata and controls

2340 lines (1879 loc) · 82.3 KB

Karrio Carrier Integration Guide (AI-Assisted)

This guide provides explicit, step-by-step instructions for building robust Karrio carrier integrations. It is designed for both human developers and AI agents, emphasizing the mandatory use of Karrio's internal tooling to ensure consistency and correctness.

Core Principles

IMPORTANT: All development, especially when performed by an AI agent, MUST adhere to these principles:

  1. Tooling is Mandatory: Use the Karrio CLI (./bin/cli) for all code generation and scaffolding tasks. Under no circumstances should you create integration files or directories manually. If a tool fails, the priority is to debug the tool.
  2. Environment First: Always work within the activated Karrio development environment. This ensures all required dependencies and tools are available and correctly configured.
  3. Pattern Replication: Replicate the design patterns, file structures, and coding styles of existing integrations. For standard carriers, refer to modules/connectors/ups. For hub carriers (multi-carrier APIs), refer to community/plugins/easyship.
  4. Reuse, Don't Reinvent: Extensively use the shared libraries in karrio.lib, karrio.core.units, and karrio.core.models.
  5. Never Edit Generated Schemas: Generated schema files in karrio/schemas/[carrier_name]/ are NEVER edited manually. Only update source schema files and regenerate.
  6. Never Modify mapper.py: The mapper.py file is generated by the CLI and follows a standard template. Do not modify this file manually. It should only contain delegation methods that call provider functions.
  7. 🚨 MANDATORY Schema Usage: All provider functions MUST use generated schema types for request creation and response parsing. Never manipulate raw dictionaries or manual data structures.

Integration Patterns

Karrio supports two primary integration patterns:

1. Direct Carrier Pattern

Used for: UPS, FedEx, DHL, Canada Post - carriers with direct APIs

Characteristics:

  • Single carrier integration
  • Static service definitions
  • Direct API communication
  • is_hub=False in metadata

Structure Location: modules/connectors/[carrier_name]/

2. Hub Carrier Pattern

Used for: Easyship, ShipEngine, EshipPer - multi-carrier aggregators

Characteristics:

  • Multi-carrier aggregation
  • Dynamic service discovery
  • Single API, multiple underlying carriers
  • is_hub=True in metadata

Structure Location: community/plugins/[carrier_name]/

Phase 1: Setup and Scaffolding

Step 1: Activate Development Environment

All commands must be run from the project root. First, activate the environment:

source ./bin/activate-env

This command makes the Karrio CLI (kcli) and other development tools available in your shell session. This step is required.

Step 2: Bootstrap the Carrier Extension

Use the sdk add-extension command to generate the complete directory structure and boilerplate for your new integration.

./bin/cli sdk add-extension \
  --path plugins \
  --carrier-slug xship \
  --display-name "X Ship" \
  --features "rating,shipping,tracking" \
  --no-is-xml-api \
  --version "2025.5" \
  --confirm

Available Features: address, callback, document, duties, insurance, manifest, pickup, rating, shipping, tracking, webhook Use: --no-is-xml-api for JSON API and --is-xml-api for XML APIs Try Help if you encounter an error: ./bin/cli sdk add-extension --help (at the root of the project and make sure the env var has been activated.)

⚠️ CRITICAL: This single command creates the entire plugin structure. If this command fails, do not proceed with manual creation. The failure indicates an issue with the development environment or the CLI tool itself that must be resolved first.

Phase 2: Schema and Code Generation

Step 3: Populate API Schemas

  1. Analyze API: Study the carrier's API documentation. If they provide an OpenAPI/Swagger specification, use it as the source of truth.

  2. Create Sample Files: From the API documentation, create raw JSON (or XML) sample files for each API operation. Place these files in the schemas/ directory of your newly created plugin.

Required Schema Files:

  • error_response.json (always required)
  • rate_request.json + rate_response.json (if rating enabled)
  • shipment_request.json + shipment_response.json (if shipping enabled)
  • tracking_request.json + tracking_response.json (if tracking enabled)

For XML APIs: Use .xsd extension instead of .json

Note for AI Agents: If an OpenAPI spec is provided, extract the schemas for each required endpoint and save them as individual JSON files in the schemas/ directory. Ensure the JSON represents actual API request/response examples, not JSON Schema definitions.

Step 4: Configure Schema Generation

Before generating schemas, you need to configure the generation script with the correct parameters based on your carrier's API field format.

Edit the generate script to choose the correct CLI parameters:

File: modules/connectors/[carrier_name]/generate (or community/plugins/[carrier_name]/generate)

SCHEMAS=./schemas
LIB_MODULES=./karrio/schemas/[carrier_name]
find "${LIB_MODULES}" -name "*.py" -exec rm -r {} \;
touch "${LIB_MODULES}/__init__.py"

generate_schema() {
    echo "Generating $1..."
    "${ROOT}/bin/cli" codegen generate "$1" "$2" [PARAMETERS]
}

# Generate each required schema file
generate_schema "${SCHEMAS}/error_response.json" "${LIB_MODULES}/error_response.py"
generate_schema "${SCHEMAS}/rate_request.json" "${LIB_MODULES}/rate_request.py"
generate_schema "${SCHEMAS}/rate_response.json" "${LIB_MODULES}/rate_response.py"
# Add other schema files as needed...

Choose the correct [PARAMETERS] based on your API field format:

API Field Format Parameters Example APIs Reason
snake_case --nice-property-names Easyship Converts to Python snake_case
camelCase --no-nice-property-names UPS, FedEx, SEKO Preserves camelCase naming
PascalCase --no-append-type-suffix --no-nice-property-names Some enterprise APIs Prevents conflicts with class names

Examples:

# For APIs with camelCase fields (like UPS, FedEx)
"${ROOT}/bin/cli" codegen generate "$1" "$2" --no-nice-property-names

# For APIs with snake_case fields (like Easyship)
"${ROOT}/bin/cli" codegen generate "$1" "$2" --nice-property-names

# For APIs with PascalCase fields that might conflict with class names
"${ROOT}/bin/cli" codegen generate "$1" "$2" --no-append-type-suffix --no-nice-property-names

Step 5: Generate Python Data Classes

  1. Make generate Executable:

    # For direct carriers
chmod +x modules/connectors/[carrier_name]/generate

# For hub carriers
chmod +x community/plugins/[carrier_name]/generate

  2. Run the Generation Script:

    # For direct carriers
./bin/run-generate-on modules/connectors/[carrier_name]

# For hub carriers
./bin/run-generate-on community/plugins/[carrier_name]

    🚨 CRITICAL:

    • ALWAYS use ./bin/run-generate-on command from the project root
    • NEVER run ./generate directly from within the carrier directory
    • The run-generate-on script properly sets up the environment and paths

This command will populate the karrio/schemas/[carrier_name]/ directory with generated Python files (.py).

⚠️ CRITICAL RULES:

  • If this step fails, do not proceed. Debug the tool before continuing.
  • NEVER manually edit generated schema files. If changes are needed, update the source JSON/XSD files and re-run the generation.
  • Generated files are overwritten on each generation run.
  • Always verify generated classes before proceeding to implementation.
  • The generate script format MUST match the original template pattern - no custom paths or bash functions

Step 6: Verify Generated Schema Types

After successful schema generation, verify the generated Python classes:

# Check generated files
ls -la modules/connectors/[carrier_name]/karrio/schemas/[carrier_name]/
# OR
ls -la community/plugins/[carrier_name]/karrio/schemas/[carrier_name]/

# Verify imports work correctly
python -c "
import karrio.schemas.[carrier_name].rate_request as req
import karrio.schemas.[carrier_name].rate_response as res
print('Schema imports successful')
print('Request type:', req.RateRequestType if hasattr(req, 'RateRequestType') else 'Check naming')
print('Response type:', res.RateResponseType if hasattr(res, 'RateResponseType') else 'Check naming')
"

Important Notes:

  • Class names depend on your generation parameters
  • With --no-append-type-suffix: Classes named exactly as in schema (e.g., RateRequest)
  • With default settings: Classes have Type suffix (e.g., RateRequestType)
  • Generated classes use @attr.s(auto_attribs=True) and jstruct decorators

Phase 3: Implementation

Step 7: Configure Connection Settings

Edit the settings file to define carrier-specific credentials:

File: karrio/mappers/[carrier_name]/settings.py

"""Karrio [Carrier Name] client settings."""

import attr
import karrio.providers.[carrier_name].utils as provider_utils

@attr.s(auto_attribs=True)
class Settings(provider_utils.Settings):
    """[Carrier Name] connection settings."""

    # Add carrier specific API connection properties here
    api_key: str                    # For JSON APIs
    # OR
    username: str                   # For XML APIs
    password: str                   # For XML APIs

    account_number: str = None      # Optional field

    # generic properties (DO NOT MODIFY)
    id: str = None
    test_mode: bool = False
    carrier_id: str = "[carrier_name]"
    account_country_code: str = None
    metadata: dict = {}
    config: dict = {}

Step 8: Configure Provider Utilities

Edit the utils file to define server URLs and authentication:

File: karrio/providers/[carrier_name]/utils.py

import base64
import datetime
import karrio.lib as lib
import karrio.core as core

class Settings(core.Settings):
    """[Carrier Name] connection settings."""

    # Add carrier specific API connection properties
    api_key: str                    # For JSON APIs
    # OR
    username: str                   # For XML APIs
    password: str                   # For XML APIs

    account_number: str = None

    @property
    def carrier_name(self):
        return "[carrier_name]"

    @property
    def server_url(self):
        return (
            "https://api.sandbox.carrier.com"    # Test environment
            if self.test_mode
            else "https://api.carrier.com"       # Production
        )

    # For XML APIs using Basic Auth
    @property
    def authorization(self):
        pair = f"{self.username}:{self.password}"
        return base64.b64encode(pair.encode("utf-8")).decode("ascii")

    # For OAuth APIs (like UPS/FedEx)
    @property
    def access_token(self):
        """Retrieve access token with caching"""
        cache_key = f"{self.carrier_name}|{self.api_key}|{self.secret_key}"
        now = datetime.datetime.now() + datetime.timedelta(minutes=30)

        auth = self.connection_cache.get(cache_key) or {}
        token = auth.get("access_token")
        expiry = lib.to_date(auth.get("expiry"), current_format="%Y-%m-%d %H:%M:%S")

        if token is not None and expiry is not None and expiry > now:
            return token

        self.connection_cache.set(cache_key, lambda: login(self))
        new_auth = self.connection_cache.get(cache_key)

        return new_auth["access_token"]

Step 9: Configure Service Units

Edit the units file to define carrier-specific services and options:

File: karrio/providers/[carrier_name]/units.py

import karrio.lib as lib
import karrio.core.units as units

class PackagingType(lib.StrEnum):
    """Carrier specific packaging types"""
    carrier_envelope = "ENVELOPE"
    carrier_pak = "PAK"
    carrier_box = "BOX"

    """Unified Packaging type mapping"""
    envelope = carrier_envelope
    pak = carrier_pak
    small_box = carrier_box
    your_packaging = carrier_box

class ShippingService(lib.StrEnum):
    """Carrier specific services"""
    carrier_standard = "Standard Service"
    carrier_express = "Express Service"
    carrier_overnight = "Overnight Service"

class ShippingOption(lib.Enum):
    """Carrier specific options"""
    carrier_insurance = lib.OptionEnum("insurance", float)
    carrier_signature_required = lib.OptionEnum("signature_required", bool)
    carrier_saturday_delivery = lib.OptionEnum("saturday_delivery", bool)

    """Unified Option type mapping"""
    insurance = carrier_insurance
    signature_required = carrier_signature_required
    saturday_delivery = carrier_saturday_delivery

def shipping_options_initializer(
    options: dict,
    package_options: units.ShippingOptions = None,
) -> units.ShippingOptions:
    """Apply default values to the given options."""
    if package_options is not None:
        options.update(package_options.content)

    def items_filter(key: str) -> bool:
        return key in ShippingOption

    return units.ShippingOptions(options, ShippingOption, items_filter=items_filter)

class TrackingStatus(lib.Enum):
    """Maps carrier tracking status codes to normalized Karrio statuses."""
    on_hold = ["ON_HOLD"]
    delivered = ["DELIVERED"]
    in_transit = ["IN_TRANSIT"]
    delivery_failed = ["DELIVERY_FAILED"]
    out_for_delivery = ["OUT_FOR_DELIVERY"]
    pending = ["PENDING", "CREATED", "LABEL_PRINTED"]
    picked_up = ["PICKED_UP", "COLLECTED"]
    delivery_delayed = ["DELAYED", "RESCHEDULED"]
    ready_for_pickup = ["READY_FOR_PICKUP", "AT_LOCATION"]


class TrackingIncidentReason(lib.Enum):
    """Maps carrier exception codes to normalized incident reasons.

    IMPORTANT: This enum is required for tracking implementations.
    It maps carrier-specific exception/status codes to standardized
    incident reasons for tracking events. The reason field helps
    identify why a delivery exception occurred.

    Categories of reasons:
    - carrier_*: Issues caused by the carrier
    - consignee_*: Issues caused by the recipient
    - customs_*: Customs-related delays
    - weather_*: Weather/force majeure events
    """
    # Carrier-caused issues
    carrier_damaged_parcel = ["DAMAGED", "DMG"]
    carrier_sorting_error = ["MISROUTED", "MSR"]
    carrier_address_not_found = ["ADDRESS_NOT_FOUND", "ANF"]
    carrier_parcel_lost = ["LOST", "LP"]
    carrier_not_enough_time = ["LATE", "NO_TIME"]
    carrier_vehicle_issue = ["VEHICLE_BREAKDOWN", "VB"]

    # Consignee-caused issues
    consignee_refused = ["REFUSED", "RJ"]
    consignee_business_closed = ["BUSINESS_CLOSED", "BC"]
    consignee_not_available = ["NOT_AVAILABLE", "NA"]
    consignee_not_home = ["NOT_HOME", "NH"]
    consignee_incorrect_address = ["WRONG_ADDRESS", "IA"]
    consignee_access_restricted = ["ACCESS_RESTRICTED", "AR"]

    # Customs-related issues
    customs_delay = ["CUSTOMS_DELAY", "CD"]
    customs_documentation = ["CUSTOMS_DOCS", "CM"]
    customs_duties_unpaid = ["DUTIES_UNPAID", "DU"]

    # Weather/Force majeure
    weather_delay = ["WEATHER", "WE"]
    natural_disaster = ["NATURAL_DISASTER", "ND"]

    # Unknown
    unknown = []

Step 10: Implement the API Proxy

Edit the proxy file to handle HTTP communication:

File: karrio/mappers/[carrier_name]/proxy.py

"""Karrio [Carrier Name] client proxy."""

import karrio.lib as lib
import karrio.api.proxy as proxy
import karrio.mappers.[carrier_name].settings as provider_settings

class Proxy(proxy.Proxy):
    settings: provider_settings.Settings

    def get_rates(self, request: lib.Serializable) -> lib.Deserializable[str]:
        response = lib.request(
            url=f"{self.settings.server_url}/rates",
            data=lib.to_json(request.serialize()),  # Use request.serialize() for XML
            trace=self.trace_as("json"),             # Use "xml" for XML APIs
            method="POST",
            headers={
                "Content-Type": "application/json",  # Use "text/xml" for XML APIs
                "Authorization": f"Bearer {self.settings.access_token}"  # Or f"Basic {self.settings.authorization}"
            },
        )

        return lib.Deserializable(response, lib.to_dict)  # Use lib.to_element for XML

    def create_shipment(self, request: lib.Serializable) -> lib.Deserializable[str]:
        response = lib.request(
            url=f"{self.settings.server_url}/shipments",
            data=lib.to_json(request.serialize()),
            trace=self.trace_as("json"),
            method="POST",
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {self.settings.access_token}"
            },
        )

        return lib.Deserializable(response, lib.to_dict)

    def get_tracking(self, request: lib.Serializable) -> lib.Deserializable:
        def _get_tracking(tracking_number: str):
            return tracking_number, lib.request(
                url=f"{self.settings.server_url}/tracking/{tracking_number}",
                trace=self.trace_as("json"),
                method="GET",
                headers={"Authorization": f"Bearer {self.settings.access_token}"},
            )

        # Use concurrent requests for multiple tracking numbers
        responses = lib.run_concurently(_get_tracking, request.serialize())

        return lib.Deserializable(
            responses,
            lambda res: [
                (num, lib.to_dict(track)) for num, track in res if any(track.strip())
            ],
        )

Step 11: Implement Provider Logic

This is where the core business logic resides. CRITICAL: Always use the generated schema types from Step 5/6.

⚠️ IMPORTANT: The mapper.py file is generated by the CLI and should NEVER be modified manually. It contains only delegation methods that call your provider functions. All actual implementation logic goes in the karrio/providers/[carrier_name]/ directory.

🚨 MANDATORY: Generated Schema Usage Patterns

CRITICAL RULE: All provider functions MUST use the generated schema types for both request creation and response parsing. This ensures type safety, reduces errors, and maintains consistency across all integrations.

❌ WRONG: Manual Dictionary Manipulation

# NEVER DO THIS - No type safety, error-prone
def rate_request(payload, settings):
    return lib.Serializable({
        "shipper": {
            "addressLine1": payload.shipper.address_line1,
            "city": payload.shipper.city,
            # ... manual field mapping
        }
    }, lib.to_dict)

def _extract_details(data, settings):
    # NEVER DO THIS - Fragile dictionary access
    service = data.get("serviceCode", "")
    total = data.get("totalCharge", 0)
    # ... manual field extraction

✅ CORRECT: Generated Schema Type Usage

# ALWAYS DO THIS - Type-safe, robust, maintainable
import karrio.schemas.[carrier_name].rate_request as carrier_req
import karrio.schemas.[carrier_name].rate_response as carrier_res

def rate_request(payload, settings):
    # Create request using generated schema type
    request = carrier_req.RateRequestType(
        shipper=carrier_req.AddressType(
            addressLine1=shipper.address_line1,
            city=shipper.city,
            # ... typed field mapping
        ),
        recipient=carrier_req.AddressType(
            addressLine1=recipient.address_line1,
            city=recipient.city,
            # ... typed field mapping
        ),
        packages=[
            carrier_req.PackageType(
                weight=package.weight.value,
                weightUnit=provider_units.WeightUnit[package.weight.unit].value,
                # ... typed field mapping
            )
            for package in packages
        ]
    )
    return lib.Serializable(request, lib.to_dict)

def _extract_details(data, settings):
    # Convert to typed object using generated schema
    rate = lib.to_object(carrier_res.RateType, data)

    # Access fields through typed attributes
    return models.RateDetails(
        carrier_id=settings.carrier_id,
        carrier_name=settings.carrier_name,
        service=rate.serviceCode if hasattr(rate, 'serviceCode') else "",
        total_charge=lib.to_money(rate.totalCharge),
        currency=rate.currency or "USD",
        transit_days=rate.transitDays if hasattr(rate, 'transitDays') else None,
        # ... typed field access
    )

Schema Usage Rules

  1. Import Generated Types: Always import request and response schema modules
  2. Create Typed Requests: Use schema classes like RateRequestType() for all requests
  3. Parse Typed Responses: Use lib.to_object(SchemaType, data) for all response parsing
  4. Access Typed Attributes: Use object attributes instead of dictionary keys
  5. Handle Missing Fields: Use hasattr() checks for optional fields
  6. Use Settings Properties: Always use settings.carrier_id and settings.carrier_name

Common Schema Patterns

For JSON APIs (default generation):

# Import with Type suffix
import karrio.schemas.carrier.rate_request as carrier_req
import karrio.schemas.carrier.rate_response as carrier_res

# Use Type suffix classes
request = carrier_req.RateRequestType(...)
rate = lib.to_object(carrier_res.RateResponseType, data)

For XML APIs (with --no-append-type-suffix):

# Import without Type suffix
import karrio.schemas.carrier.rate_request as carrier_req
import karrio.schemas.carrier.rate_response as carrier_res

# Use classes without Type suffix
request = carrier_req.RateRequest(...)
rate = lib.to_object(carrier_res.RateResponse, data)

For each feature, implement two functions:

Rating Implementation

File: karrio/providers/[carrier_name]/rate.py

"""Karrio [Carrier Name] rate API implementation."""

import typing
import karrio.lib as lib
import karrio.core.units as units
import karrio.core.models as models
import karrio.providers.[carrier_name].error as error
import karrio.providers.[carrier_name].utils as provider_utils
import karrio.providers.[carrier_name].units as provider_units
# CRITICAL: Always import and use the generated schema types
import karrio.schemas.[carrier_name].rate_request as [carrier_name]_req
import karrio.schemas.[carrier_name].rate_response as [carrier_name]_res

def parse_rate_response(
    _response: lib.Deserializable[str],
    settings: provider_utils.Settings,
) -> typing.Tuple[typing.List[models.RateDetails], typing.List[models.Message]]:
    response = _response.deserialize()
    messages = error.parse_error_response(response, settings)

    # For JSON APIs
    rate_objects = response.get("rates", []) if hasattr(response, 'get') else []
    rates = [_extract_details(rate, settings) for rate in rate_objects]

    # For XML APIs
    # rate_elements = response.xpath(".//rate") if hasattr(response, 'xpath') else []
    # rates = [_extract_details(rate, settings) for rate in rate_elements]

    return rates, messages

def _extract_details(
    data: dict,  # Use lib.Element for XML APIs
    settings: provider_utils.Settings,
) -> models.RateDetails:
    """Extract rate details from carrier response data"""

    # Convert the carrier data to a proper object for easy attribute access
    rate = lib.to_object([carrier_name]_res.RateResponseType, data)  # Remove 'Type' for XML

    return models.RateDetails(
        carrier_id=settings.carrier_id,
        carrier_name=settings.carrier_name,
        service=rate.serviceCode if hasattr(rate, 'serviceCode') else "",
        total_charge=lib.to_money(rate.totalCharge),
        currency=rate.currency if hasattr(rate, 'currency') else "USD",
        transit_days=int(rate.transitDays) if hasattr(rate, 'transitDays') and rate.transitDays else None,
        meta=dict(
            service_name=rate.serviceName if hasattr(rate, 'serviceName') else "",
        ),
    )

def rate_request(
    payload: models.RateRequest,
    settings: provider_utils.Settings,
) -> lib.Serializable:
    """Create a rate request for the carrier API"""

    # Convert karrio models to carrier-specific format
    shipper = lib.to_address(payload.shipper)
    recipient = lib.to_address(payload.recipient)
    packages = lib.to_packages(payload.parcels)
    services = lib.to_services(payload.services, provider_units.ShippingService)
    options = lib.to_shipping_options(
        payload.options,
        package_options=packages.options,
        initializer=provider_units.shipping_options_initializer,
    )

    # Create the carrier-specific request object
    request = [carrier_name]_req.RateRequestType(  # Remove 'Type' for XML
        # Map shipper details
        shipper={
            "addressLine1": shipper.address_line1,
            "city": shipper.city,
            "postalCode": shipper.postal_code,
            "countryCode": shipper.country_code,
            "stateCode": shipper.state_code,
            "personName": shipper.person_name,
            "companyName": shipper.company_name,
            "phoneNumber": shipper.phone_number,
            "email": shipper.email,
        },
        # Map recipient details
        recipient={
            "addressLine1": recipient.address_line1,
            "city": recipient.city,
            "postalCode": recipient.postal_code,
            "countryCode": recipient.country_code,
            "stateCode": recipient.state_code,
            "personName": recipient.person_name,
            "companyName": recipient.company_name,
            "phoneNumber": recipient.phone_number,
            "email": recipient.email,
        },
        # Map package details
        packages=[
            {
                "weight": package.weight.value,
                "weightUnit": provider_units.WeightUnit[package.weight.unit].value,
                "length": package.length.value if package.length else None,
                "width": package.width.value if package.width else None,
                "height": package.height.value if package.height else None,
                "dimensionUnit": provider_units.DimensionUnit[package.dimension_unit].value if package.dimension_unit else None,
                "packagingType": provider_units.PackagingType[package.packaging_type or 'your_packaging'].value,
            }
            for package in packages
        ],
        # Add service codes if specified
        services=[s.value_or_key for s in services] if services else None,
        # Add account information
        customerNumber=settings.account_number,
    )

    return lib.Serializable(request, lib.to_dict)  # Use lib.to_xml for XML APIs

Tracking Implementation

File: karrio/providers/[carrier_name]/tracking.py

"""Karrio [Carrier Name] tracking API implementation."""

import typing
import karrio.lib as lib
import karrio.core.models as models
import karrio.providers.[carrier_name].error as error
import karrio.providers.[carrier_name].utils as provider_utils
import karrio.providers.[carrier_name].units as provider_units
import karrio.schemas.[carrier_name].tracking_response as [carrier_name]_res


def _match_status(code: str) -> typing.Optional[str]:
    """Match code against TrackingStatus enum values."""
    if not code:
        return None
    for status in list(provider_units.TrackingStatus):
        if code in status.value:
            return status.name
    return None


def _match_reason(code: str) -> typing.Optional[str]:
    """Match code against TrackingIncidentReason enum values."""
    if not code:
        return None
    for reason in list(provider_units.TrackingIncidentReason):
        if code in reason.value:
            return reason.name
    return None


def parse_tracking_response(
    _response: lib.Deserializable,
    settings: provider_utils.Settings,
) -> typing.Tuple[typing.List[models.TrackingDetails], typing.List[models.Message]]:
    response = _response.deserialize()

    tracking_details = []
    for tracking_number, tracking_data in response:
        if not tracking_data:
            continue

        tracking = lib.to_object([carrier_name]_res.TrackingResponseType, tracking_data)

        events = [
            models.TrackingEvent(
                date=lib.fdate(event.date, "%Y-%m-%d"),
                description=event.description,
                location=event.location,
                code=event.status_code,
                time=lib.flocaltime(event.time, "%H:%M:%S"),
                # REQUIRED: timestamp in ISO 8601 format
                timestamp=lib.fiso_timestamp(
                    lib.fdate(event.date, "%Y-%m-%d"),
                    lib.ftime(event.time, "%H:%M:%S"),
                ),
                # REQUIRED: normalized status at event level
                status=_match_status(event.status_code),
                # Incident reason for exception events
                reason=_match_reason(event.status_code),
            )
            for event in (tracking.events or [])
        ]

        # Determine overall status from latest event
        latest_event = events[0] if events else None
        status = latest_event.status or provider_units.TrackingStatus.in_transit.name

        detail = models.TrackingDetails(
            carrier_id=settings.carrier_id,
            carrier_name=settings.carrier_name,
            tracking_number=tracking_number,
            events=events,
            status=status,
            delivered=status == "delivered",
        )
        tracking_details.append(detail)

    return tracking_details, []

def tracking_request(
    payload: models.TrackingRequest,
    settings: provider_utils.Settings,
) -> lib.Serializable:
    """Create a tracking request object."""
    return lib.Serializable(payload.tracking_numbers)

IMPORTANT TrackingEvent Fields:

Field Type Required Description
date str Yes Event date (e.g., "2024-01-15")
time str Yes Event time (e.g., "14:30:00")
description str Yes Human-readable event description
code str Yes Carrier-specific status code
location str No Event location
timestamp str Yes ISO 8601 timestamp (e.g., "2024-01-15T14:30:00")
status str Yes Normalized status from TrackingStatus enum
reason str No Incident reason from TrackingIncidentReason enum

Usage of lib.fiso_timestamp:

# Combines date and time into ISO 8601 timestamp
timestamp = lib.fiso_timestamp(
    lib.fdate(event.date, "%Y-%m-%d"),
    lib.ftime(event.time, "%H:%M:%S"),
)
# Result: "2024-01-15T14:30:00"

Multi-Piece/Multi-Package Shipment Support

CRITICAL: Before implementing shipment creation, you MUST determine how the carrier API handles multi-package shipments. This affects the entire request/response structure.

Step 1: Analyze Carrier API Documentation

Check the carrier API documentation to determine which pattern applies:

Look For Pattern Implementation
Single endpoint accepts packages[] array Bundled All packages in one request
Response has PackageResults or pieceResponses Bundled Parse individual package results
Response has masterTrackingNumber Bundled Use master as primary tracking
Must call endpoint once per package Per-Package Create list of requests
Each package gets separate label Per-Package Use lib.to_multi_piece_shipment()
Step 2: Implement Correct Pattern

Pattern A: Bundled Request (FedEx, UPS, DHL Express style)

def shipment_request(payload, settings):
    packages = lib.to_packages(payload.parcels)

    # All packages in single request
    request = carrier_req.ShipmentRequestType(
        packages=[
            carrier_req.PackageType(
                weight=pkg.weight.KG,
                dimensions=carrier_req.DimensionsType(
                    length=pkg.length.CM,
                    width=pkg.width.CM,
                    height=pkg.height.CM,
                ),
            )
            for pkg in packages
        ],
        # ... other fields
    )
    return lib.Serializable(request, lib.to_dict)

def parse_shipment_response(_response, settings):
    response = _response.deserialize()

    # Extract master tracking
    tracking_number = response.masterTrackingNumber

    # Extract all package results
    packages = lib.failsafe(lambda: response.PackageResults) or []
    tracking_ids = [pkg.TrackingID for pkg in packages if pkg.TrackingID]

    # Bundle all labels
    labels = [pkg.Label for pkg in packages if pkg.Label]
    label = lib.bundle_base64(labels, "PDF") if len(labels) > 1 else next(iter(labels), None)

    return models.ShipmentDetails(
        tracking_number=tracking_number,
        docs=models.Documents(label=label),
        meta=dict(tracking_numbers=tracking_ids),
    ), messages

Pattern B: Per-Package Request (Canada Post, USPS style)

def shipment_request(payload, settings):
    packages = lib.to_packages(payload.parcels)

    # Create list of requests, one per package
    request = [
        carrier_req.ShipmentType(
            parcel=carrier_req.ParcelType(
                weight=pkg.weight.KG,
                dimensions=carrier_req.DimensionsType(...),
            ),
            # ... common fields for each package
        )
        for pkg in packages
    ]
    return lib.Serializable(request, _serialize_requests)

def parse_shipment_response(_response, settings):
    responses = _response.deserialize()  # List of responses
    messages = error.parse_error_response(responses, settings)

    # Extract details from each package response
    shipment_details = [
        (f"{idx}", _extract_shipment(response, settings))
        for idx, response in enumerate(responses, start=1)
        if _is_valid_response(response)
    ]

    # Use lib.to_multi_piece_shipment() to aggregate
    shipment = lib.to_multi_piece_shipment(shipment_details)
    return shipment, messages
Step 3: Ensure Proper Label Bundling

For multi-package shipments, always bundle labels:

# For bundled pattern - bundle from package results
labels = [pkg.Label for pkg in packages if pkg.Label]
label = lib.bundle_base64(labels, label_type) if len(labels) > 1 else next(iter(labels), None)

# For per-package pattern - lib.to_multi_piece_shipment() handles bundling automatically
Step 4: Populate Meta Fields

Always include tracking numbers for all packages in meta:

meta=dict(
    tracking_numbers=tracking_ids,           # List of all package tracking numbers
    shipment_identifiers=shipment_ids,       # List of all shipment IDs (if applicable)
    carrier_tracking_link=tracking_url,      # Link for master/primary tracking
)

Shipment Implementation

File: karrio/providers/[carrier_name]/shipment/create.py

"""Karrio [Carrier Name] shipment creation implementation."""

import typing
import karrio.lib as lib
import karrio.core.models as models
import karrio.providers.[carrier_name].error as error
import karrio.providers.[carrier_name].utils as provider_utils
import karrio.providers.[carrier_name].units as provider_units
import karrio.schemas.[carrier_name].shipment_request as [carrier_name]_req
import karrio.schemas.[carrier_name].shipment_response as [carrier_name]_res

def parse_shipment_response(
    _response: lib.Deserializable,
    settings: provider_utils.Settings,
) -> typing.Tuple[models.ShipmentDetails, typing.List[models.Message]]:
    response = _response.deserialize()
    messages = error.parse_error_response(response, settings)

    shipment = lib.to_object([carrier_name]_res.ShipmentResponseType, response)

    return models.ShipmentDetails(
        carrier_id=settings.carrier_id,
        carrier_name=settings.carrier_name,
        tracking_number=shipment.trackingNumber,
        shipment_identifier=shipment.shipmentId,
        label=shipment.labelData,  # Base64 encoded label
        invoice=shipment.invoiceData if hasattr(shipment, 'invoiceData') else None,
        meta=dict(
            service_name=shipment.serviceName if hasattr(shipment, 'serviceName') else "",
            label_type=shipment.labelType if hasattr(shipment, 'labelType') else "PDF",
        ),
    ), messages

def shipment_request(
    payload: models.ShipmentRequest,
    settings: provider_utils.Settings,
) -> lib.Serializable:
    """Create a shipment request for the carrier API"""

    # Convert karrio models to carrier-specific format
    shipper = lib.to_address(payload.shipper)
    recipient = lib.to_address(payload.recipient)
    packages = lib.to_packages(payload.parcels)
    service = lib.to_services(payload.service, provider_units.ShippingService).first
    options = lib.to_shipping_options(
        payload.options,
        package_options=packages.options,
        initializer=provider_units.shipping_options_initializer,
    )

    request = [carrier_name]_req.ShipmentRequestType(
        # Map shipper details
        shipper={
            "addressLine1": shipper.address_line1,
            "city": shipper.city,
            "postalCode": shipper.postal_code,
            "countryCode": shipper.country_code,
            "stateCode": shipper.state_code,
            "personName": shipper.person_name,
            "companyName": shipper.company_name,
            "phoneNumber": shipper.phone_number,
            "email": shipper.email,
        },
        # Map recipient details
        recipient={
            "addressLine1": recipient.address_line1,
            "city": recipient.city,
            "postalCode": recipient.postal_code,
            "countryCode": recipient.country_code,
            "stateCode": recipient.state_code,
            "personName": recipient.person_name,
            "companyName": recipient.company_name,
            "phoneNumber": recipient.phone_number,
            "email": recipient.email,
        },
        # Map package details
        packages=[
            {
                "weight": package.weight.value,
                "weightUnit": provider_units.WeightUnit[package.weight.unit].value,
                "length": package.length.value if package.length else None,
                "width": package.width.value if package.width else None,
                "height": package.height.value if package.height else None,
                "dimensionUnit": provider_units.DimensionUnit[package.dimension_unit].value if package.dimension_unit else None,
                "packagingType": provider_units.PackagingType[package.packaging_type or 'your_packaging'].value,
            }
            for package in packages
        ],
        # Add service code
        serviceCode=service.value_or_key,
        # Add account information
        customerNumber=settings.account_number,
        # Add label details
        labelFormat=payload.label_type or "PDF",
    )

    return lib.Serializable(request, lib.to_dict)  # Use lib.to_xml for XML APIs

Manifest Implementation

File: karrio/providers/[carrier_name]/manifest.py

"""Karrio [Carrier Name] manifest creation implementation."""

import typing
import karrio.lib as lib
import karrio.core.models as models
import karrio.providers.[carrier_name].error as error
import karrio.providers.[carrier_name].utils as provider_utils
import karrio.schemas.[carrier_name].manifest_request as [carrier_name]_req
import karrio.schemas.[carrier_name].manifest_response as [carrier_name]_res

def parse_manifest_response(
    _response: lib.Deserializable[dict],
    settings: provider_utils.Settings,
) -> typing.Tuple[models.ManifestDetails, typing.List[models.Message]]:
    response = _response.deserialize()
    messages = error.parse_error_response(response, settings)
    manifest = _extract_details(response, settings) if not messages else None

    return manifest, messages

def _extract_details(
    data: dict,
    settings: provider_utils.Settings,
) -> models.ManifestDetails:
    manifest = lib.to_object([carrier_name]_res.ManifestResponseType, data)

    return models.ManifestDetails(
        carrier_id=settings.carrier_id,
        carrier_name=settings.carrier_name,
        manifest_id=manifest.manifestId if hasattr(manifest, 'manifestId') else "",
        doc=models.ManifestDocument(manifest=manifest.manifestData) if hasattr(manifest, 'manifestData') else None,
        meta=dict(
            status=manifest.status if hasattr(manifest, 'status') else "",
        ),
    ) if manifest else None

def manifest_request(
    payload: models.ManifestRequest,
    settings: provider_utils.Settings,
) -> lib.Serializable:
    request = [carrier_name]_req.ManifestRequestType(
        accountNumber=settings.account_number,
        shipments=[
            {"trackingNumber": identifier}
            for identifier in payload.shipment_identifiers
        ],
    )

    return lib.Serializable(request, lib.to_dict)

Duties & Taxes Implementation

File: karrio/providers/[carrier_name]/duties.py

"""Karrio [Carrier Name] duties and taxes calculation implementation."""

import typing
import karrio.lib as lib
import karrio.core.models as models
import karrio.providers.[carrier_name].error as error
import karrio.providers.[carrier_name].utils as provider_utils
import karrio.schemas.[carrier_name].duties_taxes_request as [carrier_name]_req
import karrio.schemas.[carrier_name].duties_taxes_response as [carrier_name]_res

def parse_duties_response(
    _response: lib.Deserializable[dict],
    settings: provider_utils.Settings,
) -> typing.Tuple[models.DutiesDetails, typing.List[models.Message]]:
    response = _response.deserialize()
    messages = error.parse_error_response(response, settings)
    duties = _extract_details(response, settings) if not messages else None

    return duties, messages

def _extract_details(
    data: dict,
    settings: provider_utils.Settings,
) -> models.DutiesDetails:
    duties = lib.to_object([carrier_name]_res.DutiesTaxesResponseType, data)

    return models.DutiesDetails(
        carrier_id=settings.carrier_id,
        carrier_name=settings.carrier_name,
        duties=lib.to_money(duties.dutiesAmount) if hasattr(duties, 'dutiesAmount') else 0,
        taxes=lib.to_money(duties.taxesAmount) if hasattr(duties, 'taxesAmount') else 0,
        total=lib.to_money(duties.totalAmount) if hasattr(duties, 'totalAmount') else 0,
        currency=duties.currency if hasattr(duties, 'currency') else "USD",
        meta=dict(
            breakdown=duties.breakdown if hasattr(duties, 'breakdown') else {},
        ),
    ) if duties else None

def duties_request(
    payload: models.DutiesRequest,
    settings: provider_utils.Settings,
) -> lib.Serializable:
    shipper = lib.to_address(payload.shipper)
    recipient = lib.to_address(payload.recipient)

    request = [carrier_name]_req.DutiesTaxesRequestType(
        originCountry=shipper.country_code,
        destinationCountry=recipient.country_code,
        items=[
            {
                "description": item.description,
                "quantity": item.quantity,
                "value": item.value,
                "hsCode": item.hs_code,
            }
            for item in (payload.commodities or [])
        ],
    )

    return lib.Serializable(request, lib.to_dict)

Insurance Implementation

File: karrio/providers/[carrier_name]/insurance.py

"""Karrio [Carrier Name] insurance application implementation."""

import typing
import karrio.lib as lib
import karrio.core.models as models
import karrio.providers.[carrier_name].error as error
import karrio.providers.[carrier_name].utils as provider_utils
import karrio.schemas.[carrier_name].insurance_request as [carrier_name]_req
import karrio.schemas.[carrier_name].insurance_response as [carrier_name]_res

def parse_insurance_response(
    _response: lib.Deserializable[dict],
    settings: provider_utils.Settings,
) -> typing.Tuple[models.InsuranceDetails, typing.List[models.Message]]:
    response = _response.deserialize()
    messages = error.parse_error_response(response, settings)
    insurance = _extract_details(response, settings) if not messages else None

    return insurance, messages

def _extract_details(
    data: dict,
    settings: provider_utils.Settings,
) -> models.InsuranceDetails:
    insurance = lib.to_object([carrier_name]_res.InsuranceResponseType, data)

    return models.InsuranceDetails(
        carrier_id=settings.carrier_id,
        carrier_name=settings.carrier_name,
        insurance_id=insurance.policyId if hasattr(insurance, 'policyId') else "",
        premium=lib.to_money(insurance.premium) if hasattr(insurance, 'premium') else 0,
        coverage=lib.to_money(insurance.coverage) if hasattr(insurance, 'coverage') else 0,
        currency=insurance.currency if hasattr(insurance, 'currency') else "USD",
        meta=dict(
            policy_number=insurance.policyNumber if hasattr(insurance, 'policyNumber') else "",
        ),
    ) if insurance else None

def insurance_request(
    payload: models.InsuranceRequest,
    settings: provider_utils.Settings,
) -> lib.Serializable:
    request = [carrier_name]_req.InsuranceRequestType(
        shipmentId=payload.shipment_identifier,
        coverage=payload.coverage_amount,
        currency=payload.currency or "USD",
    )

    return lib.Serializable(request, lib.to_dict)

Webhook Registration Implementation

File: karrio/providers/[carrier_name]/webhook/register.py

"""Karrio [Carrier Name] webhook registration implementation."""

import typing
import karrio.lib as lib
import karrio.core.models as models
import karrio.providers.[carrier_name].error as error
import karrio.providers.[carrier_name].utils as provider_utils
import karrio.schemas.[carrier_name].webhook_request as [carrier_name]_req
import karrio.schemas.[carrier_name].webhook_response as [carrier_name]_res

def parse_webhook_registration_response(
    _response: lib.Deserializable[dict],
    settings: provider_utils.Settings,
) -> typing.Tuple[models.WebhookDetails, typing.List[models.Message]]:
    response = _response.deserialize()
    messages = error.parse_error_response(response, settings)
    webhook = _extract_details(response, settings) if not messages else None

    return webhook, messages

def _extract_details(
    data: dict,
    settings: provider_utils.Settings,
) -> models.WebhookDetails:
    webhook = lib.to_object([carrier_name]_res.WebhookResponseType, data)

    return models.WebhookDetails(
        carrier_id=settings.carrier_id,
        carrier_name=settings.carrier_name,
        webhook_id=webhook.webhookId if hasattr(webhook, 'webhookId') else "",
        url=webhook.url if hasattr(webhook, 'url') else "",
        events=webhook.events if hasattr(webhook, 'events') else [],
        meta=dict(
            status=webhook.status if hasattr(webhook, 'status') else "",
        ),
    ) if webhook else None

def webhook_registration_request(
    payload: models.WebhookRegistrationRequest,
    settings: provider_utils.Settings,
) -> lib.Serializable:
    request = [carrier_name]_req.WebhookRequestType(
        url=payload.url,
        events=payload.events or ["shipment.created", "shipment.delivered"],
    )

    return lib.Serializable(request, lib.to_dict)

Callback/OAuth Implementation (for carriers with OAuth flows)

File: karrio/providers/[carrier_name]/callback/oauth.py

"""Karrio [Carrier Name] OAuth callback implementation."""

import typing
import karrio.lib as lib
import karrio.core.models as models
import karrio.providers.[carrier_name].error as error
import karrio.providers.[carrier_name].utils as provider_utils

def parse_oauth_callback_response(
    _response: lib.Deserializable[dict],
    settings: provider_utils.Settings,
) -> typing.Tuple[models.OAuthDetails, typing.List[models.Message]]:
    response = _response.deserialize()
    messages = error.parse_error_response(response, settings)

    oauth = models.OAuthDetails(
        carrier_id=settings.carrier_id,
        carrier_name=settings.carrier_name,
        access_token=response.get("access_token"),
        refresh_token=response.get("refresh_token"),
        expires_in=response.get("expires_in"),
        token_type=response.get("token_type", "Bearer"),
    ) if not messages else None

    return oauth, messages

def oauth_callback_request(
    payload: models.OAuthCallbackRequest,
    settings: provider_utils.Settings,
) -> lib.Serializable:
    # Handle OAuth code exchange
    request = dict(
        grant_type="authorization_code",
        code=payload.code,
        redirect_uri=payload.redirect_uri,
        client_id=settings.client_id,
        client_secret=settings.client_secret,
    )

    return lib.Serializable(request, lib.to_dict)

Step 12: Configure Plugin Metadata

For Direct Carriers, edit:

File: karrio/plugins/[carrier_name]/__init__.py

import karrio.core.metadata as metadata
import karrio.mappers.[carrier_name] as mappers
import karrio.providers.[carrier_name].units as units
import karrio.providers.[carrier_name].utils as utils

METADATA = metadata.PluginMetadata(
    status="production-ready",       # or "development", "beta"
    id="[carrier_name]",
    label="[Carrier Display Name]",
    # Integrations
    Mapper=mappers.Mapper,
    Proxy=mappers.Proxy,
    Settings=mappers.Settings,
    # Data Units
    is_hub=False,                    # Direct carrier
    options=units.ShippingOption,
    services=units.ShippingService,
    connection_configs=utils.ConnectionConfig,
    # Extra info
    website="https://www.carrier.com",
    description="Carrier description",
)

For Hub Carriers, set is_hub=True:

METADATA = metadata.PluginMetadata(
    # ... same as above but:
    is_hub=True,                     # Hub carrier
    # ... rest of configuration
)

Phase 4: Testing and Validation

Step 13: Run Schema Generation Tests

Before writing tests, ensure schema generation works:

# Regenerate schemas to verify everything works
./bin/run-generate-on modules/connectors/[carrier_name]
# OR
./bin/run-generate-on community/plugins/[carrier_name]

Step 14: Install and Test the Integration

Install the new integration:

# For direct carriers
source ./bin/activate-env && pip install -e modules/connectors/[carrier_name]

# For hub carriers
source ./bin/activate-env && pip install -e community/plugins/[carrier_name]

Run the tests:

# For direct carriers
source ./bin/activate-env && python -m unittest discover -v -f modules/connectors/[carrier_name]/tests

# For hub carriers
source ./bin/activate-env && python -m unittest discover -v -f community/plugins/[carrier_name]/tests

Step 15: Test Implementation - CRITICAL PATTERNS

The CLI automatically generates comprehensive test files following MANDATORY patterns. AI agents must strictly follow these patterns without deviation.

🚨 ABSOLUTE TESTING RULES

⚠️ CRITICAL: These patterns are MANDATORY and must be replicated exactly. Only the data content changes between carriers, never the structure, naming, or assertion patterns.

1. Test Class Structure (NEVER CHANGE)

"""[Carrier Name] carrier [feature] tests."""

import unittest
from unittest.mock import patch, ANY
from .fixture import gateway
import logging
import karrio.sdk as karrio
import karrio.lib as lib
import karrio.core.models as models

logger = logging.getLogger(__name__)

class Test[CompactCarrierName][Feature](unittest.TestCase):
    def setUp(self):
        self.maxDiff = None
        self.[Feature]Request = models.[Feature]Request(**[Feature]Payload)

Class Naming Convention:

  • Rating: TestDHLExpressRating, TestFedExRating, TestShipEngineRating
  • Shipment: TestDHLExpressShipment, TestFedExShipment, TestShipEngineShipment
  • Tracking: TestDHLExpressTracking, TestFedExTracking, TestShipEngineTracking

2. Test Method Structure (EXACT PATTERN REQUIRED)

For EVERY feature, implement exactly these 4 test methods:

A. Request Creation Test

def test_create_[feature]_request(self):
    request = gateway.mapper.create_[feature]_request(self.[Feature]Request)
    print(f"Generated request: {lib.to_dict(request.serialize())}")  # MANDATORY DEBUG PRINT
    self.assertEqual(lib.to_dict(request.serialize()), [Feature]Request)

B. HTTP Endpoint Test

def test_[action_verb](self):
    with patch("karrio.mappers.[carrier_id].proxy.lib.request") as mock:
        mock.return_value = "{}"  # Use "<r></r>" for XML APIs
        karrio.[Feature].[action](self.[Feature]Request).from_(gateway)
        print(f"Called URL: {mock.call_args[1]['url']}")  # MANDATORY DEBUG PRINT
        self.assertEqual(
            mock.call_args[1]["url"],
            f"{gateway.settings.server_url}/[endpoint]"
        )

C. Response Parsing Test

def test_parse_[feature]_response(self):
    with patch("karrio.mappers.[carrier_id].proxy.lib.request") as mock:
        mock.return_value = [Feature]Response
        parsed_response = (
            karrio.[Feature].[action](self.[Feature]Request)
            .from_(gateway)
            .parse()
        )
        print(f"Parsed response: {lib.to_dict(parsed_response)}")  # MANDATORY DEBUG PRINT
        self.assertListEqual(lib.to_dict(parsed_response), Parsed[Feature]Response)

D. Error Handling Test

def test_parse_error_response(self):
    with patch("karrio.mappers.[carrier_id].proxy.lib.request") as mock:
        mock.return_value = ErrorResponse
        parsed_response = (
            karrio.[Feature].[action](self.[Feature]Request)
            .from_(gateway)
            .parse()
        )
        print(f"Error response: {lib.to_dict(parsed_response)}")  # MANDATORY DEBUG PRINT
        self.assertListEqual(lib.to_dict(parsed_response), ParsedErrorResponse)

3. Exact Method Names by Feature (NEVER DEVIATE)

Feature Test Methods (EXACT NAMES)
Rating test_create_rate_request, test_get_rates, test_parse_rate_response, test_parse_error_response
Shipment test_create_shipment_request, test_create_shipment, test_parse_shipment_response, test_create_shipment_cancel_request, test_cancel_shipment, test_parse_shipment_cancel_response, test_parse_error_response
Tracking test_create_tracking_request, test_get_tracking, test_parse_tracking_response, test_parse_error_response
Pickup test_create_pickup_request, test_schedule_pickup, test_update_pickup, test_cancel_pickup, test_parse_pickup_response, test_parse_error_response
Address test_create_address_validation_request, test_validate_address, test_parse_address_validation_response, test_parse_error_response

4. Test Data Structure (MANDATORY PATTERN)

Every test file must end with these exact data structures:

if __name__ == "__main__":
    unittest.main()

# 1. KARRIO INPUT PAYLOAD (standardized format)
[Feature]Payload = {
    # Standard Karrio model format - ADAPT CONTENT ONLY
}

# 2. CARRIER REQUEST FORMAT (generated schema format)
[Feature]Request = {
    # Carrier-specific request format - ADAPT TO CARRIER API ONLY
}

# 3. CARRIER RESPONSE MOCK (carrier API response)
[Feature]Response = """
# Mock carrier API response - ADAPT TO CARRIER API ONLY
"""

# 4. ERROR RESPONSE MOCK
ErrorResponse = """
# Mock carrier error response - ADAPT TO CARRIER API ONLY
"""

# 5. PARSED SUCCESS RESPONSE (Karrio format)
Parsed[Feature]Response = [
    [
        {
            "carrier_id": "[carrier_id]",
            "carrier_name": "[carrier_id]",
            # Success response data - ADAPT CONTENT ONLY
        }
    ],
    []  # Empty errors array
]

# 6. PARSED ERROR RESPONSE (Karrio format)
ParsedErrorResponse = [
    [],  # Empty success data
    [
        {
            "carrier_id": "[carrier_id]",
            "carrier_name": "[carrier_id]",
            "code": "error_code",
            "message": "Error message",
            "details": {
                "details": "Additional error details"
            }
        }
    ]
]

5. Endpoint URL Patterns (MUST MATCH CARRIER API)

Test the exact carrier API endpoints:

Feature Common Endpoint Patterns
Rating /rates, /api/rates, /v1/rates
Shipment /shipments, /api/shipments, /v1/shipments
Shipment Cancel /shipments/{id}/cancel, /shipments/cancel
Tracking /tracking, /track, /v1/tracking
Pickup /pickups, /pickup/schedule, /v1/pickups
Address /address/validate, /validate-address

6. Response Format Patterns (NEVER CHANGE)

Response Type Format Pattern
Success [data_array_or_object, []]
Error [[], [error_objects]] or [None, [error_objects]]
Mixed [partial_data, [warning_objects]]

7. Mock Response Patterns

JSON APIs:

# Success response
mock.return_value = """{
  "rates": [
    {
      "serviceCode": "express",
      "totalCharge": 25.99
    }
  ]
}"""

# Error response
mock.return_value = """{
  "error": {
    "code": "rate_error",
    "message": "Unable to get rates"
  }
}"""

XML APIs:

# Success response
mock.return_value = """<?xml version="1.0"?>
<rate-response>
    <rate>
        <service-code>express</service-code>
        <total-charge>25.99</total-charge>
    </rate>
</rate-response>"""

# Error response
mock.return_value = """<?xml version="1.0"?>
<error-response>
    <error>
        <code>rate_error</code>
        <message>Unable to get rates</message>
    </error>
</error-response>"""

🎯 ADAPTATION GUIDELINES FOR AI AGENTS

What CHANGES Between Carriers:

  1. Carrier-specific data formats in request/response objects
  2. API endpoint URLs (/rates vs /api/v1/rate-quotes)
  3. Field names (serviceCode vs service_type)
  4. Mock response content (actual carrier API format)

What NEVER CHANGES:

  1. File names: test_rate.py, test_shipment.py, test_tracking.py
  2. Class names: Test[CompactName][Feature]
  3. Test method names: Exactly as specified in the table above
  4. Import statements: Always identical pattern
  5. Assertion methods: Always assertEqual, assertListEqual
  6. Mock patterns: Always patch karrio.mappers.[carrier_id].proxy.lib.request
  7. Response format: Always [data, errors] tuple structure
  8. Debug print statements: Always include before assertions

Testing Rules Summary:

  • Framework: Python's built-in unittest (never pytest)
  • Print Statements: Always add before assertions for debugging
  • Assertions: Use assertListEqual with full dict data
  • Variable Data: Use ANY for IDs, dates, and dynamic values
  • Structure: Never modify the test structure or naming conventions
  • Data Only: Only adapt the payload/response content to match carrier API

⚠️ VIOLATION WARNING: Deviating from these patterns will result in integration failure. The patterns are designed for consistency across 50+ carrier integrations.

Step 16: Integration Success Criteria

Your integration is complete when ALL of the following criteria are met:

1. Carrier Tests Pass

# All carrier-specific tests must pass
source ./bin/activate-env

# For direct carriers
python -m unittest discover -v -f modules/connectors/[carrier_name]/tests

# For hub carriers
python -m unittest discover -v -f community/plugins/[carrier_name]/tests

2. SDK Tests Pass

# All SDK tests must continue to pass
source ./bin/activate-env && ./bin/run-sdk-tests

This ensures your integration doesn't break existing functionality.

3. Plugin Registration

# Plugin appears in list
./bin/cli plugins list | grep [carrier_name]

# Plugin details are accessible
./bin/cli plugins show [carrier_name]

4. Schema Generation Works

# Schema regeneration should work without errors
./bin/run-generate-on modules/connectors/[carrier_name]
# OR
./bin/run-generate-on community/plugins/[carrier_name]

5. Installation Works

# Fresh installation should work
pip uninstall karrio-[carrier_name] -y
pip install -e modules/connectors/[carrier_name]
# OR
pip install -e community/plugins/[carrier_name]

6. Import Test

# All imports should work without errors
python -c "
import karrio.schemas.[carrier_name] as schemas
import karrio.providers.[carrier_name] as provider
import karrio.mappers.[carrier_name] as mapper
print('All imports successful')
"

⚠️ CRITICAL: Do not consider the integration complete until ALL criteria pass. Each failure indicates a configuration or implementation issue that must be resolved.

Phase 5: Advanced Configuration

Hub Carrier Specific Implementation

For hub carriers like Easyship or ShipEngine, additional considerations:

  1. Dynamic Service Discovery:
class ShippingService(lib.StrEnum):
    # Base services
    hub_standard = "Hub Standard"

    @classmethod
    def discover_from_response(cls, response: dict):
        """Dynamically discover services from API response"""
        for rate in response.get("rates", []):
            service_key = f"hub_{rate['carrier'].lower()}_{rate['service'].lower()}"
            service_name = f"{rate['carrier']} {rate['service']}"
            if not hasattr(cls, service_key):
                setattr(cls, service_key, service_name)
  1. Multi-Carrier Rate Handling:
def parse_rate_response(response, settings):
    rates = []
    for rate_data in response.get("rates", []):
        rate = models.RateDetails(
            carrier_id="hub_name",
            carrier_name="hub_name",
            service=f"{rate_data['carrier']}_{rate_data['service']}",
            total_charge=lib.to_money(rate_data['total']),
            currency=rate_data['currency'],
            meta={
                "actual_carrier": rate_data['carrier'],
                "actual_service": rate_data['service'],
                "carrier_id": rate_data['carrier_id'],
            }
        )
        rates.append(rate)
    return rates, []

XML API Specific Implementation

For XML APIs like Canada Post:

  1. XML Serialization:
return lib.Serializable(request, lib.to_xml)
  1. XML Deserialization:
return lib.Deserializable(response, lib.to_element)
  1. XML Error Parsing:
def parse_error_response(response: lib.Element, settings):
    errors = []
    error_elements = response.xpath("//error")
    for error in error_elements:
        errors.append(models.Message(
            carrier_id=settings.carrier_id,
            carrier_name=settings.carrier_name,
            code=error.find("code").text,
            message=error.find("message").text,
        ))
    return errors

CLI Commands Reference

Essential Commands

# Activate environment (ALWAYS FIRST)
source ./bin/activate-env

# Create new extension
./bin/cli sdk add-extension --path [location] --carrier-slug [name] --display-name "[Display Name]"

# Generate schemas
./bin/run-generate-on [extension_path]

# Run tests
source ./bin/activate-env && python -m unittest discover -v -f [extension_path]/tests

# Install extension
source ./bin/activate-env && pip install -e [extension_path]

Code Generation Commands

# Generate from JSON schema
kcli codegen generate schema.json output.py

# Generate from XML schema
kcli codegen x --src-dir schemas --out-dir output

# Create object tree for understanding structure
kcli codegen tree --module=karrio.schemas.ups.shipping_request --class-name=ShipmentRequestType --module-alias=ups

Common Patterns & Best Practices

1. Error Handling Pattern

def parse_error_response(response, settings) -> List[Message]:
    errors = response.get("errors", [])  # Adjust based on API structure

    return [
        models.Message(
            carrier_id=settings.carrier_id,
            carrier_name=settings.carrier_name,
            code=error.get("code", ""),
            message=error.get("message", ""),
            details=error.get("details", {}),
        )
        for error in errors
    ]

2. Address Mapping Pattern

def _build_address(address: models.Address) -> dict:
    return {
        "addressLine1": address.address_line1,
        "addressLine2": address.address_line2,
        "city": address.city,
        "stateCode": address.state_code,
        "postalCode": address.postal_code,
        "countryCode": address.country_code,
        "companyName": address.company_name,
        "personName": address.person_name,
        "phoneNumber": address.phone_number,
        "email": address.email,
    }

3. Service Mapping Pattern

def _map_service_code(service_code: str) -> str:
    service_map = {
        "CARRIER_EXPRESS": "carrier_express",
        "CARRIER_STANDARD": "carrier_standard",
        "CARRIER_GROUND": "carrier_ground",
    }
    return service_map.get(service_code, service_code.lower())

4. Weight/Dimension Conversion Pattern

def _build_weight(package: models.Package) -> dict:
    return {
        "value": package.weight.value,
        "unit": "KG" if package.weight.unit == "KG" else "LB"
    }

def _build_dimensions(package: models.Package) -> dict:
    if not package.length:
        return None
    return {
        "length": package.length.value,
        "width": package.width.value,
        "height": package.height.value,
        "unit": "CM" if package.dimension_unit == "CM" else "IN"
    }

Validation Checklist

Before considering your integration complete:

  • Environment: source ./bin/activate-env executed
  • Scaffolding: Used ./bin/cli sdk add-extension (never manual creation)
  • Schema Configuration: Correct CLI parameters chosen based on API field format
  • Schema Generation: Used ./bin/run-generate-on (never manual editing)
  • Schema Files: All required schema files in schemas/ directory
  • Generated Code: Python files generated in karrio/schemas/[carrier_name]/
  • Schema Verification: Generated types import correctly and have expected names
  • Mapper Untouched: mapper.py file left unmodified from CLI generation
  • Settings: Carrier credentials configured in settings
  • Units: Services, packaging, and options defined
  • Proxy: HTTP communication implemented
  • Provider Logic: Rate, shipping, tracking implementations using generated types
  • Metadata: Plugin metadata configured with correct is_hub flag
  • Tests: All carrier tests pass with python -m unittest discover
  • SDK Tests: All SDK tests pass with ./bin/run-sdk-tests
  • Plugin Registration: Plugin appears in ./bin/cli plugins list
  • Plugin Details: Plugin details accessible with ./bin/cli plugins show
  • Installation: Integration installs with pip install -e

Troubleshooting

Schema Generation Fails

  1. Check schema file format (valid JSON/XML)
  2. Verify generate script is executable (chmod +x generate)
  3. Ensure ./bin/activate-env was run
  4. Check for syntax errors in schema files
  5. Verify correct CLI parameters for API field format:
    • Use --nice-property-names for snake_case APIs
    • Use --no-nice-property-names for camelCase APIs
    • Use --no-append-type-suffix --no-nice-property-names for PascalCase APIs
  6. Test CLI command directly: 
    ./bin/cli codegen generate schemas/rate_request.json test_output.py --no-nice-property-names

Import Errors

  1. Verify extension is installed: pip install -e [path]
  2. Check __init__.py files exist in all directories
  3. Verify imports match generated file names

Test Failures

  1. Check mock data matches expected format
  2. Verify URL endpoints in tests
  3. Ensure proper request/response transformation
  4. Add print statements before assertions for debugging

API Communication Issues

  1. Verify authentication headers
  2. Check request format (JSON vs XML)
  3. Validate API endpoints
  4. Review trace logs for debugging

Success Criteria Failures

  1. Plugin not appearing in list:

    # Reinstall the plugin
pip install -e modules/connectors/[carrier_name]
# Check for import errors
python -c "import karrio.plugins.[carrier_name]"

  2. SDK tests failing:

    # Run specific test to identify issue
python -m unittest karrio.test.providers -v
# Check for circular imports or missing dependencies

  3. Schema import errors:

    # Regenerate schemas with correct parameters
./bin/run-generate-on modules/connectors/[carrier_name]
# Verify schema file format matches API documentation

AI Agent Learnings & Time-Saving Tips

Based on implementing the ShipEngine integration, here are critical observations that would save time for AI agents building Karrio carrier integrations:

🔍 Schema Format Discovery

Problem: Determining whether to use JSON Schema definitions vs. plain JSON examples Solution: Always check existing integrations first - most use plain JSON examples (like /easyship/schemas/)

# Quick check - if you see "$schema" it's JSON Schema format
head -5 schemas/*.json | grep -l "\$schema"

# Convert to plain JSON examples using test data for better generation

🏗️ CLI Parameter Selection Strategy

Problem: Choosing correct --nice-property-names flag Quick Rule:

  • snake_case APIs (field_name) → --nice-property-names
  • camelCase APIs (fieldName) → --no-nice-property-names
  • PascalCase APIs (FieldName) → --no-append-type-suffix --no-nice-property-names

Test Early: Create a simple schema and generate to verify the output format matches expectations.

🧪 Test-Driven Development Workflow

Critical Pattern: Always implement in this exact order:

  1. Write test data first (use actual API examples from documentation)
  2. Create provider implementation using dict access (response.get('field')) not object attributes
  3. Fix test expectations to match actual output (not documentation examples)
  4. Use debugging to verify actual vs expected formats:
print('ACTUAL:', lib.to_dict(parsed_response))

🔄 Hub Carrier Specific Gotchas

Problem: Hub carriers need special handling for dynamic services Solutions:

  • Set is_hub=True in plugin metadata (critical!)
  • Add static test services to enum: shipengine_ups_ups_ground = "UPS Ground via ShipEngine"
  • Handle service parsing fallbacks: service.value_or_key if service else payload.service
  • Use composite service identifiers: f"shipengine_{carrier_code}_{service_code}"

🚫 Common Implementation Pitfalls

  1. lib.to_services() Returns None: Service not in enum → Add static services for testing
  2. Empty String Filtering: lib.to_dict() filters empty strings → Update test expectations
  3. Mapper.py Modifications: Never modify beyond method implementations
  4. Schema Object Access: Use dict access (data.get()) instead of object attributes to avoid AttributeError
  5. Address Field Consistency: Always include address_line2, address_line3 in schema examples

🎯 Test Debugging Shortcuts

Format Mismatch Issues:

# Debug actual implementation output
with patch('karrio.mappers.{carrier}.proxy.lib.request') as mock:
    mock.return_value = test_response_data
    actual = carrier_operation().parse()
    print('ACTUAL FORMAT:', json.dumps(lib.to_dict(actual), indent=2))

Service Parsing Issues:

# Test service enum recognition
service = lib.to_services('service_name', ShippingService).first
print('SERVICE FOUND:', service is not None)

⚡ Rapid Implementation Strategy

  1. Start with Rate Implementation - simplest to debug
  2. Copy Test Patterns - Use /modules/connectors/fedex/tests/ as template for proper 4-method structure
  3. Use Plain JSON Schemas - Faster generation, easier debugging
  4. Test Continuously - Run tests after each method implementation
  5. Debug with Print Statements - Add temporary debug prints to understand data flow

🏃‍♂️ Time-Saving Commands

# Quick test specific feature
python -m unittest tests.{carrier}.test_{feature} -v

# Fast schema regeneration
./bin/run-generate-on community/plugins/{carrier}

# Verify plugin registration
./bin/cli plugins list | grep {carrier}

# Check all tests pass
cd community/plugins/{carrier} && python -m unittest discover tests -v

📋 Final Integration Validation

Before declaring complete:

  • All 4-method test pattern implemented per feature
  • 100% test pass rate
  • Plugin appears in ./bin/cli plugins list
  • Schema generation works with ./bin/run-generate-on
  • Installation works with pip install -e

🎯 REQUIRED: Functional Programming Patterns

CRITICAL RULE: All provider implementations MUST follow functional programming patterns to maintain code consistency, readability, and the DRY principle throughout the Karrio codebase.

Core Functional Principles

  1. Prefer List Comprehensions Over For Loops: Use list comprehensions for data transformation
  2. Use Functional Utilities: Leverage karrio.lib functions for common operations
  3. Maintain Immutability: Avoid mutating objects when possible
  4. Composition Over Imperative Code: Chain operations using functional patterns
  5. Keep Request Instantiation Trees: Maintain clear, declarative request building patterns

❌ WRONG: Imperative Patterns

# NEVER DO THIS - Imperative, verbose, non-functional
def _extract_details(data, settings):
    rate_objects = data.get("rates", [])
    rates = []
    for rate_data in rate_objects:
        rate = {}
        rate["service"] = rate_data.get("service_code", "")
        rate["total"] = 0

        # Manual total calculation with loops
        amounts = []
        if rate_data.get("shipping_amount"):
            amounts.append(rate_data["shipping_amount"])
        if rate_data.get("insurance_amount"):
            amounts.append(rate_data["insurance_amount"])

        for amount in amounts:
            if amount and amount.get("amount"):
                rate["total"] += float(amount["amount"])

        rates.append(rate)
    return rates

✅ CORRECT: Functional Patterns

# ALWAYS DO THIS - Functional, concise, declarative
import karrio.lib as lib

def _extract_details(data, settings):
    rate_response_obj = lib.to_object(carrier_res.RateResponseType, data)

    return [
        _extract_rate_details(lib.to_dict(rate), settings)
        for rate in (rate_response_obj.rate_response.rates or [])
        if rate_response_obj.rate_response
    ]

def _extract_rate_details(rate_data: dict, settings) -> models.RateDetails:
    rate = lib.to_object(carrier_res.RateType, rate_data)

    # Functional amount calculation with list comprehension
    amounts = [rate.shipping_amount, rate.insurance_amount, rate.confirmation_amount, rate.other_amount]

    total_amount = sum(
        lib.to_money(amount.amount)
        for amount in amounts if amount and amount.amount
    )

    # Functional currency extraction
    currency = next(
        (amount.currency for amount in amounts if amount and amount.currency),
        "USD"
    )

    return models.RateDetails(
        carrier_id=settings.carrier_id,
        carrier_name=settings.carrier_name,
        service=rate.service_code,
        total_charge=lib.to_money(total_amount),
        currency=currency,
        # ... other fields
    )

Essential karrio.lib Functions

Always use these functional utilities:

# Data conversion and safety
lib.to_object(SchemaType, data)    # Convert dict to typed object
lib.to_dict(typed_object)          # Convert object to dict
lib.to_money(value)                # Safe money conversion
lib.to_address(address_data)       # Parse address safely
lib.to_packages(parcels)           # Parse packages safely
lib.to_services(services, ServiceEnum)  # Parse services safely

# Date and time handling
lib.fdate(date_string, format)     # Parse date safely
lib.flocaltime(date_string, format) # Parse local time safely

# Safe operations
lib.failsafe(lambda: risky_operation())  # Safe execution with fallback
lib.identity(value if condition else None)  # Conditional value passing

# Text processing
lib.text(value)                    # Safe string conversion

Functional List Processing Patterns

Use these patterns consistently:

# Event processing with functional patterns
events = [
    models.TrackingEvent(
        date=lib.fdate(event.occurred_at, "%Y-%m-%dT%H:%M:%SZ") if event.occurred_at else None,
        description=event.description or '',
        location=", ".join(filter(None, [
            event.city_locality,
            event.state_province,
            event.country_code
        ])),
    )
    for event in (tracking_details.events or [])
]

# Package conversion with functional patterns
packages = [
    carrier_req.PackageType(
        weight=carrier_req.WeightType(
            value=package.weight.value,
            unit=provider_units.WeightUnit[package.weight.unit].value,
        ),
        dimensions=lib.identity(
            carrier_req.DimensionsType(
                unit=provider_units.DimensionUnit[package.dimension_unit or "IN"].value,
                length=package.length.value,
                width=package.width.value,
                height=package.height.value,
            ) if all([package.length, package.width, package.height]) else None
        ),
    )
    for package in packages
]

# Error message aggregation with functional patterns
messages = sum(
    [
        error.parse_error_response(response, settings, tracking_number=tracking_number)
        for tracking_number, response in responses
    ],
    start=[],
)

Request Building Patterns

Maintain clear request instantiation trees like easyship:

def rate_request(payload, settings):
    # Parse inputs functionally
    shipper = lib.to_address(payload.shipper)
    recipient = lib.to_address(payload.recipient)
    packages = lib.to_packages(payload.parcels)

    # Build request tree declaratively
    request = carrier_req.RateRequestType(
        rate_options=carrier_req.RateOptionsType(
            calculate_tax_amount=True,
            preferred_currency="USD",
        ),
        shipment=carrier_req.ShipmentType(
            ship_to=carrier_req.AddressType(
                name=recipient.person_name,
                address_line1=recipient.address_line1,
                city=recipient.city,
                # ... all fields mapped declaratively
            ),
            ship_from=carrier_req.AddressType(
                name=shipper.person_name,
                address_line1=shipper.address_line1,
                city=shipper.city,
                # ... all fields mapped declaratively
            ),
            packages=[
                _build_package(package) for package in packages
            ],
        ),
    )

    return lib.Serializable(request, lib.to_dict)

def _build_package(package):
    """Helper function to maintain clean request tree structure."""
    return carrier_req.PackageType(
        weight=carrier_req.WeightType(
            value=package.weight.value,
            unit=provider_units.WeightUnit[package.weight.unit].value,
        ),
        dimensions=lib.identity(
            carrier_req.DimensionsType(
                unit=provider_units.DimensionUnit[package.dimension_unit or "IN"].value,
                length=package.length.value,
                width=package.width.value,
                height=package.height.value,
            ) if all([package.length, package.width, package.height]) else None
        ),
    )

Functional Pattern Rules

  1. No For Loops: Use list comprehensions instead of explicit for loops
  2. Use sum() for Aggregation: For collecting lists or calculating totals
  3. Use filter() and map(): For data transformation
  4. Use next() for First Match: For finding first element matching criteria
  5. Use lib.identity(): For conditional assignments
  6. Use lib.failsafe(): For operations that might fail
  7. Chain Operations: Combine multiple functional operations instead of intermediate variables

Following these functional patterns ensures your integration code is consistent with the rest of the Karrio codebase and maximizes code reuse and maintainability.

Summary

This guide provides comprehensive instructions for building Karrio carrier integrations. The key principles are:

  1. Always use Karrio CLI tooling - never create files manually
  2. Configure schema generation correctly - choose proper CLI parameters based on API field format
  3. Never edit generated schema files - only update source schemas and regenerate
  4. Always use generated types - import and use schema classes in all provider implementations
  5. Follow established patterns - study existing integrations
  6. Test thoroughly - ensure all success criteria pass
  7. Validate everything - use the comprehensive checklist before considering complete
  8. Use plain JSON schemas - faster generation and easier debugging than JSON Schema format
  9. Debug with dict access - use response.get() instead of object attributes to avoid errors
  10. Test early and often - implement one feature at a time with immediate testing

Success Criteria Reminder:

  • Carrier tests pass
  • SDK tests pass with ./bin/run-sdk-tests
  • Plugin appears in ./bin/cli plugins list
  • Plugin details accessible with ./bin/cli plugins show [carrier_name]

By following this guide strictly, you will create robust, maintainable carrier integrations that integrate seamlessly with the Karrio platform.