diff --git a/src/a2a/compat/v0_3/grpc_transport.py b/src/a2a/compat/v0_3/grpc_transport.py
index 26faaa3d3..425b0d2f0 100644
--- a/src/a2a/compat/v0_3/grpc_transport.py
+++ b/src/a2a/compat/v0_3/grpc_transport.py
@@ -33,6 +33,7 @@
 from a2a.compat.v0_3.extension_headers import add_legacy_extension_header
 from a2a.types import a2a_pb2
 from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER
+from a2a.utils.task import apply_history_length
 from a2a.utils.telemetry import SpanKind, trace_class
 
 
@@ -130,8 +131,11 @@ async def send_message(
         which = resp_proto.WhichOneof('payload')
         if which == 'task':
             return a2a_pb2.SendMessageResponse(
-                task=conversions.to_core_task(
-                    proto_utils.FromProto.task(resp_proto.task)
+                task=apply_history_length(
+                    conversions.to_core_task(
+                        proto_utils.FromProto.task(resp_proto.task)
+                    ),
+                    request.configuration,
                 )
             )
         if which == 'msg':
@@ -217,7 +221,12 @@ async def get_task(
             req_proto,
             metadata=self._get_grpc_metadata(context),
         )
-        return conversions.to_core_task(proto_utils.FromProto.task(resp_proto))
+        # v0.3 has no way to ask for an empty history (0 means unlimited), so
+        # the v1.0 `history_length` semantics are applied to the result here.
+        return apply_history_length(
+            conversions.to_core_task(proto_utils.FromProto.task(resp_proto)),
+            request,
+        )
 
     @_handle_grpc_exception
     async def list_tasks(
diff --git a/src/a2a/compat/v0_3/jsonrpc_transport.py b/src/a2a/compat/v0_3/jsonrpc_transport.py
index f3369910b..e09c0e288 100644
--- a/src/a2a/compat/v0_3/jsonrpc_transport.py
+++ b/src/a2a/compat/v0_3/jsonrpc_transport.py
@@ -40,6 +40,7 @@
 )
 from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER
 from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP
+from a2a.utils.task import apply_history_length
 from a2a.utils.telemetry import SpanKind, trace_class
 
 
@@ -105,8 +106,11 @@ async def send_message(
 
         if kind == 'task':
             return SendMessageResponse(
-                task=conversions.to_core_task(
-                    types_v03.Task.model_validate(result_dict)
+                task=apply_history_length(
+                    conversions.to_core_task(
+                        types_v03.Task.model_validate(result_dict)
+                    ),
+                    request.configuration,
                 )
             )
         if kind == 'message':
@@ -164,8 +168,13 @@ async def get_task(
         json_rpc_response = JSONRPC20Response(**response_data)
         if json_rpc_response.error:
             raise self._create_jsonrpc_error(json_rpc_response.error)
-        return conversions.to_core_task(
-            types_v03.Task.model_validate(json_rpc_response.result)
+        # v0.3 has no way to ask for an empty history (0 means unlimited), so
+        # the v1.0 `history_length` semantics are applied to the result here.
+        return apply_history_length(
+            conversions.to_core_task(
+                types_v03.Task.model_validate(json_rpc_response.result)
+            ),
+            request,
         )
 
     async def list_tasks(
diff --git a/src/a2a/compat/v0_3/rest_transport.py b/src/a2a/compat/v0_3/rest_transport.py
index cdeed2867..4d4c22622 100644
--- a/src/a2a/compat/v0_3/rest_transport.py
+++ b/src/a2a/compat/v0_3/rest_transport.py
@@ -46,6 +46,7 @@
 )
 from a2a.utils.constants import PROTOCOL_VERSION_0_3, VERSION_HEADER
 from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP, MethodNotFoundError
+from a2a.utils.task import apply_history_length
 from a2a.utils.telemetry import SpanKind, trace_class
 
 
@@ -107,8 +108,11 @@ async def send_message(
         which = resp_proto.WhichOneof('payload')
         if which == 'task':
             return SendMessageResponse(
-                task=conversions.to_core_task(
-                    proto_utils.FromProto.task(resp_proto.task)
+                task=apply_history_length(
+                    conversions.to_core_task(
+                        proto_utils.FromProto.task(resp_proto.task)
+                    ),
+                    request.configuration,
                 )
             )
         if which == 'msg':
@@ -165,7 +169,12 @@ async def get_task(
         resp_proto = ParseDict(
             response_data, a2a_v0_3_pb2.Task(), ignore_unknown_fields=True
         )
-        return conversions.to_core_task(proto_utils.FromProto.task(resp_proto))
+        # v0.3 has no way to ask for an empty history (0 means unlimited), so
+        # the v1.0 `history_length` semantics are applied to the result here.
+        return apply_history_length(
+            conversions.to_core_task(proto_utils.FromProto.task(resp_proto)),
+            request,
+        )
 
     async def list_tasks(
         self,
diff --git a/tests/compat/v0_3/test_grpc_transport.py b/tests/compat/v0_3/test_grpc_transport.py
index 402a57000..f12a86792 100644
--- a/tests/compat/v0_3/test_grpc_transport.py
+++ b/tests/compat/v0_3/test_grpc_transport.py
@@ -7,6 +7,7 @@
 from a2a.compat.v0_3 import a2a_v0_3_pb2
 from a2a.compat.v0_3.grpc_transport import CompatGrpcTransport
 from a2a.types.a2a_pb2 import (
+    GetTaskRequest,
     Message,
     Role,
     SendMessageRequest,
@@ -66,3 +67,32 @@ def test_compat_grpc_transport_no_extension_metadata():
 
     assert 'a2a-extensions' not in metadata
     assert 'x-a2a-extensions' not in metadata
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'history_length, expected_ids',
+    [
+        (None, ['m0', 'm1', 'm2']),
+        (0, []),
+        (1, ['m2']),
+        (3, ['m0', 'm1', 'm2']),
+        (5, ['m0', 'm1', 'm2']),
+    ],
+)
+async def test_get_task_history_length(
+    history_length: int | None, expected_ids: list[str]
+) -> None:
+    transport = CompatGrpcTransport(
+        channel=AsyncMock(spec=Channel), agent_card=None
+    )
+    task = a2a_v0_3_pb2.Task(id='task-1', context_id='ctx-1')
+    task.status.state = a2a_v0_3_pb2.TaskState.TASK_STATE_WORKING
+    for i in range(3):
+        task.history.add(message_id=f'm{i}', role=a2a_v0_3_pb2.Role.ROLE_USER)
+    transport.stub = MagicMock(GetTask=AsyncMock(return_value=task))
+    req = GetTaskRequest(id='task-1')
+    if history_length is not None:
+        req.history_length = history_length
+    result = await transport.get_task(req)
+    assert [message.message_id for message in result.history] == expected_ids
diff --git a/tests/compat/v0_3/test_jsonrpc_transport.py b/tests/compat/v0_3/test_jsonrpc_transport.py
index 70291f005..923498b6a 100644
--- a/tests/compat/v0_3/test_jsonrpc_transport.py
+++ b/tests/compat/v0_3/test_jsonrpc_transport.py
@@ -565,3 +565,80 @@ async def test_compat_jsonrpc_transport_mirrors_extension_header(
     headers = kwargs['headers']
     assert headers['A2A-Extensions'] == 'foo,bar'
     assert headers['X-A2A-Extensions'] == 'foo,bar'
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'history_length, expected_ids',
+    [
+        (None, ['m0', 'm1', 'm2']),
+        (0, []),
+        (1, ['m2']),
+        (3, ['m0', 'm1', 'm2']),
+        (5, ['m0', 'm1', 'm2']),
+    ],
+)
+async def test_get_task_history_length(
+    transport: CompatJsonRpcTransport,
+    history_length: int | None,
+    expected_ids: list[str],
+) -> None:
+    # v0.3 treats zero as unlimited; v1.0 requires an empty local history.
+    payload = {
+        'id': 'task-1',
+        'contextId': 'ctx-1',
+        'status': {'state': 'working'},
+        'history': [
+            {'messageId': f'm{i}', 'role': 'user', 'parts': []}
+            for i in range(3)
+        ],
+    }
+    with patch.object(
+        transport, '_send_request', new_callable=AsyncMock
+    ) as request:
+        request.return_value = {'jsonrpc': '2.0', 'id': '0', 'result': payload}
+        req = GetTaskRequest(id='task-1')
+        if history_length is not None:
+            req.history_length = history_length
+        task = await transport.get_task(req)
+    assert [message.message_id for message in task.history] == expected_ids
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'history_length, expected_ids',
+    [
+        (None, ['m0', 'm1', 'm2']),
+        (0, []),
+        (1, ['m2']),
+        (3, ['m0', 'm1', 'm2']),
+        (5, ['m0', 'm1', 'm2']),
+    ],
+)
+async def test_send_message_history_length(
+    transport: CompatJsonRpcTransport,
+    history_length: int | None,
+    expected_ids: list[str],
+) -> None:
+    # v0.3 treats zero as unlimited; v1.0 requires an empty local history.
+    payload = {
+        'id': 'task-1',
+        'contextId': 'ctx-1',
+        'status': {'state': 'working'},
+        'history': [
+            {'messageId': f'm{i}', 'role': 'user', 'parts': []}
+            for i in range(3)
+        ],
+    }
+    payload['kind'] = 'task'
+    with patch.object(
+        transport, '_send_request', new_callable=AsyncMock
+    ) as request:
+        request.return_value = {'jsonrpc': '2.0', 'id': '0', 'result': payload}
+        req = SendMessageRequest()
+        if history_length is not None:
+            req.configuration.history_length = history_length
+        response = await transport.send_message(req)
+        assert response.HasField('task')
+        task = response.task
+    assert [message.message_id for message in task.history] == expected_ids
diff --git a/tests/compat/v0_3/test_rest_transport.py b/tests/compat/v0_3/test_rest_transport.py
index 0952aaa80..f265a3f0a 100644
--- a/tests/compat/v0_3/test_rest_transport.py
+++ b/tests/compat/v0_3/test_rest_transport.py
@@ -690,3 +690,79 @@ async def test_compat_rest_transport_execute_request(
     mock_send_http_request.assert_called_once_with(
         transport.httpx_client, mock_request, transport._handle_http_error
     )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'history_length, expected_ids',
+    [
+        (None, ['m0', 'm1', 'm2']),
+        (0, []),
+        (1, ['m2']),
+        (3, ['m0', 'm1', 'm2']),
+        (5, ['m0', 'm1', 'm2']),
+    ],
+)
+async def test_get_task_history_length(
+    transport: CompatRestTransport,
+    history_length: int | None,
+    expected_ids: list[str],
+) -> None:
+    # v0.3 treats zero as unlimited; v1.0 requires an empty local history.
+    payload = {
+        'id': 'task-1',
+        'contextId': 'ctx-1',
+        'status': {'state': 'TASK_STATE_WORKING'},
+        'history': [
+            {'messageId': f'm{i}', 'role': 'ROLE_USER', 'parts': []}
+            for i in range(3)
+        ],
+    }
+    with patch.object(
+        transport, '_execute_request', new_callable=AsyncMock
+    ) as request:
+        request.return_value = payload
+        req = GetTaskRequest(id='task-1')
+        if history_length is not None:
+            req.history_length = history_length
+        task = await transport.get_task(req)
+    assert [message.message_id for message in task.history] == expected_ids
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'history_length, expected_ids',
+    [
+        (None, ['m0', 'm1', 'm2']),
+        (0, []),
+        (1, ['m2']),
+        (3, ['m0', 'm1', 'm2']),
+        (5, ['m0', 'm1', 'm2']),
+    ],
+)
+async def test_send_message_history_length(
+    transport: CompatRestTransport,
+    history_length: int | None,
+    expected_ids: list[str],
+) -> None:
+    # v0.3 treats zero as unlimited; v1.0 requires an empty local history.
+    payload = {
+        'id': 'task-1',
+        'contextId': 'ctx-1',
+        'status': {'state': 'TASK_STATE_WORKING'},
+        'history': [
+            {'messageId': f'm{i}', 'role': 'ROLE_USER', 'parts': []}
+            for i in range(3)
+        ],
+    }
+    with patch.object(
+        transport, '_execute_request', new_callable=AsyncMock
+    ) as request:
+        request.return_value = {'task': payload}
+        req = SendMessageRequest()
+        if history_length is not None:
+            req.configuration.history_length = history_length
+        response = await transport.send_message(req)
+        assert response.HasField('task')
+        task = response.task
+    assert [message.message_id for message in task.history] == expected_ids