はじめに
この記事は、QiitaのModel Context Protocol（以下、MCP）解説シリーズの第11回です。
今回は、MCPのResources機能に焦点を当て、外部のテキストファイルを読み取り可能にする具体的な実装方法をさらに深く掘り下げます。
📚 Resourcesの基本：なぜファイルパスではないのか？
Resourcesは、LLMに提供したい 「静的なデータ」 の抽象化された表現です。多くの人が「ファイルを公開するなら、ファイルパスを渡せばいいのでは？」と考えるかもしれません。しかし、MCPは単純なファイルパスを使いません。その理由は以下の通りです。
1. セキュリティの確保
ファイルパスを直接公開すると、LLMがサーバー上の任意のファイルにアクセスするリスクが生じます。MCPでは、公開するリソースを明示的に定義することで、アクセスを限定し、セキュリティを確保します。
2. 抽象化による柔軟性
MCPは、ファイルだけでなく、データベースのエントリやWeb上のドキュメントなど、さまざまな情報源を同じResourcesという概念で扱えるようにします。これにより、LLMは情報源の種類を意識することなく、一貫した方法でデータにアクセスできます。
3. URI による統一的なアドレス指定
ResourcesはURIスキームを使用して識別されるため、file://, http://, db://など、様々なデータソースを統一的に扱えます。
つまり、Resourcesは「ファイルのパス」ではなく、 「特定の情報を識別するためのハンドル（取っ手）」 なのです。
📝 実装：個別のテキストファイルを公開する
特定のテキストファイルをResourcesとして公開する最も基本的な方法を学びます。
🛠️ プロジェクトのセットアップ
Pythonを使って実装します。
# プロジェクトディレクトリを作成
mkdir mcp-resources-example
cd mcp-resources-example
# 仮想環境を作成・有効化
python -m venv mcp-env
source mcp-env/bin/activate # Windows: mcp-env\Scripts\activate
# 必要なパッケージをインストール
pip install mcp
次に、公開したいテキストファイルを作成します。
# サンプルファイルを作成
mkdir resources
echo "これは社内の製品マニュアルです。
製品の概要：
- 高性能なデータ処理エンジン
- 直感的なユーザーインターフェース
- 豊富なAPI連携機能
よくある質問：
Q: 導入にはどのくらい時間がかかりますか？
A: 通常1-2週間程度で完了します。
Q: サポート体制はどうなっていますか？
A: 24時間365日のサポートを提供しています。" > resources/product_manual.txt
echo "システム開発ガイドライン
1. コーディング規約
- PEP 8に準拠したPythonコードの記述
- 適切なコメントとドキュメントの作成
- テストコードの必須化
2. セキュリティガイドライン
- 入力値の検証を必須とする
- 認証・認可の適切な実装
- ログ記録の徹底
3. パフォーマンス要件
- レスポンス時間は2秒以内
- 同時接続数1000以上に対応
- 可用性99.9%以上を維持" > resources/development_guide.txt
💻 サーバーのコード実装
server.pyというファイルを作成し、以下のコードを実装してください。
import asyncio
import logging
from pathlib import Path
from typing import List, Dict, Any
from urllib.parse import urlparse
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
from pydantic import AnyUrl
# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# リソースディレクトリのパス
RESOURCES_DIR = Path("./resources").resolve()
# MCPサーバーを作成
server = Server("resources-server")
class ResourceManager:
"""リソース管理クラス"""
def __init__(self, resources_dir: Path):
self.resources_dir = resources_dir
self._resource_cache: Dict[str, str] = {}
def get_available_resources(self) -> List[types.Resource]:
"""利用可能なリソースの一覧を取得"""
resources = []
if not self.resources_dir.exists():
logger.warning(f"Resources directory does not exist: {self.resources_dir}")
return resources
# テキストファイルを検索
for file_path in self.resources_dir.glob("*.txt"):
try:
# ファイル情報を取得
file_stats = file_path.stat()
resources.append(
types.Resource(
uri=AnyUrl(f"file://{file_path}"),
name=file_path.stem, # 拡張子なしのファイル名
description=f"Text file: {file_path.name} ({file_stats.st_size} bytes)",
mimeType="text/plain",
)
)
except Exception as e:
logger.error(f"Error processing file {file_path}: {e}")
return resources
def read_resource(self, uri: str) -> str:
"""リソースの内容を読み取り"""
try:
# URIからファイルパスを抽出
parsed_uri = urlparse(uri)
if parsed_uri.scheme != "file":
raise ValueError(f"Unsupported URI scheme: {parsed_uri.scheme}")
file_path = Path(parsed_uri.path)
# セキュリティチェック：リソースディレクトリ配下のファイルのみ
try:
file_path.resolve().relative_to(self.resources_dir)
except ValueError:
raise ValueError("Access denied: File outside resources directory")
if not file_path.exists():
raise FileNotFoundError(f"Resource not found: {file_path}")
# キャッシュをチェック
cache_key = str(file_path)
if cache_key in self._resource_cache:
return self._resource_cache[cache_key]
# ファイル内容を読み取り
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
# キャッシュに保存（小さなファイルのみ）
if len(content) < 100000: # 100KB未満
self._resource_cache[cache_key] = content
return content
except Exception as e:
raise RuntimeError(f"Failed to read resource {uri}: {e}")
# リソースマネージャーを初期化
resource_manager = ResourceManager(RESOURCES_DIR)
@server.list_resources()
async def handle_list_resources() -> List[types.Resource]:
"""利用可能なリソースの一覧を返す"""
try:
resources = resource_manager.get_available_resources()
logger.info(f"Listed {len(resources)} resources")
return resources
except Exception as e:
logger.error(f"Error listing resources: {e}")
return []
@server.read_resource()
async def handle_read_resource(uri: AnyUrl) -> str:
"""指定されたリソースの内容を読み取る"""
try:
content = resource_manager.read_resource(str(uri))
logger.info(f"Read resource: {uri}")
return content
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
raise
# 基本的なツールも提供
@server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
"""利用可能なツールの一覧を返す"""
return [
types.Tool(
name="search_resources",
description="Search for specific content within available resources",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query to find in resources",
}
},
"required": ["query"],
},
),
types.Tool(
name="summarize_resource",
description="Get a summary of a specific resource",
inputSchema={
"type": "object",
"properties": {
"resource_name": {
"type": "string",
"description": "Name of the resource to summarize",
}
},
"required": ["resource_name"],
},
),
]
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> List[types.TextContent]:
"""ツールの実行を処理"""
try:
if name == "search_resources":
return await search_resources_tool(arguments)
elif name == "summarize_resource":
return await summarize_resource_tool(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
logger.error(f"Error in tool {name}: {e}")
return [types.TextContent(type="text", text=f"Error: {str(e)}")]
async def search_resources_tool(arguments: dict | None) -> List[types.TextContent]:
"""リソース内容検索ツール"""
if not arguments or "query" not in arguments:
raise ValueError("Query argument is required")
query = arguments["query"].lower()
results = []
resources = resource_manager.get_available_resources()
for resource in resources:
try:
content = resource_manager.read_resource(str(resource.uri))
if query in content.lower():
# マッチした部分の前後を抽出
lines = content.split('\n')
matching_lines = [
f"Line {i+1}: {line.strip()}"
for i, line in enumerate(lines)
if query in line.lower()
]
if matching_lines:
results.append(f"📄 {resource.name}:\n" + "\n".join(matching_lines[:5]))
except Exception as e:
logger.error(f"Error searching in {resource.name}: {e}")
continue
if results:
result_text = f"Search results for '{arguments['query']}':\n\n" + "\n\n".join(results)
else:
result_text = f"No matches found for '{arguments['query']}'"
return [types.TextContent(type="text", text=result_text)]
async def summarize_resource_tool(arguments: dict | None) -> List[types.TextContent]:
"""リソース要約ツール"""
if not arguments or "resource_name" not in arguments:
raise ValueError("Resource name argument is required")
resource_name = arguments["resource_name"]
resources = resource_manager.get_available_resources()
# 指定されたリソースを検索
target_resource = None
for resource in resources:
if resource.name == resource_name:
target_resource = resource
break
if not target_resource:
raise ValueError(f"Resource not found: {resource_name}")
try:
content = resource_manager.read_resource(str(target_resource.uri))
# シンプルな要約作成
lines = content.split('\n')
non_empty_lines = [line.strip() for line in lines if line.strip()]
summary_parts = []
summary_parts.append(f"リソース名: {target_resource.name}")
summary_parts.append(f"ファイルサイズ: {len(content)} 文字")
summary_parts.append(f"行数: {len(non_empty_lines)} 行")
# 最初の数行を抜粋
if non_empty_lines:
summary_parts.append("\n内容の抜粋:")
for i, line in enumerate(non_empty_lines[:5]):
summary_parts.append(f" {i+1}. {line}")
if len(non_empty_lines) > 5:
summary_parts.append(f" ... (残り {len(non_empty_lines) - 5} 行)")
result_text = "\n".join(summary_parts)
except Exception as e:
raise RuntimeError(f"Failed to summarize resource: {e}")
return [types.TextContent(type="text", text=result_text)]
async def main():
"""サーバーのメイン関数"""
# リソースディレクトリを作成
RESOURCES_DIR.mkdir(exist_ok=True)
logger.info(f"Resources directory: {RESOURCES_DIR}")
logger.info(f"Available resources: {len(resource_manager.get_available_resources())}")
# stdio transport
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="resources-server",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
asyncio.run(main())
🚀 動作の確認
-
サーバーを起動する:
python server.py -
Claude Desktopでの設定:
claude_desktop_config.jsonに以下を追加：{ "mcpServers": { "resources": { "command": "python", "args": ["/path/to/your/project/server.py"] } } } -
Claude Desktopでのテスト:
以下のような質問を試してみてください：
利用可能なリソースを一覧表示してください。product_manualリソースの内容を読み込んで要約してください。「サポート」というキーワードでリソースを検索してください。
🧠 Resourcesのより高度な使い方
動的リソースの実装
静的ファイルだけでなく、動的に生成されるリソースも実装できます：
import datetime
import json
from typing import Optional
class DynamicResourceManager(ResourceManager):
"""動的リソースを含むリソース管理クラス"""
def get_available_resources(self) -> List[types.Resource]:
"""静的リソース + 動的リソースを返す"""
resources = super().get_available_resources()
# 動的リソースを追加
resources.extend([
types.Resource(
uri=AnyUrl("dynamic://system_status"),
name="system_status",
description="Current system status and metrics",
mimeType="application/json",
),
types.Resource(
uri=AnyUrl("dynamic://current_time"),
name="current_time",
description="Current date and time information",
mimeType="text/plain",
),
])
return resources
def read_resource(self, uri: str) -> str:
"""静的リソース + 動的リソースの読み取り"""
parsed_uri = urlparse(uri)
if parsed_uri.scheme == "dynamic":
return self._read_dynamic_resource(parsed_uri.netloc)
else:
return super().read_resource(uri)
def _read_dynamic_resource(self, resource_type: str) -> str:
"""動的リソースの内容を生成"""
if resource_type == "system_status":
status = {
"timestamp": datetime.datetime.now().isoformat(),
"status": "healthy",
"resources_count": len(self.get_available_resources()),
"memory_usage": "Normal",
"uptime": "24h 15m 30s"
}
return json.dumps(status, indent=2, ensure_ascii=False)
elif resource_type == "current_time":
now = datetime.datetime.now()
return f"""現在時刻情報
日時: {now.strftime('%Y年%m月%d日 %H:%M:%S')}
タイムゾーン: JST
Unix時刻: {int(now.timestamp())}
曜日: {now.strftime('%A')}
"""
else:
raise ValueError(f"Unknown dynamic resource: {resource_type}")
大容量ファイルの効率的な処理
大きなファイルを扱う場合は、チャンク読み込みを実装：
async def handle_read_resource(uri: AnyUrl) -> str:
"""大容量ファイル対応の読み込み"""
try:
parsed_uri = urlparse(str(uri))
file_path = Path(parsed_uri.path)
# ファイルサイズをチェック
file_size = file_path.stat().st_size
if file_size > 1024 * 1024: # 1MB以上
# 大きなファイルは最初の部分のみ読み込み
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read(50000) # 最初の50KB
return f"""[Large file preview - showing first 50KB of {file_size} bytes]
{content}
[... file continues ...]
"""
else:
# 通常の読み込み
return resource_manager.read_resource(str(uri))
except Exception as e:
logger.error(f"Error reading large resource {uri}: {e}")
raise
🔐 セキュリティとパフォーマンス
セキュリティ対策
- パス正規化: 相対パスやシンボリックリンクを適切に処理
- ファイルタイプ制限: 許可されたファイルタイプのみアクセス可能
- アクセス制御: ユーザー権限に基づくリソースアクセス制御
def is_safe_resource(self, file_path: Path) -> bool:
"""リソースの安全性をチェック"""
try:
# パス正規化
resolved_path = file_path.resolve()
# ディレクトリ制限チェック
resolved_path.relative_to(self.resources_dir)
# ファイルタイプチェック
allowed_extensions = {'.txt', '.md', '.json', '.csv'}
if file_path.suffix.lower() not in allowed_extensions:
return False
# 隠しファイルの除外
if any(part.startswith('.') for part in file_path.parts):
return False
return True
except (ValueError, OSError):
return False
パフォーマンス最適化
- キャッシング: 頻繁にアクセスされるリソースをメモリにキャッシュ
- 遅延読み込み: リソース一覧取得時はメタデータのみ
- 圧縮: 大きなテキストファイルの圧縮転送
🎯 まとめ
今回学んだ内容：
MCPにおけるResourcesの重要性
- セキュリティファースト: 明示的なリソース定義による安全なデータアクセス
- 抽象化: ファイル、データベース、APIなどの統一的な扱い
- 柔軟性: 静的・動的リソースの両方に対応
実装のポイント
-
適切なURI設計:
file://,dynamic://などのスキーム活用 - エラーハンドリング: 堅牢なエラー処理とログ記録
- パフォーマンス: キャッシュと最適化の実装
- 拡張性: 新しいリソースタイプの容易な追加
次のステップ
この基盤を使って、以下のような高度なリソース管理システムを構築できます：
- データベースリソース: SQLクエリ結果をリソースとして提供
- APIリソース: 外部APIからのデータをリソース化
- 動的ドキュメント: テンプレートベースの動的文書生成
- バージョン管理: リソースの履歴管理と差分表示
次回は、MCPのもう一つの核となる機能であるToolsに焦点を当て、より複雑なツールの実装方法を解説します。お楽しみに！