RevComm Tech Blog

コミュニケーションを再発明し 人が人を想う社会を創る

Python × Salesforce CPQ APIで商談~見積品目登録プロセスの自動化

こんにちは。レブコムのコーポレートエンジニアリングチームの@ken-1200です。
この記事は、RevComm Advent Calendar 2024 の 20 日目の記事です。

1. はじめに

  • 記事の目的
    • 本記事では、Salesforce CPQ APIを活用して、商談から見積作成、見積品目の登録までのプロセスを自動化する手順をご紹介します
  • 対象読者
    • Salesforce CPQを利用するエンジニアの方を主な読者と想定しています

2. 開発の背景・モチベーション

  • なぜ開発に至ったのか
    • 営業プロセスでは、商談・見積・見積品目などの情報入力や修正が手動で行われ、工数がかかっていました。さらに、オンライン申込対応時には、商品の追加・削除・価格調整を都度行う必要があり、手作業によるミスや作業遅延が発生しがちでした
    • これらの課題解決のため、Salesforce CPQ APIを用いた自動化によって、業務フローを効率化し、正確性とスピードの向上を目指しました

3. 前提条件

  • Salesforce環境の準備
    • Salesforce CPQが有効化されていること
    • 必要なAPIアクセス権限(ユーザー権限設定)が設定されていること
  • 開発環境のセットアップ
    • Salesforce Sandbox環境
    • プログラミング言語:Python 3.10以上を推奨します
  • 基本的な知識
    • Salesforce CPQの基本概念
    • REST APIの基礎知識

4. Salesforce CPQ APIの概要

  • APIの種類
    • REST APIとSOAP APIの2種類が存在しますが、軽量で汎用性が高く、JSON形式のデータ交換が容易なREST APIを選択します
  • 認証方法
    • 一般的にはOAuth 2.0を使用することで、トークンベースの認証・認可が可能です
  • エンドポイントとリソース
    • Salesforce CPQには特定のエンドポイントを介して見積や商品情報へアクセスできます。以下は主な例です
      • QuoteReader: /services/apexrest/SBQQ/ServiceRouter?reader=SBQQ.QuoteAPI.QuoteReader&uid={quote_id}
        • 指定したQuote IDの詳細情報を取得します
      • ProductLoader: /services/apexrest/SBQQ/ServiceRouter?loader=SBQQ.ProductAPI.ProductLoader&uid={product_id}
        • 指定した商品IDに対する商品情報やオプションを取得します
      • QuoteProductAdder: /services/apexrest/SBQQ/ServiceRouter?loader=SBQQ.QuoteAPI.QuoteProductAdder
        • 見積に商品を追加するために使用します
      • QuoteCalculator: /services/apexrest/SBQQ/ServiceRouter?loader=SBQQ.QuoteAPI.QuoteCalculator
        • 見積品目を追加後に、見積全体の価格計算を行います
      • QuoteSave: /services/apexrest/SBQQ/ServiceRouter
        • 設定した見積や見積品目を保存および確定します
    • これらのエンドポイントを組み合わせることで、商談作成→見積生成→見積品目追加→価格再計算→保存という一連の流れを自動化できます
  • SalesforceRestApiClientクラスについて
    • Salesforce CPQ APIやSalesforce標準APIへのアクセスを簡潔にするために、本記事では共通的に利用できるSalesforceRestApiClientクラスを用いています
from collections.abc import Mapping
from typing import Any

import httpx

class SalesforceRestApiClient:
    """Salesforce REST APIクライアントクラス

    Salesforce APIを呼び出すための基本クラスです
    """

    def __init__(self, path: str, additional_headers: dict | None = None):
        self.base_url = "https://your-instance.salesforce.com"  # SalesforceインスタンスURLを指定してください
        self.path = path
        # ここでは例としてAuthorizationヘッダを省略していますが、
        # 実際にはOAuth2トークンや有効な認証ヘッダを設定してください
        self.headers = {"Authorization": "Bearer your_access_token", "Content-Type": "application/json"}
        if additional_headers:
            self.headers.update(additional_headers)

    async def get(self) -> httpx.Response:
        """GETリクエストを送信します"""
        async with httpx.AsyncClient() as client:
            return await client.get(f"{self.base_url}{self.path}", headers=self.headers, timeout=30)

    async def patch(self, json: Mapping[str, Any]) -> httpx.Response:
        """PATCHリクエストを送信します"""
        async with httpx.AsyncClient() as client:
            return await client.patch(f"{self.base_url}{self.path}", headers=self.headers, json=json, timeout=30)

    async def post(self, json: Mapping[str, Any]) -> httpx.Response:
        """POSTリクエストを送信します"""
        async with httpx.AsyncClient() as client:
            return await client.post(f"{self.base_url}{self.path}", headers=self.headers, json=json, timeout=30)

5. 商談(Opportunity)の作成

  • 必要なデータ
    • 商談名、ステージ、取引先情報など。必要に応じて追加してください
  • APIリクエストの構築

    • POSTメソッドを用いて、指定のエンドポイントへJSON形式でデータを送信します。Salesforce APIはContent-Lengthヘッダが必要となる場合があるため、事前にJSON文字列の長さを計算して設定します
      • エンドポイント例

        path="/services/data/vXX.X/sobjects/Opportunity"

      • ヘッダー例

        headers={"Content-Length": str(len(json.dumps(data)))}

  • サンプルコード

    • 以下はPythonによる実装例です。非同期HTTPクライアント(httpx)を用いてSalesforce APIにPOSTリクエストを送信し、商談を作成します
import asyncio

class SalesforceOpportunity:

    async def create_opportunity(self, data: Mapping[str, Any]) -> httpx.Response:
        """Salesforceの商談をAPIで作成します

        Args:
            data (Mapping[str, Any]): 作成する商談の情報を含んだ辞書型データ

        Returns:
            httpx.Response: Salesforce APIからのレスポンス
        """
        # APIクライアントを初期化(必要なヘッダを設定)
        sf_api_client = SalesforceRestApiClient(
            path="/services/data/vXX.X/sobjects/Opportunity",
            additional_headers={
                "Content-Length": str(len(json.dumps(data))),
            },
        )
        return await sf_api_client.post(json=data)

if __name__ == "__main__":
    # 実行例:商談を作成します
    async def main():
        sf = SalesforceOpportunity()
        response = await sf.create_opportunity(
            {
                "Name": "Test Opportunity",
                "StageName": "Prospecting",
                "CloseDate": "2024-12-31",
                "AccountId": "0015g00000A3X7dAAF",  # 有効なAccountIdを設定してください
            }
        )
        print(f"{response.status_code=}")
        print(f"{response.json()=}")

    asyncio.run(main())
  • エラーハンドリング
    • Salesforce APIへのPOST時には、201 Createdが成功時の典型的なステータスコードです。エラー時には400404などが返り、レスポンスボディ内にerrorCodefieldsなどの詳細が含まれます
    • 以下はエラーが発生した場合の例です

        [
            {
                "message": "不正な種別の ID 値: 0015g00000A3X7dAAF",
                "errorCode": "MALFORMED_ID",
                "fields": ["AccountId"]
            }
        ]
      
    • このようなエラーに対しては、ログ出力やリトライ、適切なエラーメッセージのユーザー通知などを行います。

6. 見積(Quote)の作成

  • 商談との関連付け
    • 見積と商談は、SBQQ__Opportunity2__cフィールドで関連付けます
  • 必要なフィールド
    • 商談ID、価格表ID、期限日など。要件に応じて追加フィールドやカスタムフィールドを設定します
  • APIリクエストの詳細

    • POSTリクエストを使用して、SBQQ__Quote__cオブジェクトにデータを送信します
      • エンドポイント例

        path="/services/data/vXX.X/sobjects/SBQQ__Quote__c"

      • ヘッダー例

        headers={"Content-Length": str(len(json.dumps(data)))}

  • サンプルコード

    • 以下は、Pythonを使用して見積を作成するサンプルコードです
import asyncio

class SalesforceQuote:

    async def create_quote(self, data: Mapping[str, Any]) -> httpx.Response:
        """Salesforceの見積をAPIで作成します

        Args:
            data (Mapping[str, Any]): 作成する見積の情報を含んだ辞書型データ

        Returns:
            httpx.Response: Salesforce APIからのレスポンス
        """
        # APIクライアントを初期化(必要なヘッダを設定)
        sf_api_client = SalesforceRestApiClient(
            path="/services/data/vXX.X/sobjects/SBQQ__Quote__c",
            additional_headers={
                "Content-Length": str(len(json.dumps(data))),
            },
        )
        return await sf_api_client.post(json=data)

if __name__ == "__main__":
    # 実行例:見積を作成します
    async def main():
        sf = SalesforceQuote()
        # 以下は例としてのフィールド設定です。実際のIDや値は有効なものを指定してください。
        response = await sf.create_quote(
            {
                "SBQQ__BillingCity__c": "Chiyoda",
                "SBQQ__BillingPostalCode__c": "100-0000",
                "SBQQ__BillingState__c": "Tokyo",
                "SBQQ__BillingStreet__c": "1-1-1",
                "SBQQ__EndDate__c": "2024-12-31",
                "SBQQ__Opportunity2__c": "0065g00000B3X7dAAF",  # 商談ID
                "SBQQ__PricebookId__c": "01s5g00000A3X7dAAF",  # 価格表ID
                "SBQQ__PrimaryContact__c": None,
                "SBQQ__Primary__c": True,
                "SBQQ__QuoteTemplateId__c": "a1s5g0000003X7dAAF",  # テンプレートID
                "SBQQ__StartDate__c": "2024-01-01",
                "SBQQ__SubscriptionTerm__c": 12,
            }
        )
        print(f"{response.status_code=}")
        print(f"{response.json()=}")

    asyncio.run(main())
  • エラーハンドリング
    • エラーが発生した場合、400 Bad Requestなどのステータスコードとともに、errorCodemessageが返されます。以下は一般的なエラー応答例です

        [
            {
                "message": "invalid cross reference id",
                "errorCode": "INVALID_CROSS_REFERENCE_KEY",
                "fields": []
            }
        ]
      
    • このようなエラーに対しては、ログ出力やIDの再確認、必要なデータフィールドの修正を行います

7. 見積品目(Quote Line Items)の登録

  • 手順

    1. 見積の読み取り
    2. 商品の読み取り
    3. 商品の追加
    4. 見積の計算
    5. 見積の保存

    これらのステップを通じて、見積品目を自動的に追加できます

  • 製品情報の準備

    • 製品ID、数量、価格などのデータ。必要に応じて追加フィールドを設定できます
  • APIリクエストの構築
    • 見積IDを基に見積品目を追加するには、CPQ固有のエンドポイントを使用します。QuoteReaderProductLoaderQuoteProductAdderQuoteCalculatorQuoteSaverといったCPQ APIエンドポイントを順番に呼び出すことで、一連の処理を自動化できます
  • バルク操作の考慮
    • 複数の見積品目を一度に登録する場合、一括処理用のコンテキストをまとめて送信することで、パフォーマンスを最適化できます
    • 商品を一括追加した後、見積を再計算し、最後に保存する流れで処理を完結させます
  • サンプルコード
    • 以下のサンプルコードは、見積に商品を追加する一連の流れをCPQ APIで実現します
import asyncio
        
class SalesforceCpqQuote:

    async def get_quote(self, quote_id: str) -> httpx.Response:
        """Salesforceの見積をCPQ APIで取得します"""
        sf_api_client = SalesforceRestApiClient(
            path=f"/services/apexrest/SBQQ/ServiceRouter?reader=SBQQ.QuoteAPI.QuoteReader&uid={quote_id}",
        )
        return await sf_api_client.get()

    async def get_product(self, product_id: str, data: Mapping[str, str]) -> httpx.Response:
        """Salesforceの商品をCPQ APIで取得します"""
        sf_api_client = SalesforceRestApiClient(
            path=f"/services/apexrest/SBQQ/ServiceRouter?loader=SBQQ.ProductAPI.ProductLoader&uid={product_id}",
            additional_headers={
                "Content-Length": str(len(json.dumps(data))),
            },
        )
        return await sf_api_client.patch(json=data)

    async def add_product(self, data: Mapping[str, str]) -> httpx.Response:
        """Salesforceの見積品目をCPQ APIで作成します"""
        sf_api_client = SalesforceRestApiClient(
            path="/services/apexrest/SBQQ/ServiceRouter?loader=SBQQ.QuoteAPI.QuoteProductAdder",
            additional_headers={
                "Content-Length": str(len(json.dumps(data))),
            },
        )
        return await sf_api_client.patch(json=data)

    async def calculate_quote(self, data: Mapping[str, str]) -> httpx.Response:
        """Salesforceの見積をCPQ APIで計算します"""
        sf_api_client = SalesforceRestApiClient(
            path="/services/apexrest/SBQQ/ServiceRouter?loader=SBQQ.QuoteAPI.QuoteCalculator",
            additional_headers={
                "Content-Length": str(len(json.dumps(data))),
            },
        )
        return await sf_api_client.patch(json=data)

    async def save_quote(self, data: Mapping[str, str]) -> httpx.Response:
        """Salesforceの見積をCPQ APIで保存します"""
        sf_api_client = SalesforceRestApiClient(
            path="/services/apexrest/SBQQ/ServiceRouter",
            additional_headers={
                "Content-Length": str(len(json.dumps(data))),
            },
        )
        return await sf_api_client.post(json=data)

class CreateQuoteLineItem:

    def __init__(self) -> None:
        self.salesforce_cpq_quote = SalesforceCpqQuote()

    async def execute(
        self,
        quote_id: str,
        product_id: str,
        pricebook_id: str,
        currency_code: str,
        product_counts: dict,
    ):
        """処理の流れ:

        1. 見積の読み取り
        2. 商品の読み込み
        3. 商品の追加
        4. 見積の計算
        5. 見積の保存
        """
        # 見積の読み取り
        cpq_quote = await self.get_quote(quote_id)
        print(f"Successfully get quote: {cpq_quote}")

        # 商品の読み込み
        product = await self.get_product(product_id, pricebook_id, currency_code)
        print(f"Successfully get product: {product}")

        # 商品の追加
        add_product_to_quote = await self.add_product_to_quote(product_counts, cpq_quote, product)
        print(f"Successfully add product: {add_product_to_quote}")

        # 見積の計算
        calculate_quote = await self.calculate_quote(add_product_to_quote)
        print(f"Successfully calculate quote: {calculate_quote}")

        # 見積の保存
        save_quote = await self.save_quote(calculate_quote)
        print(f"Successfully save quote: {save_quote}")

    async def get_quote(self, quote_id: str) -> dict:
        """Salesforceの見積を取得"""
        quote_response = await self.salesforce_cpq_quote.get_quote(quote_id)
        return json.loads(quote_response.json())

    async def get_product(self, product_id: str, pricebook_id: str, currency_code: str) -> dict:
        """Salesforceの商品を取得"""
        product_data = {
            "context": json.dumps(ProductGetContext(pricebookId=pricebook_id, currencyCode=currency_code).dict())
        }
        product_response = await self.salesforce_cpq_quote.get_product(product_id, product_data)
        return json.loads(product_response.json())

    async def add_product_to_quote(self, product_counts: dict, quote: dict, product: dict) -> dict:
        """Salesforceの商品を見積に追加"""
        # ProductModelやConfigurationModelなど
        product_model = ProductModel(**product)
        list_of_product_model = []
        list_of_configuration_model = []

        # バンドル商品の子商品を追加
        for mb_op in product_model.options:
            product_id = mb_op.record["SBQQ__OptionalSKU__c"]
            quantity = product_counts.get(product_id)
            if quantity:
                mb_op.record["SBQQ__Quantity__c"] = quantity
                cf_model = ConfigurationModel(
                    configuredProductId=product_id,
                    optionId=mb_op.record["Id"],
                    optionData=mb_op.record,
                    configurationData=mb_op.record,
                    inheritedConfigurationData=None,
                    optionConfigurations=[],
                    configured=False,
                    changedByProductActions=False,
                    isDynamicOption=False,
                    isUpgrade=False,
                    disabledOptionIds=[],
                    hiddenOptionIds=[],
                    listPrice=None,
                    priceEditable=False,
                    validationMessages=[],
                    dynamicOptionKey=None,
                )
                list_of_configuration_model.append(cf_model.dict())

        # バンドル商品本体へのオプション追加
        if product_model.configuration is not None:
            if product_model.configuration.optionConfigurations is not None:
                product_model.configuration.optionConfigurations.extend(list_of_configuration_model)
            product_model.configuration.configured = True
            list_of_product_model.append(product_model.dict())

        # 見積と商品モデルをcontextにセット
        context = ProductAddContext(
            quote={k: v for k, v in quote.items() if k != "ui_original_record"},
            products=list_of_product_model,
        )
        add_product_response = await self.salesforce_cpq_quote.add_product({"context": json.dumps(context.dict())})
        return json.loads(add_product_response.json())

    async def calculate_quote(self, quote: dict) -> dict:
        """Salesforceの見積を計算"""
        calculate_quote_data = {
            "context": json.dumps({"quote": {k: v for k, v in quote.items() if k != "ui_original_record"}})
        }
        calculate_quote_response = await self.salesforce_cpq_quote.calculate_quote(calculate_quote_data)
        return json.loads(calculate_quote_response.json())

    async def save_quote(self, quote: dict) -> dict:
        """Salesforceの見積を保存"""
        save_quote_data = {
            "saver": "SBQQ.QuoteAPI.QuoteSaver",
            "model": json.dumps({k: v for k, v in quote.items() if k != "ui_original_record"}),
        }
        save_quote_response = await self.salesforce_cpq_quote.save_quote(save_quote_data)
        return json.loads(save_quote_response.json())

if __name__ == "__main__":
    """見積品目の追加を実行する例です。実際には有効なIDと通貨コードを設定してください"""

    async def main():
        quote_id = "a0B5g00000DwJtEEAV"  # 見積ID
        product_id = "01t5g00000B1Q0PAK"  # 商品バンドルID
        pricebook_id = "01s5g0000008Q5eAAE"  # 価格表ID
        currency_code = "JPY"  # 通貨コード
        product_counts = {
            "01t5g00000B1Q0MKA": 1,  # 見積商品IDと数量
            "01t5g00000B1Q0NAAV": 2,
        }
        await CreateQuoteLineItem().execute(
            quote_id,
            product_id,
            pricebook_id,
            currency_code,
            product_counts,
        )

    asyncio.run(main())
  • CPQ API のリクエストモデル定義
    • 以下は、CPQ APIとのやりとりで使用するデータモデルの例です。pydanticを用いてスキーマを定義し、バリデーションやコメントを明確にしています。これらのモデルは、受け取ったJSONデータを明確な型情報のあるPythonオブジェクトとして扱うことで、可読性を向上させます
from pydantic import BaseModel, Field
        
class ConfigurationModel(BaseModel):
    configuredProductId: str = Field(..., title="商品ID", description="The Product2.Id", example="01t6F00000B8XZTAA3")
    optionId: str | None = Field(
        default=None, title="オプションID", description="The SBQQ__ProductOption__c.Id", example="01t6F00000B8XZTAA3"
    )
    optionData: dict = Field(
        ...,
        title="オプションデータ",
        description="Editable data about the option, such as quantity or discount",
        example={"Id": "01t6F00000B8XZTAA3"},
    )
    configurationData: dict = Field(
        ...,
        title="構成データ",
        description="Stores the values of the configuration attributes.",
        example={"Id": "01t6F00000B8XZTAA3"},
    )
    inheritedConfigurationData: dict | None = Field(
        default=None,
        title="継承された構成データ",
        description="Stores the values of the inherited configuration attributes.",
        example={"Id": "01t6F00000B8XZTAA3"},
    )
    optionConfigurations: list = Field(
        ...,
        title="オプション構成",
        description="Stores the options selected on this product.",
        example=[{"Id": "01t6F00000B8XZTAA3"}],
    )
    configured: bool = Field(
        ..., title="構成済み", description="Indicates whether the product has been configured.", example=False
    )
    changedByProductActions: bool = Field(
        ...,
        title="商品アクションによる変更",
        description="Indicates whether a product action changed the configuration of this bundle.",
        example=False,
    )
    isDynamicOption: bool = Field(
        ...,
        title="動的オプション",
        description="Indicates whether the product was configured using a dynamic lookup.",
        example=False,
    )
    isUpgrade: bool = Field(
        ..., title="アップグレード", description="Queries whether this product is an upgrade.", example=False
    )
    disabledOptionIds: list = Field(
        default=None,
        title="無効なオプションID",
        description="The option IDs that are disabled.",
        example=["01t6F00000B8XZTAA3"],
    )
    hiddenOptionIds: list = Field(
        default=None,
        title="非表示オプションID",
        description="The option IDs that are hidden.",
        example=["01t6F00000B8XZTAA3"],
    )
    listPrice: float | None = Field(default=None, title="定価", description="The list price.", example=0.0)
    priceEditable: bool = Field(
        ..., title="価格編集可能", description="Indicates whether the price is editable.", example=False
    )
    validationMessages: list = Field(
        ..., title="検証メッセージ", description="Validation messages.", example=["Error message"]
    )
    dynamicOptionKey: str | None = Field(
        default=None,
        title="動的オプションキー",
        description="Internal property for dynamic options.",
        example="01t6F00000B8XZTAA3",
    )

class OptionModel(BaseModel):
    record: dict = Field(
        ...,
        title="オプション",
        description="The record that this model represents.",
        example={"Id": "01t6F00000B8XZTAA3"},
    )
    externalConfigurationData: dict | None = Field(
        default=None,
        title="外部構成データ",
        description="Internal property for the external configurator feature.",
        example={"Id": "01t6F00000B8XZTAA3"},
    )
    configurable: bool = Field(
        ..., title="構成可能", description="Indicates whether the option is configurable.", example=False
    )
    configurationRequired: bool = Field(
        ...,
        title="構成必須",
        description="Indicates whether the configuration of the option is required.",
        example=False,
    )
    quantityEditable: bool = Field(
        ..., title="数量編集可能", description="Indicates whether the quantity is editable.", example=False
    )
    priceEditable: bool = Field(
        ..., title="価格編集可能", description="Indicates whether the price is editable.", example=False
    )
    productQuantityScale: float | None = Field(
        default=None,
        title="商品数量スケール",
        description="Returns the value of the quantity scale field for the product being configured.",
        example=0.0,
    )
    priorOptionExists: bool | None = Field(
        default=None,
        title="前のオプションが存在する",
        description="Checks if this option is an asset on the account that the quote is associated with.",
        example=False,
    )
    dependentIds: list = Field(
        ...,
        title="依存するオプションID",
        description="The option IDs that depend on this option.",
        example=["01t6F00000B8XZTAA3"],
    )
    controllingGroups: dict = Field(
        ...,
        title="制御グループ",
        description="The option IDs that this option depends on.",
        example={"Id": "01t6F00000B8XZTAA3"},
    )
    exclusionGroups: dict = Field(
        ...,
        title="排他グループ",
        description="The option IDs that this option is exclusive with.",
        example={"Id": "01t6F00000B8XZTAA3"},
    )
    reconfigureDimensionWarning: str = Field(
        ...,
        title="再構成次元警告",
        description="Reconfigures the warning label for an option with segments.",
        example="01t6F00000B8XZTAA3",
    )
    hasDimension: bool = Field(
        ..., title="次元がある", description="Indicates whether this option has dimensions or segments.", example=False
    )
    isUpgrade: bool = Field(
        ...,
        title="アップグレード",
        description="Indicates whether the product option is related to an upgrade product.",
        example=False,
    )
    dynamicOptionKey: str | None = Field(
        default=None,
        title="動的オプションキー",
        description="Internal property for dynamic options.",
        example="01t6F00000B8XZTAA3",
    )

class FeatureModel(BaseModel):
    record: dict = Field(
        ..., title="機能", description="The record that this model represents.", example={"Id": "01t6F00000B8XZTAA3"}
    )
    instructionsText: str | None = Field(
        default=None,
        title="指示テキスト",
        description="Instruction label for the feature.",
        example="01t6F00000B8XZTAA3",
    )
    containsUpgrades: bool = Field(
        ...,
        title="アップグレードが含まれている",
        description="This feature is related to an upgrade product.",
        example=False,
    )

class ConfigAttributeModel(BaseModel):
    name: str | None = Field(
        default=None,
        title="名前",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.Name.",
        example="01t6F00000B8XZTAA3",
    )
    targetFieldName: str = Field(
        ...,
        title="ターゲットフィールド名",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__TargetField__c.",
        example="01t6F00000B8XZTAA3",
    )
    displayOrder: float | None = Field(
        default=None,
        title="表示順",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__DisplayOrder__c.",
        example=0.0,
    )
    columnOrder: str = Field(
        ...,
        title="カラム順",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__ColumnOrder__c.",
        example="01t6F00000B8XZTAA3",
    )
    required: bool = Field(
        ...,
        title="必須",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__Required__c.",
        example=False,
    )
    featureId: str = Field(
        ...,
        title="機能ID",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__Feature__c.",
        example="01t6F00000B8XZTAA3",
    )
    position: str = Field(
        ...,
        title="位置",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__Position__c.",
        example="01t6F00000B8XZTAA3",
    )
    appliedImmediately: bool = Field(
        ...,
        title="直ちに適用",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__AppliedImmediately__c.",
        example=False,
    )
    applyToProductOptions: bool = Field(
        ...,
        title="商品オプションに適用",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__ApplyToProductOptions__c.",
        example=False,
    )
    autoSelect: bool = Field(
        ...,
        title="自動選択",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__AutoSelect__c.",
        example=False,
    )
    shownValues: list | None = Field(
        default=None,
        title="表示値",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__ShownValues__c.",
        example=["01t6F00000B8XZTAA3"],
    )
    hiddenValues: list | None = Field(
        default=None,
        title="非表示値",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__HiddenValues__c.",
        example=["01t6F00000B8XZTAA3"],
    )
    hidden: bool = Field(
        ...,
        title="非表示",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.SBQQ__Hidden__c.",
        example=False,
    )
    noSuchFieldName: str | None = Field(
        default=None,
        title="存在しないフィールド名",
        description="If no field with the target name exists, the target name is stored here.",
        example="01t6F00000B8XZTAA3",
    )
    myId: str = Field(
        ...,
        title="ID",
        description="Corresponds directly to SBQQ__ConfigurationAttribute__c.Id.",
        example="01t6F00000B8XZTAA3",
    )

class ConstraintModel(BaseModel):
    record: dict = Field(
        ..., title="制約", description="The record that this model represents.", example={"Id": "01t6F00000B8XZTAA3"}
    )
    priorOptionExists: bool = Field(
        ...,
        title="前のオプションが存在する",
        description="Checks if this option is an asset on the account that the quote is associated with.",
        example=False,
    )

class ProductModel(BaseModel):
    record: dict = Field(
        ..., title="商品", description="The record that this model represents.", example={"Id": "01t6F00000B8XZTAA3"}
    )
    upgradedAssetId: str | None = Field(
        default=None,
        title="SBQQ__QuoteLine__c.SBQQ__UpgradedAsset__c.Id",
        description="Provides a source for SBQQ__QuoteLine__c.SBQQ__UpgradedAsset__c.",
        example="01t6F00000B8XZTAA3",
    )
    currencySymbol: str = Field(
        ..., title="通貨シンボル", description="The symbol for the currency in use.", example="¥"
    )
    currencyCode: str = Field(
        ..., title="通貨コード", description="The ISO code for the currency in use.", example="JPY"
    )
    featureCategories: list = Field(
        ...,
        title="機能カテゴリ",
        description="Allows users to sort product features by category.",
        example=["01t6F00000B8XZTAA3"],
    )
    options: list[OptionModel] = Field(
        ...,
        title="オプション",
        description="A list of all available options for this product.",
        example=["01t6F00000B8XZTAA3"],
    )
    features: list[FeatureModel] = Field(
        ..., title="機能", description="All features available for this product", example=["01t6F00000B8XZTAA3"]
    )
    configuration: ConfigurationModel = Field(
        ...,
        title="構成",
        description="An object representing this product’s current configuration.",
        example={"Id": "01t6F00000B8XZTAA3"},
    )
    configurationAttributes: list[ConfigAttributeModel] = Field(
        ...,
        title="構成属性",
        description="All configuration attributes available for this product.",
        example=["01t6F00000B8XZTAA3"],
    )
    inheritedConfigurationAttributes: list[ConfigAttributeModel] | None = Field(
        default=None,
        title="継承された構成属性",
        description="All configuration attributes that this product inherits from ancestor products.",
        example=["01t6F00000B8XZTAA3"],
    )
    constraints: list[ConstraintModel] = Field(
        ..., title="制約", description="Option constraints on this product.", example=["01t6F00000B8XZTAA3"]
    )

class ProductGetContext(BaseModel):
    pricebookId: str = Field(
        ..., title="価格表ID", description="The ID of the price book to use.", example="01s6F00000CneeJQAR"
    )
    currencyCode: str = Field(
        ..., title="通貨コード", description="The ISO code for the currency in use.", example="JPY"
    )

class ProductAddContext(BaseModel):
    ignoreCalculate: bool = Field(default=True, title="計算無視", description="計算無視", example=True)
    quote: dict = Field(..., title="見積", description="見積モデル", example={})
    products: list = Field(..., title="商品リスト", description="商品モデル", example=[])
    groupKey: int = Field(default=0, title="グループキー", description="グループキー", example=0)
  • エラーハンドリング
    • エラーが発生した場合はHTTPステータスコードとエラーメッセージが返されます。たとえば、500 Internal Server Errorなどが返された場合、レスポンス本文にはerrorCodemessageフィールドが含まれます。

        [
            {
                "errorCode": "APEX_ERROR",
                "message": "System.AssertException: Assertion Failed: Unsupported quote object: a0B5g00000DwJtEEAV\n\n(System Code)",
            }
        ]
      

8. ポイントの振り返り

1. 商談(Opportunity)の作成

  • 必要なデータ(商談名、ステージ、取引先ID、CloseDateなど)を準備します
  • POST /services/data/vXX.X/sobjects/Opportunity エンドポイントを用い、JSON形式でデータを送信します

2. 見積(Quote)の作成

  • 商談ID、価格表ID、期間や開始日などの必須項目を指定します
  • POST /services/data/vXX.X/sobjects/SBQQ__Quote__c を使用し、JSON形式でデータを送信します

3. 見積品目(Quote Line Items)の登録

  • CPQ API固有のフロー(QuoteReader, ProductLoader, QuoteProductAdder, QuoteCalculator, QuoteSaver)を順序立てて実行します
  • 製品IDや数量、オプション構成などを事前に用意し、バンドル構成商品に対応します
  • 複数商品の一括追加時は、リクエストをまとめて送信し、パフォーマンスを最適化します

9. 自動化により得られた効果

自動化により、以下のような効果を得られました。

  • 手動作業が減少し、ヒューマンエラーが抑制されました
  • 営業担当者がコア業務に集中できる環境が整い、業務効率が向上しました
  • 見積作成から承認までのリードタイムが短縮され、顧客対応スピードと満足度が向上しました

10. 苦労した点・ハマりどころ

開発過程では、以下のような課題に直面しました。

  • Salesforce CPQ APIに関するドキュメントや事例が少なく、適切なエンドポイント選定やデータモデル理解までに試行錯誤が必要でした
  • 関連IDや依存関係の正確な把握が難しく、エラーの解消に時間がかかりました
  • 適切なエラーハンドリングやPydanticでのデータモデル定義など、Python実装上のベストプラクティスを探りながら開発を進める必要がありました

11. まとめ

本記事では、Salesforce CPQ APIを用いて、商談から見積作成、見積品目の登録までを自動化する具体的な手順とポイントをご紹介しました。これにより、手作業を減らしてヒューマンエラーを抑え、業務スピードを向上させることで、顧客満足度を高められる可能性が見えてきたかと思います。

この記事が少しでも参考になり、読者の方々の開発や業務改善にお役立ていただければ幸いです。

12. 参考文献