diff --git a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_retrieval.py b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_retrieval.py
index afd3bc90627d..58fad304771e 100644
--- a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_retrieval.py
+++ b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_retrieval.py
@@ -36,7 +36,7 @@ async def main():
             # Invoke the agent
             # The chat history is maintained in the session
             async for response in bedrock_agent.invoke(
-                input_text=user_input,
+                messages=user_input,
                 thread=thread,
             ):
                 print(f"Bedrock agent: {response}")
diff --git a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_simple_chat.py b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_simple_chat.py
index 5fc1c8e60d3b..9ce616df851f 100644
--- a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_simple_chat.py
+++ b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_simple_chat.py
@@ -33,7 +33,7 @@ async def main():
             # Invoke the agent
             # The chat history is maintained in the session
             response = await bedrock_agent.get_response(
-                input_text=user_input,
+                messages=user_input,
                 thread=thread,
             )
             print(f"Bedrock agent: {response}")
diff --git a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_simple_chat_streaming.py b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_simple_chat_streaming.py
index c64e711d48ff..92f770763d92 100644
--- a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_simple_chat_streaming.py
+++ b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_simple_chat_streaming.py
@@ -29,7 +29,7 @@ async def main():
             # Invoke the agent
             # The chat history is maintained in the thread
             print("Bedrock agent: ", end="")
-            async for response in bedrock_agent.invoke_stream(input_text=user_input, thread=thread):
+            async for response in bedrock_agent.invoke_stream(messages=user_input, thread=thread):
                 print(response, end="")
                 thread = response.thread
             print()
diff --git a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_code_interpreter.py b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_code_interpreter.py
index 58e6ab5c4f4a..fdd31f5a9ee7 100644
--- a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_code_interpreter.py
+++ b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_code_interpreter.py
@@ -41,7 +41,7 @@ async def main():
     try:
         # Invoke the agent
         async for response in bedrock_agent.invoke(
-            input_text=ASK,
+            messages=ASK,
             thread=thread,
         ):
             print(f"Response:\n{response}")
diff --git a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_code_interpreter_streaming.py b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_code_interpreter_streaming.py
index 2367116c6773..345a9a54b8a3 100644
--- a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_code_interpreter_streaming.py
+++ b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_code_interpreter_streaming.py
@@ -42,7 +42,7 @@ async def main():
         # Invoke the agent
         print("Response: ")
         async for response in bedrock_agent.invoke_stream(
-            input_text=ASK,
+            messages=ASK,
             thread=thread,
         ):
             print(response, end="")
diff --git a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function.py b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function.py
index 1623c930e84a..282284c302fd 100644
--- a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function.py
+++ b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function.py
@@ -53,7 +53,7 @@ async def main():
     try:
         # Invoke the agent
         async for response in bedrock_agent.invoke(
-            input_text="What is the weather in Seattle?",
+            messages="What is the weather in Seattle?",
             thread=thread,
         ):
             print(f"Response:\n{response}")
diff --git a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function_simple.py b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function_simple.py
index f1ea261bcb9d..2e267f3eb943 100644
--- a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function_simple.py
+++ b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function_simple.py
@@ -44,7 +44,7 @@ async def main():
     try:
         # Invoke the agent
         async for response in bedrock_agent.invoke(
-            input_text="What is the weather in Seattle?",
+            messages="What is the weather in Seattle?",
             thread=thread,
         ):
             print(f"Response:\n{response}")
diff --git a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function_streaming.py b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function_streaming.py
index 58521322b333..f2b823da1243 100644
--- a/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function_streaming.py
+++ b/python/samples/concepts/agents/bedrock_agent/bedrock_agent_with_kernel_function_streaming.py
@@ -54,7 +54,7 @@ async def main():
         # Invoke the agent
         print("Response: ")
         async for response in bedrock_agent.invoke_stream(
-            input_text="What is the weather in Seattle?",
+            messages="What is the weather in Seattle?",
             thread=thread,
         ):
             print(response, end="")
diff --git a/python/semantic_kernel/agents/agent.py b/python/semantic_kernel/agents/agent.py
index efa8aa50c710..3cc40692116b 100644
--- a/python/semantic_kernel/agents/agent.py
+++ b/python/semantic_kernel/agents/agent.py
@@ -207,7 +207,13 @@ def _configure_plugins(cls, data: Any) -> Any:
         return data
 
     @abstractmethod
-    def get_response(self, *args, **kwargs) -> Awaitable[AgentResponseItem[ChatMessageContent]]:
+    def get_response(
+        self,
+        *,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent],
+        thread: AgentThread | None = None,
+        **kwargs,
+    ) -> Awaitable[AgentResponseItem[ChatMessageContent]]:
         """Get a response from the agent.
 
         This method returns the final result of the agent's execution
@@ -219,28 +225,64 @@ def get_response(self, *args, **kwargs) -> Awaitable[AgentResponseItem[ChatMessa
         objects. Streaming only the final result is not feasible because the timing of
         the final result's availability is unknown, and blocking the caller until then
         is undesirable in streaming scenarios.
+
+        Args:
+            messages: The message(s) to send to the agent.
+            thread: The conversation thread associated with the message(s).
+            kwargs: Additional keyword arguments.
+
+        Returns:
+            An agent response item.
         """
         pass
 
     @abstractmethod
-    def invoke(self, *args, **kwargs) -> AsyncIterable[AgentResponseItem[ChatMessageContent]]:
+    def invoke(
+        self,
+        *,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent],
+        thread: AgentThread | None = None,
+        **kwargs,
+    ) -> AsyncIterable[AgentResponseItem[ChatMessageContent]]:
         """Invoke the agent.
 
         This invocation method will return the intermediate steps and the final results
         of the agent's execution as a stream of ChatMessageContent objects to the caller.
 
         Note: A ChatMessageContent object contains an entire message.
+
+        Args:
+            messages: The message(s) to send to the agent.
+            thread: The conversation thread associated with the message(s).
+            kwargs: Additional keyword arguments.
+
+        Yields:
+            An agent response item.
         """
         pass
 
     @abstractmethod
-    def invoke_stream(self, *args, **kwargs) -> AsyncIterable[AgentResponseItem[StreamingChatMessageContent]]:
+    def invoke_stream(
+        self,
+        *,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent],
+        thread: AgentThread | None = None,
+        **kwargs,
+    ) -> AsyncIterable[AgentResponseItem[StreamingChatMessageContent]]:
         """Invoke the agent as a stream.
 
         This invocation method will return the intermediate steps and final results of the
         agent's execution as a stream of StreamingChatMessageContent objects to the caller.
 
         Note: A StreamingChatMessageContent object contains a chunk of a message.
+
+        Args:
+            messages: The message(s) to send to the agent.
+            thread: The conversation thread associated with the message(s).
+            kwargs: Additional keyword arguments.
+
+        Yields:
+            An agent response item.
         """
         pass
 
diff --git a/python/semantic_kernel/agents/autogen/autogen_conversable_agent.py b/python/semantic_kernel/agents/autogen/autogen_conversable_agent.py
index d0788b549a60..f816b4f8ffa0 100644
--- a/python/semantic_kernel/agents/autogen/autogen_conversable_agent.py
+++ b/python/semantic_kernel/agents/autogen/autogen_conversable_agent.py
@@ -129,7 +129,10 @@ def __init__(self, conversable_agent: ConversableAgent, **kwargs: Any) -> None:
     @trace_agent_get_response
     @override
     async def get_response(
-        self, messages: str | ChatMessageContent | list[str | ChatMessageContent], thread: AgentThread | None = None
+        self,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent],
+        thread: AgentThread | None = None,
+        **kwargs: Any,
     ) -> AgentResponseItem[ChatMessageContent]:
         """Get a response from the agent.
 
@@ -137,6 +140,7 @@ async def get_response(
             messages: The input chat message content either as a string, ChatMessageContent or
                 a list of strings or ChatMessageContent.
             thread: The thread to use for the conversation. If None, a new thread will be created.
+            kwargs: Additional keyword arguments
 
         Returns:
             An AgentResponseItem of type ChatMessageContent object with the response and the thread.
@@ -153,6 +157,7 @@ async def get_response(
 
         reply = await self.conversable_agent.a_generate_reply(
             messages=[message.to_dict() for message in chat_history.messages],
+            **kwargs,
         )
 
         logger.info("Called AutoGenConversableAgent.a_generate_reply.")
@@ -245,7 +250,9 @@ async def invoke(
     @override
     def invoke_stream(
         self,
-        message: str,
+        *,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent],
+        thread: AgentThread | None = None,
         kernel: "Kernel | None" = None,
         arguments: KernelArguments | None = None,
         **kwargs: Any,
diff --git a/python/semantic_kernel/agents/bedrock/bedrock_agent.py b/python/semantic_kernel/agents/bedrock/bedrock_agent.py
index 990dca527aff..61fa261b1ec6 100644
--- a/python/semantic_kernel/agents/bedrock/bedrock_agent.py
+++ b/python/semantic_kernel/agents/bedrock/bedrock_agent.py
@@ -262,7 +262,7 @@ async def create_and_prepare_agent(
     async def get_response(
         self,
         *,
-        input_text: str,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent],
         thread: AgentThread | None = None,
         agent_alias: str | None = None,
         arguments: KernelArguments | None = None,
@@ -272,7 +272,7 @@ async def get_response(
         """Get a response from the agent.
 
         Args:
-            input_text (str): The input text.
+            messages (str | ChatMessageContent | list[str | ChatMessageContent]): The messages.
             thread (AgentThread, optional): The thread. This is used to maintain the session state in the service.
             agent_alias (str, optional): The agent alias.
             arguments (KernelArguments, optional): The kernel arguments to override the current arguments.
@@ -282,8 +282,11 @@ async def get_response(
         Returns:
             A chat message content with the response.
         """
+        if not isinstance(messages, str) and not isinstance(messages, ChatMessageContent):
+            raise AgentInvokeException("Messages must be a string or a ChatMessageContent for BedrockAgent.")
+
         thread = await self._ensure_thread_exists_with_messages(
-            messages=[input_text],
+            messages=messages,
             thread=thread,
             construct_thread=lambda: BedrockAgentThread(bedrock_runtime_client=self.bedrock_runtime_client),
             expected_type=BedrockAgentThread,
@@ -302,7 +305,7 @@ async def get_response(
         kwargs.setdefault("sessionState", {})
 
         for _ in range(self.function_choice_behavior.maximum_auto_invoke_attempts):
-            response = await self._invoke_agent(thread.id, input_text, agent_alias, **kwargs)
+            response = await self._invoke_agent(thread.id, messages, agent_alias, **kwargs)
 
             events: list[dict[str, Any]] = []
             for event in response.get("completion", []):
@@ -355,7 +358,7 @@ async def get_response(
     @override
     async def invoke(
         self,
-        input_text: str,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent],
         thread: AgentThread | None = None,
         *,
         agent_alias: str | None = None,
@@ -366,7 +369,7 @@ async def invoke(
         """Invoke an agent.
 
         Args:
-            input_text (str): The input text.
+            messages (str | ChatMessageContent | list[str | ChatMessageContent]): The messages.
             thread (AgentThread, optional): The thread. This is used to maintain the session state in the service.
             agent_alias (str, optional): The agent alias.
             arguments (KernelArguments, optional): The kernel arguments to override the current arguments.
@@ -376,8 +379,11 @@ async def invoke(
         Returns:
             An async iterable of chat message content.
         """
+        if not isinstance(messages, str) and not isinstance(messages, ChatMessageContent):
+            raise AgentInvokeException("Messages must be a string or a ChatMessageContent for BedrockAgent.")
+
         thread = await self._ensure_thread_exists_with_messages(
-            messages=[input_text],
+            messages=messages,
             thread=thread,
             construct_thread=lambda: BedrockAgentThread(bedrock_runtime_client=self.bedrock_runtime_client),
             expected_type=BedrockAgentThread,
@@ -396,7 +402,7 @@ async def invoke(
         kwargs.setdefault("sessionState", {})
 
         for _ in range(self.function_choice_behavior.maximum_auto_invoke_attempts):
-            response = await self._invoke_agent(thread.id, input_text, agent_alias, **kwargs)
+            response = await self._invoke_agent(thread.id, messages, agent_alias, **kwargs)
 
             events: list[dict[str, Any]] = []
             for event in response.get("completion", []):
@@ -451,7 +457,7 @@ async def invoke(
     @override
     async def invoke_stream(
         self,
-        input_text: str,
+        messages: str | ChatMessageContent | list[str | ChatMessageContent],
         thread: AgentThread | None = None,
         *,
         agent_alias: str | None = None,
@@ -462,7 +468,7 @@ async def invoke_stream(
         """Invoke an agent with streaming.
 
         Args:
-            input_text (str): The input text.
+            messages (str | ChatMessageContent | list[str | ChatMessageContent]): The messages.
             thread (AgentThread, optional): The thread. This is used to maintain the session state in the service.
             agent_alias (str, optional): The agent alias.
             arguments (KernelArguments, optional): The kernel arguments to override the current arguments.
@@ -472,8 +478,11 @@ async def invoke_stream(
         Returns:
             An async iterable of streaming chat message content
         """
+        if not isinstance(messages, str) and not isinstance(messages, ChatMessageContent):
+            raise AgentInvokeException("Messages must be a string or a ChatMessageContent for BedrockAgent.")
+
         thread = await self._ensure_thread_exists_with_messages(
-            messages=[input_text],
+            messages=messages,
             thread=thread,
             construct_thread=lambda: BedrockAgentThread(bedrock_runtime_client=self.bedrock_runtime_client),
             expected_type=BedrockAgentThread,
@@ -492,7 +501,7 @@ async def invoke_stream(
         kwargs.setdefault("sessionState", {})
 
         for request_index in range(self.function_choice_behavior.maximum_auto_invoke_attempts):
-            response = await self._invoke_agent(thread.id, input_text, agent_alias, **kwargs)
+            response = await self._invoke_agent(thread.id, messages, agent_alias, **kwargs)
 
             all_function_call_messages: list[StreamingChatMessageContent] = []
             for event in response.get("completion", []):
diff --git a/python/semantic_kernel/agents/bedrock/bedrock_agent_base.py b/python/semantic_kernel/agents/bedrock/bedrock_agent_base.py
index d37db32b7d93..b0391ef6ec8c 100644
--- a/python/semantic_kernel/agents/bedrock/bedrock_agent_base.py
+++ b/python/semantic_kernel/agents/bedrock/bedrock_agent_base.py
@@ -15,6 +15,8 @@
 from semantic_kernel.agents.bedrock.models.bedrock_agent_model import BedrockAgentModel
 from semantic_kernel.agents.bedrock.models.bedrock_agent_status import BedrockAgentStatus
 from semantic_kernel.connectors.ai.function_choice_behavior import FunctionChoiceBehavior, FunctionChoiceType
+from semantic_kernel.contents.chat_message_content import ChatMessageContent
+from semantic_kernel.contents.utils.author_role import AuthorRole
 from semantic_kernel.utils.async_utils import run_in_executor
 from semantic_kernel.utils.feature_stage_decorator import experimental
 
@@ -349,7 +351,7 @@ async def list_associated_agent_knowledge_bases(self, **kwargs) -> dict[str, Any
     async def _invoke_agent(
         self,
         thread_id: str,
-        input_text: str,
+        message: str | ChatMessageContent,
         agent_alias: str | None = None,
         **kwargs,
     ) -> dict[str, Any]:
@@ -357,6 +359,9 @@ async def _invoke_agent(
         if not self.agent_model.agent_id:
             raise ValueError("Agent does not exist. Please create the agent before invoking it.")
 
+        if isinstance(message, ChatMessageContent) and message.role != AuthorRole.USER:
+            raise ValueError("Only user messages are supported for invoking a Bedrock agent.")
+
         agent_alias = agent_alias or self.WORKING_DRAFT_AGENT_ALIAS
 
         try:
@@ -367,7 +372,7 @@ async def _invoke_agent(
                     agentAliasId=agent_alias,
                     agentId=self.agent_model.agent_id,
                     sessionId=thread_id,
-                    inputText=input_text,
+                    inputText=message if isinstance(message, str) else message.content,
                     **kwargs,
                 ),
             )
diff --git a/python/semantic_kernel/agents/channels/bedrock_agent_channel.py b/python/semantic_kernel/agents/channels/bedrock_agent_channel.py
index a14c71a8bbd0..03d1353a52e7 100644
--- a/python/semantic_kernel/agents/channels/bedrock_agent_channel.py
+++ b/python/semantic_kernel/agents/channels/bedrock_agent_channel.py
@@ -67,7 +67,7 @@ async def invoke(self, agent: "Agent", **kwargs: Any) -> AsyncIterable[tuple[boo
         await self._ensure_last_message_is_user()
 
         async for response in agent.invoke(
-            input_text=self.messages[-1].content,
+            messages=self.messages[-1].content,
             thread=self.thread,
             sessionState=await self._parse_chat_history_to_session_state(),
         ):
@@ -105,7 +105,7 @@ async def invoke_stream(
 
         full_message: list[StreamingChatMessageContent] = []
         async for response_chunk in agent.invoke_stream(
-            input_text=self.messages[-1].content,
+            messages=self.messages[-1].content,
             thread=self.thread,
             sessionState=await self._parse_chat_history_to_session_state(),
         ):
diff --git a/python/tests/unit/agents/bedrock_agent/test_bedrock_agent.py b/python/tests/unit/agents/bedrock_agent/test_bedrock_agent.py
index 9e8a46945b30..3531cd809dcc 100644
--- a/python/tests/unit/agents/bedrock_agent/test_bedrock_agent.py
+++ b/python/tests/unit/agents/bedrock_agent/test_bedrock_agent.py
@@ -466,7 +466,7 @@ async def test_bedrock_agent_get_response(
         mock_invoke_agent.return_value = bedrock_agent_non_streaming_simple_response
         mock_start.return_value = "test_session_id"
 
-        response = await agent.get_response(input_text="test_input_text", thread=thread)
+        response = await agent.get_response(messages="test_input_text", thread=thread)
         assert response.message.content == simple_response
 
         mock_invoke_agent.assert_called_once()
@@ -495,7 +495,7 @@ async def test_bedrock_agent_get_response_exception(
         mock_start.return_value = "test_session_id"
 
         with pytest.raises(AgentInvokeException):
-            await agent.get_response(input_text="test_input_text")
+            await agent.get_response(messages="test_input_text")
 
 
 # Test case to verify the invocation of BedrockAgent
diff --git a/python/tests/unit/agents/bedrock_agent/test_bedrock_agent_channel.py b/python/tests/unit/agents/bedrock_agent/test_bedrock_agent_channel.py
index 9f4e0c3d3038..4f72e70ad0cf 100644
--- a/python/tests/unit/agents/bedrock_agent/test_bedrock_agent_channel.py
+++ b/python/tests/unit/agents/bedrock_agent/test_bedrock_agent_channel.py
@@ -173,7 +173,7 @@ async def test_invoke_inserts_placeholders_when_history_needs_to_alternate(mock_
     mock_channel.messages.append(ChatMessageContent(role=AuthorRole.ASSISTANT, content="Assistant 1"))
 
     # Mock agent.invoke to return an async generator
-    async def mock_invoke(input_text: str, thread: AgentThread, sessionState=None, **kwargs):
+    async def mock_invoke(messages: str, thread: AgentThread, sessionState=None, **kwargs):
         # We just yield one message as if the agent responded
         yield AgentResponseItem(
             message=ChatMessageContent(role=AuthorRole.ASSISTANT, content="Mock Agent Response"),
@@ -231,7 +231,7 @@ async def test_invoke_stream_appends_response_message(mock_channel, mock_agent):
     mock_channel.messages.append(ChatMessageContent(role=AuthorRole.USER, content="Last user message"))
 
     async def mock_invoke_stream(
-        input_text: str, thread: AgentThread, sessionState=None, **kwargs
+        messages: str, thread: AgentThread, sessionState=None, **kwargs
     ) -> AsyncIterable[StreamingChatMessageContent]:
         yield AgentResponseItem(
             message=StreamingChatMessageContent(