Building Digital Twins for AI Agents

CREATE VIEW products AS
SELECT p.product_id,
       p.name,
       BOOL_OR(p.active AND il.quantity_on_hand > 0) AS in_stock,
       SUM(il.quantity_on_hand - il.quantity_reserved) AS available_to_sell,
       COUNT(DISTINCT il.warehouse_location) AS num_fulfillment_sites,
       CASE
         WHEN SUM(il.quantity_on_hand - il.quantity_reserved) > 100 THEN 'High'
         WHEN SUM(il.quantity_on_hand - il.quantity_reserved) > 0 THEN 'Medium'
         ELSE 'Low'
       END AS stock_level
FROM product_catalog p
JOIN inventory_levels il ON p.product_id = il.product_id
GROUP BY p.product_id, p.name;

CREATE INDEX products_primary ON products(product_id);

COMMENT ON VIEW products IS 'Definitive, operationally trusted representation of a product and its current availability across the fulfillment network. Serves as the single source of truth for determining product availability, stock levels, and fulfillment capabilities in real time. Central to inventory management, fulfillment routing, and product-level operational decisions.';
COMMENT ON COLUMN products.product_id IS 'Globally unique identifier for the product. Serves as the primary key for referencing product-related operations, inventory levels, and catalog information across the business.';
COMMENT ON COLUMN products.name IS 'Canonical name of the product used in customer-facing systems and internal operations. Provides human-readable context and is essential for search, navigation, and user interfaces.';
COMMENT ON COLUMN products.primary_warehouse IS 'Warehouse location marked as the primary fulfillment site for this product. Selected based on highest priority location flag, this value drives preferred routing decisions and serves as a fallback for stock aggregation and replenishment.';
COMMENT ON COLUMN products.in_stock IS 'Boolean flag indicating whether the product is currently available at any fulfillment site with non-zero salable inventory. Used in determining product availability for purchase, surfacing in-stock indicators, and powering real-time stock checks.';
COMMENT ON COLUMN products.available_to_sell IS 'Total quantity of unreserved, sellable inventory across all fulfillment locations. Computed as the sum of quantity on hand minus quantity reserved. Drives availability logic for order placement, backorder prevention, and stock-level decisioning.';
COMMENT ON COLUMN products.num_fulfillment_sites IS 'Count of distinct warehouse locations currently stocking this product. Reflects the breadth of fulfillment coverage and is a proxy for fulfillment flexibility and resiliency.';
COMMENT ON COLUMN products.stock_level IS 'Qualitative stock level classification derived from available-to-sell quantity. Used for merchandising, alerts, and operational reporting. High indicates ample inventory, Medium signals moderate availability, and Low warns of impending stockout or zero availability.';

uv pip install mcp-materialize
uv run mcp-materialize --mz-dsn "$DSN" --transport http

materialize = MCPServer(lambda: streamablehttp("http://mcp-materialize:8001/mcp"))
with materialize:
        tools = materialize.list_tools_sync()
	purchase_agent = Agent(
    		system_prompt="Select and reserve ingredients for a customer’s recipe using live availability. If an ingredient isn't available find an alternative and and check with the customer",
    		tools=tools
	)

	response = purchase_agent("I want to make lasagna")

CREATE VIEW customers AS
WITH merged AS (
    SELECT
        COALESCE(c.customer_id, b.customer_id, s.customer_id) AS customer_id,
        COALESCE(c.name, 'Unknown') AS name,
        COALESCE(c.email, s.email) AS email,
        b.phone,
        b.address,
        GREATEST(c.updated_at, b.updated_at, s.updated_at) AS last_updated,
        COALESCE(st.crm_trust, 0) * 1.5 +
        COALESCE(st.billing_trust, 0) * 1.2 +
        COALESCE(st.support_trust, 0) +
        (CASE WHEN c.email IS NOT NULL THEN 5 ELSE 0 END) +
        (CASE WHEN b.phone IS NOT NULL THEN 3 ELSE 0 END) +
        (CASE WHEN b.address IS NOT NULL THEN 2 ELSE 0 END) AS score,
        ARRAY(SELECT DISTINCT a.allergen FROM customer_allergies a WHERE a.customer_id = COALESCE(c.customer_id, b.customer_id, s.customer_id)) AS allergens
    FROM crm_customers c
    FULL OUTER JOIN billing_customers b USING (customer_id)
    FULL OUTER JOIN support_customers s USING (customer_id)
    LEFT JOIN source_trust st ON st.customer_id = COALESCE(c.customer_id, b.customer_id, s.customer_id)
)
SELECT DISTINCT ON (customer_id)
    customer_id,
    name,
    email,
    phone,
    address,
    last_updated,
    score,
    allergens
FROM merged
ORDER BY customer_id, score DESC, last_updated DESC;

CREATE INDEX customers_primary ON customers(customer_id);

COMMENT ON VIEW customers IS 'Provides a unified and current profile of each customer, combining identity, contact details, and behavioral signals from multiple operational systems. Includes trust-based scoring and allergy information to support precise, context-aware decisions.';

COMMENT ON COLUMN customers.customer_id IS 'The unified identifier for a customer, resolved across CRM, billing, and support systems.';
COMMENT ON COLUMN customers.name IS 'Customer name prioritized from CRM and used for personalization and support.';
COMMENT ON COLUMN customers.email IS 'Customer email derived with fallback to support systems, used for notifications and validation.';
COMMENT ON COLUMN customers.phone IS 'Customer phone number as stored in billing systems, used for contact and verification.';
COMMENT ON COLUMN customers.address IS 'Most up-to-date known address for the customer, used for shipping and delivery logic.';
COMMENT ON COLUMN customers.last_updated IS 'Timestamp of most recent activity across all customer-related systems.';
COMMENT ON COLUMN customers.score IS 'Composite trust score based on source confidence and field completeness, used to prioritize high-quality records.';
COMMENT ON COLUMN customers.allergens IS 'List of known customer allergens based on medical or preference records, used for product filtering.'