エージェントコンテキスト (AgentCtx)
AgentCtx はNekro Agentプラグイン開発において最も重要な概念の一つです。エージェントがタスクを実行する際に必要なすべてのコンテキスト情報をカプル化し、プラグインが関連データと機能にアクセスするための統一されたインターフェースを提供します。
AgentCtxとは？
AgentCtx（エージェントコンテキスト）は、現在の実行環境のすべての主要情報を含むコンテキストオブジェクトです。チャットソフトからのメッセージを処理する場合でも、Webhookイベントに応答する場合でも、AgentCtx はプラグインがNekro Agentのコア機能と対話するための統一されたインターフェースを提供します。
プラグイン開発では、AgentCtx は通常、サンドボックスメソッドの最初のパラメータとして提供されます（慣例的に _ctx を変数名として使用します）。
コアプロパティ
基本情報プロパティ
python
# サンドボックスとチャット識別子
_ctx.from_chat_key: Optional[str] # ソースチャットの一意の識別子
_ctx.chat_key: Optional[str] # チャットチャネルの一意のID（from_chat_keyと同じ）
_ctx.container_key: Optional[str] # サンドボックスコンテナの一意の識別子
# チャットチャネル情報
_ctx.channel_id: Optional[str] # チャネルの元プラットフォームID（QQグループ番号やユーザーIDなど）
_ctx.channel_name: Optional[str] # チャネル名（QQグループ名やユーザー名など）
_ctx.channel_type: Optional[str] # チャネルタイプ（'group'や'private'など）
_ctx.adapter_key: Optional[str] # アダプター識別子（'onebot_v11'など）
# Webhook関連（Webhookによってトリガーされた場合）
_ctx.webhook_request: Optional[WebhookRequest] # Webhookリクエストデータ重要なプロパティアクセサ
chat_key プロパティ
python
@property
def chat_key(self) -> str:
"""チャットチャネルの一意のID"""これは現在のチャットチャネルの一意の識別子で、通常は adapter_key と channel_id で構成されます。
使用例：
python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_chat_info", "現在のチャットチャネル情報を取得")
async def get_chat_info(_ctx: AgentCtx) -> str:
return f"現在のチャットチャネル: {_ctx.chat_key}"
# 出力例: "現在のチャットチャネル: onebot_v11-group_12345678"adapter プロパティ
python
@property
def adapter(self) -> "BaseAdapter":
"""メッセージ関連のアダプターインスタンス"""このアダプターインスタンスを通じて、プラグインはアダプター関連情報を取得したり、アダプター関連メソッドを呼び出したりできます。
db_chat_channel と db_user プロパティ
python
@property
def db_chat_channel(self) -> Optional["DBChatChannel"]:
"""現在のチャットチャネルのデータベースインスタンス"""
@property
def db_user(self) -> Optional["DBUser"]:
"""このエージェントをトリガーしたデータベースユーザーインスタンス"""これらのプロパティは、高度なデータ操作のための基礎となるデータベースモデルへのアクセスを提供します。
コア機能モジュール
ファイルシステム (fs)
_ctx.fs は、プラグインとAI間で安全にファイルを転送するための強力なファイル処理機能を提供します。
python
@property
def fs(self) -> FileSystem:
"""ファイルシステムツール"""主なメソッド：
mixed_forward_file - プラグインからAIへファイルを転送
プラグインがファイル（画像、ドキュメントなど）を生成し、それをAIに返す必要がある場合に使用します：
python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "generate_chart", "チャートを生成してAIに返す")
async def generate_chart(_ctx: AgentCtx, data: str) -> str:
# プラグインがAPIを通じて画像を取得
image_url = "https://example.com/chart.png"
# AIが使用できるサンドボックスパスに変換
sandbox_path = await _ctx.fs.mixed_forward_file(image_url, file_name="chart.png")
# サンドボックスパスをAIに返す
return sandbox_path # "/app/uploads/chart.png"get_file - AIからプラグインへファイルを転送
AIがプラグインを呼び出し、サンドボックスファイルパスをパラメータとして渡す場合、プラグインはこのメソッドを使用してホストマシン上の実際のパスを取得します：
python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "analyze_image", "画像コンテンツを分析")
async def analyze_image(_ctx: AgentCtx, image_path: str) -> str:
# AIが提供したサンドボックスパス: "/app/shared/photo.jpg"
# ホストマシンがアクセス可能な実際のパスに変換
host_path = _ctx.fs.get_file(image_path)
# これでファイルを読み取り可能
with open(host_path, "rb") as f:
file_content = f.read()
# 画像分析を実行...
return "画像分析完了"メッセージモジュール (ms)
_ctx.ms は、基礎となるメッセージモジュールへの直接アクセスを提供し、主に手動で chat_key を指定し、能動的にAI応答をトリガーする必要がある高度なシナリオで使用されます。
python
@property
def ms(self):
"""メッセージモジュール"""主なメソッド：
send_text - 指定チャネルにテキストメッセージを送信
python
async def send_text(self, chat_key: str, content: str, ctx: AgentCtx) -> None:
"""指定されたチャットチャネルにテキストメッセージを送信"""push_system - システムメッセージをプッシュしてAI応答をトリガー
python
async def push_system(self, chat_key: str, message: str, ctx: AgentCtx, trigger_agent: bool = False) -> None:
"""メッセージをシステムにプッシュし、オプションでAI応答をトリガー"""パラメータ説明：
chat_key: ターゲットチャットチャネル識別子message: プッシュするシステムメッセージコンテンツctx: エージェントコンテキストtrigger_agent: AIによる返信生成をトリガーするかどうか（デフォルトFalse）
使用シナリオ：
他のチャットチャネルにメッセージを送信
python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "notify_admin", "管理者に通知")
async def notify_admin(_ctx: AgentCtx, message: str) -> str:
# 監視チャネルにステータス更新を送信
monitor_chat_key = "onebot_v11-group_987654321"
await _ctx.ms.send_text(monitor_chat_key, f"システム通知: {message}", _ctx)
return "管理者に通知しました"チャットチャネルを超えてAI応答をトリガー
python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "report_to_monitoring", "監視グループに報告")
async def report_to_monitoring(_ctx: AgentCtx, incident: str) -> str:
# イベントを監視グループにプッシュし、AI分析をトリガー
monitoring_chat = "onebot_v11-group_monitor123"
analysis_prompt = f"システムイベントが検出されました: {incident}\n影響を分析し、対応提案を提供してください。"
await _ctx.ms.push_system(
chat_key=monitoring_chat,
message=analysis_prompt,
ctx=_ctx,
trigger_agent=True # 監視グループでAI分析をトリガー
)
return "監視グループに報告しました。AIが分析中です"便利メソッド
メッセージ送信メソッド
send_text - テキストメッセージを送信
python
async def send_text(self, content: str, *, record: bool = True):
"""現在のチャットチャネルにテキストメッセージを送信"""パラメータ説明：
content: 送信するテキストコンテンツrecord: 会話履歴に記録するかどうか（デフォルトTrue）
使用例：
python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_notification", "通知を送信")
async def send_notification(_ctx: AgentCtx, message: str) -> str:
await _ctx.send_text(f"📢 通知: {message}")
return "通知を送信しました"send_image - 画像を送信
python
async def send_image(self, file_path: str, *, record: bool = True):
"""現在のチャットチャネルに画像を送信"""使用例：
python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_chart", "チャートを送信")
async def send_chart(_ctx: AgentCtx, chart_url: str) -> str:
# サンドボックスパスに変換
image_path = await _ctx.fs.mixed_forward_file(chart_url, file_name="chart.png")
# 画像を送信
await _ctx.send_image(image_path)
return "チャートを送信しました"send_file - ファイルを送信
python
async def send_file(self, file_path: str, *, record: bool = True):
"""現在のチャットチャネルにファイルを送信"""使用例：
python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_report", "レポートを送信")
async def send_report(_ctx: AgentCtx, report_content: str) -> str:
# 共有ディレクトリにファイルを作成
file_on_host = _ctx.fs.shared_path / "report.txt"
with open(file_on_host, "w", encoding="utf-8") as f:
f.write(report_content)
# サンドボックスパスに変換
file_path = _ctx.fs.forward_file(file_on_host)
# ファイルを送信
await _ctx.send_file(file_path)
return "レポートを送信しました"push_system - 現在のチャネルにシステムメッセージをプッシュ
python
async def push_system(self, message: str, *, trigger_agent: bool = False):
"""現在のチャットチャネルにシステムメッセージをプッシュし、オプションでAI応答をトリガー"""パラメータ説明：
message: プッシュするシステムメッセージコンテンツtrigger_agent: AIによる返信生成をトリガーするかどうか（デフォルトFalse）
これは _ctx.ms.push_system() の便利な形式で、現在の chat_key を自動的に使用します。
使用例：
python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "schedule_analysis", "スケジュール分析")
async def schedule_analysis(_ctx: AgentCtx, report_data: str) -> str:
# 現在のセッションのAI分析を非同期にトリガー（スケジュールタスク完了後など）
analysis_prompt = f"スケジュールされたデータ収集が完了しました。以下はレポートデータです:\n{report_data}\n分析して要点を要約してください。"
await _ctx.push_system(analysis_prompt, trigger_agent=True)
return "データ収集完了、AI分析を開始しました"
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "cross_chat_notify", "クロスグループ通知")
async def cross_chat_notify(_ctx: AgentCtx, target_chat: str, event: str) -> str:
# イベントを他のチャットグループにプッシュし、AI処理をトリガー
notification = f"{_ctx.channel_name}からのイベント通知: {event}\nこのイベントの重要性を評価してください。"
await _ctx.ms.push_system(
chat_key=target_chat,
message=notification,
ctx=_ctx,
trigger_agent=True
)
return f"イベント通知を {target_chat} に送信しました"設定アクセス
get_core_config - コア設定を取得
python
async def get_core_config(self) -> CoreConfig:
"""現在有効なコア設定インスタンスを取得"""コア設定は、3層の設定（システム基本設定 → アダプター設定 → チャットチャネル設定）を混合して生成され、優先度はこの順序で高くなります。
使用例：
python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "check_cloud_status", "クラウドサービス状態を確認")
async def check_cloud_status(_ctx: AgentCtx) -> str:
config = await _ctx.get_core_config()
if config.ENABLE_NEKRO_CLOUD:
return "クラウドサービスが有効です"
else:
return "クラウドサービスが無効です"アダプター固有メソッド
get_onebot_v11_bot - OneBot V11ボットを取得
python
async def get_onebot_v11_bot(self) -> OneBotV11Bot:
"""OneBot V11ボットインスタンスを取得（OneBot V11アダプターにのみ適用）"""使用例：
python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_private_msg", "プライベートメッセージを送信")
async def send_private_message(_ctx: AgentCtx, user_id: int, message: str) -> str:
if _ctx.adapter_key == "onebot_v11":
bot = await _ctx.get_onebot_v11_bot()
await bot.send_private_msg(user_id=user_id, message=message)
return f"ユーザー {user_id} にプライベートメッセージを送信しました"
else:
return "現在のアダプターはこの機能をサポートしていません"ベストプラクティス
1. エラーハンドリング
常に主要なプロパティが存在するかチェックしてください：
python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_channel_info", "チャネル情報を取得")
async def get_channel_info(_ctx: AgentCtx) -> str:
if not _ctx.channel_id:
return "エラー: チャネル情報を取得できません"
return f"チャネルID: {_ctx.channel_id}, チャネル名: {_ctx.channel_name}"2. ファイルパス処理
すべてのファイル操作に fs ツールを使用してください：
python
# 正しいアプローチ
sandbox_path = await _ctx.fs.mixed_forward_file(url, file_name="image.png")
# 間違ったアプローチ - URLやホストパスを直接AIに返さないでください
# return url # AIは外部URLにアクセスできない場合があります
# return "/host/path/file.png" # AIはホストパスにアクセスできません3. メッセージ記録制御
ユーザー向けのプロンプトメッセージのみの場合、record=False を設定できます：
python
# 進行状況プロンプトメッセージはAI会話コンテキストに記録されない
await _ctx.send_text("処理中、お待ちください...", record=False)
# AIの参加や結果の認識が必要なプロンプトメッセージはAI会話コンテキストに記録される
await _ctx.send_text("処理完了！結果は以下の通りです...", record=True)4. アダプター互換性
アダプター固有機能を使用する前にアダプタータイプを確認してください：
python
if _ctx.adapter_key == "onebot_v11":
# OneBot V11固有機能を使用
bot = await _ctx.get_onebot_v11_bot()
# ...
else:
# 汎用機能を使用
await _ctx.send_text("この機能にはOneBot V11アダプターが必要です")5. AI応答を能動的にトリガーする使用シナリオ
push_system と trigger_agent=True パラメータを合理的に使用してください：
python
# ✅ AIトリガーに適したシナリオ：
# 1. 非同期データ処理完了後、AIに結果の分析が必要
await _ctx.push_system("データ収集完了、トレンドを分析してください...", trigger_agent=True)
# 2. チャットチャネルを超えるイベント通知
await _ctx.ms.push_system(other_chat, "緊急イベントを受信、対応提案が必要", _ctx, trigger_agent=True)
# 3. スケジュールタスクがAI要約をトリガー
await _ctx.push_system("日次レポートデータ準備完了、要約を生成してください", trigger_agent=True)
# 4. WebhookイベントがAI応答を必要とする
await _ctx.push_system("外部システム状態が変更、影響を評価してください", trigger_agent=True)
# ❌ AIトリガーに適さないシナリオ：
# - 純粋なステータス通知やログ記録
await _ctx.push_system("プラグイン状態更新完了", trigger_agent=False)
# - シンプルなユーザー操作確認
await _ctx.push_system("操作完了", trigger_agent=False)注意点：
trigger_agent=TrueはAIモデル呼び出しクォータを消費しますので、合理的に使用してください- 主に非同期トリガー（スケジュールタスク、Webhooksなど）とチャットを超えるアプリケーションシナリオで使用されます
- 同じ実行フローでの自己トリガーを避け、循環呼び出しを防止してください
- プッシュされたメッセージコンテンツがAIに価値あるコンテキスト情報を提供できることを確認してください
AgentCtx の合理的な使用を通じて、プラグインはNekro Agentの様々なコンポーネントとシームレスに統合し、強力で安定した機能拡張を提供できます。