Agent Context (AgentCtx)

AgentCtx is one of the most important concepts in Nekro Agent plugin development. It encapsulates all context information needed by the Agent when executing tasks, providing plugins with a unified interface to access relevant data and functionality.

What is AgentCtx?

AgentCtx (Agent Context) is a context object that contains all key information of the current execution environment. Whether processing messages from chat software or responding to Webhook events, AgentCtx provides a unified interface for plugins to interact with Nekro Agent's core functionality.

In plugin development, AgentCtx is typically provided as the first parameter of sandbox methods (conventionally using _ctx as the variable name).

Core Properties

Basic Information Properties

# Sandbox and chat identifiers
_ctx.from_chat_key: Optional[str]     # Unique identifier of the source chat
_ctx.chat_key: Optional[str]          # Unique ID of the chat channel (same as from_chat_key)
_ctx.container_key: Optional[str]     # Unique identifier of the sandbox container

# Chat channel information
_ctx.channel_id: Optional[str]        # Original platform ID of the channel (such as QQ group number or user ID)
_ctx.channel_name: Optional[str]      # Channel name (such as QQ group name or username)
_ctx.channel_type: Optional[str]      # Channel type (such as 'group' or 'private')
_ctx.adapter_key: Optional[str]       # Adapter identifier (such as 'onebot_v11')

# Webhook related (when triggered by Webhook)
_ctx.webhook_request: Optional[WebhookRequest]  # Webhook request data

Important Property Accessors

chat_key Property

@property
def chat_key(self) -> str:
    """Unique ID of the chat channel"""

This is the unique identifier of the current chat channel, typically composed of adapter_key and channel_id.

Usage Example:

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_chat_info", "Get current chat channel information")
async def get_chat_info(_ctx: AgentCtx) -> str:
    return f"Current chat channel: {_ctx.chat_key}"
    # Output example: "Current chat channel: onebot_v11-group_12345678"

adapter Property

@property
def adapter(self) -> "BaseAdapter":
    """Message-related adapter instance"""

Through this adapter instance, plugins can obtain adapter-related information or call adapter-related methods.

db_chat_channel and db_user Properties

@property
def db_chat_channel(self) -> Optional["DBChatChannel"]:
    """Database instance of the current chat channel"""

@property
def db_user(self) -> Optional["DBUser"]:
    """Database user instance that triggered this Agent"""

These properties provide access to underlying database models for advanced data operations.

Core Functionality Modules

File System (fs)

_ctx.fs provides powerful file handling capabilities for securely transferring files between plugins and AI.

@property
def fs(self) -> FileSystem:
    """File system tools"""

Main Methods:

mixed_forward_file - Plugin Transfers Files to AI

Used when a plugin needs to generate a file (such as an image, document) and return it to the AI:

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "generate_chart", "Generate chart and return to AI")
async def generate_chart(_ctx: AgentCtx, data: str) -> str:
    # Plugin obtains an image through API
    image_url = "https://example.com/chart.png"

    # Convert it to a sandbox path usable by AI
    sandbox_path = await _ctx.fs.mixed_forward_file(image_url, file_name="chart.png")

    # Return sandbox path to AI
    return sandbox_path  # "/app/uploads/chart.png"

get_file - AI Transfers Files to Plugin

When AI calls a plugin and passes a sandbox file path as a parameter, the plugin uses this method to get the real path on the host machine:

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "analyze_image", "Analyze image content")
async def analyze_image(_ctx: AgentCtx, image_path: str) -> str:
    # Sandbox path provided by AI: "/app/shared/photo.jpg"
    # Convert to real path accessible by host machine
    host_path = _ctx.fs.get_file(image_path)

    # Now the file can be read
    with open(host_path, "rb") as f:
        file_content = f.read()
        # Perform image analysis...

    return "Image analysis completed"

Message Module (ms)

_ctx.ms provides direct access to the underlying message module, mainly used for advanced scenarios that require manually specifying chat_key and actively triggering AI responses.

@property
def ms(self):
    """Message module"""

Main Methods:

send_text - Send Text Message to Specified Channel

async def send_text(self, chat_key: str, content: str, ctx: AgentCtx) -> None:
    """Send text message to specified chat channel"""

push_system - Push System Message and Trigger AI Response

async def push_system(self, chat_key: str, message: str, ctx: AgentCtx, trigger_agent: bool = False) -> None:
    """Push message to system, optionally triggering AI response"""

Parameter Description:

• chat_key: Target chat channel identifier
• message: System message content to push
• ctx: Agent context
• trigger_agent: Whether to trigger AI to generate a reply (default False)

Usage Scenarios:

Sending Messages to Other Chat Channels

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "notify_admin", "Notify administrator")
async def notify_admin(_ctx: AgentCtx, message: str) -> str:
    # Send status update to monitoring channel
    monitor_chat_key = "onebot_v11-group_987654321"

    await _ctx.ms.send_text(monitor_chat_key, f"System notification: {message}", _ctx)

    return "Administrator notified"

Triggering AI Response Across Chat Channels

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "report_to_monitoring", "Report to monitoring group")
async def report_to_monitoring(_ctx: AgentCtx, incident: str) -> str:
    # Push event to monitoring group and trigger AI analysis
    monitoring_chat = "onebot_v11-group_monitor123"
    analysis_prompt = f"System event detected: {incident}\nPlease analyze impact and provide handling suggestions."

    await _ctx.ms.push_system(
        chat_key=monitoring_chat,
        message=analysis_prompt,
        ctx=_ctx,
        trigger_agent=True  # Trigger AI analysis in monitoring group
    )

    return "Reported to monitoring group, AI is analyzing"

Convenience Methods

Message Sending Methods

send_text - Send Text Message

async def send_text(self, content: str, *, record: bool = True):
    """Send text message to current chat channel"""

Parameter Description:

• content: Text content to send
• record: Whether to record in conversation history (default True)

Usage Example:

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_notification", "Send notification")
async def send_notification(_ctx: AgentCtx, message: str) -> str:
    await _ctx.send_text(f"📢 Notification: {message}")
    return "Notification sent"

send_image - Send Image

async def send_image(self, file_path: str, *, record: bool = True):
    """Send image to current chat channel"""

Usage Example:

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_chart", "Send chart")
async def send_chart(_ctx: AgentCtx, chart_url: str) -> str:
    # Convert to sandbox path
    image_path = await _ctx.fs.mixed_forward_file(chart_url, file_name="chart.png")

    # Send image
    await _ctx.send_image(image_path)

    return "Chart sent"

send_file - Send File

async def send_file(self, file_path: str, *, record: bool = True):
    """Send file to current chat channel"""

Usage Example:

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_report", "Send report")
async def send_report(_ctx: AgentCtx, report_content: str) -> str:
    # Create file in shared directory
    file_on_host = _ctx.fs.shared_path / "report.txt"
    with open(file_on_host, "w", encoding="utf-8") as f:
        f.write(report_content)

    # Convert to sandbox path
    file_path = _ctx.fs.forward_file(file_on_host)

    # Send file
    await _ctx.send_file(file_path)

    return "Report sent"

push_system - Push System Message to Current Channel

async def push_system(self, message: str, *, trigger_agent: bool = False):
    """Push system message to current chat channel, optionally triggering AI response"""

Parameter Description:

• message: System message content to push
• trigger_agent: Whether to trigger AI to generate a reply (default False)

This is a convenient form of _ctx.ms.push_system(), automatically using the current chat_key.

Usage Example:

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "schedule_analysis", "Scheduled analysis")
async def schedule_analysis(_ctx: AgentCtx, report_data: str) -> str:
    # Asynchronously trigger AI analysis of current session (such as after scheduled task completion)
    analysis_prompt = f"Scheduled data collection completed, below is report data:\n{report_data}\nPlease analyze and summarize key points."

    await _ctx.push_system(analysis_prompt, trigger_agent=True)

    return "Data collection completed, AI analysis started"

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "cross_chat_notify", "Cross-group notification")
async def cross_chat_notify(_ctx: AgentCtx, target_chat: str, event: str) -> str:
    # Push event to other chat group and trigger AI processing
    notification = f"Event notification from {_ctx.channel_name}: {event}\nPlease evaluate the importance of this event."

    await _ctx.ms.push_system(
        chat_key=target_chat,
        message=notification,
        ctx=_ctx,
        trigger_agent=True
    )

    return f"Event notification sent to {target_chat}"

Configuration Access

get_core_config - Get Core Configuration

async def get_core_config(self) -> CoreConfig:
    """Get currently effective core configuration instance"""

Core configuration is generated by mixing three layers of configuration: system basic settings → adapter settings → chat channel settings, with priority increasing in order.

Usage Example:

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "check_cloud_status", "Check cloud service status")
async def check_cloud_status(_ctx: AgentCtx) -> str:
    config = await _ctx.get_core_config()

    if config.ENABLE_NEKRO_CLOUD:
        return "Cloud service enabled"
    else:
        return "Cloud service disabled"

Adapter-Specific Methods

get_onebot_v11_bot - Get OneBot V11 Bot

async def get_onebot_v11_bot(self) -> OneBotV11Bot:
    """Get OneBot V11 Bot instance (only applicable to OneBot V11 adapter)"""

Usage Example:

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_private_msg", "Send private message")
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"Private message sent to user {user_id}"
    else:
        return "Current adapter does not support this feature"

Best Practices

1. Error Handling

Always check if key properties exist:

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_channel_info", "Get channel information")
async def get_channel_info(_ctx: AgentCtx) -> str:
    if not _ctx.channel_id:
        return "Error: Unable to get channel information"

    return f"Channel ID: {_ctx.channel_id}, Channel name: {_ctx.channel_name}"

2. File Path Handling

Use fs tools for all file operations:

# Correct approach
sandbox_path = await _ctx.fs.mixed_forward_file(url, file_name="image.png")

# Wrong approach - do not directly return URLs or host paths to AI
# return url  # AI may have difficulty accessing external URLs
# return "/host/path/file.png"  # AI cannot access host paths

3. Message Recording Control

For user-facing prompt messages only, you can set record=False:

# Progress prompt messages are not recorded to AI conversation context
await _ctx.send_text("Processing, please wait...", record=False)

# Prompt messages that require AI participation or awareness of results are recorded to AI conversation context
await _ctx.send_text("Processing completed! Results are as follows:...", record=True)

4. Adapter Compatibility

Check adapter type before using adapter-specific features:

if _ctx.adapter_key == "onebot_v11":
    # Use OneBot V11 specific features
    bot = await _ctx.get_onebot_v11_bot()
    # ...
else:
    # Use universal features
    await _ctx.send_text("This feature requires OneBot V11 adapter")

5. Usage Scenarios for Actively Triggering AI Responses

Reasonably use push_system and trigger_agent=True parameters:

# ✅ Scenarios suitable for triggering AI:

# 1. Asynchronous data processing completed, need AI to analyze results
await _ctx.push_system("Data collection completed, please analyze trends...", trigger_agent=True)

# 2. Cross-chat channel event notifications
await _ctx.ms.push_system(other_chat, "Emergency event received, need handling suggestions", _ctx, trigger_agent=True)

# 3. Scheduled tasks trigger AI summaries
await _ctx.push_system("Daily report data ready, please generate summary", trigger_agent=True)

# 4. Webhook events need AI response
await _ctx.push_system("External system status changed, please evaluate impact", trigger_agent=True)

# ❌ Scenarios not suitable for triggering AI:
# - Pure status notifications or log recording
await _ctx.push_system("Plugin status update completed", trigger_agent=False)

# - Simple user operation confirmations
await _ctx.push_system("Operation completed", trigger_agent=False)

Notes:

• trigger_agent=True will consume AI model call quotas, please use it reasonably
• Mainly used for asynchronous triggering (scheduled tasks, Webhooks, etc.) and cross-chat application scenarios
• Avoid self-triggering in the same execution flow to prevent circular calls
• Ensure pushed message content can provide valuable contextual information for AI

Through reasonable use of AgentCtx, plugins can seamlessly integrate with various components of Nekro Agent, providing powerful and stable functional extensions.