Amazon Bedrock から MCP サーバー経由で Salesforce を触る

1uv init --python 3.12 bedrock-case-agent
2cd bedrock-case-agent
3uv add strands-agents python-dotenv

1bedrock-case-agent/
2├── .env              # 4 で作成。接続情報なので、リポジトリに含めない
3├── agent.py          # 6 で作成
4├── sf_auth.py        # 5 で作成
5├── reset_cases.apex  # 7 で作成 (試し直すとき用)
6└── data/
7    └── cases.json    # 3 で作成

1https://api.salesforce.com/platform/mcp/v1/d/<My Domain 名>/develop/platform/sobject-reads
2https://api.salesforce.com/platform/mcp/v1/d/<My Domain 名>/develop/platform/sobject-mutations

1{
2  "records": [
3    {
4      "attributes": { "type": "Case", "referenceId": "Case01" },
5      "Subject": "本番環境で全ユーザーがログインできない",
6      "Description": "本日 9:30 頃から全ユーザーが管理ポータルにログインできず、受注処理が止まっています。至急対応をお願いします。",
7      "Status": "New", "Priority": "Medium", "SuppliedCompany": "BEDROCK-DEMO"
8    },
9    {
10      "attributes": { "type": "Case", "referenceId": "Case02" },
11      "Subject": "夜間バッチ後に取引先の住所データが消えている",
12      "Description": "昨夜の連携バッチの後、取引先の約 300 件で住所が空になりました。請求書の発送に影響します。",
13      "Status": "New", "Priority": "Medium", "SuppliedCompany": "BEDROCK-DEMO"
14    },
15    {
16      "attributes": { "type": "Case", "referenceId": "Case03" },
17      "Subject": "心当たりのない海外 IP からのログイン成功通知が届いた",
18      "Description": "今朝、海外の IP アドレスからのログイン成功通知が届きました。チームの誰も海外にはいません。",
19      "Status": "New", "Priority": "Medium", "SuppliedCompany": "BEDROCK-DEMO"
20    },
21    {
22      "attributes": { "type": "Case", "referenceId": "Case04" },
23      "Subject": "営業ダッシュボードの表示が遅い",
24      "Description": "ダッシュボードの表示に 30 秒以上かかるようになりました。使えてはいますが、朝会で待たされています。",
25      "Status": "New", "Priority": "Medium", "SuppliedCompany": "BEDROCK-DEMO"
26    },
27    {
28      "attributes": { "type": "Case", "referenceId": "Case05" },
29      "Subject": "二要素認証の設定手順書はありますか",
30      "Description": "新入社員向けに、二要素認証の設定手順の資料があれば共有してください。急ぎではありません。",
31      "Status": "New", "Priority": "Medium", "SuppliedCompany": "BEDROCK-DEMO"
32    },
33    {
34      "attributes": { "type": "Case", "referenceId": "Case06" },
35      "Subject": "見積書テンプレートに会社ロゴを入れたい",
36      "Description": "見積書のテンプレートに、自社ロゴを入れられるようにしてほしいという要望です。",
37      "Status": "New", "Priority": "Medium", "SuppliedCompany": "BEDROCK-DEMO"
38    }
39  ]
40}

1sf org login web --alias my-de
2sf data import tree --files data/cases.json --target-org my-de

1sf data query --query "SELECT CaseNumber, Subject, Priority FROM Case WHERE SuppliedCompany = 'BEDROCK-DEMO' ORDER BY CaseNumber" --target-org my-de

1# Salesforce
2SF_CONSUMER_KEY=<コンシューマー鍵>
3SF_LOGIN_URL=https://login.salesforce.com
4SF_MCP_READS_URL=<sobject-reads のサーバー URL>
5SF_MCP_MUTATIONS_URL=<sobject-mutations のサーバー URL>
6
7# Amazon Bedrock
8AWS_BEARER_TOKEN_BEDROCK=<Bedrock API キー>
9AWS_REGION=ap-northeast-1
10BEDROCK_MODEL_ID=jp.amazon.nova-2-lite-v1:0

1"""Salesforce Hosted MCP Server 用のアクセストークンを、認可コードフロー + PKCE で取得する。
2
3  uv run sf_auth.py login   ブラウザで認可し、トークンを保存する
4  uv run sf_auth.py token   アクセストークンを出力する (MCP Inspector での疎通確認用)
5"""
6import base64
7import hashlib
8import http.server
9import json
10import os
11import secrets
12import sys
13import time
14import urllib.error
15import urllib.parse
16import urllib.request
17import webbrowser
18from pathlib import Path
19
20from dotenv import load_dotenv
21
22load_dotenv()
23
24LOGIN_URL = os.environ.get("SF_LOGIN_URL", "https://login.salesforce.com").rstrip("/")
25CLIENT_ID = os.environ.get("SF_CONSUMER_KEY", "")
26REDIRECT_URI = "http://localhost:8765/callback"  # 外部クライアントアプリケーションのコールバック URL と同じにする
27SCOPES = "mcp_api refresh_token"
28TOKEN_FILE = Path("~/.config/bedrock-case-agent/sf_token.json").expanduser()
29
30
31def _b64url(data: bytes) -> str:
32    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
33
34
35def _post_token(form: dict) -> dict:
36    req = urllib.request.Request(
37        f"{LOGIN_URL}/services/oauth2/token",
38        data=urllib.parse.urlencode(form).encode(),
39        headers={"Content-Type": "application/x-www-form-urlencoded"},
40    )
41    try:
42        with urllib.request.urlopen(req, timeout=30) as res:
43            return json.loads(res.read())
44    except urllib.error.HTTPError as e:
45        sys.exit(f"トークンの取得に失敗しました (HTTP {e.code}): {e.read().decode()}")
46
47
48def _save(tokens: dict) -> None:
49    tokens["saved_at"] = int(time.time())
50    TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True)
51    TOKEN_FILE.write_text(json.dumps(tokens))
52    TOKEN_FILE.chmod(0o600)  # 自分だけが読めるようにする
53
54
55def login() -> None:
56    if not CLIENT_ID:
57        sys.exit(".env に SF_CONSUMER_KEY (コンシューマー鍵) を設定してください。")
58
59    # PKCE: code_verifier から code_challenge を作り、認可画面に渡す
60    verifier = _b64url(secrets.token_bytes(64))
61    challenge = _b64url(hashlib.sha256(verifier.encode()).digest())
62    state = secrets.token_urlsafe(16)
63    params = {
64        "response_type": "code",
65        "client_id": CLIENT_ID,
66        "redirect_uri": REDIRECT_URI,
67        "scope": SCOPES,
68        "code_challenge": challenge,
69        "code_challenge_method": "S256",
70        "state": state,
71    }
72
73    result: dict = {}
74
75    class CallbackHandler(http.server.BaseHTTPRequestHandler):
76        def do_GET(self):
77            url = urllib.parse.urlparse(self.path)
78            if url.path == "/callback":
79                result.update(urllib.parse.parse_qsl(url.query))
80            self.send_response(200)
81            self.send_header("Content-Type", "text/plain; charset=utf-8")
82            self.end_headers()
83            self.wfile.write("認可が完了しました。ターミナルに戻ってください。".encode())
84
85        def log_message(self, *args):
86            pass
87
88    server = http.server.HTTPServer(("localhost", 8765), CallbackHandler)
89    webbrowser.open(f"{LOGIN_URL}/services/oauth2/authorize?" + urllib.parse.urlencode(params))
90    while "code" not in result and "error" not in result:
91        server.handle_request()  # コールバックが届くまで待つ
92    server.server_close()
93
94    if "error" in result or result.get("state") != state:
95        sys.exit(f"認可に失敗しました: {result.get('error_description', result)}")
96
97    # PKCE: トークンを取るときに code_verifier を送る
98    tokens = _post_token({
99        "grant_type": "authorization_code",
100        "code": result["code"],
101        "client_id": CLIENT_ID,
102        "redirect_uri": REDIRECT_URI,
103        "code_verifier": verifier,
104    })
105    _save(tokens)
106    print(f"トークンを保存しました: {TOKEN_FILE}")
107
108
109def get_access_token(max_age_sec: int = 3600) -> str:
110    if not TOKEN_FILE.exists():
111        sys.exit("トークンがありません。先に `uv run sf_auth.py login` を実行してください。")
112    tokens = json.loads(TOKEN_FILE.read_text())
113    if time.time() - tokens["saved_at"] > max_age_sec:  # 保存から時間が経っていたら取り直す
114        new = _post_token({
115            "grant_type": "refresh_token",
116            "refresh_token": tokens["refresh_token"],
117            "client_id": CLIENT_ID,
118        })
119        tokens = {**tokens, **new}
120        _save(tokens)
121    return tokens["access_token"]
122
123
124if __name__ == "__main__":
125    command = sys.argv[1] if len(sys.argv) > 1 else ""
126    if command == "login":
127        login()
128    elif command == "token":
129        print(get_access_token())
130    else:
131        sys.exit("使い方: uv run sf_auth.py login | token")

1uv run sf_auth.py login

1import os
2
3from dotenv import load_dotenv
4from strands import Agent
5from strands.models import BedrockModel
6from strands.tools.mcp import MCPClient
7
8from sf_auth import get_access_token
9
10load_dotenv()  # .env の値 (Bedrock の API キーを含む) を環境変数に読み込む
11
12headers = {"Authorization": f"Bearer {get_access_token()}"}
13
14# 読み取り用: sobject-reads の soqlQuery だけを渡す
15reads = MCPClient(
16    url=os.environ["SF_MCP_READS_URL"],  # sobject-reads のサーバー URL
17    headers=headers,
18    tool_filters={"allowed": ["soqlQuery"]},
19)
20
21# 書き込み用: sobject-mutations の updateSobjectRecord だけを渡す
22mutations = MCPClient(
23    url=os.environ["SF_MCP_MUTATIONS_URL"],  # sobject-mutations のサーバー URL
24    headers=headers,
25    tool_filters={"allowed": ["updateSobjectRecord"]},
26)
27
28model = BedrockModel(
29    model_id=os.environ["BEDROCK_MODEL_ID"],  # 例: jp.amazon.nova-2-lite-v1:0
30    region_name=os.environ["AWS_REGION"],  # 例: ap-northeast-1
31    temperature=0,
32    # Nova 2 の推論 (考えてから答える機能) をオンにする。オフのままだと、更新のツールを呼ばないことがあった
33    additional_request_fields={"reasoningConfig": {"type": "enabled", "maxReasoningEffort": "low"}},
34)
35
36SYSTEM_PROMPT = """あなたは Salesforce のサポートケース (Case) をトリアージする日本語の AI エージェントです。
37
38# 手順
391. 次の SOQL を soqlQuery で実行して、未対応のケースを取得する (そのまま使うこと)。
40   SELECT Id, CaseNumber, Subject, Description, Priority FROM Case
41   WHERE Status = 'New' AND SuppliedCompany = 'BEDROCK-DEMO'
42   ORDER BY CaseNumber LIMIT 10
432. 各ケースの件名と説明から、優先度を High、Medium、Low のどれかに判断する。
44   - High: 業務停止、データ消失、セキュリティ、法令、本番障害など緊急性が高いもの
45   - Medium: 不具合や設定の問題だが、業務は継続できるもの
46   - Low: 操作方法の質問、軽微な要望、情報提供の依頼など
473. 判断した優先度が、1 で取得した Priority と異なるケースは、すべて updateSobjectRecord で更新する。
48   High への変更も Low への変更も、同じように 1 件ずつ更新する。
49   sobject-name は "Case"、id は 1 の結果の Id をそのまま使い、body は {"Priority": "Low"} の形にする。
50   Priority 以外の項目は変更しない。
514. 更新が終わったら、1 と同じ SOQL をもう一度 soqlQuery で実行し、Priority が判断どおりになったかを確かめる。
52   判断どおりになっていないケースがあれば、もう一度 updateSobjectRecord で更新してから、もう一度確かめる。
535. 最後に、4 で確かめた結果を Markdown の表でまとめる (列: CaseNumber、件名、更新前の Priority、判定、理由、更新後の Priority)。
54"""
55
56agent = Agent(model=model, system_prompt=SYSTEM_PROMPT, tools=[reads, mutations])
57agent("未対応のケースの優先度を判断して、Priority を更新してください。")

1uv run agent.py

1// BEDROCK-DEMO の目印が付いたケースを、優先度 Medium、状況 New に戻す
2List<Case> cases = [SELECT Id FROM Case WHERE SuppliedCompany = 'BEDROCK-DEMO'];
3for (Case c : cases) {
4    c.Priority = 'Medium';
5    c.Status = 'New';
6}
7update cases;

1sf apex run --file reset_cases.apex --target-org my-de