centralize logic

Author: pgrayy

src/strands/_identifier.py

@@ -0,0 +1,30 @@
+"""Strands identifier utilities."""
+
+import enum
+import os
+
+
+class Identifier(enum.Enum):
+    """Strands identifier types."""
+
+    AGENT = "agent"
+    SESSION = "session"
+
+
+def validate(id_: str, type_: Identifier) -> str:
+    """Validate strands id.
+
+    Args:
+        id_: Id to validate.
+        type_: Type of the identifier (e.g., session id, agent id, etc.)
+
+    Returns:
+        Validated id.
+
+    Raises:
+        ValueError: If id contains path separators.
+    """
+    if os.path.basename(id_) != id_:
+        raise ValueError(f"{type_.value}_id={id_} | id cannot contain path separators")
+
+    return id_

src/strands/agent/agent.py

@@ -19,6 +19,7 @@
 from opentelemetry import trace as trace_api
 from pydantic import BaseModel
 
+from .. import _identifier
 from ..event_loop.event_loop import event_loop_cycle, run_tool
 from ..handlers.callback_handler import PrintingCallbackHandler, null_callback_handler
 from ..hooks import (
@@ -40,7 +41,6 @@
 from ..types.exceptions import ContextWindowOverflowException
 from ..types.tools import ToolResult, ToolUse
 from ..types.traces import AttributeValue
-from . import identifier
 from .agent_result import AgentResult
 from .conversation_manager import (
     ConversationManager,
@@ -258,7 +258,7 @@ def __init__(
         self.messages = messages if messages is not None else []
 
         self.system_prompt = system_prompt
-        self.agent_id = identifier.validate(agent_id or _DEFAULT_AGENT_ID)
+        self.agent_id = _identifier.validate(agent_id or _DEFAULT_AGENT_ID, _identifier.Identifier.AGENT)
         self.name = name or _DEFAULT_AGENT_NAME
         self.description = description
 

src/strands/agent/identifier.py

@@ -7,10 +7,9 @@
 import tempfile
 from typing import Any, Optional, cast
 
-from ..agent.identifier import validate as validate_agent_id
+from .. import _identifier
 from ..types.exceptions import SessionException
 from ..types.session import Session, SessionAgent, SessionMessage
-from .identifier import validate as validate_session_id
 from .repository_session_manager import RepositorySessionManager
 from .session_repository import SessionRepository
 
@@ -61,7 +60,7 @@ def _get_session_path(self, session_id: str) -> str:
         Raises:
             ValueError: If session id contains a path separator.
         """
-        session_id = validate_session_id(session_id)
+        session_id = _identifier.validate(session_id, _identifier.Identifier.SESSION)
         return os.path.join(self.storage_dir, f"{SESSION_PREFIX}{session_id}")
 
     def _get_agent_path(self, session_id: str, agent_id: str) -> str:
@@ -75,7 +74,7 @@ def _get_agent_path(self, session_id: str, agent_id: str) -> str:
             ValueError: If session id or agent id contains a path separator.
         """
         session_path = self._get_session_path(session_id)
-        agent_id = validate_agent_id(agent_id)
+        agent_id = _identifier.validate(agent_id, _identifier.Identifier.AGENT)
         return os.path.join(session_path, "agents", f"{AGENT_PREFIX}{agent_id}")
 
     def _get_message_path(self, session_id: str, agent_id: str, message_id: int) -> str:

src/strands/session/file_session_manager.py

@@ -8,10 +8,9 @@
 from botocore.config import Config as BotocoreConfig
 from botocore.exceptions import ClientError
 
-from ..agent.identifier import validate as validate_agent_id
+from .. import _identifier
 from ..types.exceptions import SessionException
 from ..types.session import Session, SessionAgent, SessionMessage
-from .identifier import validate as validate_session_id
 from .repository_session_manager import RepositorySessionManager
 from .session_repository import SessionRepository
 
@@ -90,7 +89,7 @@ def _get_session_path(self, session_id: str) -> str:
         Raises:
             ValueError: If session id contains a path separator.
         """
-        session_id = validate_session_id(session_id)
+        session_id = _identifier.validate(session_id, _identifier.Identifier.SESSION)
         return f"{self.prefix}/{SESSION_PREFIX}{session_id}/"
 
     def _get_agent_path(self, session_id: str, agent_id: str) -> str:
@@ -104,7 +103,7 @@ def _get_agent_path(self, session_id: str, agent_id: str) -> str:
             ValueError: If session id or agent id contains a path separator.
         """
         session_path = self._get_session_path(session_id)
-        agent_id = validate_agent_id(agent_id)
+        agent_id = _identifier.validate(agent_id, _identifier.Identifier.AGENT)
         return f"{session_path}agents/{AGENT_PREFIX}{agent_id}/"
 
     def _get_message_path(self, session_id: str, agent_id: str, message_id: int) -> str:

src/strands/session/identifier.py

@@ -258,7 +258,7 @@ def test_agent__init__deeply_nested_tools(tool_decorated, tool_module, tool_impo
     ],
 )
 def test_agent__init__invalid_id(agent_id):
-    with pytest.raises(ValueError):
+    with pytest.raises(ValueError, match=f"agent_id={agent_id} | id cannot contain path separators"):
         Agent(agent_id=agent_id)
 
 

src/strands/session/s3_session_manager.py

@@ -376,7 +376,7 @@ def test_permission_error_handling(file_manager):
     ],
 )
 def test__get_session_path_invalid_session_id(session_id, file_manager):
-    with pytest.raises(ValueError, match="session id cannot contain path separators"):
+    with pytest.raises(ValueError, match=f"session_id={session_id} | id cannot contain path separators"):
         file_manager._get_session_path(session_id)
 
 
@@ -388,5 +388,5 @@ def test__get_session_path_invalid_session_id(session_id, file_manager):
     ],
 )
 def test__get_agent_path_invalid_agent_id(agent_id, file_manager):
-    with pytest.raises(ValueError, match="agent id cannot contain path separators"):
+    with pytest.raises(ValueError, match=f"agent_id={agent_id} | id cannot contain path separators"):
         file_manager._get_agent_path("session1", agent_id)