Ready to Run
SDK Examples
Production-ready code samples for common use cases. Copy, paste, and customize for your integration. Available in both Python and TypeScript.
Health Score Analysis
Analyze products for health impact based on materials and ingredients.
HealthMaterialsAnalysis
import asyncio
from justkalm import JustKalmClient
async def analyze_health_impact():
async with JustKalmClient() as client:
# Create product with materials
product = await client.products.create({
"name": "Organic Cotton T-Shirt",
"brand": "EcoWear",
"category": "apparel",
"materials": {
"organic_cotton": 0.95,
"elastane": 0.05
}
})
# Get comprehensive health analysis
valuation = await client.valuations.create(
product.id,
include_health=True
)
# Access health details
health = valuation.health_details
print(f"Overall Score: {health.overall_score}/100")
print(f"Chemical Safety: {health.chemical_safety}/100")
print(f"Material Toxicity: {health.material_toxicity}/100")
for risk in health.identified_risks:
print(f"⚠️ {risk.severity}: {risk.description}")
asyncio.run(analyze_health_impact())Sustainability Rating
Calculate environmental impact and sustainability scores for products.
SustainabilityEnvironmentESG
import asyncio
from justkalm import JustKalmClient
async def get_sustainability_rating():
async with JustKalmClient() as client:
product = await client.products.create({
"name": "Recycled Polyester Jacket",
"brand": "GreenOutdoor",
"category": "apparel",
"materials": {
"recycled_polyester": 0.80,
"organic_cotton": 0.15,
"recycled_nylon": 0.05
},
"certifications": ["GRS", "OEKO-TEX"]
})
valuation = await client.valuations.create(
product.id,
include_sustainability=True
)
sus = valuation.sustainability_details
print(f"Sustainability Score: {sus.overall_score}/100")
print(f"Carbon Footprint: {sus.carbon_footprint_kg} kg CO2e")
print(f"Water Usage: {sus.water_usage_liters} liters")
print(f"Recyclability: {sus.recyclability_score}/100")
for cert in sus.recognized_certifications:
print(f"✓ {cert.name}: +{cert.impact_score} points")
asyncio.run(get_sustainability_rating())Batch Processing
Process large volumes of products efficiently with parallel execution.
BatchPerformanceBulk
import asyncio
from justkalm import JustKalmClient
async def batch_process_products():
async with JustKalmClient() as client:
# Define products to process
products_data = [
{"name": f"Product {i}", "brand": "BatchBrand", "category": "apparel"}
for i in range(100)
]
# Use batch processor for efficiency
async with client.batch(
concurrency=5,
chunk_size=20,
on_error="continue"
) as batch:
results = await batch.create_products(products_data)
print(f"✓ Created: {results.successful} products")
print(f"✗ Failed: {results.failed} products")
# Process valuations in batch
product_ids = [r.product.id for r in results.succeeded]
async with client.batch(concurrency=10) as batch:
valuations = await batch.create_valuations(
product_ids,
include_health=True
)
avg_score = sum(v.health_score for v in valuations) / len(valuations)
print(f"Average Health Score: {avg_score:.1f}/100")
asyncio.run(batch_process_products())Webhook Integration
Handle real-time updates via webhooks with signature verification.
WebhooksReal-timeEvents
from fastapi import FastAPI, Request, HTTPException
from justkalm import WebhookHandler
app = FastAPI()
webhook = WebhookHandler(secret="whsec_your_secret")
@app.post("/webhooks/justkalm")
async def handle_webhook(request: Request):
payload = await request.body()
signature = request.headers.get("x-justkalm-signature")
try:
event = webhook.verify_and_parse(payload, signature)
except ValueError as e:
raise HTTPException(status_code=401, detail=str(e))
if event.type == "product.created":
product = event.data
print(f"New product: {product.name}")
# Sync to your database
elif event.type == "valuation.completed":
valuation = event.data
print(f"Valuation ready: {valuation.id}")
# Notify customer
elif event.type == "health.alert":
alert = event.data
print(f"⚠️ Health Alert: {alert.message}")
# Take action
return {"received": True}E-commerce Integration
Integrate health scores and valuations into your online store.
E-commerceShopifyAPI
import asyncio
from justkalm import JustKalmClient
from your_store import ShopifyClient
async def sync_product_to_store(product_id: str):
async with JustKalmClient() as jk:
shopify = ShopifyClient()
# Get JustKalm valuation
valuation = await jk.valuations.create(
product_id,
include_health=True,
include_sustainability=True
)
# Format for display
metafields = {
"justkalm_health_score": valuation.health_score,
"justkalm_sustainability": valuation.sustainability_score,
"justkalm_verified": True,
"justkalm_badge_url": valuation.badge_url,
"justkalm_report_url": valuation.report_url
}
# Update Shopify product
shopify.update_product_metafields(
product_id,
metafields
)
print(f"Synced {product_id} with scores:")
print(f" Health: {valuation.health_score}/100")
print(f" Sustainability: {valuation.sustainability_score}/100")
asyncio.run(sync_product_to_store("prod_abc123"))Real-time Streaming
Stream AI insights and valuations as they're generated.
StreamingReal-timeAI
import asyncio
from justkalm import JustKalmClient
async def stream_valuation(product_id: str):
async with JustKalmClient() as client:
# Stream valuation generation
async for chunk in client.valuations.stream(
product_id,
include_insights=True
):
if chunk.type == "health_score":
print(f"Health Score: {chunk.data.score}/100")
elif chunk.type == "sustainability_score":
print(f"Sustainability: {chunk.data.score}/100")
elif chunk.type == "insight":
print(f"💡 {chunk.data.text}")
elif chunk.type == "risk_identified":
print(f"⚠️ Risk: {chunk.data.description}")
elif chunk.type == "complete":
print("\n✓ Valuation complete")
print(f"Final value: ${chunk.data.adjusted_value:.2f}")
break
asyncio.run(stream_valuation("prod_abc123"))Market Value Analysis
Get AI-powered market valuations with confidence scores.
ValuationMarketPricing
import asyncio
from justkalm import JustKalmClient
async def analyze_market_value(product_id: str):
async with JustKalmClient() as client:
valuation = await client.valuations.create(
product_id,
include_market_analysis=True,
include_comparables=True
)
print(f"Product: {valuation.product_name}")
print(f"Condition: {valuation.condition}")
print()
print("=== Market Analysis ===")
print(f"Base Value: ${valuation.base_value:.2f}")
print(f"Adjusted Value: ${valuation.adjusted_value:.2f}")
print(f"Confidence: {valuation.confidence:.1%}")
print()
print("=== Value Adjustments ===")
for factor in valuation.factors:
sign = "+" if factor.impact > 0 else ""
print(f" {factor.name}: {sign}{factor.impact:.1%}")
print()
print("=== Comparable Products ===")
for comp in valuation.comparables[:3]:
print(f" {comp.name}: ${comp.sold_price:.2f} ({comp.condition})")
asyncio.run(analyze_market_value("prod_abc123"))Mobile App Integration
Integrate health scanning into React Native mobile apps.
MobileReact NativeScanner
# Backend API for mobile app
from fastapi import FastAPI
from justkalm import JustKalmClient
app = FastAPI()
@app.post("/api/scan")
async def scan_product(barcode: str, image_url: str = None):
async with JustKalmClient() as client:
# Lookup or create product from barcode
product = await client.products.lookup_or_create(
barcode=barcode,
image_url=image_url
)
# Get quick valuation for mobile
valuation = await client.valuations.create(
product.id,
include_health=True,
mode="fast" # Optimized for mobile
)
return {
"product": {
"id": product.id,
"name": product.name,
"brand": product.brand,
"image": product.image_url
},
"health": {
"score": valuation.health_score,
"grade": valuation.health_grade,
"color": valuation.health_color
},
"quick_insights": valuation.top_insights[:3]
}More Examples on GitHub
Browse our complete examples repository with full projects for Next.js, FastAPI, mobile apps, and more.