Skip to content

katana_public_api_client.helpers

katana_public_api_client.helpers

Domain classes for the Katana API client.

These domain classes provide ergonomic, domain-specific methods that reduce boilerplate and serve as the foundation for MCP tools.

Example

async with KatanaClient() as client: ... # Product catalog operations ... products = await client.products.list(is_sellable=True) ... product = await client.products.get(123) ... results = await client.products.search("widget")

Classes

Base(client)

Base class for all domain classes.

Provides common functionality and access to the KatanaClient instance.

Parameters:

  • client (KatanaClient) –

    The KatanaClient instance to use for API calls.

Parameters:

  • client (KatanaClient) –

    The KatanaClient instance to use for API calls.

Source code in katana_public_api_client/helpers/base.py
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions

Materials(client)

Bases: Base

Material catalog management.

Provides CRUD operations for materials in the Katana catalog.

Example

async with KatanaClient() as client: ... # CRUD operations ... materials = await client.materials.list() ... material = await client.materials.get(123) ... new_material = await client.materials.create({"name": "Steel"})

Source code in katana_public_api_client/helpers/base.py
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions
create(material_data) async

Create a new material.

Parameters:

Returns:

Example

from katana_public_api_client.models import CreateMaterialRequest new_material = await client.materials.create( ... CreateMaterialRequest(name="Steel") ... )

Source code in katana_public_api_client/helpers/materials.py
async def create(self, material_data: CreateMaterialRequest) -> KatanaMaterial:
    """Create a new material.

    Args:
        material_data: CreateMaterialRequest model with material details.

    Returns:
        Created KatanaMaterial domain model object.

    Example:
        >>> from katana_public_api_client.models import CreateMaterialRequest
        >>> new_material = await client.materials.create(
        ...     CreateMaterialRequest(name="Steel")
        ... )
    """
    response = await create_material.asyncio_detailed(
        client=self._client,
        body=material_data,
    )
    attrs_material = unwrap_as(response, Material)
    return material_to_katana(attrs_material)
delete(material_id) async

Delete a material.

Parameters:

  • material_id (int) –

    The material ID to delete.

Example

await client.materials.delete(123)

Source code in katana_public_api_client/helpers/materials.py
async def delete(self, material_id: int) -> None:
    """Delete a material.

    Args:
        material_id: The material ID to delete.

    Example:
        >>> await client.materials.delete(123)
    """
    await delete_material.asyncio_detailed(
        client=self._client,
        id=material_id,
    )
get(material_id) async

Get a specific material by ID.

Parameters:

  • material_id (int) –

    The material ID.

Returns:

Example

material = await client.materials.get(123)

Source code in katana_public_api_client/helpers/materials.py
async def get(self, material_id: int) -> KatanaMaterial:
    """Get a specific material by ID.

    Args:
        material_id: The material ID.

    Returns:
        KatanaMaterial domain model object.

    Example:
        >>> material = await client.materials.get(123)
    """
    response = await get_material.asyncio_detailed(
        client=self._client,
        id=material_id,
    )
    attrs_material = unwrap_as(response, Material)
    return material_to_katana(attrs_material)
list(**filters) async

List all materials with optional filters.

Parameters:

  • **filters (Any, default: {} ) –

    Filtering parameters.

Returns:

  • list[KatanaMaterial]

    List of KatanaMaterial domain model objects.

Example

materials = await client.materials.list(limit=100)

Source code in katana_public_api_client/helpers/materials.py
async def list(self, **filters: Any) -> builtins.list[KatanaMaterial]:
    """List all materials with optional filters.

    Args:
        **filters: Filtering parameters.

    Returns:
        List of KatanaMaterial domain model objects.

    Example:
        >>> materials = await client.materials.list(limit=100)
    """
    response = await get_all_materials.asyncio_detailed(
        client=self._client,
        **filters,
    )
    attrs_materials = unwrap_data(response)
    return materials_to_katana(attrs_materials)
update(material_id, material_data) async

Update an existing material.

Parameters:

  • material_id (int) –

    The material ID to update.

  • material_data (UpdateMaterialRequest) –

    UpdateMaterialRequest model with fields to update.

Returns:

Example

from katana_public_api_client.models import UpdateMaterialRequest updated = await client.materials.update( ... 123, UpdateMaterialRequest(name="Aluminum") ... )

Source code in katana_public_api_client/helpers/materials.py
async def update(
    self, material_id: int, material_data: UpdateMaterialRequest
) -> KatanaMaterial:
    """Update an existing material.

    Args:
        material_id: The material ID to update.
        material_data: UpdateMaterialRequest model with fields to update.

    Returns:
        Updated KatanaMaterial domain model object.

    Example:
        >>> from katana_public_api_client.models import UpdateMaterialRequest
        >>> updated = await client.materials.update(
        ...     123, UpdateMaterialRequest(name="Aluminum")
        ... )
    """
    response = await update_material.asyncio_detailed(
        client=self._client,
        id=material_id,
        body=material_data,
    )
    attrs_material = unwrap_as(response, Material)
    return material_to_katana(attrs_material)

Products(client)

Bases: Base

Product catalog management.

Provides CRUD operations and search for products in the Katana catalog.

Example

async with KatanaClient() as client: ... # Search products ... products = await client.products.search("widget") ... ... # CRUD operations ... products = await client.products.list(is_sellable=True) ... product = await client.products.get(123) ... new_product = await client.products.create({"name": "Widget"})

Source code in katana_public_api_client/helpers/base.py
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions
create(product_data) async

Create a new product.

Parameters:

Returns:

Example

from katana_public_api_client.models import CreateProductRequest new_product = await client.products.create( ... CreateProductRequest( ... name="New Widget", ... sku="WIDGET-NEW", ... is_sellable=True, ... variants=[], ... ) ... )

Source code in katana_public_api_client/helpers/products.py
async def create(self, product_data: CreateProductRequest) -> KatanaProduct:
    """Create a new product.

    Args:
        product_data: CreateProductRequest model with product details.

    Returns:
        Created KatanaProduct domain model object.

    Example:
        >>> from katana_public_api_client.models import CreateProductRequest
        >>> new_product = await client.products.create(
        ...     CreateProductRequest(
        ...         name="New Widget",
        ...         sku="WIDGET-NEW",
        ...         is_sellable=True,
        ...         variants=[],
        ...     )
        ... )
    """
    response = await create_product.asyncio_detailed(
        client=self._client,
        body=product_data,
    )
    attrs_product = unwrap_as(response, Product)
    return product_to_katana(attrs_product)
delete(product_id) async

Delete a product.

Parameters:

  • product_id (int) –

    The product ID to delete.

Example

await client.products.delete(123)

Source code in katana_public_api_client/helpers/products.py
async def delete(self, product_id: int) -> None:
    """Delete a product.

    Args:
        product_id: The product ID to delete.

    Example:
        >>> await client.products.delete(123)
    """
    await delete_product.asyncio_detailed(
        client=self._client,
        id=product_id,
    )
get(product_id) async

Get a specific product by ID.

Parameters:

  • product_id (int) –

    The product ID.

Returns:

Example

product = await client.products.get(123)

Source code in katana_public_api_client/helpers/products.py
async def get(self, product_id: int) -> KatanaProduct:
    """Get a specific product by ID.

    Args:
        product_id: The product ID.

    Returns:
        KatanaProduct domain model object.

    Example:
        >>> product = await client.products.get(123)
    """
    response = await get_product.asyncio_detailed(
        client=self._client,
        id=product_id,
    )
    attrs_product = unwrap_as(response, Product)
    return product_to_katana(attrs_product)
list(**filters) async

List all products with optional filters.

Parameters:

  • **filters (Any, default: {} ) –

    Filtering parameters (e.g., is_sellable, is_producible, include_deleted).

Returns:

  • list[KatanaProduct]

    List of KatanaProduct domain model objects.

Example

products = await client.products.list(is_sellable=True, limit=100)

Source code in katana_public_api_client/helpers/products.py
async def list(self, **filters: Any) -> builtins.list[KatanaProduct]:
    """List all products with optional filters.

    Args:
        **filters: Filtering parameters (e.g., is_sellable, is_producible, include_deleted).

    Returns:
        List of KatanaProduct domain model objects.

    Example:
        >>> products = await client.products.list(is_sellable=True, limit=100)
    """
    response = await get_all_products.asyncio_detailed(
        client=self._client,
        **filters,
    )
    attrs_products = unwrap_data(response)
    return products_to_katana(attrs_products)
search(query, limit=50) async

Search products by name and category with fuzzy matching and ranking.

Used by: MCP tool search_products

Features: - Multi-token matching with AND logic - Fuzzy matching for typo tolerance - Relevance-based ranking (exact name > prefix > substring > fuzzy)

Note: The Katana API 'name' parameter only does exact matches, so we fetch all products and perform client-side searching.

Parameters:

  • query (str) –

    Search query to match against product names (case-insensitive).

  • limit (int, default: 50 ) –

    Maximum number of results to return.

Returns:

  • list[KatanaProduct]

    List of matching KatanaProduct domain model objects, sorted by relevance.

Example

products = await client.products.search("part a1", limit=10) for product in products: ... print(f"{product.id}: {product.name}")

Source code in katana_public_api_client/helpers/products.py
async def search(self, query: str, limit: int = 50) -> builtins.list[KatanaProduct]:
    """Search products by name and category with fuzzy matching and ranking.

    Used by: MCP tool search_products

    Features:
    - Multi-token matching with AND logic
    - Fuzzy matching for typo tolerance
    - Relevance-based ranking (exact name > prefix > substring > fuzzy)

    Note: The Katana API 'name' parameter only does exact matches, so we
    fetch all products and perform client-side searching.

    Args:
        query: Search query to match against product names (case-insensitive).
        limit: Maximum number of results to return.

    Returns:
        List of matching KatanaProduct domain model objects, sorted by relevance.

    Example:
        >>> products = await client.products.search("part a1", limit=10)
        >>> for product in products:
        ...     print(f"{product.id}: {product.name}")
    """
    from katana_public_api_client.helpers.search import search_and_rank

    # Fetch all products (the API doesn't support partial/fuzzy search)
    response = await get_all_products.asyncio_detailed(
        client=self._client,
        limit=1000,  # Fetch up to 1000 products for searching
    )
    attrs_products = unwrap_data(response)

    # Convert to domain models
    domain_products = products_to_katana(attrs_products)

    return search_and_rank(
        query=query,
        items=domain_products,
        field_extractor=lambda p: {
            "name": (p.name or "", 100),
            "category": (p.category_name or "", 30),
        },
        limit=limit,
    )
update(product_id, product_data) async

Update an existing product.

Parameters:

  • product_id (int) –

    The product ID to update.

  • product_data (UpdateProductRequest) –

    UpdateProductRequest model with fields to update.

Returns:

Example

from katana_public_api_client.models import UpdateProductRequest updated = await client.products.update( ... 123, UpdateProductRequest(name="Updated Name") ... )

Source code in katana_public_api_client/helpers/products.py
async def update(
    self, product_id: int, product_data: UpdateProductRequest
) -> KatanaProduct:
    """Update an existing product.

    Args:
        product_id: The product ID to update.
        product_data: UpdateProductRequest model with fields to update.

    Returns:
        Updated KatanaProduct domain model object.

    Example:
        >>> from katana_public_api_client.models import UpdateProductRequest
        >>> updated = await client.products.update(
        ...     123, UpdateProductRequest(name="Updated Name")
        ... )
    """
    response = await update_product.asyncio_detailed(
        client=self._client,
        id=product_id,
        body=product_data,
    )
    attrs_product = unwrap_as(response, Product)
    return product_to_katana(attrs_product)

Services(client)

Bases: Base

Service catalog management.

Provides CRUD operations for services in the Katana catalog.

Example

async with KatanaClient() as client: ... # CRUD operations ... services = await client.services.list() ... service = await client.services.get(123) ... new_service = await client.services.create({"name": "Assembly"})

Source code in katana_public_api_client/helpers/base.py
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions
create(service_data) async

Create a new service.

Parameters:

Returns:

Example

from katana_public_api_client.models import CreateServiceRequest new_service = await client.services.create( ... CreateServiceRequest(name="Assembly") ... )

Source code in katana_public_api_client/helpers/services.py
async def create(self, service_data: CreateServiceRequest) -> KatanaService:
    """Create a new service.

    Args:
        service_data: CreateServiceRequest model with service details.

    Returns:
        Created KatanaService domain model object.

    Example:
        >>> from katana_public_api_client.models import CreateServiceRequest
        >>> new_service = await client.services.create(
        ...     CreateServiceRequest(name="Assembly")
        ... )
    """
    response = await create_service.asyncio_detailed(
        client=self._client,
        body=service_data,
    )
    attrs_service = unwrap_as(response, Service)
    return service_to_katana(attrs_service)
delete(service_id) async

Delete a service.

Parameters:

  • service_id (int) –

    The service ID to delete.

Example

await client.services.delete(123)

Source code in katana_public_api_client/helpers/services.py
async def delete(self, service_id: int) -> None:
    """Delete a service.

    Args:
        service_id: The service ID to delete.

    Example:
        >>> await client.services.delete(123)
    """
    await delete_service.asyncio_detailed(
        client=self._client,
        id=service_id,
    )
get(service_id) async

Get a specific service by ID.

Parameters:

  • service_id (int) –

    The service ID.

Returns:

Example

service = await client.services.get(123)

Source code in katana_public_api_client/helpers/services.py
async def get(self, service_id: int) -> KatanaService:
    """Get a specific service by ID.

    Args:
        service_id: The service ID.

    Returns:
        KatanaService domain model object.

    Example:
        >>> service = await client.services.get(123)
    """
    response = await get_service.asyncio_detailed(
        client=self._client,
        id=service_id,
    )
    attrs_service = unwrap_as(response, Service)
    return service_to_katana(attrs_service)
list(**filters) async

List all services with optional filters.

Parameters:

  • **filters (Any, default: {} ) –

    Filtering parameters.

Returns:

  • list[KatanaService]

    List of KatanaService domain model objects.

Example

services = await client.services.list(limit=100)

Source code in katana_public_api_client/helpers/services.py
async def list(self, **filters: Any) -> builtins.list[KatanaService]:
    """List all services with optional filters.

    Args:
        **filters: Filtering parameters.

    Returns:
        List of KatanaService domain model objects.

    Example:
        >>> services = await client.services.list(limit=100)
    """
    response = await get_all_services.asyncio_detailed(
        client=self._client,
        **filters,
    )
    attrs_services = unwrap_data(response)
    return services_to_katana(attrs_services)
update(service_id, service_data) async

Update an existing service.

Parameters:

  • service_id (int) –

    The service ID to update.

  • service_data (UpdateServiceRequest) –

    UpdateServiceRequest model with fields to update.

Returns:

Example

from katana_public_api_client.models import UpdateServiceRequest updated = await client.services.update( ... 123, UpdateServiceRequest(name="QA Testing") ... )

Source code in katana_public_api_client/helpers/services.py
async def update(
    self, service_id: int, service_data: UpdateServiceRequest
) -> KatanaService:
    """Update an existing service.

    Args:
        service_id: The service ID to update.
        service_data: UpdateServiceRequest model with fields to update.

    Returns:
        Updated KatanaService domain model object.

    Example:
        >>> from katana_public_api_client.models import UpdateServiceRequest
        >>> updated = await client.services.update(
        ...     123, UpdateServiceRequest(name="QA Testing")
        ... )
    """
    response = await update_service.asyncio_detailed(
        client=self._client,
        id=service_id,
        body=service_data,
    )
    attrs_service = unwrap_as(response, Service)
    return service_to_katana(attrs_service)

Variants(client)

Bases: Base

Variant catalog management.

Provides CRUD operations and search for product variants in the Katana catalog.

Note: Caching is handled by the MCP server's CatalogCache (SQLite + FTS5), not by this client-level helper. The client is a pure API wrapper.

Example

async with KatanaClient() as client: ... variants = await client.variants.list() ... variant = await client.variants.get(123) ... results = await client.variants.search("part a1")

Source code in katana_public_api_client/helpers/base.py
def __init__(self, client: KatanaClient) -> None:
    """Initialize with a client instance.

    Args:
        client: The KatanaClient instance to use for API calls.
    """
    self._client = client
Functions
create(variant_data) async

Create a new variant.

Note: Clears the variant cache after creation.

Parameters:

Returns:

Example

from katana_public_api_client.models import CreateVariantRequest new_variant = await client.variants.create( ... CreateVariantRequest(name="Large", product_id=123) ... )

Source code in katana_public_api_client/helpers/variants.py
async def create(self, variant_data: CreateVariantRequest) -> KatanaVariant:
    """Create a new variant.

    Note: Clears the variant cache after creation.

    Args:
        variant_data: CreateVariantRequest model with variant details.

    Returns:
        Created Variant object.

    Example:
        >>> from katana_public_api_client.models import CreateVariantRequest
        >>> new_variant = await client.variants.create(
        ...     CreateVariantRequest(name="Large", product_id=123)
        ... )
    """
    response = await create_variant.asyncio_detailed(
        client=self._client,
        body=variant_data,
    )
    attrs_variant = unwrap_as(response, Variant)
    return variant_to_katana(attrs_variant)
delete(variant_id) async

Delete a variant.

Note: Clears the variant cache after deletion.

Parameters:

  • variant_id (int) –

    The variant ID to delete.

Example

await client.variants.delete(123)

Source code in katana_public_api_client/helpers/variants.py
async def delete(self, variant_id: int) -> None:
    """Delete a variant.

    Note: Clears the variant cache after deletion.

    Args:
        variant_id: The variant ID to delete.

    Example:
        >>> await client.variants.delete(123)
    """
    await delete_variant.asyncio_detailed(
        client=self._client,
        id=variant_id,
    )
get(variant_id) async

Get a specific variant by ID.

Parameters:

  • variant_id (int) –

    The variant ID.

Returns:

Example

variant = await client.variants.get(123) print(variant.get_display_name()) print(f"Profit margin: {variant.profit_margin}%")

Source code in katana_public_api_client/helpers/variants.py
async def get(self, variant_id: int) -> KatanaVariant:
    """Get a specific variant by ID.

    Args:
        variant_id: The variant ID.

    Returns:
        KatanaVariant object.

    Example:
        >>> variant = await client.variants.get(123)
        >>> print(variant.get_display_name())
        >>> print(f"Profit margin: {variant.profit_margin}%")
    """
    response = await get_variant.asyncio_detailed(
        client=self._client,
        id=variant_id,
    )
    attrs_variant = unwrap_as(response, Variant)
    return variant_to_katana(attrs_variant)
list(**filters) async

List all variants with optional filters.

Parameters:

  • **filters (Any, default: {} ) –

    Filtering parameters.

Returns:

Example

variants = await client.variants.list(limit=100) for v in variants: ... print(f"{v.get_display_name()}: {v.profit_margin}%")

Source code in katana_public_api_client/helpers/variants.py
async def list(self, **filters: Any) -> List[KatanaVariant]:
    """List all variants with optional filters.

    Args:
        **filters: Filtering parameters.

    Returns:
        List of KatanaVariant objects.

    Example:
        >>> variants = await client.variants.list(limit=100)
        >>> for v in variants:
        ...     print(f"{v.get_display_name()}: {v.profit_margin}%")
    """
    response = await get_all_variants.asyncio_detailed(
        client=self._client,
        **filters,
    )
    attrs_variants = unwrap_data(response)
    return variants_to_katana(attrs_variants)
search(query, limit=50) async

Search variants by SKU or name with tokenization and fuzzy matching.

Fetches all variants from the API and searches client-side. For cached/persistent search, use the MCP server's CatalogCache.

Parameters:

  • query (str) –

    Search query (e.g., "part a1 160")

  • limit (int, default: 50 ) –

    Maximum number of results to return

Returns:

  • list[KatanaVariant]

    List of matching Variant objects, sorted by relevance

Example

variants = await client.variants.search("part a1", limit=10) for variant in variants: ... print(f"{variant.sku}: {variant.product_or_material_name}")

Source code in katana_public_api_client/helpers/variants.py
async def search(self, query: str, limit: int = 50) -> List[KatanaVariant]:
    """Search variants by SKU or name with tokenization and fuzzy matching.

    Fetches all variants from the API and searches client-side.
    For cached/persistent search, use the MCP server's CatalogCache.

    Args:
        query: Search query (e.g., "part a1 160")
        limit: Maximum number of results to return

    Returns:
        List of matching Variant objects, sorted by relevance

    Example:
        >>> variants = await client.variants.search("part a1", limit=10)
        >>> for variant in variants:
        ...     print(f"{variant.sku}: {variant.product_or_material_name}")
    """
    from katana_public_api_client.helpers.search import search_and_rank

    if not query or not query.strip():
        return []

    # Fetch all variants from API (no client-side caching)
    response = await get_all_variants.asyncio_detailed(
        client=self._client,
        extend=[GetAllVariantsExtendItem.PRODUCT_OR_MATERIAL],
    )
    all_variants = variants_to_katana(unwrap_data(response))

    return search_and_rank(
        query=query,
        items=all_variants,
        field_extractor=lambda v: {
            "sku": (v.sku or "", 100),
            "name": (v.get_display_name(), 30),
            "parent_name": (v.product_or_material_name or "", 20),
        },
        limit=limit,
    )
update(variant_id, variant_data) async

Update an existing variant.

Note: Clears the variant cache after update.

Parameters:

  • variant_id (int) –

    The variant ID to update.

  • variant_data (UpdateVariantRequest) –

    UpdateVariantRequest model with fields to update.

Returns:

Example

from katana_public_api_client.models import UpdateVariantRequest updated = await client.variants.update( ... 123, UpdateVariantRequest(name="XL") ... )

Source code in katana_public_api_client/helpers/variants.py
async def update(
    self, variant_id: int, variant_data: UpdateVariantRequest
) -> KatanaVariant:
    """Update an existing variant.

    Note: Clears the variant cache after update.

    Args:
        variant_id: The variant ID to update.
        variant_data: UpdateVariantRequest model with fields to update.

    Returns:
        Updated Variant object.

    Example:
        >>> from katana_public_api_client.models import UpdateVariantRequest
        >>> updated = await client.variants.update(
        ...     123, UpdateVariantRequest(name="XL")
        ... )
    """
    response = await update_variant.asyncio_detailed(
        client=self._client,
        id=variant_id,
        body=variant_data,
    )
    attrs_variant = unwrap_as(response, Variant)
    return variant_to_katana(attrs_variant)