feat(agent): introduce AgentBase Protocol as the interface for agent classes to implement

Author: awsarron

src/strands/__init__.py

@@ -2,11 +2,13 @@
 
 from . import agent, models, telemetry, types
 from .agent.agent import Agent
+from .agent.base import AgentBase
 from .tools.decorator import tool
 from .types.tools import ToolContext
 
 __all__ = [
     "Agent",
+    "AgentBase",
     "agent",
     "models",
     "tool",

src/strands/agent/__init__.py

@@ -8,6 +8,7 @@
 
 from .agent import Agent
 from .agent_result import AgentResult
+from .base import AgentBase
 from .conversation_manager import (
     ConversationManager,
     NullConversationManager,
@@ -17,6 +18,7 @@
 
 __all__ = [
     "Agent",
+    "AgentBase",
     "AgentResult",
     "ConversationManager",
     "NullConversationManager",

src/strands/agent/agent.py

@@ -89,7 +89,7 @@ class _DefaultCallbackHandlerSentinel:
 
 
 class Agent:
-    """Core Agent interface.
+    """Core Agent implementation.
 
     An agent orchestrates the following workflow:
 

src/strands/agent/base.py

@@ -0,0 +1,66 @@
+"""Agent Interface.
+
+Defines the minimal interface that all agent types must implement.
+"""
+
+from typing import Any, AsyncIterator, Protocol, runtime_checkable
+
+from ..types.agent import AgentInput
+from .agent_result import AgentResult
+
+
+@runtime_checkable
+class AgentBase(Protocol):
+    """Protocol defining the interface for all agent types in Strands.
+
+    This protocol defines the minimal contract that all agent implementations
+    must satisfy.
+    """
+
+    async def invoke_async(
+        self,
+        prompt: AgentInput = None,
+        **kwargs: Any,
+    ) -> AgentResult:
+        """Asynchronously invoke the agent with the given prompt.
+
+        Args:
+            prompt: Input to the agent.
+            **kwargs: Additional arguments.
+
+        Returns:
+            AgentResult containing the agent's response.
+        """
+        ...
+
+    def __call__(
+        self,
+        prompt: AgentInput = None,
+        **kwargs: Any,
+    ) -> AgentResult:
+        """Synchronously invoke the agent with the given prompt.
+
+        Args:
+            prompt: Input to the agent.
+            **kwargs: Additional arguments.
+
+        Returns:
+            AgentResult containing the agent's response.
+        """
+        ...
+
+    def stream_async(
+        self,
+        prompt: AgentInput = None,
+        **kwargs: Any,
+    ) -> AsyncIterator[Any]:
+        """Stream agent execution asynchronously.
+
+        Args:
+            prompt: Input to the agent.
+            **kwargs: Additional arguments.
+
+        Yields:
+            Events representing the streaming execution.
+        """
+        ...