diff --git a/src/google/adk/a2a/converters/request_converter.py b/src/google/adk/a2a/converters/request_converter.py
new file mode 100644
index 00000000..293df46e
--- /dev/null
+++ b/src/google/adk/a2a/converters/request_converter.py
@@ -0,0 +1,90 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import sys
+from typing import Any
+
+try:
+  from a2a.server.agent_execution import RequestContext
+except ImportError as e:
+  if sys.version_info < (3, 10):
+    raise ImportError(
+        'A2A Tool requires Python 3.10 or above. Please upgrade your Python'
+        ' version.'
+    ) from e
+  else:
+    raise e
+
+from google.genai import types as genai_types
+
+from ...runners import RunConfig
+from ...utils.feature_decorator import working_in_progress
+from .part_converter import convert_a2a_part_to_genai_part
+from .utils import _from_a2a_context_id
+from .utils import _get_adk_metadata_key
+
+
+def _get_user_id(request: RequestContext, user_id_from_context: str) -> str:
+  # Get user from call context if available (auth is enabled on a2a server)
+  if request.call_context and request.call_context.user:
+    return request.call_context.user.user_name
+
+  # Get user from context id if available
+  if user_id_from_context:
+    return user_id_from_context
+
+  # Get user from message metadata if available (client is an ADK agent)
+  if request.message.metadata:
+    user_id = request.message.metadata.get(_get_adk_metadata_key('user_id'))
+    if user_id:
+      return f'ADK_USER_{user_id}'
+
+  # Get user from task if available (client is a an ADK agent)
+  if request.current_task:
+    user_id = request.current_task.metadata.get(
+        _get_adk_metadata_key('user_id')
+    )
+    if user_id:
+      return f'ADK_USER_{user_id}'
+  return (
+      f'temp_user_{request.task_id}'
+      if request.task_id
+      else f'TEMP_USER_{request.message.messageId}'
+  )
+
+
+@working_in_progress
+def convert_a2a_request_to_adk_run_args(
+    request: RequestContext,
+) -> dict[str, Any]:
+
+  if not request.message:
+    raise ValueError('Request message cannot be None')
+
+  _, user_id, session_id = _from_a2a_context_id(request.context_id)
+
+  return {
+      'user_id': _get_user_id(request, user_id),
+      'session_id': session_id,
+      'new_message': genai_types.Content(
+          role='user',
+          parts=[
+              convert_a2a_part_to_genai_part(part)
+              for part in request.message.parts
+          ],
+      ),
+      'run_config': RunConfig(),
+  }
diff --git a/src/google/adk/a2a/converters/utils.py b/src/google/adk/a2a/converters/utils.py
index fe5f2e92..ecbff1e1 100644
--- a/src/google/adk/a2a/converters/utils.py
+++ b/src/google/adk/a2a/converters/utils.py
@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 ADK_METADATA_KEY_PREFIX = "adk_"
+ADK_CONTEXT_ID_PREFIX = "ADK"
 
 
 def _get_adk_metadata_key(key: str) -> str:
@@ -32,3 +33,39 @@ def _get_adk_metadata_key(key: str) -> str:
   if not key:
     raise ValueError("Metadata key cannot be empty or None")
   return f"{ADK_METADATA_KEY_PREFIX}{key}"
+
+
+def _to_a2a_context_id(app_name: str, user_id: str, session_id: str) -> str:
+  """Converts app name, user id and session id to an A2A context id.
+
+  Args:
+    app_name: The app name.
+    user_id: The user id.
+    session_id: The session id.
+
+  Returns:
+    The A2A context id.
+  """
+  return [ADK_CONTEXT_ID_PREFIX, app_name, user_id, session_id].join("$")
+
+
+def _from_a2a_context_id(context_id: str) -> tuple[str, str, str]:
+  """Converts an A2A context id to app name, user id and session id.
+  if context_id is None, return None, None, None
+  if context_id is not None, but not in the format of
+  ADK$app_name$user_id$session_id, return None, None, None
+
+  Args:
+    context_id: The A2A context id.
+
+  Returns:
+    The app name, user id and session id.
+  """
+  if not context_id:
+    return None, None, None
+
+  prefix, app_name, user_id, session_id = context_id.split("$")
+  if prefix == "ADK" and app_name and user_id and session_id:
+    return app_name, user_id, session_id
+
+  return None, None, None
diff --git a/tests/unittests/a2a/converters/test_request_converter.py b/tests/unittests/a2a/converters/test_request_converter.py
new file mode 100644
index 00000000..02c6400f
--- /dev/null
+++ b/tests/unittests/a2a/converters/test_request_converter.py
@@ -0,0 +1,497 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+from unittest.mock import Mock
+from unittest.mock import patch
+
+import pytest
+
+# Skip all tests in this module if Python version is less than 3.10
+pytestmark = pytest.mark.skipif(
+    sys.version_info < (3, 10), reason="A2A tool requires Python 3.10+"
+)
+
+# Import dependencies with version checking
+try:
+  from a2a.server.agent_execution import RequestContext
+  from google.adk.a2a.converters.request_converter import _get_user_id
+  from google.adk.a2a.converters.request_converter import convert_a2a_request_to_adk_run_args
+  from google.adk.runners import RunConfig
+  from google.genai import types as genai_types
+except ImportError as e:
+  if sys.version_info < (3, 10):
+    # Create dummy classes to prevent NameError during test collection
+    # Tests will be skipped anyway due to pytestmark
+    class DummyTypes:
+      pass
+
+    a2a_types = DummyTypes()
+    genai_types = DummyTypes()
+    RequestContext = DummyTypes()
+    RunConfig = DummyTypes()
+    _get_user_id = lambda x, y: None
+    convert_a2a_request_to_adk_run_args = lambda x: None
+  else:
+    raise e
+
+
+class TestGetUserId:
+  """Test cases for _get_user_id function."""
+
+  def test_get_user_id_from_call_context(self):
+    """Test getting user ID from call context when auth is enabled."""
+    # Arrange
+    mock_user = Mock()
+    mock_user.user_name = "authenticated_user"
+
+    mock_call_context = Mock()
+    mock_call_context.user = mock_user
+
+    request = Mock(spec=RequestContext)
+    request.call_context = mock_call_context
+    request.message = Mock()
+    request.current_task = None
+    request.task_id = "task123"
+
+    # Act
+    result = _get_user_id(request, "context_user")
+
+    # Assert
+    assert result == "authenticated_user"
+
+  def test_get_user_id_from_context_when_no_call_context(self):
+    """Test getting user ID from context when call context is not available."""
+    # Arrange
+    request = Mock(spec=RequestContext)
+    request.call_context = None
+    request.message = Mock()
+    request.current_task = None
+    request.task_id = "task123"
+
+    # Act
+    result = _get_user_id(request, "context_user")
+
+    # Assert
+    assert result == "context_user"
+
+  def test_get_user_id_from_context_when_call_context_has_no_user(self):
+    """Test getting user ID from context when call context has no user."""
+    # Arrange
+    mock_call_context = Mock()
+    mock_call_context.user = None
+
+    request = Mock(spec=RequestContext)
+    request.call_context = mock_call_context
+    request.message = Mock()
+    request.current_task = None
+    request.task_id = "task123"
+
+    # Act
+    result = _get_user_id(request, "context_user")
+
+    # Assert
+    assert result == "context_user"
+
+  def test_get_user_id_from_message_metadata(self):
+    """Test getting user ID from message metadata when context user is not available."""
+    # Arrange
+    mock_message = Mock()
+    mock_message.metadata = {"adk_user_id": "message_user"}
+
+    request = Mock(spec=RequestContext)
+    request.call_context = None
+    request.message = mock_message
+    request.current_task = None
+    request.task_id = "task123"
+
+    # Act
+    result = _get_user_id(request, "")
+
+    # Assert
+    assert result == "ADK_USER_message_user"
+
+  def test_get_user_id_from_task_metadata(self):
+    """Test getting user ID from task metadata when message metadata is not available."""
+    # Arrange
+    mock_message = Mock()
+    mock_message.metadata = None
+
+    mock_task = Mock()
+    mock_task.metadata = {"adk_user_id": "task_user"}
+
+    request = Mock(spec=RequestContext)
+    request.call_context = None
+    request.message = mock_message
+    request.current_task = mock_task
+    request.task_id = "task123"
+
+    # Act
+    result = _get_user_id(request, "")
+
+    # Assert
+    assert result == "ADK_USER_task_user"
+
+  def test_get_user_id_fallback_to_task_id(self):
+    """Test fallback to task ID when no other user ID is available."""
+    # Arrange
+    mock_message = Mock()
+    mock_message.metadata = None
+    mock_message.messageId = "msg456"
+
+    request = Mock(spec=RequestContext)
+    request.call_context = None
+    request.message = mock_message
+    request.current_task = None
+    request.task_id = "task123"
+
+    # Act
+    result = _get_user_id(request, "")
+
+    # Assert
+    assert result == "temp_user_task123"
+
+  def test_get_user_id_fallback_to_message_id(self):
+    """Test fallback to message ID when no task ID is available."""
+    # Arrange
+    mock_message = Mock()
+    mock_message.metadata = None
+    mock_message.messageId = "msg456"
+
+    request = Mock(spec=RequestContext)
+    request.call_context = None
+    request.message = mock_message
+    request.current_task = None
+    request.task_id = None
+
+    # Act
+    result = _get_user_id(request, "")
+
+    # Assert
+    assert result == "TEMP_USER_msg456"
+
+  def test_get_user_id_message_metadata_empty(self):
+    """Test getting user ID when message metadata exists but doesn't contain user_id."""
+    # Arrange
+    mock_message = Mock()
+    mock_message.metadata = {"other_key": "other_value"}
+    mock_message.messageId = "msg456"
+
+    request = Mock(spec=RequestContext)
+    request.call_context = None
+    request.message = mock_message
+    request.current_task = None
+    request.task_id = "task123"
+
+    # Act
+    result = _get_user_id(request, "")
+
+    # Assert
+    assert result == "temp_user_task123"
+
+  def test_get_user_id_task_metadata_empty(self):
+    """Test getting user ID when task metadata exists but doesn't contain user_id."""
+    # Arrange
+    mock_message = Mock()
+    mock_message.metadata = None
+    mock_message.messageId = "msg456"
+
+    mock_task = Mock()
+    mock_task.metadata = {"other_key": "other_value"}
+
+    request = Mock(spec=RequestContext)
+    request.call_context = None
+    request.message = mock_message
+    request.current_task = mock_task
+    request.task_id = "task123"
+
+    # Act
+    result = _get_user_id(request, "")
+
+    # Assert
+    assert result == "temp_user_task123"
+
+
+class TestConvertA2aRequestToAdkRunArgs:
+  """Test cases for convert_a2a_request_to_adk_run_args function."""
+
+  @patch(
+      "google.adk.a2a.converters.request_converter.convert_a2a_part_to_genai_part"
+  )
+  @patch("google.adk.a2a.converters.request_converter._from_a2a_context_id")
+  @patch("google.adk.a2a.converters.request_converter._get_user_id")
+  def test_convert_a2a_request_basic(
+      self, mock_get_user_id, mock_from_context_id, mock_convert_part
+  ):
+    """Test basic conversion of A2A request to ADK run args."""
+    # Arrange
+    mock_part1 = Mock()
+    mock_part2 = Mock()
+
+    mock_message = Mock()
+    mock_message.parts = [mock_part1, mock_part2]
+
+    request = Mock(spec=RequestContext)
+    request.message = mock_message
+    request.context_id = "ADK$app$user$session"
+
+    mock_from_context_id.return_value = (
+        "app_name",
+        "user_from_context",
+        "session123",
+    )
+    mock_get_user_id.return_value = "final_user"
+
+    # Create proper genai_types.Part objects instead of mocks
+    mock_genai_part1 = genai_types.Part(text="test part 1")
+    mock_genai_part2 = genai_types.Part(text="test part 2")
+    mock_convert_part.side_effect = [mock_genai_part1, mock_genai_part2]
+
+    # Act
+    result = convert_a2a_request_to_adk_run_args(request)
+
+    # Assert
+    assert result is not None
+    assert result["user_id"] == "final_user"
+    assert result["session_id"] == "session123"
+    assert isinstance(result["new_message"], genai_types.Content)
+    assert result["new_message"].role == "user"
+    assert result["new_message"].parts == [mock_genai_part1, mock_genai_part2]
+    assert isinstance(result["run_config"], RunConfig)
+
+    # Verify calls
+    mock_from_context_id.assert_called_once_with("ADK$app$user$session")
+    mock_get_user_id.assert_called_once_with(request, "user_from_context")
+    assert mock_convert_part.call_count == 2
+    mock_convert_part.assert_any_call(mock_part1)
+    mock_convert_part.assert_any_call(mock_part2)
+
+  def test_convert_a2a_request_no_message_raises_error(self):
+    """Test that conversion raises ValueError when message is None."""
+    # Arrange
+    request = Mock(spec=RequestContext)
+    request.message = None
+
+    # Act & Assert
+    with pytest.raises(ValueError, match="Request message cannot be None"):
+      convert_a2a_request_to_adk_run_args(request)
+
+  @patch(
+      "google.adk.a2a.converters.request_converter.convert_a2a_part_to_genai_part"
+  )
+  @patch("google.adk.a2a.converters.request_converter._from_a2a_context_id")
+  @patch("google.adk.a2a.converters.request_converter._get_user_id")
+  def test_convert_a2a_request_empty_parts(
+      self, mock_get_user_id, mock_from_context_id, mock_convert_part
+  ):
+    """Test conversion with empty parts list."""
+    # Arrange
+    mock_message = Mock()
+    mock_message.parts = []
+
+    request = Mock(spec=RequestContext)
+    request.message = mock_message
+    request.context_id = "ADK$app$user$session"
+
+    mock_from_context_id.return_value = (
+        "app_name",
+        "user_from_context",
+        "session123",
+    )
+    mock_get_user_id.return_value = "final_user"
+
+    # Act
+    result = convert_a2a_request_to_adk_run_args(request)
+
+    # Assert
+    assert result is not None
+    assert result["user_id"] == "final_user"
+    assert result["session_id"] == "session123"
+    assert isinstance(result["new_message"], genai_types.Content)
+    assert result["new_message"].role == "user"
+    assert result["new_message"].parts == []
+    assert isinstance(result["run_config"], RunConfig)
+
+    # Verify convert_part wasn't called
+    mock_convert_part.assert_not_called()
+
+  @patch(
+      "google.adk.a2a.converters.request_converter.convert_a2a_part_to_genai_part"
+  )
+  @patch("google.adk.a2a.converters.request_converter._from_a2a_context_id")
+  @patch("google.adk.a2a.converters.request_converter._get_user_id")
+  def test_convert_a2a_request_none_context_id(
+      self, mock_get_user_id, mock_from_context_id, mock_convert_part
+  ):
+    """Test conversion when context_id is None."""
+    # Arrange
+    mock_part = Mock()
+    mock_message = Mock()
+    mock_message.parts = [mock_part]
+
+    request = Mock(spec=RequestContext)
+    request.message = mock_message
+    request.context_id = None
+
+    mock_from_context_id.return_value = (None, None, None)
+    mock_get_user_id.return_value = "fallback_user"
+
+    # Create proper genai_types.Part object instead of mock
+    mock_genai_part = genai_types.Part(text="test part")
+    mock_convert_part.return_value = mock_genai_part
+
+    # Act
+    result = convert_a2a_request_to_adk_run_args(request)
+
+    # Assert
+    assert result is not None
+    assert result["user_id"] == "fallback_user"
+    assert result["session_id"] is None
+    assert isinstance(result["new_message"], genai_types.Content)
+    assert result["new_message"].role == "user"
+    assert result["new_message"].parts == [mock_genai_part]
+    assert isinstance(result["run_config"], RunConfig)
+
+    # Verify calls
+    mock_from_context_id.assert_called_once_with(None)
+    mock_get_user_id.assert_called_once_with(request, None)
+
+  @patch(
+      "google.adk.a2a.converters.request_converter.convert_a2a_part_to_genai_part"
+  )
+  @patch("google.adk.a2a.converters.request_converter._from_a2a_context_id")
+  @patch("google.adk.a2a.converters.request_converter._get_user_id")
+  def test_convert_a2a_request_invalid_context_id(
+      self, mock_get_user_id, mock_from_context_id, mock_convert_part
+  ):
+    """Test conversion when context_id is invalid format."""
+    # Arrange
+    mock_part = Mock()
+    mock_message = Mock()
+    mock_message.parts = [mock_part]
+
+    request = Mock(spec=RequestContext)
+    request.message = mock_message
+    request.context_id = "invalid_format"
+
+    mock_from_context_id.return_value = (None, None, None)
+    mock_get_user_id.return_value = "fallback_user"
+
+    # Create proper genai_types.Part object instead of mock
+    mock_genai_part = genai_types.Part(text="test part")
+    mock_convert_part.return_value = mock_genai_part
+
+    # Act
+    result = convert_a2a_request_to_adk_run_args(request)
+
+    # Assert
+    assert result is not None
+    assert result["user_id"] == "fallback_user"
+    assert result["session_id"] is None
+    assert isinstance(result["new_message"], genai_types.Content)
+    assert result["new_message"].role == "user"
+    assert result["new_message"].parts == [mock_genai_part]
+    assert isinstance(result["run_config"], RunConfig)
+
+    # Verify calls
+    mock_from_context_id.assert_called_once_with("invalid_format")
+    mock_get_user_id.assert_called_once_with(request, None)
+
+
+class TestIntegration:
+  """Integration test cases combining both functions."""
+
+  @patch(
+      "google.adk.a2a.converters.request_converter.convert_a2a_part_to_genai_part"
+  )
+  def test_end_to_end_conversion_with_auth_user(self, mock_convert_part):
+    """Test end-to-end conversion with authenticated user."""
+    # Arrange
+    mock_user = Mock()
+    mock_user.user_name = "auth_user"
+
+    mock_call_context = Mock()
+    mock_call_context.user = mock_user
+
+    mock_part = Mock()
+    mock_message = Mock()
+    mock_message.parts = [mock_part]
+
+    request = Mock(spec=RequestContext)
+    request.call_context = mock_call_context
+    request.message = mock_message
+    request.context_id = "ADK$myapp$context_user$mysession"
+    request.current_task = None
+    request.task_id = "task123"
+
+    # Create proper genai_types.Part object instead of mock
+    mock_genai_part = genai_types.Part(text="test part")
+    mock_convert_part.return_value = mock_genai_part
+
+    # Act
+    result = convert_a2a_request_to_adk_run_args(request)
+
+    # Assert
+    assert result is not None
+    assert (
+        result["user_id"] == "auth_user"
+    )  # Should use authenticated user, not context user
+    assert result["session_id"] == "mysession"
+    assert isinstance(result["new_message"], genai_types.Content)
+    assert result["new_message"].role == "user"
+    assert result["new_message"].parts == [mock_genai_part]
+    assert isinstance(result["run_config"], RunConfig)
+
+  @patch(
+      "google.adk.a2a.converters.request_converter.convert_a2a_part_to_genai_part"
+  )
+  @patch("google.adk.a2a.converters.request_converter._from_a2a_context_id")
+  def test_end_to_end_conversion_with_fallback_user(
+      self, mock_from_context_id, mock_convert_part
+  ):
+    """Test end-to-end conversion with fallback user ID."""
+    # Arrange
+    mock_part = Mock()
+    mock_message = Mock()
+    mock_message.parts = [mock_part]
+    mock_message.messageId = "msg789"
+    mock_message.metadata = None
+
+    request = Mock(spec=RequestContext)
+    request.call_context = None
+    request.message = mock_message
+    request.context_id = "invalid_format"
+    request.current_task = None
+    request.task_id = None
+
+    # Mock the utils function to return None values for invalid context
+    mock_from_context_id.return_value = (None, None, None)
+
+    # Create proper genai_types.Part object instead of mock
+    mock_genai_part = genai_types.Part(text="test part")
+    mock_convert_part.return_value = mock_genai_part
+
+    # Act
+    result = convert_a2a_request_to_adk_run_args(request)
+
+    # Assert
+    assert result is not None
+    assert (
+        result["user_id"] == "TEMP_USER_msg789"
+    )  # Should fallback to message ID
+    assert result["session_id"] is None
+    assert isinstance(result["new_message"], genai_types.Content)
+    assert result["new_message"].role == "user"
+    assert result["new_message"].parts == [mock_genai_part]
+    assert isinstance(result["run_config"], RunConfig)