--- title: "Bright DataとOpenHands SDKでエンタープライズTPRMエージェントを構築する" slug: openhands-sdk-with-bright-data date: 2026-01-28T07:39:06+00:00 modified: 2026-01-28T07:39:09+00:00 permalink: https://brightdata.jp/blog/ai/openhands-sdk-with-bright-data type: blog --- [ ブログ ](https://brightdata.jp/blog "ブログ") / [AI](https://brightdata.jp/blog/ai) [AI](https://brightdata.jp/blog/ai) # Bright DataとOpenHands SDKでエンタープライズTPRMエージェントを構築する このガイドでは、ベンダーリスクスクリーニングを自動化するため、Bright Data APIとOpenHands SDKを用いたスケーラブルなTPRMエージェントの構築方法を示します。 17 分読 [ ![Arindam Majumder](https://media.brightdata.jp/2025/07/Arindam-Majumder-50x50.jpg) ](https://brightdata.com/blog/authors/arindam-majumder) [Arindam Majumder Technical Writer ](https://brightdata.com/blog/authors/arindam-majumder) ![OpenHands Agent SDK with Bright Data](https://media.brightdata.jp/2025/12/OpenHands-Agent-SDK-with-Bright-Data.png) このガイドでは以下を学びます： - サードパーティリスク管理（TPRM）とは何か、手動スクリーニングが失敗する理由 - ベンダーのネガティブメディアを調査する自律型AIエージェントの構築方法 - 信頼性が高く最新のデータ収集を実現[するBright DataのSERP](/products/serp-api) [APIとWeb Unlockerの](/products/web-unlocker)統合方法 - – エージェント用スクリプト生成に[OpenHands SDKを](https://docs.openhands.dev/sdk/getting-started)、リスク分析にOpenAIを活用する方法 - 裁判所記録などの複雑なシナリオに対応するため、[Browser API](/products/scraping-browser)でエージェントを強化する方法 さあ始めましょう！ ## 手動によるベンダー審査の問題点 企業コンプライアンスチームは、ウェブ全体にわたる数百のサードパーティベンダーのリスクシグナルを監視するという不可能な課題に直面しています。従来のアプローチには以下が含まれます： - 各ベンダー名と「訴訟」「破産」「詐欺」などのキーワードを組み合わせて**手動でGoogle検索** - ニュース記事や裁判記録にアクセスする際の**有料壁やCAPTCHAの壁** - 調査結果を記録する標準化されたプロセスがなく、**文書化が不統一** - **継続的な監視がなく**、ベンダー審査はオンボーディング時に一度行われるだけで、その後は二度と実施されない この手法が失敗する3つの重大な理由： 1. **規模**：1人のアナリストが1日に徹底調査できるベンダーはせいぜい5～10社 2. **アクセス**：裁判所記録や有料ニュースサイトなどの保護された情報源は自動アクセスをブロック 3. **継続性**：オンボーディング時点での評価では、その後発生するリスクを見逃す ## 解決策：自律型TPRMエージェント TPRMエージェントは3つの専門レイヤーでベンダー調査ワークフロー全体を自動化： - **発見（SERP API）**：訴訟、規制措置、財務的苦境などの危険信号をGoogleで検索 - **アクセス（Web Unlocker）**：有料コンテンツやCAPTCHAで保護された関連結果に対し、エージェントが障壁を迂回して全文を抽出 - **アクション（OpenAI + OpenHands SDK）**：エージェントはOpenAIでコンテンツのリスク深刻度を分析し、OpenHands SDKを用いてPython監視スクリプトを生成。これにより毎日新たなネガティブメディアをチェック このシステムにより、数時間かかっていた手動調査が数分の自動分析に短縮されます。 ## 前提条件 開始前に以下の環境を確認してください： - Python 3.12 以上（OpenHands SDK に必須） - [APIアクセス権限付きのBright Dataアカウント](/)（無料トライアルあり） - リスク[分析用のOpenAI APIキー](https://platform.openai.com/api-keys) - [OpenHands Cloudアカウント](https://openhands.dev/)、またはエージェントスクリプト生成用の独自のLLM APIキー - PythonおよびREST APIの基本的な知識 ## プロジェクトアーキテクチャ TPRMエージェントは3段階のパイプラインに従います： ```none ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ 発見 │────▶│ アクセス │────▶│ アクション │ │ (SERP API) │ │ (Web Unlocker) │ │ (OpenAI + SDK) │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ Google検索 ペイウォールの回避 リスク分析 レッドフラグの検出 およびCAPTCHA スクリプト生成 ``` 以下のプロジェクト構造を作成: ```none tprm-agent/ ├── src/ │ ├── __init__.py │ ├── config.py # 設定 │ ├── discovery.py # SERP API 統合 │ ├── access.py # Web Unlocker 統合 │ ├── actions.py # OpenAI + OpenHands SDK │ ├── agent.py # メインオーケストレーション │ └── browser.py # ブラウザAPI（拡張機能） ├── api/ │ └── main.py # FastAPIエンドポイント ├── scripts/ │ └── generated/ # 自動生成モニタリングスクリプト ├── .env ├── requirements.txt └── README.md ``` ## 環境設定 仮想環境を作成し、必要な依存関係をインストールします: ```none python -m venv venv source venv/bin/activate # Windows: venvScriptsactivate pip install requests fastapi uvicorn python-dotenv pydantic openai beautifulsoup4 playwright openhands-sdk openhands-tools ``` API認証情報を保存する`.envファイル`を作成: ```none # Bright Data API トークン (SERP API 用) BRIGHT_DATA_API_TOKEN=your_api_token # Bright Data SERP ゾーン BRIGHT_DATA_SERP_ZONE=your_serp_zone_name # Bright Data Web Unlocker 認証情報 BRIGHT_DATA_CUSTOMER_ID=your_customer_id BRIGHT_DATA_UNLOCKER_ZONE=your_unlocker_zone_name BRIGHT_DATA_UNLOCKER_PASSWORD=your_zone_password # OpenAI (リスク分析用) OPENAI_API_KEY=your_openai_api_key # OpenHands (エージェントスクリプト生成用) # OpenHands Cloudを使用: openhands/claude-sonnet-4-5-20260929 # または独自モデル: anthropic/claude-sonnet-4-5-20260929 LLM_API_KEY=your_llm_api_key LLM_MODEL=openhands/claude-sonnet-4-5-20260929 ``` ## Bright Dataの設定 ### ステップ1: Bright Dataアカウントの作成 [Bright Data](/)にサインアップし、ダッシュボードに移動します。 ### ステップ2: SERP APIゾーンの設定 1. **プロキシ＆スクレイピングインフラ**に移動 2. 「**追加**」をクリックし、**SERP API**を選択 3. ゾーンに名前を付けます（例:`tprm_serp`） 4. **設定**＞**API**トークンからゾーン名とAPI**トークンを**コピーしてメモしてください SERP APIはGoogleからブロックされずに構造化された検索結果を返します。解析済みJSON出力を得るには、検索URLに`brd_json=1`を追加してください。 ![A Bright Data Dashboard SERP API ](https://media.brightdata.com/2025/12/SkNJOhufWg.png) ### ステップ3: Web Unlockerゾーンの設定 1. **\[追加\]**をクリックし、**Web Unlocker**を選択 2. ゾーンに名前を付けます（例：`tprm_unlocker`） 3. ゾーン認証情報をコピー（ユーザー名形式：`brd-customer-CUSTOMER_ID-zone-ZONE_NAME`） Web Unlockerはプロキシエンドポイントを通じてCAPTCHA、フィンガープリンティング、IPローテーションを自動的に処理します。 ![A Bright Data Dashboard Web Unlocker API](https://media.brightdata.com/2025/12/Hksdu2uGWx.png) ## ディスカバリー層の構築（SERP API） ディスカバリーレイヤーはSERP APIを使用してベンダーに関するネガティブメディアをGoogleで検索します。`src/discovery.py`を作成: ```none import requests from typing import Optional from dataclasses import dataclass from urllib.parse import quote_plus from config import settings @dataclass class SearchResult: title: str url: str snippet: str source: str class DiscoveryClient: """Bright Data SERP API (Direct API) を使用して不利なメディアを検索します。""" RISK_CATEGORIES = { "litigation": ["訴訟", "訴訟", "訴えられた", "裁判", "法的措置"], "financial": ["破産", "債務超過", "債務", "財政難", "債務不履行"], "fraud": ["詐欺", "詐欺行為", "調査", "起訴", "スキャンダル"], "regulatory": ["違反", "罰金", "制裁", "コンプライアンス", "規制当局"], "operational": ["recall", "safety issue", "supply chain", "disruption"], } def __init__(self): self.api_url = "https://api.brightdata.com/request" self.headers = { "Content-Type": "application/json", "Authorization": f"Bearer {settings.BRIGHT_DATA_API_TOKEN}", } def _build_queries(self, vendor_name: str, categories: Optional[list] = None) -> list[str]: """各リスクカテゴリの検索クエリを構築します。""" categories = categories or list(self.RISK_CATEGORIES.keys()) queries = [] for category in categories: keywords = self.RISK_CATEGORIES.get(category, []) keyword_str = " OR ".join(keywords) query = f'"{vendor_name}" ({keyword_str})' queries.append(query) return queries def search(self, query: str) -> list[SearchResult]: """Bright Data SERP APIを使用して単一の検索クエリを実行します。""" try: # パース済みJSON用brd_json=1付きGoogle検索URLを構築 encoded_query = quote_plus(query) google_url = f"https://www.google.com/search?q={encoded_query}&hl=en&gl=us&brd_json=1" payload = { "ゾーン": settings.BRIGHT_DATA_SERP_ゾーン, "url": google_url, "format": "raw", } response = requests.post( self.api_url, headers=self.headers, json=payload, timeout=30, ) response.raise_for_status() data = response.json() results = [] organic = data.get("organic", []) for item in organic: results.append( SearchResult( title=item.get("title", ""), url=item.get("link", ""), snippet=item.get("description", ""), source=item.get("displayed_link", ""), ) ) return results except Exception as e: print(f"検索エラー: {e}") return [] def discover_adverse_media( self, vendor_name: str, categories: Optional[list] = None, ) -> dict[str, list[SearchResult]]: """すべてのリスクカテゴリにわたる悪意のあるメディアを検索します。""" queries = self._build_queries(vendor_name, categories) category_names = categories or list(self.RISK_CATEGORIES.keys()) categorized_results = {} for category, query in zip(category_names, queries): print(f" 検索中: {category}...") results = self.search(query) categorized_results[category] = results return categorized_results def filter_relevant_results( self, results: dict[str, list[SearchResult]], vendor_name: str ) -> dict[str, list[SearchResult]]: """関連性のない結果を除外する。""" filtered = {} vendor_lower = vendor_name.lower() for category, items in results.items(): relevant = [] for item in items: if ( vendor_lower in item.title.lower() or vendor_lower in item.snippet.lower() ): relevant.append(item) filtered[category] = relevant return filtered ``` SERP APIはオーガニック検索結果を構造化されたJSON形式で返すため、各検索結果のタイトル、URL、スニペットを簡単にパースできます。 ## アクセス層の構築（Web Unlocker） ディスカバリー層が関連URLを発見すると、アクセス層はWeb Unlocker APIを使用して完全なコンテンツを取得します。`src/access.py`を作成: ```none import requests from bs4 import BeautifulSoup from dataclasses import dataclass from typing import Optional from config import settings @dataclass class ExtractedContent: url: str title: str text: str publish_date: Optional[str] author: Optional[str] success: bool error: Optional[str] = None class AccessClient: """Bright Data Web Unlocker (APIベース)を使用して保護されたコンテンツにアクセスします。""" def __init__(self): self.api_url = "https://api.brightdata.com/request" self.headers = { "Content-Type": "application/json", "Authorization": f"Bearer {settings.BRIGHT_DATA_API_TOKEN}", } def fetch_url(self, url: str) -> ExtractedContent: """Web Unlocker APIを使用してURLからコンテンツを取得・抽出します。""" try: payload = { "ゾーン": settings.BRIGHT_DATA_UNLOCKER_ZONE, "url": url, "format": "raw", } response = requests.post( self.api_url, headers=self.headers, json=payload, timeout=60, ) response.raise_for_status() # Web Unlocker APIはHTMLを直接返す html_content = response.text content = self._extract_content(html_content, url) return content except requests.Timeout: return ExtractedContent( url=url, title="", text="", publish_date=None, author=None, success=False, error="Request timed out", ) except Exception as e: return ExtractedContent( url=url, title="", text="", publish_date=None, author=None, success=False, error=str(e), ) def _extract_content(self, html: str, url: str) -> ExtractedContent: """HTMLから記事コンテンツを抽出する""" soup = BeautifulSoup(html, "html.parser") # 不要な要素を除去 for element in soup(["script", "style", "nav", "footer", "header", "aside"]): element.decompose() # タイトルを抽出 title = "" if soup.title: title = soup.title.string or "" elif soup.find("h1"): title = soup.find("h1").get_text(strip=True) # メインコンテンツを抽出 article = soup.find("article") or soup.find("main") or soup.find("body") text = article.get_text(separator="n", strip=True) if article else "" # テキストの長さを制限 text = text[:10000] if len(text) > 10000 else text # 公開日を取得する publish_date = None date_meta = soup.find("meta", {"property": "article:published_time"}) if date_meta: publish_date = date_meta.get("content") # 著者を抽出しようとする author = None author_meta = soup.find("meta", {"name": "author"}) if author_meta: author = author_meta.get("content") return ExtractedContent( url=url, title=title, text=text, publish_date=publish_date, author=author, success=True, ) def fetch_multiple(self, urls: list[str]) -> list[ExtractedContent]: """複数のURLを順次取得する。""" results = [] for url in urls: print(f" 取得中: {url[:60]}...") content = self.fetch_url(url) if not content.success: print(f" エラー: {content.error}") results.append(content) return results ``` Web UnlockerはCAPTCHA、ブラウザフィンガープリント、IPローテーションを自動的に処理します。プロキシ経由でリクエストをルーティングするだけで、残りはすべて処理されます。 ## アクションレイヤーの構築 (OpenAI + OpenHands SDK) アクションレイヤーは、OpenAIでリスクの深刻度を分析し、[OpenHands SDKでBright](https://docs.openhands.dev/sdk/getting-started)Data Web Unlocker APIを利用する監視スクリプトを生成します。OpenHands SDKはエージェント機能を提供します：エージェントは推論、ファイル編集、コマンド実行を行い、本番環境対応のスクリプトを作成できます。 `src/actions.py` を作成: ```none import os import json from datetime import datetime, UTC from dataclasses import dataclass, asdict from openai import OpenAI from pydantic import SecretStr from openhands.sdk import LLM, Agent, Conversation, Tool from openhands.tools.terminal import TerminalTool from openhands.tools.file_editor import FileEditorTool from config import settings @dataclass class RiskAssessment: vendor_name: str category: str severity: str summary: str key_findings: list[str] sources: list[str] recommended_actions: list[str] assessed_at: str @dataclass class MonitoringScript: vendor_name: str script_path: str urls_monitored: list[str] check_frequency: str created_at: str class ActionsClient: """OpenAIでリスクを分析し、OpenHands SDKで監視スクリプトを生成する。""" def __init__(self): # リスク分析用 OpenAI self.openai_client = OpenAI(api_key=settings.OPENAI_API_KEY) # エージェント型スクリプト生成用 OpenHands self.llm = LLM( model=settings.LLM_MODEL, api_key=SecretStr(settings.LLM_API_KEY), ) self.workspace = os.path.join(os.getcwd(), "scripts", "generated") os.makedirs(self.workspace, exist_ok=True) def analyze_risk( self, vendor_name: str, category: str, content: list[dict], ) -> RiskAssessment: """OpenAIを使用して抽出されたコンテンツのリスク深刻度を分析する。""" content_summary = "nn".join( [f"ソース: {c['url']}nタイトル: {c['title']}nコンテンツ: {c['text'][:2000]}" for c in content] ) prompt = f"""サードパーティリスク評価のため、"{vendor_name}"に関する以下のコンテンツを分析してください。 カテゴリ: {category} コンテンツ: {content_summary} 以下のJSON形式で応答してください: {{ "severity": "low|medium|high|critical", "summary": "調査結果の概要（2～3文）", "key_findings": ["調査結果1", "調査結果2", ...], "recommended_actions": ["推奨措置1", "推奨措置2", ...] }} 考慮事項: - 深刻度は潜在的なビジネス影響に基づくこと - 重大 = 即時対応が必要（進行中の不正、破産申請） - 高 = 調査が必要な重大なリスク - 中 = 監視に値する顕著な懸念 - 低 = 軽微な問題または過去の事象 """ response = self.openai_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, ) response_text = response.choices[0].message.content try: result = json.loads(response_text) except (json.JSONDecodeError, ValueError): result = { "severity": "medium", "summary": "リスク評価をパースできませんでした", "key_findings": [], "recommended_actions": ["手動レビューが必要です"], } return RiskAssessment( vendor_name=vendor_name, category=category, severity=result.get("severity", "medium"), summary=result.get("summary", ""), key_findings=result.get("key_findings", []), sources=[c["url"] for c in content], recommended_actions=result.get("recommended_actions", []), assessed_at=datetime.now(UTC).isoformat(), ) def generate_monitoring_script( self, vendor_name: str, urls: list[str], check_keywords: list[str], ) -> MonitoringScript: """OpenHands SDK エージェントを使用した Python 監視スクリプトを生成します。""" script_name = f"monitor_{vendor_name.lower().replace(' ', '_')}.py" script_path = os.path.join(self.workspace, script_name) prompt = f"""{script_path} に以下の機能を持つ Python 監視スクリプトを作成します: 1. 毎日これらの URL をチェックし、新規コンテンツを確認: {urls[:5]} 2. 以下のキーワードを検索: {check_keywords} 3. 関連する新規コンテンツが見つかった場合、アラートを送信（コンソールに出力） 4. 全てのチェック結果を'monitoring_log.json'という名前のJSONファイルに記録 スクリプトは必ずBright Data Web Unlocker APIを使用してペイウォールとCAPTCHAを回避すること: - APIエンドポイント: https://api.brightdata.com/request - Bearerトークンには環境変数BRIGHT_DATA_API_TOKENを使用 - ゾーン名には環境変数BRIGHT_DATA_UNLOCKER_ZONEを使用 - JSONペイロード付きPOSTリクエストを送信: {{"zone": "zone_name", "url": "target_url", "format": "raw"}} - ヘッダーを追加: "Authorization": "Bearer " - ヘッダーを追加: "Content-Type": "application/json" スクリプトは以下を実行する: - python-dotenvを使用して環境変数からBright Data認証情報を読み込む - 全てのHTTPリクエストにBright Data Web Unlocker APIを使用（単純なrequests.getではない） - try/exceptでエラーを適切に処理 - 直接実行可能なmain()関数を実装 - cronによるスケジュール実行をサポート - 変更検出のためのコンテンツハッシュを保存 完全なスクリプトを{script_path}に記述してください。 """ # ターミナルとファイルエディタツールを備えたOpenHandsエージェントを作成 agent = Agent( llm=self.llm, tools=[ Tool(name=TerminalTool.name), Tool(name=FileEditorTool.name), ], ) # エージェントを実行してスクリプトを生成 conversation = Conversation(agent=agent, workspace=self.workspace) conversation.send_message(prompt) conversation.run() return MonitoringScript( vendor_name=vendor_name, script_path=script_path, urls_monitored=urls[:5], check_frequency="daily", created_at=datetime.now(UTC).isoformat(), ) def export_assessment(self, assessment: RiskAssessment, output_path: str) -> None: """リスク評価をJSONファイルにエクスポートする。""" with open(output_path, "w") as f: json.dump(asdict(assessment), f, indent=2) ``` OpenHands SDKを単純なプロンプトベースのコード生成よりも使用する主な利点は、エージェントが作業を反復できる点です。スクリプトをテストし、エラーを修正し、正しく動作するまで改良を重ねることができます。 ## エージェントのオーケストレーション それではすべてを組み合わせてみましょう。`src/agent.py`を作成します: ```none from dataclasses import dataclass from datetime import datetime, UTC from typing import Optional from discovery import DiscoveryClient, SearchResult from access import AccessClient, ExtractedContent from actions import ActionsClient, RiskAssessment, MonitoringScript @dataclass class InvestigationResult: vendor_name: str started_at: str completed_at: str total_sources_found: int total_sources_accessed: int risk_assessments: list[RiskAssessment] monitoring_scripts: list[MonitoringScript] errors: list[str] class TPRMAgent: """サードパーティリスク管理調査のための自律エージェント。""" def __init__(self): self.discovery = DiscoveryClient() self.access = AccessClient() self.actions = ActionsClient() def investigate( self, vendor_name: str, categories: Optional[list[str]] = None, generate_monitors: bool = True, ) -> InvestigationResult: """ベンダーの完全な調査を実行する。""" started_at = datetime.now(UTC).isoformat() errors = [] risk_assessments = [] monitoring_scripts = [] # ステージ1: 発見 (SERP API) print(f"[発見] {vendor_name}に関するネガティブメディアを検索中...") try: raw_results = self.discovery.discover_adverse_media(vendor_name, categories) filtered_results = self.discovery.filter_relevant_results(raw_results, vendor_name) except Exception as e: errors.append(f"調査失敗: {str(e)}") return InvestigationResult( vendor_name=vendor_name, started_at=started_at, completed_at=datetime.now(UTC).isoformat(), total_sources_found=0, total_sources_accessed=0, risk_assessments=[], monitoring_scripts=[], errors=errors, ) total_sources = sum(len(results) for results in filtered_results.values()) print(f"[Discovery] 関連ソース {total_sources} 件を発見") # ステージ2: アクセス (Web Unlocker) print(f"[Access] ソースからコンテンツを抽出中...") all_urls = [] url_to_category = {} for category, results in filtered_results.items(): for result in results: all_urls.append(result.url) url_to_category[result.url] = category try: extracted_content = self.access.fetch_multiple(all_urls) successful_extractions = [c for c in extracted_content if c.success] except Exception as e: error_msg = f"アクセス失敗: {str(e)}" print(f"[アクセス] {error_msg}") errors.append(error_msg) successful_extractions = [] print(f"[アクセス] {len(successful_extractions)} 件のソースを正常に抽出しました") # ステージ3: アクション - リスク分析 (OpenAI) print(f"[アクション] リスク分析中...") category_content = {} for content in successful_extractions: category = url_to_category.get(content.url, "unknown") if category not in category_content: category_content[category] = [] category_content[category].append({ "url": content.url, "title": content.title, "text": content.text, }) for category, content_list in category_content.items(): if not content_list: continue try: assessment = self.actions.analyze_risk(vendor_name, category, content_list) risk_assessments.append(assessment) except Exception as e: errors.append(f"{category} のリスク分析に失敗しました: {str(e)}") # ステージ 3: アクション - 監視スクリプトの生成 if generate_monitors and successful_extractions: print(f"[アクション] 監視スクリプトを生成中...") try: urls_to_monitor = [c.url for c in successful_extractions[:10]] keywords = [vendor_name, "lawsuit", "bankruptcy", "fraud"] script = self.actions.generate_monitoring_script( vendor_name, urls_to_monitor, keywords ) monitoring_scripts.append(script) except Exception as e: errors.append(f"スクリプト生成失敗: {str(e)}") completed_at = datetime.now(UTC).isoformat() print(f"[Complete] 調査完了") return InvestigationResult( vendor_name=vendor_name, started_at=started_at, completed_at=completed_at, total_sources_found=total_sources, total_sources_accessed=len(successful_extractions), risk_assessments=risk_assessments, monitoring_scripts=monitoring_scripts, errors=errors, ) def main(): """使用例。""" agent = TPRMAgent() result = agent.investigate("Acme Corp") print(f"n{'='*50}") print(f"調査完了: {result.vendor_name}") print(f"検出ソース数: {result.total_sources_found}") print(f"アクセス済みソース数: {result.total_sources_accessed}") print(f"リスク評価数: {len(result.risk_assessments)}") print(f"監視スクリプト数: {len(result.monitoring_scripts)}") for assessment in result.risk_assessments: print(f"n[{assessment.category.upper()}] 深刻度: {assessment.severity}") print(f"概要: {assessment.summary}") if __name__ == "__main__": main() ``` エージェントは3つのレイヤーすべてを調整し、エラーを適切に処理して包括的な調査結果を生成します。 ## 設定 アプリケーションを正常に実行するために必要なすべてのシークレットとキーを設定するため、`src/config.py`を作成します: ```none import os from dotenv import load_dotenv load_dotenv() class Settings: # SERP API BRIGHT_DATA_API_TOKEN: str = os.getenv("BRIGHT_DATA_API_TOKEN", "") BRIGHT_DATA_SERP_ZONE: str = os.getenv("BRIGHT_DATA_SERP_ZONE", "") # Web Unlocker BRIGHT_DATA_CUSTOMER_ID: str = os.getenv("BRIGHT_DATA_CUSTOMER_ID", "") BRIGHT_DATA_UNLOCKER_ZONE: str = os.getenv("BRIGHT_DATA_UNLOCKER_ZONE", "") BRIGHT_DATA_UNLOCKER_PASSWORD: str = os.getenv("BRIGHT_DATA_UNLOCKER_PASSWORD", "") # OpenAI (リスク分析用) OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "") # OpenHands（エージェントスクリプト生成用） LLM_API_KEY: str = os.getenv("LLM_API_KEY", "") LLM_MODEL: str = os.getenv("LLM_MODEL", "openhands/claude-sonnet-4-5-20260929") settings = Settings() ``` ## APIレイヤーの構築 [FastAPI](https://fastapi.tiangolo.com/)を使用して、RESTエンドポイント経由でエージェントを公開する`api/main.py`を作成します: ```none from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel from typing import Optional import uuid import sys sys.path.insert(0, 'src') from agent import TPRMAgent, InvestigationResult app = FastAPI( title="TPRM Agent API", description="自律型サードパーティリスク管理エージェント", version="1.0.0",) investigations: dict[str, InvestigationResult] = {} agent = TPRMAgent() class InvestigationRequest(BaseModel): vendor_name: str categories: Optional[list[str]] = None generate_monitors: bool = True class InvestigationResponse(BaseModel): investigation_id: str status: str message: str @app.post("/investigate", response_model=InvestigationResponse) def start_investigation( request: InvestigationRequest, background_tasks: BackgroundTasks, ): """新規ベンダー調査を開始します。""" investigation_id = str(uuid.uuid4()) def run_investigation(): result = agent.investigate( vendor_name=request.vendor_name, categories=request.categories, generate_monitors=request.generate_monitors, ) investigations[investigation_id] = result background_tasks.add_task(run_investigation) return InvestigationResponse( investigation_id=investigation_id, status="started", message=f"{request.vendor_name} の調査を開始しました", ) @app.get("/investigate/{investigation_id}") def get_investigation(investigation_id: str): """調査結果を取得します。""" if investigation_id not in investigations: raise HTTPException(status_code=404, detail="調査が見つかりません、またはまだ進行中です") return investigations[investigation_id] @app.get("/reports/{vendor_name}") def get_reports(vendor_name: str): """ベンダーの全レポートを取得します。""" vendor_reports = [ result for result in investigations.values() if result.vendor_name.lower() == vendor_name.lower() ] if not vendor_reports: raise HTTPException(status_code=404, detail="このベンダーのレポートが見つかりません") return vendor_reports @app.get("/health") def health_check(): """ヘルスチェックエンドポイント。""" return {"status": "healthy"} ``` APIをローカルで実行: ```none python -m uvicorn API.main:app --reload ``` インタラクティブなAPIドキュメントを閲覧するには`http://localhost:8000/docs`にアクセスしてください。 ![FastAPI DOcumentation](https://media.brightdata.com/2025/12/Sy2lY2_GWe.png) ## Browser API（スクレイピングブラウザ）による機能強化 フォーム送信が必要な裁判所登録システムやJavaScript多用サイトなど複雑なシナリオでは、Bright DataのBrowser API（スクレイピングブラウザ）でエージェントを強化できます。Web Unlocker APIやSERP APIと同様の方法で設定可能です。 ![A Bright Data Dashboard Web Unlocker API](https://media.brightdata.com/2025/12/rJWwKndzWl.png) Browser APIは、Chrome DevTools Protocol（CDP）を介したPlaywright経由で制御可能なクラウドホスト型ブラウザを提供します。これは以下のような場合に有用です： - フォーム送信やナビゲーションを必要とする**裁判所登録検索** - 動的コンテンツ読み込みを伴う**JavaScript多用サイト** - **多段階認証フロー** - コンプライアンス文書用の**スクリーンショット取得** ### 設定 `.envにBrowser` API認証情報を追加: ```none # Browser API BRIGHT_DATA_BROWSER_USER: str = os.getenv("BRIGHT_DATA_BROWSER_USER", "") BRIGHT_DATA_BROWSER_PASSWORD: str = os.getenv("BRIGHT_DATA_BROWSER_PASSWORD", "") ``` ### ブラウザクライアントの実装 `src/browser.py` を作成: ```none import asyncio from playwright.async_api import async_playwright from dataclasses import dataclass from typing import Optional from config import settings @dataclass class BrowserContent: url: str title: str text: str screenshot_path: Optional[str] success: bool error: Optional[str] = None class BrowserClient: """Bright Data Browser API (スクレイピングブラウザ) を使用して動的コンテンツにアクセスします。 以下の用途に使用します: - 完全なレンダリングが必要な JavaScript ヘビーなサイト - 複数ステップのフォーム (例: 裁判所登録検索) - クリック、スクロール、またはインタラクションが必要なサイト - コンプライアンス文書用のスクリーンショット取得 """ def __init__(self): # CDP接続用WebSocketエンドポイント構築 auth = f"{settings.BRIGHT_DATA_BROWSER_USER}:{settings.BRIGHT_DATA_BROWSER_PASSWORD}" self.endpoint_url = f"wss://{auth}@brd.superproxy.io:9222" async def fetch_dynamic_page( self, url: str, wait_for_selector: Optional[str] = None, take_screenshot: bool = False, screenshot_path: Optional[str] = None, ) -> BrowserContent: """Browser APIを使用して動的ページからコンテンツを取得します。""" async with async_playwright() as playwright: try: print(f"Bright Data スクレイピングブラウザに接続中...") browser = await playwright.chromium.connect_over_cdp(self.endpoint_url) try: page = await browser.new_page() print(f"{url} に移動中...") await page.goto(url, timeout=120000) # 指定されたセレクタを待機（指定されている場合） if wait_for_selector: await page.wait_for_selector(wait_for_selector, timeout=30000) # ページコンテンツを取得 title = await page.title() # テキストを抽出 text = await page.evaluate("() => document.body.innerText") # スクリーンショットを要求された場合取得 if take_screenshot and screenshot_path: await page.screenshot(path=screenshot_path, full_page=True) return BrowserContent( url=url, title=title, text=text[:10000], screenshot_path=screenshot_path if take_screenshot else None, success=True, ) finally: await browser.close() except Exception as e: return BrowserContent( url=url, title="", text="", screenshot_path=None, success=False, error=str(e), ) async def fill_and_submit_form( self, url: str, form_data: dict[str, str], submit_selector: str, result_selector: str, ) -> BrowserContent: """フォームに入力し結果を取得 - 裁判所の登録業務に有用""" async with async_playwright() as playwright: try: browser = await playwright.chromium.connect_over_cdp(self.endpoint_url) try: page = await browser.new_page() await page.goto(url, timeout=120000) # フォームフィールド入力 for selector, value in form_data.items(): await page.fill(selector, value) # フォーム送信 await page.click(submit_selector) # 結果待ち await page.wait_for_selector(result_selector, timeout=30000) title = await page.title() text = await page.evaluate("() => document.body.innerText") return BrowserContent( url=url, title=title, text=text[:10000], screenshot_path=None, success=True, ) finally: await browser.close() except Exception as e: return BrowserContent( url=url, title="", text="", screenshot_path=None, success=False, error=str(e), ) async def scroll_and_collect( self, url: str, scroll_count: int = 5, wait_between_scrolls: float = 1.0, ) -> BrowserContent: """無限スクロールページを処理する。""" async with async_playwright() as playwright: try: browser = await playwright.chromium.connect_over_cdp(self.endpoint_url) try: page = await browser.new_page() await page.goto(url, timeout=120000) # 複数回スクロールダウン for i in range(scroll_count): await page.evaluate("window.scrollTo(0, document.body.scrollHeight)") await asyncio.sleep(wait_between_scrolls) title = await page.title() text = await page.evaluate("() => document.body.innerText") return BrowserContent( url=url, title=title, text=text[:10000], screenshot_path=None, success=True, ) finally: await browser.close() except Exception as e: return BrowserContent( url=url, title="", text="", screenshot_path=None, success=False, error=str(e), ) # 裁判所登録検索の使用例 async def example_court_search(): client = BrowserClient() # 例: 裁判所記録を検索 result = await client.fill_and_submit_form( url="https://example-court-registry.gov/search", form_data={ "#party-name": "Acme Corp", "#case-type": "civil", }, submit_selector="#search-button", result_selector=".search-results", ) if result.success: print(f"裁判記録が見つかりました: {result.text[:500]}") else: print(f"エラー: {result.error}") if __name__ == "__main__": asyncio.run(example_court_search()) ``` ### ブラウザAPIとWeb Unlockerの使い分け シナリオ使用例シンプルなHTTPリクエストWeb Unlocker静的HTMLページWeb Unlocker読み込み時のCAPTCHAWeb UnlockerJavaScriptでレンダリングされたコンテンツブラウザ APIフォーム送信ブラウザ APIマルチステップナビゲーションブラウザ APIスクリーンショットが必要ブラウザ API## Railway を使用したデプロイ TPRMエージェントは、大規模な依存関係を持つPythonアプリケーションをサポートするRailwayまたはRenderを使用して本番環境にデプロイできます。 [Railwayは](https://railway.com/)、OpenHands SDKのような重い依存関係を持つPythonアプリケーションをデプロイする最も簡単な選択肢です。これを利用するには、サインアップしてアカウントを作成する必要があります。 ### ステップ1: Railway CLIをグローバルにインストール ```none npm i -g @railway/cli ``` ### ステップ2:`Procfileファイル`を追加 アプリケーションのルートフォルダに`新しいProcfileファイル`を作成し、以下の内容を追加します。これはデプロイの構成または起動コマンドとして機能します。 ```none web: uvicorn API.main:app --host 0.0.0.0 --port $PORT ``` ### ステップ3: プロジェクトディレクトリでRailwayにログインし初期化 ```none railway login railway init ``` ### ステップ4: デプロイ ```none railway up ``` ![Railway Initialization and Deployment](https://media.brightdata.com/2025/12/rJ36KndGbx.png)### ステップ5: 環境変数の追加 Railwayプロジェクトダッシュボード →**\[設定**\] →**\[共有変数**\] に移動し、以下の変数とその値を追加します: ```none BRIGHT_DATA_API_TOKEN BRIGHT_DATA_SERP_ZONE BRIGHT_DATA_UNLOCKER_ZONE OPENAI_API_KEY LLM_API_KEY LLM_MODEL ``` ![Adding variables to your Railway app](https://media.brightdata.com/2025/12/BkrHc2dGbl.png)Railwayは変更を自動検出し、ダッシュボード上で再デプロイを促します。「デプロイ」をクリックすると、シークレットが適用されたアプリが更新されます。 ![Redeploy Railway app after adding variables](https://media.brightdata.com/2025/12/SkAi5nuG-g.png) 再デプロイ後、サービスカードをクリックし「**設定**」を選択すると、サービスがまだ公開されていないためドメイン生成場所が表示されます。「**ドメインを生成**」をクリックして公開URLを取得してください。 ![Generate domain for Railway application](https://media.brightdata.com/2025/12/Sy8londzWx.png) ## 完全な調査の実行 ### curlを使用したローカル実行 #### FastAPIサーバーを起動: ```none # 仮想環境をアクティブ化 source venv/bin/activate # Windowsの場合: venvScriptsactivate # サーバーを実行 python -m uvicorn api.main:app --reload ``` http://localhost:8000/docs にアクセスしてインタラクティブなAPIドキュメントを閲覧してください。 #### APIリクエストの実行 - 調査を開始: ```none curl -X POST "http://localhost:8000/investigate" -H "Content-Type: application/json" -d '{ "vendor_name": "Acme Corp", "categories": ["litigation", "fraud"], "generate_monitors": true }' ``` - これにより調査IDが返されます: ```none { "investigation_id": "f6af2e0f-991a-4cb7-949e-2f316e677b5c", "status": "started", "message": "Acme Corp の調査が開始されました" } ``` - 調査ステータスの確認: ```none curl http://localhost:8000/investigate/f6af2e0f-991a-4cb7-949e-2f316e677b5c ``` ### エージェントをスクリプトとして実行 プロジェクトルートに`run_investigation.py`というファイルを作成: ```none import sys sys.path.insert(0, 'src') from agent import TPRMAgent def investigate_vendor(): """ベンダーの完全な調査を実行します。""" agent = TPRMAgent() # 調査の実行 result = agent.investigate( vendor_name="Acme Corp", categories=["litigation", "financial", "fraud"], generate_monitors=True, ) # 概要の出力 print(f"n{'='*60}") print(f"調査完了: {result.vendor_name}") print(f"{'='*60}") print(f"発見された情報源: {result.total_sources_found}") print(f"アクセスされた情報源: {result.total_sources_accessed}") print(f"リスク評価: {len(result.risk_assessments)}") print(f"監視スクリプト: {len(result.monitoring_scripts)}") # リスク評価を出力 for assessment in result.risk_assessments: print(f"n{'─'*60}") print(f"[{assessment.category.upper()}] 深刻度: {assessment.severity.upper()}") print(f"{'─'*60}") print(f"概要: {assessment.summary}") print("n主な発見事項:") for finding in assessment.key_findings: print(f" • {finding}") print("n推奨アクション:") for action in assessment.recommended_actions: print(f" → {action}") # 監視スクリプト情報の出力 for script in result.monitoring_scripts: print(f"n{'='*60}") print(f"生成された監視スクリプト") print(f"{'='*60}") print(f"パス: {script.script_path}") print(f"監視対象URL数: {len(script.urls_monitored)}") print(f"頻度: {script.check_frequency}") # エラーがあれば出力 if result.errors: print(f"n{'='*60}") print("エラー:") for error in result.errors: print(f" ⚠️ {error}") if __name__ == "__main__": investigate_vendor() ``` 調査スクリプトを新しいターミナルで実行 ```none # 仮想環境をアクティブ化 source venv/bin/activate # Windowsの場合: venvScriptsactivate # 調査スクリプトを実行 python run_investigation.py ``` エージェントは以下を実行します: 1. SERP APIを使用してGoogleでネガティブメディアを検索 2. Web Unlockerを使用してソースにアクセス 3. OpenAIを使用してコンテンツのリスク深刻度を分析 4. OpenHands SDKを使用してPython監視スクリプトを生成（cronによるスケジュール設定可能） ![Terminal running the tprm-agent app](https://media.brightdata.com/2025/12/Byytsndf-e.png)### 自動生成された監視スクリプトの実行 調査完了後、監視スクリプトが`scripts/generatedフォルダ`に生成されます: ```none cd scripts/generated python monitor_acme_corp.py ``` この監視スクリプトはBright Data Web Unlocker APIを使用して監視対象の全URLをチェックし、以下を出力します: ![Terminal monitoring the script from tprm-agent app](https://media.brightdata.com/2025/12/H1sCo2uGZg.png) ビジネスに関する正確かつ最新の情報を常に取得するため、スクリプトのcronスケジュールを自由に設定できます。 ## まとめ ベンダーのネガティブメディア調査を自動化するエンタープライズ向けTPRMエージェント構築の完全なフレームワークが完成しました。本システムは以下を実現します: - [Bright Data SERP API](/products/serp-api)を使用して複数カテゴリにわたるリスクシグナル**を発見** - [Bright Data Web Unlocker](/products/web-unlocker)でコンテンツ**にアクセス** - OpenAIによるリスク**分析**と[OpenHands SDK](https://docs.openhands.dev/sdk/getting-started)を用いた**監視スクリプト生成** - [複雑なシナリオ向けにBrowser APIで](/products/scraping-browser)機能を**強化** モジュール式アーキテクチャにより拡張が容易です： - `RISK_CATEGORIES辞書を`更新して新たなリスクカテゴリを追加 - APIレイヤーを拡張してGRCプラットフォームと連携 - バックグラウンドタスクキューで数千のベンダーに対応 - ブラウザAPI強化機能による裁判記録検索の追加 ## 次のステップ このエージェントをさらに改善するには、以下を検討してください： - **追加データソースの統合**：SEC提出書類、OFAC制裁リスト、法人登記情報 - **データベース永続化の追加**：調査履歴をPostgreSQLまたはMongoDBに保存 - **Webhook通知の実装**：高リスクベンダーを検出した際にSlackやTeamsへ通知 - **ダッシュボード構築**：ベンダーリスクスコアを可視化するReactフロントエンドを作成 - **自動スキャンのスケジュール設定**：定期的なベンダー監視にCeleryまたはAPSchedulerを使用 ## リソース - [Bright Data SERP API ドキュメント](https://docs.brightdata.com/scraping-automation/serp-api/introduction) - [Bright Data Web Unlocker ドキュメント](https://docs.brightdata.com/scraping-automation/web-unlocker/introduction) - [Bright Data Browser API ドキュメント](https://docs.brightdata.com/scraping-automation/scraping-browser/introduction) - [OpenHands SDK ドキュメント](https://docs.openhands.dev/sdk/getting-started) - [OpenAI API ドキュメント](https://platform.openai.com/docs) - [FastAPI ドキュメント](https://fastapi.tiangolo.com/) お問い合わせ無料トライアル![google social icon](/wp-content/themes/brightdata/assets/images/ic_google.svg) 目次 日本企業向けデータソリューション 日本のトップ企業が信頼するスケーラブルなウェブデータプラットフォーム。日本語24時間サポート、GDPR・個人情報保護法準拠、日本企業20社以上の導入実績。 [詳細を見る](https://brightdata.jp/solutions/data-solutions-for-japanese-companies "詳細を見る") [ ](https://news.ycombinator.com/submitlink?t=Bright+Data%E3%81%A8OpenHands+SDK%E3%81%A7%E3%82%A8%E3%83%B3%E3%82%BF%E3%83%BC%E3%83%97%E3%83%A9%E3%82%A4%E3%82%BATPRM%E3%82%A8%E3%83%BC%E3%82%B8%E3%82%A7%E3%83%B3%E3%83%88%E3%82%92%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B&u=https://brightdata.jp/blog/ai/openhands-sdk-with-bright-data) [ ](https://www.linkedin.com/shareArticle?mini=true&title=Bright+Data%E3%81%A8OpenHands+SDK%E3%81%A7%E3%82%A8%E3%83%B3%E3%82%BF%E3%83%BC%E3%83%97%E3%83%A9%E3%82%A4%E3%82%BATPRM%E3%82%A8%E3%83%BC%E3%82%B8%E3%82%A7%E3%83%B3%E3%83%88%E3%82%92%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B&url=https://brightdata.jp/blog/ai/openhands-sdk-with-bright-data) [ ](http://www.reddit.com/submit?title=Bright+Data%E3%81%A8OpenHands+SDK%E3%81%A7%E3%82%A8%E3%83%B3%E3%82%BF%E3%83%BC%E3%83%97%E3%83%A9%E3%82%A4%E3%82%BATPRM%E3%82%A8%E3%83%BC%E3%82%B8%E3%82%A7%E3%83%B3%E3%83%88%E3%82%92%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B&url=https://brightdata.jp/blog/ai/openhands-sdk-with-bright-data) ## あなたは下記にもご興味がおありかもしれません [ ![OpenHuman with Bright Data](https://media.brightdata.jp/2026/09/OpenHuman-with-Bright-Data.png) ](https://brightdata.jp/blog/ai/openhuman-with-bright-data "Bright Data CLIを通じたOpenHumanにおける本番対応ウェブアクセス") [AI ![Antonello Zanini](https://media.brightdata.jp/2022/12/Antonello-Zanini-2-50x50.jpg) Antonello Zanini Technical Writer ### Bright Data CLIを通じたOpenHumanにおける本番対応ウェブアクセス Bright Data CLIをOpenHumanに統合して、AIエージェント向けの本番対応ウェブアクセスとデータ収集を実現します。 09-Sep-2026 3 分読 ](https://brightdata.jp/blog/ai/openhuman-with-bright-data) [ ![Multimodal Web Scraping with MiniMax](https://media.brightdata.jp/2026/09/Multimodal-Web-Scraping-with-MiniMax.png) ](https://brightdata.jp/blog/%e3%82%a6%e3%82%a7%e3%83%96%e3%83%87%e3%83%bc%e3%82%bf/multimodal-web-scraping-with-minimax "MiniMaxによるマルチモーダルウェブスクレイピング") [ウェブデータ ![Antonello Zanini](https://media.brightdata.jp/2022/12/Antonello-Zanini-2-50x50.jpg) Antonello Zanini Technical Writer ### MiniMaxによるマルチモーダルウェブスクレイピング Bright Data Web UnlockerとMiniMax M3ビジョンを組み合わせて、画像やウェブページのスクリーンショットから構造化データを抽出します。 09-Sep-2026 1 分読 ](https://brightdata.jp/blog/%e3%82%a6%e3%82%a7%e3%83%96%e3%83%87%e3%83%bc%e3%82%bf/multimodal-web-scraping-with-minimax) [ ![The 10 Best CLI Tools for Codex in 2026](https://media.brightdata.jp/2026/08/The-10-Best-CLI-Tools-for-Codex-in-2026.png) ](https://brightdata.jp/blog/ai/best-cli-tools-for-codex "2026年版 Codex向けCLIツール10選 – テスト＆ランキング") [AI ![Daniel Shashko](https://media.brightdata.jp/2022/04/Daniel-Shashko-2-50x50.png) Daniel Shashko Web Data & AI Expert ### 2026年版 Codex向けCLIツール10選 – テスト＆ランキング Codexをより速く、より有能にする10のCLIツール。サンドボックス内から実際のウェブアクセスを提供するBright Data CLIから始まります。 06-Sep-2026 4 分読 ](https://brightdata.jp/blog/ai/best-cli-tools-for-codex)