dragon.ai.agent.core.base.DragonAgent

class DragonAgent

Bases: ABC

Stateless, persistent base agent running on a node.

Each DragonAgent instance owns:

• Its own Dragon input queue (wrapped by comm).

• Its own LLM proxy — a lightweight client handle pointing at a shared inference pipeline, reused across all tasks.

• A ToolRegistry of inline tools.

The agent holds no per-task state. All task context, tool history, and results are stored in a shared Dragon Distributed Dictionary (DDict) that is created per batch run and passed to the agent via the message header.

LLM proxy resolution (in priority order):

1. config.inference_queue — auto-creates a DragonQueueLLMProxy per agent process. This is the recommended path.

2. Neither — self.llm is None, agent has no LLM.

Parameters

config:

Agent identity, role, model, and inference parameters. Set config.inference_queue to the Dragon Queue feeding the inference pipeline so a proxy is auto-created.

tool_registry:

Tools available for inline invocation during LLM reasoning.

protocol:

Communication protocol class (default: DragonQueueProtocol).

mcp_servers:

Optional list of MCP server configurations for remote tools.

shutdown_event:

Event signalled by the parent to stop the agent listen loop.

__init__(config: AgentConfig, tool_registry: ToolRegistry, protocol: CommunicationProtocol = <class 'dragon.ai.agent.communication.dragon_comm.DragonQueueProtocol'>, mcp_servers: list [MCPServerConfig] | None = None, shutdown_event=None) → None

Methods

__init__(config, tool_registry, protocol, ...)
list_tools()	Return all tools available to this agent, separated by source.
listen()	Listening task from dragon queue or a port.
process(task_id, header)	Process a single task.

__init__(config: AgentConfig, tool_registry: ToolRegistry, protocol: CommunicationProtocol = <class 'dragon.ai.agent.communication.dragon_comm.DragonQueueProtocol'>, mcp_servers: list [MCPServerConfig] | None = None, shutdown_event=None) → None

abstractmethod listen() → None

Listening task from dragon queue or a port.

list_tools() → dict [str , Any ]

Return all tools available to this agent, separated by source.

Returns a dict with two keys:

• "registry" — local Python tools registered in tool_registry. Each entry is an OpenAI-compatible schema dict as returned by to_schema().

• "mcp" — remote MCP tools owned by the internal ToolDispatcher, grouped by server alias. Each alias maps to a list of OpenAI-compatible schema dicts. Empty {} when no dispatcher or no MCP servers are connected.

Example output:

{
    "registry": [
        {"type": "function", "function": {"name": "web_search", ...}},
        {"type": "function", "function": {"name": "calculate_word_count", ...}},
    ],
    "mcp": {
        "jupyter": [
            {"type": "function", "function": {"name": "jupyter__create_notebook", ...}},
        ],
    },
}

abstractmethod async process(task_id: str , header: Any ) → dict [str , Any ]

Process a single task.

Parameters

task_id:

Unique identifier for this batch run / user request.

header:

Implementation-defined header object carrying task context. For SubAgent this is a DispatchHeader.

Returns

dict

Result payload to be written to DDict under the agent’s result key.