fix: switch to enum

Author: zastrowm

src/strands/agent/agent.py

@@ -633,7 +633,7 @@ async def stream_async(
         # Using threading.Lock instead of asyncio.Lock because run_async() creates
         # separate event loops in different threads
         lock_acquired = False
-        if self._concurrent_invocation_mode == "throw":
+        if self._concurrent_invocation_mode == ConcurrentInvocationMode.THROW:
             lock_acquired = self._invocation_lock.acquire(blocking=False)
             if not lock_acquired:
                 raise ConcurrencyException(

src/strands/types/agent.py

@@ -3,21 +3,26 @@
 This module defines the types used for an Agent.
 """
 
-from typing import Literal, TypeAlias
+from enum import Enum
+from typing import TypeAlias
 
 from .content import ContentBlock, Messages
 from .interrupt import InterruptResponseContent
 
 AgentInput: TypeAlias = str | list[ContentBlock] | list[InterruptResponseContent] | Messages | None
 
-ConcurrentInvocationMode = Literal["throw", "unsafe_reentrant"]
-"""Mode controlling concurrent invocation behavior.
 
-Values:
-    throw: Raises ConcurrencyException if concurrent invocation is attempted (default).
-    unsafe_reentrant: Allows concurrent invocations without locking.
+class ConcurrentInvocationMode(str, Enum):
+    """Mode controlling concurrent invocation behavior.
 
-Warning:
-    The ``unsafe_reentrant`` mode makes no guarantees about resulting behavior and is
-    provided only for advanced use cases where the caller understands the risks.
-"""
+    Values:
+        THROW: Raises ConcurrencyException if concurrent invocation is attempted (default).
+        UNSAFE_REENTRANT: Allows concurrent invocations without locking.
+
+    Warning:
+        The ``UNSAFE_REENTRANT`` mode makes no guarantees about resulting behavior and is
+        provided only for advanced use cases where the caller understands the risks.
+    """
+
+    THROW = "throw"
+    UNSAFE_REENTRANT = "unsafe_reentrant"

tests/strands/agent/test_agent.py

@@ -2382,6 +2382,22 @@ def test_agent_concurrent_invocation_mode_stores_value():
     assert agent_reentrant._concurrent_invocation_mode == "unsafe_reentrant"
 
 
+def test_agent_concurrent_invocation_mode_accepts_enum():
+    """Test that concurrent_invocation_mode accepts enum values as well as strings."""
+    from strands.types.agent import ConcurrentInvocationMode
+
+    model = MockedModelProvider([{"role": "assistant", "content": [{"text": "hello"}]}])
+
+    # Using enum values
+    agent_throw = Agent(model=model, concurrent_invocation_mode=ConcurrentInvocationMode.THROW)
+    assert agent_throw._concurrent_invocation_mode == "throw"
+    assert agent_throw._concurrent_invocation_mode == ConcurrentInvocationMode.THROW
+
+    agent_reentrant = Agent(model=model, concurrent_invocation_mode=ConcurrentInvocationMode.UNSAFE_REENTRANT)
+    assert agent_reentrant._concurrent_invocation_mode == "unsafe_reentrant"
+    assert agent_reentrant._concurrent_invocation_mode == ConcurrentInvocationMode.UNSAFE_REENTRANT
+
+
 @pytest.mark.asyncio
 async def test_agent_sequential_invocations_work():
     """Test that sequential invocations work correctly after lock is released."""