プラグインの検出と登録
ツールレジストリ、マニフェストファイル、実行時の動的なツール読み込みを扱います。
「プラグインの検出と登録」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
エージェント向けプラグインシステム
プラグインシステムを使うと、エージェントのコアコードを変更せずに、実行時に新しいツールを検出して読み込めます。エージェントはプラグインディレクトリを読み取り、各プラグインのマニフェストを読み込んで検証し、そのツールをアクティブなツールレジストリに追加します。
これにより、モジュール性と拡張性に優れたエージェントアーキテクチャを実現できます。
プラグインマニフェスト形式
すべてのプラグインにはplugin.jsonマニフェストファイルが含まれます。これはプラグインの身分証明書であり、プラグインの種類、バージョン、提供するツール、必要なもの（Pythonパッケージや環境変数）を示します。
# plugin.json — stored in the plugin's root directory
EXAMPLE_MANIFEST = {
'name': 'weather-tools',
'display_name': 'Weather Tools',
'version': '2.1.0',
'description': 'Real-time weather and forecast tools',
'author': 'Jane Developer <jane@example.com>',
'license': 'MIT',
'entry_point': 'weather_tools.plugin', # Python module path
'tool_definitions': [
'get_current_weather',
'get_5day_forecast',
'get_weather_alerts'
],
'requires': {
'python_packages': ['requests>=2.28'],
'env_vars': ['WEATHER_API_KEY']
},
'tags': ['weather', 'forecast', 'iot'],
'min_framework_version': '1.0.0'
}
import json
print(json.dumps(EXAMPLE_MANIFEST, indent=2)[:300])プラグインディレクトリローダー
ディレクトリをスキャンしてプラグインのサブディレクトリを探します。各サブディレクトリにはplugin.jsonマニフェストが含まれていなければなりません。ローダーは各マニフェストを読み込んで検証し、プラグインレジストリにプラグインを登録します。
import os
import json
PLUGIN_DIR = 'plugins'
def discover_plugins(plugin_dir: str = PLUGIN_DIR) -> list:
discovered = []
if not os.path.isdir(plugin_dir):
print(f'Plugin directory not found: {plugin_dir}')
return []
for entry in os.scandir(plugin_dir):
if not entry.is_dir():
continue
manifest_path = os.path.join(entry.path, 'plugin.json')
if not os.path.exists(manifest_path):
print(f'Skipping {entry.name}: no plugin.json')
continue
try:
with open(manifest_path) as f:
manifest = json.load(f)
manifest['_path'] = entry.path
manifest['_name'] = entry.name
discovered.append(manifest)
print(f'Discovered: {manifest["name"]} v{manifest["version"]}')
except (json.JSONDecodeError, KeyError) as e:
print(f'Invalid manifest in {entry.name}: {e}')
return discovered
plugins = discover_plugins()
print(f'Total plugins discovered: {len(plugins)}')マニフェストの検証
プラグインを読み込む前に、スキーマに対してマニフェストを検証し、必須フィールドの不足、無効なバージョン、環境変数の不足を検出してください。無効なプラグインは拒否し、その理由をログに記録します。
import os
REQUIRED_MANIFEST_FIELDS = {'name', 'version', 'entry_point', 'tool_definitions'}
def validate_manifest(manifest: dict) -> tuple:
"""
Returns (is_valid: bool, errors: list)
"""
errors = []
# Required fields
missing = REQUIRED_MANIFEST_FIELDS - set(manifest.keys())
if missing:
errors.append(f'Missing required fields: {missing}')
# Version format
version = manifest.get('version', '')
parts = version.split('.')
if len(parts) != 3 or not all(p.isdigit() for p in parts):
errors.append(f'Invalid version format: {version}')
# Check required env vars exist
env_vars = manifest.get('requires', {}).get('env_vars', [])
for var in env_vars:
if not os.environ.get(var):
errors.append(f'Missing required env var: {var}')
# Tool definitions must be a non-empty list
tools = manifest.get('tool_definitions', [])
if not isinstance(tools, list) or len(tools) == 0:
errors.append('tool_definitions must be a non-empty list')
return len(errors) == 0, errors
if __name__ == '__main__':
good_manifest = {
'name': 'weather-tools', 'version': '1.2.0',
'entry_point': 'weather_tools.plugin', 'tool_definitions': [{'name': 'get_weather'}]
}
bad_manifest = {'name': 'broken-plugin', 'version': 'v1'}
print('Good manifest:', validate_manifest(good_manifest))
print('Bad manifest: ', validate_manifest(bad_manifest))
動的なツール読み込み
プラグインのマニフェストを検証したら、プラグインのPythonモジュールを動的にインポートし、get_tools()を呼び出してツールの定義と実行関数を取得します。Pythonのimportlibを使えば簡単に実装できます。
import importlib
import importlib.util
import sys
def load_plugin_module(manifest: dict):
"""
Import the plugin's Python module and return it.
Adds the plugin directory to sys.path if needed.
"""
plugin_path = manifest['_path']
entry_point = manifest['entry_point'] # e.g. 'weather_tools.plugin'
# Add plugin directory to path so relative imports work
if plugin_path not in sys.path:
sys.path.insert(0, plugin_path)
try:
module = importlib.import_module(entry_point)
return module
except ImportError as e:
print(f'Failed to import {entry_point}: {e}')
return None
def extract_tools_from_module(module, tool_names: list) -> dict:
"""
Returns {tool_name: {'schema': dict, 'execute': callable}}
"""
tools = {}
for name in tool_names:
schema_fn = getattr(module, f'get_{name}_schema', None)
execute_fn = getattr(module, f'execute_{name}', None)
if schema_fn and execute_fn:
tools[name] = {'schema': schema_fn(), 'execute': execute_fn}
return tools
if __name__ == '__main__':
class FakeModule:
def get_weather_schema(self):
return {'name': 'get_weather'}
def execute_weather(self, params):
return {'temp': 72}
tools = extract_tools_from_module(FakeModule(), ['weather'])
print('Extracted tools:', list(tools.keys()))
print('Weather schema:', tools['weather']['schema'])
ツールレジストリ
ツールレジストリは、利用可能なすべてのツールに関する唯一の信頼できる情報源です。ツール名をスキーマと実行関数に対応付けます。エージェントは各LLM呼び出し用のツールリストを作成するときに、レジストリを参照します。
class ToolRegistry:
def __init__(self):
self._tools: dict = {} # name -> {'schema', 'execute', 'plugin'}
self._plugins: dict = {} # plugin_name -> manifest
def register_tool(
self, name: str, schema: dict,
execute_fn, plugin_name: str
):
if name in self._tools:
print(f'WARNING: Tool {name} already registered, overwriting')
self._tools[name] = {
'schema': schema,
'execute': execute_fn,
'plugin': plugin_name
}
def register_plugin(self, manifest: dict, tools: dict):
self._plugins[manifest['name']] = manifest
for name, tool in tools.items():
self.register_tool(name, tool['schema'],
tool['execute'], manifest['name'])
print(f'Registered plugin: {manifest["name"]} '
f'({len(tools)} tools)')
def get_all_schemas(self) -> list:
return [t['schema'] for t in self._tools.values()]
def execute(self, tool_name: str, params: dict):
if tool_name not in self._tools:
raise KeyError(f'Unknown tool: {tool_name}')
return self._tools[tool_name]['execute'](params)
registry = ToolRegistry()
if __name__ == '__main__':
def get_weather(params):
return {'temp': 72, 'city': params.get('city')}
manifest = {'name': 'weather-tools'}
tools = {'get_weather': {'schema': {'name': 'get_weather'}, 'execute': get_weather}}
registry.register_plugin(manifest, tools)
print('Registered schemas:', registry.get_all_schemas())
print('Execution result:', registry.execute('get_weather', {'city': 'Paris'}))
機能による検索
エージェントが多数のプラグインを持つ場合、すべてのツールを毎回のLLM呼び出しに読み込むのではなく、機能タグでレジストリを検索できるようにする必要があります。コンテキスト内のツールが多すぎると、LLMによるツール選択の精度が低下します。
class SearchableToolRegistry(ToolRegistry):
def search(self, query: str) -> list:
"""
Search tools by name, description, or tags.
Returns list of matching tool schemas.
"""
query_lower = query.lower()
matches = []
for name, tool in self._tools.items():
schema = tool['schema']
plugin = self._plugins.get(tool['plugin'], {})
plugin_tags = plugin.get('tags', [])
if (
query_lower in name.lower() or
query_lower in schema.get('description', '').lower() or
any(query_lower in tag.lower() for tag in plugin_tags)
):
matches.append(schema)
return matches
# Usage:
registry = SearchableToolRegistry()
# After loading plugins...
weather_tools = registry.search('weather')
print(f'Weather tools: {[t["name"] for t in weather_tools]}')ディレクトリ変更時のホットリロード
開発中は、エージェントを再起動せずにファイル変更時のプラグインを再読み込みできると便利です。watchdogライブラリを使用してプラグインディレクトリを監視し、plugin.jsonファイルが変更されたときに再読み込みを実行してください。
# pip install watchdog
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class PluginReloadHandler(FileSystemEventHandler):
def __init__(self, registry, plugin_loader_fn):
self.registry = registry
self.loader = plugin_loader_fn
def on_modified(self, event):
if 'plugin.json' in event.src_path:
print(f'Plugin manifest changed: {event.src_path}')
self.reload_plugin(event.src_path)
def reload_plugin(self, manifest_path: str):
import json, os
plugin_dir = os.path.dirname(manifest_path)
try:
with open(manifest_path) as f:
manifest = json.load(f)
manifest['_path'] = plugin_dir
self.loader(manifest, self.registry)
print(f'Reloaded: {manifest["name"]}')
except Exception as e:
print(f'Reload failed: {e}')
def start_plugin_watcher(plugin_dir: str, registry):
handler = PluginReloadHandler(registry, lambda m, r: None)
observer = Observer()
observer.schedule(handler, plugin_dir, recursive=True)
observer.start()
return observer依存関係の確認と自動インストール
プラグインを読み込むときに、必要なPythonパッケージがインストールされているか確認してください。必要に応じて、pipを使って不足しているパッケージを自動インストールします。依存関係を解決できない場合は、明確なエラーをログに記録してください。
import subprocess
import sys
import importlib
def check_and_install_deps(manifest: dict, auto_install: bool = False) -> bool:
packages = manifest.get('requires', {}).get('python_packages', [])
missing = []
for pkg_spec in packages:
pkg_name = pkg_spec.split('>=')[0].split('==')[0].strip()
try:
importlib.import_module(pkg_name.replace('-', '_'))
except ImportError:
missing.append(pkg_spec)
if not missing:
return True
print(f'Missing packages for {manifest["name"]}: {missing}')
if auto_install:
for pkg in missing:
print(f'Installing {pkg}...')
result = subprocess.run(
[sys.executable, '-m', 'pip', 'install', pkg],
capture_output=True, text=True
)
if result.returncode != 0:
print(f'Install failed: {result.stderr[:200]}')
return False
return True
return False
if __name__ == '__main__':
manifest = {'name': 'demo-plugin', 'requires': {'python_packages': ['totally_fake_package_xyz']}}
ok = check_and_install_deps(manifest, auto_install=False)
print('All dependencies satisfied:', ok)
プラグインの完全なブートストラップ
検出、検証、依存関係の確認、モジュールの読み込み、レジストリへの登録を、エージェントの起動時に呼び出す単一のブートストラップ関数にまとめます。
def bootstrap_plugins(
plugin_dir: str,
registry: ToolRegistry,
auto_install_deps: bool = False
) -> dict:
results = {'loaded': [], 'failed': []}
manifests = discover_plugins(plugin_dir)
for manifest in manifests:
name = manifest.get('name', '?')
is_valid, errors = validate_manifest(manifest)
if not is_valid:
print(f'INVALID {name}: {errors}')
results['failed'].append({'name': name, 'reason': errors})
continue
deps_ok = check_and_install_deps(manifest, auto_install_deps)
if not deps_ok:
results['failed'].append({'name': name, 'reason': 'missing_deps'})
continue
module = load_plugin_module(manifest)
if not module:
results['failed'].append({'name': name, 'reason': 'import_error'})
continue
tools = extract_tools_from_module(module, manifest['tool_definitions'])
registry.register_plugin(manifest, tools)
results['loaded'].append(name)
print(f'Plugins loaded: {results["loaded"]}')
print(f'Plugins failed: {results["failed"]}')
return resultsプラグインのライフサイクルイベントのログ記録
プラグインのライフサイクルにおける重要なイベントをすべてログに記録してください。検出、検証失敗、読み込み成功、ホットリロード、アンロードが対象です。これらのログは、本番環境でプラグインシステムをデバッグおよび監査するために不可欠です。
import logging
from datetime import datetime
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
plugin_logger = logging.getLogger('plugin_system')
def log_plugin_event(event: str, plugin_name: str, details: dict = None):
entry = {
'event': event,
'plugin': plugin_name,
'timestamp': datetime.utcnow().isoformat(),
'details': details or {}
}
if event in ('VALIDATION_FAILED', 'LOAD_FAILED', 'SIGNATURE_INVALID'):
plugin_logger.warning('Plugin event: %s', entry)
else:
plugin_logger.info('Plugin event: %s', entry)
# Usage:
log_plugin_event('DISCOVERED', 'weather-tools', {'path': 'plugins/weather-tools'})
log_plugin_event('LOADED', 'weather-tools', {'tools': ['get_current_weather']})
log_plugin_event('VALIDATION_FAILED', 'bad-plugin', {'errors': ['Missing entry_point']})
if __name__ == '__main__':
import sys
plugin_logger.addHandler(logging.StreamHandler(sys.stdout))
log_plugin_event('LOADED', 'demo-plugin', {'tools': ['get_weather']})
理解度チェック
プラグインマニフェストのentry_pointフィールドには、どのような目的がありますか？
要約：プラグインの検出と登録
すばらしいです！学習した内容：
- プラグインマニフェスト：name、version、entry_point、tool_definitions、requiresを含むplugin.json
- 検出：plugin.jsonを含むサブディレクトリをスキャン
- 検証：必須フィールド、バージョン形式、環境変数、空でないツールを確認
- 動的読み込み：entry_pointのパスを使ったimportlib.import_module()
- ツールレジストリ：ツール名をスキーマとexecute関数に対応付ける中央マップ
- ホットリロード：watchdogでプラグインディレクトリを監視し、マニフェストの変更時に再読み込み
次の内容：ツールのバージョン管理と互換性管理
よくある質問
「プラグインの検出と登録」レッスンは無料ですか？
はい。「プラグインの検出と登録」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「プラグインの検出と登録」で何を学びますか？
ツールレジストリ、マニフェストファイル、実行時の動的なツール読み込みを扱います。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。
「プラグインの検出と登録」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。