はじめに
Model Context Protocol (MCP) を活用するLLMアプリケーションでは、様々なデータソースからコンテンツを取得します。これらのコンテンツには、それぞれ利用規約やライセンス条件が存在する場合があり、適切な管理が必要です。
本記事では、MCPを使用する際のライセンス情報の取り扱い方法と、利用規約遵守をサポートする実装アプローチについて考察します。
重要な注意： 本記事は技術的な実装例を示すものであり、法的助言ではありません。実際の運用にあたっては、必ず法務専門家にご相談ください。
1. ライセンス管理の課題
1.1. LLMアプリケーションにおける課題
LLMアプリケーションが複数のデータソースを利用する際、以下のような課題があります。
| 課題 | 詳細 | 影響 |
|---|---|---|
| ライセンスの多様性 | データソースごとに異なる利用規約 | 管理の複雑化 |
| 利用目的の追跡 | コンテンツがどのように使用されたか | 規約違反のリスク |
| 自動化の困難さ | 人手による確認の必要性 | スケーラビリティの制約 |
| 権利者への報告 | 利用状況の透明性確保 | 信頼関係の構築 |
1.2. 従来のアプローチの限界
手動管理の問題点：
- データソースごとに利用規約を確認する必要がある
- 開発者が規約を理解し、コードに反映させる必要がある
- 規約変更時の追従が困難
- 大規模なシステムでは実質的に管理不可能
必要なアプローチ：
- ライセンス情報のメタデータ化
- システムによる自動チェックのサポート
- 利用状況の記録と監査
2. MCPにおけるライセンス情報の表現
2.1. 現在のMCP仕様
MCPの標準仕様では、リソースに対して以下のようなメタデータを付与できます。
{
"uri": "file:///path/to/document.pdf",
"name": "Company Annual Report 2024",
"description": "Annual financial report",
"mimeType": "application/pdf"
}
ライセンス情報の追加方法：
標準仕様に含まれていませんが、カスタムメタデータとしてライセンス情報を追加できます。
{
"uri": "file:///path/to/document.pdf",
"name": "Company Annual Report 2024",
"mimeType": "application/pdf",
"annotations": {
"license": {
"type": "CC-BY-4.0",
"terms": "https://creativecommons.org/licenses/by/4.0/",
"attribution": "Company Name",
"commercialUse": true,
"modifications": true,
"shareAlike": false
}
}
}
2.2. ライセンスメタデータの設計
実用的なライセンス情報には、以下のような属性を含めることができます。
基本情報：
{
"license": {
"id": "license-001",
"type": "CC-BY-NC-4.0",
"url": "https://creativecommons.org/licenses/by-nc/4.0/",
"summary": "Non-commercial use only with attribution"
}
}
詳細な制約条件：
{
"license": {
"permissions": {
"commercial": false,
"modification": true,
"distribution": true,
"training": false
},
"requirements": {
"attribution": true,
"shareAlike": true,
"notice": true
},
"limitations": {
"liability": true,
"warranty": true
},
"validUntil": "2025-12-31T23:59:59Z",
"territorialScope": ["JP", "US", "EU"]
}
}
3. ライセンスチェックの実装
3.1. サーバーサイドでのチェック
MCPサーバー側でライセンス情報を管理し、アクセス時にチェックする方法です。
実装例：
from dataclasses import dataclass
from typing import Optional, List
from datetime import datetime
@dataclass
class License:
"""ライセンス情報を表すクラス"""
type: str
commercial_use: bool
ai_training: bool
attribution_required: bool
valid_until: Optional[datetime] = None
territorial_scope: Optional[List[str]] = None
def is_valid(self) -> bool:
"""ライセンスが有効期限内かチェック"""
if self.valid_until:
return datetime.now() < self.valid_until
return True
def allows_commercial(self) -> bool:
"""商用利用が許可されているかチェック"""
return self.commercial_use
def allows_training(self) -> bool:
"""AI学習利用が許可されているかチェック"""
return self.ai_training
class LicenseChecker:
"""ライセンス条件をチェックするクラス"""
def __init__(self):
self.licenses = {} # リソースIDとライセンスのマッピング
def register_license(self, resource_id: str, license: License):
"""リソースのライセンス情報を登録"""
self.licenses[resource_id] = license
def check_access(
self,
resource_id: str,
purpose: str,
user_context: dict
) -> tuple[bool, str]:
"""
アクセス可否をチェック
Returns:
(許可/拒否, 理由)
"""
# ライセンス情報の取得
license = self.licenses.get(resource_id)
if not license:
return True, "No license restrictions"
# 有効期限チェック
if not license.is_valid():
return False, "License expired"
# 用途チェック
if purpose == "commercial":
if not license.allows_commercial():
return False, "Commercial use not allowed"
if purpose == "training":
if not license.allows_training():
return False, "AI training use not allowed"
# 地域制限チェック
if license.territorial_scope:
user_region = user_context.get('region')
if user_region not in license.territorial_scope:
return False, f"Access not allowed from {user_region}"
return True, "Access granted"
# 使用例
checker = LicenseChecker()
# ライセンス登録
license = License(
type="CC-BY-NC-4.0",
commercial_use=False,
ai_training=False,
attribution_required=True,
valid_until=datetime(2025, 12, 31)
)
checker.register_license("resource-001", license)
# アクセスチェック
allowed, reason = checker.check_access(
resource_id="resource-001",
purpose="commercial",
user_context={"region": "JP"}
)
if not allowed:
print(f"Access denied: {reason}")
3.2. クライアントサイドでのチェック
MCPクライアント（LLMアプリケーション）側でライセンス情報を確認し、適切に処理する方法です。
class MCPClientWithLicenseCheck:
"""ライセンスチェック機能付きMCPクライアント"""
def __init__(self):
self.usage_log = []
async def get_resource_with_license(
self,
resource_uri: str,
intended_use: str
):
"""
リソースをライセンス情報と共に取得
"""
# MCPサーバーからリソースとメタデータを取得
resource = await self.fetch_resource(resource_uri)
# ライセンス情報の抽出
license_info = resource.get('annotations', {}).get('license')
if not license_info:
# ライセンス情報がない場合は警告
self.log_warning(f"No license info for {resource_uri}")
return resource
# 利用目的とライセンスの整合性チェック
if not self._check_license_compatibility(license_info, intended_use):
raise LicenseViolationError(
f"Intended use '{intended_use}' not allowed by license"
)
# 利用ログに記録
self._log_usage(resource_uri, license_info, intended_use)
# 帰属表示が必要な場合は追加
if license_info.get('requirements', {}).get('attribution'):
resource['attribution'] = license_info.get('attribution')
return resource
def _check_license_compatibility(
self,
license_info: dict,
intended_use: str
) -> bool:
"""ライセンスと利用目的の整合性をチェック"""
permissions = license_info.get('permissions', {})
if intended_use == 'commercial':
return permissions.get('commercial', False)
if intended_use == 'training':
return permissions.get('training', False)
return True
def _log_usage(
self,
resource_uri: str,
license_info: dict,
intended_use: str
):
"""利用状況をログに記録"""
self.usage_log.append({
'timestamp': datetime.now().isoformat(),
'resource': resource_uri,
'license': license_info.get('type'),
'intended_use': intended_use
})
4. 利用目的の追跡
4.1. コンテキストタグの活用
リソースが取得される際に、利用目的をタグとして付与します。
class ResourceUsageTracker:
"""リソース利用状況の追跡"""
def __init__(self):
self.usage_records = []
def track_usage(
self,
resource_id: str,
license_type: str,
context: dict
):
"""
リソースの利用を記録
context: {
'purpose': 'response_generation' | 'training' | 'analysis',
'user_id': 'user-001',
'application': 'chatbot',
'timestamp': '2024-01-01T00:00:00Z'
}
"""
record = {
'resource_id': resource_id,
'license_type': license_type,
'purpose': context.get('purpose'),
'user_id': context.get('user_id'),
'application': context.get('application'),
'timestamp': context.get('timestamp', datetime.now().isoformat())
}
self.usage_records.append(record)
# ライセンス要件に応じた処理
self._check_usage_limits(resource_id, license_type)
def _check_usage_limits(self, resource_id: str, license_type: str):
"""利用回数制限などをチェック"""
# 同一リソースの利用回数をカウント
usage_count = sum(
1 for r in self.usage_records
if r['resource_id'] == resource_id
)
# ライセンスに応じた制限チェック
if license_type == 'limited':
if usage_count > 100:
raise UsageLimitExceeded(
f"Usage limit exceeded for {resource_id}"
)
def generate_report(self, resource_id: str = None) -> dict:
"""利用レポートを生成"""
records = self.usage_records
if resource_id:
records = [
r for r in records
if r['resource_id'] == resource_id
]
return {
'total_usage': len(records),
'by_purpose': self._group_by(records, 'purpose'),
'by_license': self._group_by(records, 'license_type'),
'records': records
}
def _group_by(self, records: list, key: str) -> dict:
"""レコードをキーでグループ化"""
result = {}
for record in records:
value = record.get(key, 'unknown')
result[value] = result.get(value, 0) + 1
return result
4.2. 監査ログの生成
ライセンス関連のすべての操作を記録します。
import json
import hashlib
from datetime import datetime
class LicenseAuditLogger:
"""ライセンス監査ログ"""
def __init__(self, log_file: str = "license_audit.jsonl"):
self.log_file = log_file
def log_access_attempt(
self,
resource_id: str,
user_id: str,
license_type: str,
granted: bool,
reason: str
):
"""アクセス試行を記録"""
log_entry = {
'event': 'access_attempt',
'timestamp': datetime.now().isoformat(),
'resource_id': resource_id,
'user_id': user_id,
'license_type': license_type,
'granted': granted,
'reason': reason,
'hash': None # 後で計算
}
# 改竄防止のためのハッシュ
log_entry['hash'] = self._calculate_hash(log_entry)
self._write_log(log_entry)
def log_license_check(
self,
resource_id: str,
checks_performed: list,
result: dict
):
"""ライセンスチェックの詳細を記録"""
log_entry = {
'event': 'license_check',
'timestamp': datetime.now().isoformat(),
'resource_id': resource_id,
'checks': checks_performed,
'result': result,
'hash': None
}
log_entry['hash'] = self._calculate_hash(log_entry)
self._write_log(log_entry)
def _calculate_hash(self, log_entry: dict) -> str:
"""ログエントリのハッシュを計算"""
# hashフィールドを除外してハッシュ計算
entry_copy = {k: v for k, v in log_entry.items() if k != 'hash'}
entry_str = json.dumps(entry_copy, sort_keys=True)
return hashlib.sha256(entry_str.encode()).hexdigest()
def _write_log(self, log_entry: dict):
"""ログファイルに書き込み"""
with open(self.log_file, 'a') as f:
f.write(json.dumps(log_entry) + '\n')
def verify_log_integrity(self) -> bool:
"""ログの整合性を検証"""
with open(self.log_file, 'r') as f:
for line in f:
entry = json.loads(line)
stored_hash = entry.get('hash')
calculated_hash = self._calculate_hash(entry)
if stored_hash != calculated_hash:
print(f"Log integrity violation detected!")
return False
return True
5. 実装上の考慮事項
5.1. パフォーマンスへの影響
ライセンスチェックは、リソース取得のたびに実行されるため、パフォーマンスへの影響を考慮する必要があります。
最適化手法：
from functools import lru_cache
from datetime import datetime, timedelta
class CachedLicenseChecker:
"""キャッシュ機能付きライセンスチェッカー"""
def __init__(self, cache_ttl_seconds: int = 300):
self.cache_ttl = timedelta(seconds=cache_ttl_seconds)
self.cache = {}
def check_license(self, resource_id: str, context: dict) -> tuple[bool, str]:
"""キャッシュを使用したライセンスチェック"""
cache_key = self._make_cache_key(resource_id, context)
# キャッシュ確認
if cache_key in self.cache:
cached_result, cached_time = self.cache[cache_key]
if datetime.now() - cached_time < self.cache_ttl:
return cached_result
# 実際のチェック
result = self._perform_license_check(resource_id, context)
# キャッシュに保存
self.cache[cache_key] = (result, datetime.now())
return result
def _make_cache_key(self, resource_id: str, context: dict) -> str:
"""キャッシュキーを生成"""
relevant_context = {
'purpose': context.get('purpose'),
'user_id': context.get('user_id')
}
return f"{resource_id}:{json.dumps(relevant_context, sort_keys=True)}"
def _perform_license_check(
self,
resource_id: str,
context: dict
) -> tuple[bool, str]:
"""実際のライセンスチェック処理"""
# 実装は省略
pass
5.2. エラーハンドリング
ライセンス違反を検出した際の適切な処理が重要です。
class LicenseViolationError(Exception):
"""ライセンス違反エラー"""
pass
class LicenseExpiredError(LicenseViolationError):
"""ライセンス期限切れエラー"""
pass
def handle_resource_access(resource_id: str, purpose: str):
"""リソースアクセス処理（エラーハンドリング付き）"""
try:
# ライセンスチェック
checker = LicenseChecker()
allowed, reason = checker.check_access(resource_id, purpose, {})
if not allowed:
if "expired" in reason.lower():
raise LicenseExpiredError(reason)
else:
raise LicenseViolationError(reason)
# リソース取得
resource = fetch_resource(resource_id)
return resource
except LicenseExpiredError as e:
# 期限切れの場合は更新を促す
logger.warning(f"License expired for {resource_id}: {e}")
notify_license_renewal_needed(resource_id)
return None
except LicenseViolationError as e:
# その他の違反は記録して拒否
logger.error(f"License violation for {resource_id}: {e}")
log_violation_attempt(resource_id, purpose)
return None
5.3. ユーザーへの通知
ライセンス制約について、ユーザーに適切に通知する必要があります。
def get_resource_with_user_notification(
resource_id: str,
purpose: str,
user_interface
):
"""ユーザー通知付きリソース取得"""
checker = LicenseChecker()
allowed, reason = checker.check_access(resource_id, purpose, {})
if not allowed:
# ユーザーに理由を説明
user_interface.show_warning(
title="License Restriction",
message=f"This content cannot be used for {purpose}. Reason: {reason}",
alternatives=get_alternative_resources(resource_id)
)
return None
# ライセンス情報を取得
license_info = get_license_info(resource_id)
# 帰属表示が必要な場合
if license_info.get('attribution_required'):
attribution = license_info.get('attribution')
user_interface.add_citation(attribution)
return fetch_resource(resource_id)
6. 権利者への報告
6.1. 利用レポートの生成
定期的に権利者へ利用状況を報告します。
class UsageReporter:
"""利用状況レポート生成"""
def __init__(self, tracker: ResourceUsageTracker):
self.tracker = tracker
def generate_monthly_report(
self,
resource_id: str,
month: str
) -> dict:
"""月次レポートを生成"""
records = self._get_records_for_month(resource_id, month)
return {
'resource_id': resource_id,
'period': month,
'total_accesses': len(records),
'by_purpose': self._analyze_by_purpose(records),
'by_application': self._analyze_by_application(records),
'unique_users': len(set(r['user_id'] for r in records)),
'compliance_status': self._check_compliance(records)
}
def send_report_to_rightholder(
self,
rightholder_email: str,
report: dict
):
"""権利者にレポートを送信"""
# メール送信やAPI経由での通知
# 実装は省略
pass
6.2. リアルタイム通知
重要なイベントは即座に通知します。
class LicenseEventNotifier:
"""ライセンスイベント通知"""
def notify_usage_threshold(
self,
resource_id: str,
current_usage: int,
threshold: int
):
"""利用回数が閾値に達した場合に通知"""
if current_usage >= threshold:
message = f"Resource {resource_id} has reached {current_usage} uses (threshold: {threshold})"
self._send_notification(message)
def notify_violation_attempt(
self,
resource_id: str,
violation_type: str,
details: dict
):
"""ライセンス違反の試みを通知"""
message = f"License violation attempt detected for {resource_id}: {violation_type}"
self._send_alert(message, details)
def _send_notification(self, message: str):
"""通知を送信（通常の優先度）"""
# 実装は省略
pass
def _send_alert(self, message: str, details: dict):
"""アラートを送信（高い優先度）"""
# 実装は省略
pass
7. 標準ライセンスへの対応
7.1. Creative Commonsライセンス
class CreativeCommonsLicense:
"""Creative Commonsライセンスのサポート"""
LICENSE_TYPES = {
'CC0': {
'commercial': True,
'modification': True,
'distribution': True,
'attribution': False
},
'CC-BY': {
'commercial': True,
'modification': True,
'distribution': True,
'attribution': True
},
'CC-BY-SA': {
'commercial': True,
'modification': True,
'distribution': True,
'attribution': True,
'shareAlike': True
},
'CC-BY-NC': {
'commercial': False,
'modification': True,
'distribution': True,
'attribution': True
},
'CC-BY-NC-SA': {
'commercial': False,
'modification': True,
'distribution': True,
'attribution': True,
'shareAlike': True
},
'CC-BY-ND': {
'commercial': True,
'modification': False,
'distribution': True,
'attribution': True
},
'CC-BY-NC-ND': {
'commercial': False,
'modification': False,
'distribution': True,
'attribution': True
}
}
@classmethod
def get_permissions(cls, license_type: str) -> dict:
"""ライセンスタイプから許可内容を取得"""
return cls.LICENSE_TYPES.get(license_type, {})
@classmethod
def check_compatibility(
cls,
license_type: str,
intended_use: dict
) -> bool:
"""利用目的との互換性をチェック"""
permissions = cls.get_permissions(license_type)
if intended_use.get('commercial') and not permissions.get('commercial'):
return False
if intended_use.get('modification') and not permissions.get('modification'):
return False
return True
まとめ
MCPを使用する際のライセンス管理には、以下のようなアプローチが有効です。
実践的な実装ステップ：
- メタデータの設計： リソースにライセンス情報を付与
- チェック機構の実装： サーバーまたはクライアントでのチェック
- ログと追跡： 利用状況の記録
- 報告とコンプライアンス： 権利者への報告
重要な原則：
- 透明性： すべての利用を記録し、追跡可能にする
- 自動化： 可能な限り手動チェックを減らす
- 柔軟性： 様々なライセンス形態に対応できる設計
- ユーザー体験： ライセンス制約を明確に伝える
制限と課題：
- 完全な自動化は困難（人間による判断が必要な場合もある）
- ライセンス解釈の複雑さ（法的専門知識が必要）
- パフォーマンスとセキュリティのトレードオフ
法的な注意：
本記事は技術的な実装例を示すものであり、法的助言ではありません。ライセンス管理の実装にあたっては、必ず法務専門家に相談し、適用される法律や規制を遵守してください。
参考情報
免責事項： 本記事の内容は情報提供のみを目的としており、法的助言を構成するものではありません。実際の運用にあたっては、必ず法律専門家にご相談ください。
補足：実践的な統合例
8. 完全な実装例
以下は、MCPサーバーにライセンスチェック機能を統合した完全な例です。
import json
from datetime import datetime
from typing import Optional, Dict, List, Any
from dataclasses import dataclass, asdict
@dataclass
class LicenseInfo:
"""ライセンス情報"""
license_id: str
license_type: str
commercial_use: bool
ai_training: bool
modification: bool
attribution_required: bool
attribution_text: Optional[str] = None
valid_until: Optional[str] = None
territorial_scope: Optional[List[str]] = None
usage_limit: Optional[int] = None
url: Optional[str] = None
class LicenseManager:
"""
ライセンス管理の統合クラス
チェック、追跡、ログ、レポート機能を提供
"""
def __init__(self, config_file: str = "licenses.json"):
self.config_file = config_file
self.licenses: Dict[str, LicenseInfo] = {}
self.usage_log: List[Dict] = []
self.violation_log: List[Dict] = []
self._load_licenses()
def _load_licenses(self):
"""設定ファイルからライセンス情報を読み込み"""
try:
with open(self.config_file, 'r') as f:
data = json.load(f)
for resource_id, license_data in data.items():
self.licenses[resource_id] = LicenseInfo(**license_data)
except FileNotFoundError:
print(f"License config file not found: {self.config_file}")
def register_resource(self, resource_id: str, license_info: LicenseInfo):
"""新しいリソースのライセンス情報を登録"""
self.licenses[resource_id] = license_info
self._save_licenses()
def _save_licenses(self):
"""ライセンス情報を保存"""
data = {
resource_id: asdict(license_info)
for resource_id, license_info in self.licenses.items()
}
with open(self.config_file, 'w') as f:
json.dump(data, f, indent=2)
def check_and_log_access(
self,
resource_id: str,
user_id: str,
purpose: str,
context: Dict[str, Any]
) -> tuple[bool, str, Optional[Dict]]:
"""
アクセスチェックとログ記録を一度に実行
Returns:
(許可/拒否, 理由, ライセンス情報)
"""
# ライセンス情報取得
license_info = self.licenses.get(resource_id)
if not license_info:
# ライセンス情報がない場合は警告付きで許可
self._log_access(resource_id, user_id, purpose, True,
"No license restrictions", context)
return True, "No license restrictions", None
# 各種チェック
checks = []
# 1. 有効期限チェック
if license_info.valid_until:
expiry = datetime.fromisoformat(license_info.valid_until)
if datetime.now() > expiry:
reason = f"License expired on {license_info.valid_until}"
self._log_violation(resource_id, user_id, purpose, reason, context)
return False, reason, asdict(license_info)
checks.append("expiry_check_passed")
# 2. 用途チェック
if purpose == "commercial" and not license_info.commercial_use:
reason = "Commercial use not permitted by license"
self._log_violation(resource_id, user_id, purpose, reason, context)
return False, reason, asdict(license_info)
checks.append("purpose_check_passed")
if purpose == "training" and not license_info.ai_training:
reason = "AI training use not permitted by license"
self._log_violation(resource_id, user_id, purpose, reason, context)
return False, reason, asdict(license_info)
# 3. 地域制限チェック
if license_info.territorial_scope:
user_region = context.get('region', 'unknown')
if user_region not in license_info.territorial_scope:
reason = f"Access not allowed from region: {user_region}"
self._log_violation(resource_id, user_id, purpose, reason, context)
return False, reason, asdict(license_info)
checks.append("region_check_passed")
# 4. 利用回数制限チェック
if license_info.usage_limit:
current_usage = self._count_usage(resource_id)
if current_usage >= license_info.usage_limit:
reason = f"Usage limit reached: {current_usage}/{license_info.usage_limit}"
self._log_violation(resource_id, user_id, purpose, reason, context)
return False, reason, asdict(license_info)
checks.append("usage_limit_check_passed")
# すべてのチェックを通過
self._log_access(resource_id, user_id, purpose, True,
f"All checks passed: {', '.join(checks)}", context)
return True, "Access granted", asdict(license_info)
def _log_access(
self,
resource_id: str,
user_id: str,
purpose: str,
granted: bool,
reason: str,
context: Dict
):
"""アクセスログを記録"""
log_entry = {
'timestamp': datetime.now().isoformat(),
'event_type': 'access',
'resource_id': resource_id,
'user_id': user_id,
'purpose': purpose,
'granted': granted,
'reason': reason,
'context': context
}
self.usage_log.append(log_entry)
def _log_violation(
self,
resource_id: str,
user_id: str,
purpose: str,
reason: str,
context: Dict
):
"""ライセンス違反の試みを記録"""
violation_entry = {
'timestamp': datetime.now().isoformat(),
'resource_id': resource_id,
'user_id': user_id,
'purpose': purpose,
'reason': reason,
'context': context
}
self.violation_log.append(violation_entry)
# 重大な違反は即座に通知
self._send_violation_alert(violation_entry)
def _count_usage(self, resource_id: str) -> int:
"""特定リソースの利用回数をカウント"""
return sum(
1 for log in self.usage_log
if log['resource_id'] == resource_id and log['granted']
)
def _send_violation_alert(self, violation: Dict):
"""違反アラートを送信（実装例）"""
print(f"⚠️ LICENSE VIOLATION ALERT:")
print(f" Resource: {violation['resource_id']}")
print(f" User: {violation['user_id']}")
print(f" Reason: {violation['reason']}")
print(f" Time: {violation['timestamp']}")
def generate_usage_report(
self,
resource_id: Optional[str] = None,
start_date: Optional[str] = None,
end_date: Optional[str] = None
) -> Dict:
"""利用レポートを生成"""
logs = self.usage_log
# フィルタリング
if resource_id:
logs = [l for l in logs if l['resource_id'] == resource_id]
if start_date:
logs = [l for l in logs if l['timestamp'] >= start_date]
if end_date:
logs = [l for l in logs if l['timestamp'] <= end_date]
# 統計計算
total_accesses = len(logs)
granted = sum(1 for l in logs if l['granted'])
denied = total_accesses - granted
# 用途別集計
by_purpose = {}
for log in logs:
purpose = log.get('purpose', 'unknown')
by_purpose[purpose] = by_purpose.get(purpose, 0) + 1
# ユーザー別集計
unique_users = len(set(l['user_id'] for l in logs))
return {
'summary': {
'total_accesses': total_accesses,
'granted': granted,
'denied': denied,
'unique_users': unique_users
},
'by_purpose': by_purpose,
'period': {
'start': start_date or 'beginning',
'end': end_date or 'now'
},
'violations': len(self.violation_log)
}
def export_logs(self, filename: str):
"""ログをファイルにエクスポート"""
export_data = {
'exported_at': datetime.now().isoformat(),
'usage_logs': self.usage_log,
'violation_logs': self.violation_log
}
with open(filename, 'w') as f:
json.dump(export_data, f, indent=2)
print(f"Logs exported to {filename}")
# 使用例
def example_usage():
"""実際の使用例"""
# ライセンスマネージャーの初期化
manager = LicenseManager()
# リソースの登録
license1 = LicenseInfo(
license_id="lic-001",
license_type="CC-BY-NC-4.0",
commercial_use=False,
ai_training=False,
modification=True,
attribution_required=True,
attribution_text="© 2024 Example Corp",
valid_until="2025-12-31T23:59:59",
territorial_scope=["JP", "US"],
usage_limit=1000,
url="https://creativecommons.org/licenses/by-nc/4.0/"
)
manager.register_resource("document-001", license1)
# アクセスチェック（商用利用を試みる）
allowed, reason, license_info = manager.check_and_log_access(
resource_id="document-001",
user_id="user-123",
purpose="commercial",
context={
'region': 'JP',
'application': 'chatbot',
'ip_address': '192.168.1.1'
}
)
if not allowed:
print(f"❌ Access denied: {reason}")
if license_info and license_info.get('attribution_text'):
print(f" License: {license_info['license_type']}")
print(f" URL: {license_info.get('url', 'N/A')}")
else:
print(f"✅ Access granted: {reason}")
if license_info and license_info.get('attribution_required'):
print(f" Attribution required: {license_info['attribution_text']}")
# 非商用利用を試みる（成功するはず）
allowed2, reason2, license_info2 = manager.check_and_log_access(
resource_id="document-001",
user_id="user-123",
purpose="research",
context={
'region': 'JP',
'application': 'research_tool'
}
)
print(f"\n{'✅' if allowed2 else '❌'} Second access: {reason2}")
# レポート生成
report = manager.generate_usage_report(resource_id="document-001")
print(f"\n📊 Usage Report:")
print(json.dumps(report, indent=2))
# ログのエクスポート
manager.export_logs("license_logs.json")
if __name__ == "__main__":
example_usage()
9. MCPサーバーへの統合
実際のMCPサーバーに統合する例：
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool
class LicensedMCPServer:
"""ライセンスチェック機能を持つMCPサーバー"""
def __init__(self):
self.server = Server("licensed-content-server")
self.license_manager = LicenseManager()
self._setup_handlers()
def _setup_handlers(self):
"""ハンドラーの設定"""
@self.server.list_resources()
async def list_resources() -> list[Resource]:
"""利用可能なリソースをリスト"""
# ライセンス情報を含むリソースリストを返す
resources = []
for resource_id, license_info in self.license_manager.licenses.items():
resources.append(Resource(
uri=f"licensed://{resource_id}",
name=resource_id,
mimeType="text/plain",
description=f"License: {license_info.license_type}",
annotations={
"license": asdict(license_info)
}
))
return resources
@self.server.read_resource()
async def read_resource(uri: str) -> str:
"""リソースの読み取り（ライセンスチェック付き）"""
# URIからリソースIDを抽出
resource_id = uri.replace("licensed://", "")
# コンテキスト情報（実際のアプリケーションから取得）
context = {
'region': 'JP',
'application': 'mcp-client',
'timestamp': datetime.now().isoformat()
}
# ライセンスチェック
allowed, reason, license_info = self.license_manager.check_and_log_access(
resource_id=resource_id,
user_id="default-user", # 実際は認証から取得
purpose="read",
context=context
)
if not allowed:
raise PermissionError(f"License check failed: {reason}")
# リソースの内容を取得
content = self._fetch_resource_content(resource_id)
# 帰属表示が必要な場合は追加
if license_info and license_info.get('attribution_required'):
attribution = license_info.get('attribution_text', '')
content = f"{content}\n\n---\nSource: {attribution}"
return content
@self.server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[Any]:
"""ツール実行（用途に応じたライセンスチェック）"""
if name == "get_usage_report":
# 利用レポートの取得
resource_id = arguments.get("resource_id")
report = self.license_manager.generate_usage_report(resource_id)
return [{"type": "text", "text": json.dumps(report, indent=2)}]
elif name == "check_license":
# ライセンス情報の確認
resource_id = arguments.get("resource_id")
license_info = self.license_manager.licenses.get(resource_id)
if license_info:
return [{"type": "text", "text": json.dumps(asdict(license_info), indent=2)}]
else:
return [{"type": "text", "text": "No license information found"}]
else:
raise ValueError(f"Unknown tool: {name}")
def _fetch_resource_content(self, resource_id: str) -> str:
"""リソースの実際のコンテンツを取得（実装例）"""
# 実際のシステムでは、データベースやファイルシステムから取得
return f"Content of {resource_id}"
async def run(self):
"""サーバーの起動"""
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
self.server.create_initialization_options()
)
# サーバーの起動
if __name__ == "__main__":
import asyncio
server = LicensedMCPServer()
asyncio.run(server.run())
10. クライアント側での活用
MCPクライアント側でライセンス情報を適切に扱う例：
class LicenseAwareClient:
"""ライセンス情報を考慮するMCPクライアント"""
def __init__(self, mcp_client):
self.mcp_client = mcp_client
self.license_cache = {}
async def get_resource_safely(
self,
resource_uri: str,
intended_use: str
) -> tuple[Optional[str], Optional[Dict]]:
"""
ライセンスを確認しながらリソースを取得
Returns:
(コンテンツ, ライセンス情報)
"""
try:
# リソース一覧を取得してライセンス情報を確認
resources = await self.mcp_client.list_resources()
target_resource = None
for resource in resources:
if resource.uri == resource_uri:
target_resource = resource
break
if not target_resource:
print(f"Resource not found: {resource_uri}")
return None, None
# ライセンス情報の取得
license_info = target_resource.annotations.get('license', {})
# 意図した用途がライセンスで許可されているか確認
if not self._check_license_allows(license_info, intended_use):
print(f"⚠️ Warning: Intended use '{intended_use}' may not be allowed by license")
print(f" License type: {license_info.get('license_type', 'Unknown')}")
# ユーザーに確認を求める
if not self._get_user_confirmation(license_info):
return None, license_info
# リソースを取得
content = await self.mcp_client.read_resource(resource_uri)
# ライセンス情報をキャッシュ
self.license_cache[resource_uri] = license_info
return content, license_info
except PermissionError as e:
print(f"❌ Access denied: {e}")
return None, None
def _check_license_allows(self, license_info: Dict, intended_use: str) -> bool:
"""ライセンスが意図した用途を許可しているかチェック"""
if intended_use == "commercial":
return license_info.get('commercial_use', False)
elif intended_use == "training":
return license_info.get('ai_training', False)
elif intended_use == "modification":
return license_info.get('modification', False)
else:
return True # その他の用途は一般的に許可
def _get_user_confirmation(self, license_info: Dict) -> bool:
"""ユーザーに確認を求める（簡易実装）"""
print("\nLicense Information:")
print(f" Type: {license_info.get('license_type', 'Unknown')}")
print(f" Commercial use: {license_info.get('commercial_use', False)}")
print(f" AI training: {license_info.get('ai_training', False)}")
print(f" Attribution required: {license_info.get('attribution_required', False)}")
if license_info.get('url'):
print(f" Full license: {license_info['url']}")
response = input("\nDo you want to proceed? (yes/no): ")
return response.lower() in ['yes', 'y']
def format_citation(self, resource_uri: str) -> str:
"""リソースの引用情報をフォーマット"""
license_info = self.license_cache.get(resource_uri, {})
if not license_info.get('attribution_required'):
return ""
attribution = license_info.get('attribution_text', 'Source unknown')
license_type = license_info.get('license_type', '')
url = license_info.get('url', '')
citation = f"\n---\nSource: {attribution}"
if license_type:
citation += f"\nLicense: {license_type}"
if url:
citation += f"\nLicense URL: {url}"
return citation
# 使用例
async def example_client_usage():
"""クライアントの使用例"""
from mcp.client import ClientSession, StdioServerParameters
from mcp import stdio_client
# MCPサーバーに接続
server_params = StdioServerParameters(
command="python",
args=["licensed_mcp_server.py"]
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# ライセンス対応クライアントの作成
client = LicenseAwareClient(session)
# リソースを安全に取得
content, license_info = await client.get_resource_safely(
resource_uri="licensed://document-001",
intended_use="commercial"
)
if content:
print(f"\n✅ Content retrieved successfully")
print(content)
# 引用情報を追加
citation = client.format_citation("licensed://document-001")
if citation:
print(citation)
else:
print(f"\n❌ Could not retrieve content")
これらの実装例により、MCPを使用する際に実践的なライセンス管理を行うことができます。
注意: MCPはAnthropicが開発した比較的新しいプロトコルです。最新の情報については、公式ドキュメントを参照してください。