Skip to main content

🏗️ BaseAgent

Abstract Base Class | Domain: Core

Abstract base class that all KOSMOS agents inherit from, defining the common interface and lifecycle.

Overview​

KOSMOS V2.0 Base Agent

Abstract base class that all KOSMOS agents inherit from. Integrated with Global Tool Registry (GTR) for dynamic MCP tool discovery.

FIXES APPLIED:

  • Issue #1: Changed tool path parsing from "/" to "." (dot notation)
  • Issue #3a: Fixed tool.server_name to tool.server
  • Issue #3b: Fixed result.data to result.result in call_mcp_simple
  • Issue #3c: Removed invalid tool_path field from ToolCallResult
  • Issue #4: Use get_tool_registry() instead of direct GlobalToolRegistry instantiation
Auto-Generated

This documentation is automatically extracted from source code. Source: implementation/backend/agents/base.py (521 lines)

Enumerations​

AgentState​

Agent lifecycle states.

MemberValue

Data Models​

AgentConfig​

Configuration for an agent.

@dataclass
class AgentConfig:

AgentMessage​

Message format for inter-agent communication.

@dataclass
class AgentMessage:

AgentMetrics​

Metrics for agent performance tracking.

@dataclass
class AgentMetrics:

BaseAgent​

Abstract base class for all KOSMOS agents.

All 11 KOSMOS agents inherit from this class and implement the required abstract methods.

Integrated with:

  • Global Tool Registry (GTR) for dynamic MCP tool discovery
  • Circuit Breaker pattern for resilient tool calls
  • Automatic tool capability filtering by category

Inherits from: ABC

Methods​

__init__(config: AgentConfig)​

async initialize() → None​

Initialize the agent and its dependencies.

async process(message: AgentMessage) → Dict[str, Any] (abstract)​

Process an incoming message.

async call_tool(tool_name: str, params: Dict[str, Any]) → Any​

Execute a local tool with automatic metrics tracking.

async call_mcp(tool_path: str, params: Dict[str, Any], timeout: Optional[float]) → ToolCallResult​

Call an MCP tool via the Global Tool Registry.

Args: tool_path: Full tool path using DOT notation (e.g., "postgres-mcp.query") params: Tool parameters timeout: Optional timeout in seconds

Returns: ToolCallResult with success status and result/error

async call_mcp_simple(server: str, tool: str, params: Dict[str, Any]) → Any​

Simplified MCP call that returns data directly or raises exception.

Args: server: MCP server name (e.g., "postgres-mcp") tool: Tool name (e.g., "query") params: Tool parameters

Returns: Tool result data

Raises: RuntimeError: If tool call fails

get_available_mcp_tools() → List[Dict[str, Any]]​

Get list of MCP tools available to this agent.

get_tools_for_llm() → List[Dict[str, Any]]​

Get tool definitions formatted for LLM function calling.

Returns list suitable for OpenAI/Anthropic tool definitions.

async delegate_to(agent_id: str, payload: Dict[str, Any], timeout: float) → Optional[Dict[str, Any]]​

Delegate a task to another agent.

async emit_event(event_type: str, payload: Dict[str, Any]) → None​

Emit an event to the event stream.

async shutdown() → None​

Gracefully shutdown the agent.

get_status() → Dict[str, Any]​

Get agent status information.

get_circuit_breaker_status() → Dict[str, Any]​

Get circuit breaker status for all MCP servers this agent uses.