diff --git a/src/runloop_api_client/resources/axons.py b/src/runloop_api_client/resources/axons.py
index 9758057de..f39c994a3 100644
--- a/src/runloop_api_client/resources/axons.py
+++ b/src/runloop_api_client/resources/axons.py
@@ -227,10 +227,12 @@ def subscribe_sse(
         """
         if not id:
             raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")
+        default_headers: Headers = {"Accept": "text/event-stream"}
+        merged_headers = default_headers if extra_headers is None else {**default_headers, **extra_headers}
         return self._get(
             path_template("/v1/axons/{id}/subscribe/sse", id=id),
             options=make_request_options(
-                extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout
+                extra_headers=merged_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout
             ),
             cast_to=AxonEventView,
             stream=True,
@@ -437,10 +439,12 @@ async def subscribe_sse(
         """
         if not id:
             raise ValueError(f"Expected a non-empty value for `id` but received {id!r}")
+        default_headers: Headers = {"Accept": "text/event-stream"}
+        merged_headers = default_headers if extra_headers is None else {**default_headers, **extra_headers}
         return await self._get(
             path_template("/v1/axons/{id}/subscribe/sse", id=id),
             options=make_request_options(
-                extra_headers=extra_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout
+                extra_headers=merged_headers, extra_query=extra_query, extra_body=extra_body, timeout=timeout
             ),
             cast_to=AxonEventView,
             stream=True,
diff --git a/src/runloop_api_client/sdk/__init__.py b/src/runloop_api_client/sdk/__init__.py
index 126b112c8..136080fe8 100644
--- a/src/runloop_api_client/sdk/__init__.py
+++ b/src/runloop_api_client/sdk/__init__.py
@@ -5,7 +5,9 @@
 
 from __future__ import annotations
 
+from .axon import Axon
 from .sync import (
+    AxonOps,
     AgentOps,
     DevboxOps,
     ScorerOps,
@@ -23,6 +25,7 @@
 from .agent import Agent
 from ._types import ScenarioPreview
 from .async_ import (
+    AsyncAxonOps,
     AsyncAgentOps,
     AsyncDevboxOps,
     AsyncScorerOps,
@@ -45,6 +48,7 @@
 from .benchmark import Benchmark
 from .blueprint import Blueprint
 from .execution import Execution
+from .async_axon import AsyncAxon
 from .mcp_config import McpConfig
 from .async_agent import AsyncAgent
 from .async_devbox import AsyncDevbox, AsyncNamedShell
@@ -78,6 +82,8 @@
     # Management interfaces
     "AgentOps",
     "AsyncAgentOps",
+    "AxonOps",
+    "AsyncAxonOps",
     "BenchmarkOps",
     "AsyncBenchmarkOps",
     "DevboxOps",
@@ -103,6 +109,8 @@
     # Resource classes
     "Agent",
     "AsyncAgent",
+    "Axon",
+    "AsyncAxon",
     "AsyncSecret",
     "Benchmark",
     "AsyncBenchmark",
diff --git a/src/runloop_api_client/sdk/_types.py b/src/runloop_api_client/sdk/_types.py
index 82e4bbb9a..ac678f126 100644
--- a/src/runloop_api_client/sdk/_types.py
+++ b/src/runloop_api_client/sdk/_types.py
@@ -5,9 +5,11 @@
     InputContext,
     ScenarioView,
     AgentListParams,
+    AxonCreateParams,
     DevboxListParams,
     ObjectListParams,
     AgentCreateParams,
+    AxonPublishParams,
     DevboxCreateParams,
     ObjectCreateParams,
     ScenarioListParams,
@@ -186,6 +188,14 @@ class SDKAgentListParams(AgentListParams, BaseRequestOptions):
     pass
 
 
+class SDKAxonCreateParams(AxonCreateParams, LongRequestOptions):
+    pass
+
+
+class SDKAxonPublishParams(AxonPublishParams, LongRequestOptions):
+    pass
+
+
 class SDKScenarioListParams(ScenarioListParams, BaseRequestOptions):
     pass
 
diff --git a/src/runloop_api_client/sdk/async_.py b/src/runloop_api_client/sdk/async_.py
index 6dcb6dff4..c2a237da8 100644
--- a/src/runloop_api_client/sdk/async_.py
+++ b/src/runloop_api_client/sdk/async_.py
@@ -14,6 +14,7 @@
     BaseRequestOptions,
     LongRequestOptions,
     SDKAgentListParams,
+    SDKAxonCreateParams,
     SDKDevboxListParams,
     SDKObjectListParams,
     SDKScorerListParams,
@@ -38,6 +39,7 @@
 from .._types import Timeout, NotGiven, not_given
 from .._client import DEFAULT_MAX_RETRIES, AsyncRunloop
 from ._helpers import detect_content_type
+from .async_axon import AsyncAxon
 from .async_agent import AsyncAgent
 from .async_devbox import AsyncDevbox
 from .async_scorer import AsyncScorer
@@ -521,6 +523,33 @@ async def upload_from_bytes(
         return obj
 
 
+class AsyncAxonOps:
+    """[Beta] Create and manage axons (async). Access via ``runloop.axon``.
+
+    Example:
+        >>> runloop = AsyncRunloopSDK()
+        >>> axon = await runloop.axon.create()
+        >>> await axon.publish(event_type="test", origin="USER_EVENT", payload="{}", source="sdk")
+    """
+
+    def __init__(self, client: AsyncRunloop) -> None:
+        self._client = client
+
+    async def create(self, **params: Unpack[SDKAxonCreateParams]) -> AsyncAxon:
+        """[Beta] Create a new axon."""
+        response = await self._client.axons.create(**params)
+        return AsyncAxon(self._client, response.id)
+
+    def from_id(self, axon_id: str) -> AsyncAxon:
+        """Get an AsyncAxon instance for an existing axon ID."""
+        return AsyncAxon(self._client, axon_id)
+
+    async def list(self, **options: Unpack[BaseRequestOptions]) -> list[AsyncAxon]:
+        """[Beta] List all active axons."""
+        result = await self._client.axons.list(**options)
+        return [AsyncAxon(self._client, axon.id) for axon in result.axons]
+
+
 class AsyncScorerOps:
     """Create and manage custom scorers (async). Access via ``runloop.scorer``.
 
@@ -1205,6 +1234,8 @@ class AsyncRunloopSDK:
     :vartype api: AsyncRunloop
     :ivar agent: High-level async interface for agent management.
     :vartype agent: AsyncAgentOps
+    :ivar axon: [Beta] High-level async interface for axon management
+    :vartype axon: AsyncAxonOps
     :ivar benchmark: High-level async interface for benchmark management
     :vartype benchmark: AsyncBenchmarkOps
     :ivar devbox: High-level async interface for devbox management
@@ -1238,6 +1269,7 @@ class AsyncRunloopSDK:
 
     api: AsyncRunloop
     agent: AsyncAgentOps
+    axon: AsyncAxonOps
     benchmark: AsyncBenchmarkOps
     devbox: AsyncDevboxOps
     blueprint: AsyncBlueprintOps
@@ -1289,6 +1321,7 @@ def __init__(
         )
 
         self.agent = AsyncAgentOps(self.api)
+        self.axon = AsyncAxonOps(self.api)
         self.benchmark = AsyncBenchmarkOps(self.api)
         self.devbox = AsyncDevboxOps(self.api)
         self.blueprint = AsyncBlueprintOps(self.api)
diff --git a/src/runloop_api_client/sdk/async_axon.py b/src/runloop_api_client/sdk/async_axon.py
new file mode 100644
index 000000000..dda6a3037
--- /dev/null
+++ b/src/runloop_api_client/sdk/async_axon.py
@@ -0,0 +1,56 @@
+"""Axon resource class for asynchronous operations."""
+
+from __future__ import annotations
+
+from typing_extensions import Unpack, override
+
+from ._types import (
+    BaseRequestOptions,
+    SDKAxonPublishParams,
+)
+from .._client import AsyncRunloop
+from .._streaming import AsyncStream
+from ..types.axon_view import AxonView
+from ..types.axon_event_view import AxonEventView
+from ..types.publish_result_view import PublishResultView
+
+
+class AsyncAxon:
+    """[Beta] Wrapper around asynchronous axon operations.
+
+    Axons are event communication channels that support publishing events
+    and subscribing to event streams via server-sent events (SSE).
+    Obtain instances via ``runloop.axon.create()`` or ``runloop.axon.from_id()``.
+
+    Example:
+        >>> runloop = AsyncRunloopSDK()
+        >>> axon = await runloop.axon.create()
+        >>> await axon.publish(event_type="task_done", origin="AGENT_EVENT", payload="{}", source="my-agent")
+        >>> async with await axon.subscribe_sse() as stream:
+        ...     async for event in stream:
+        ...         print(event.event_type, event.payload)
+    """
+
+    def __init__(self, client: AsyncRunloop, axon_id: str) -> None:
+        self._client = client
+        self._id = axon_id
+
+    @override
+    def __repr__(self) -> str:
+        return f"<AsyncAxon id={self._id!r}>"
+
+    @property
+    def id(self) -> str:
+        return self._id
+
+    async def get_info(self, **options: Unpack[BaseRequestOptions]) -> AxonView:
+        """[Beta] Retrieve the latest axon information."""
+        return await self._client.axons.retrieve(self._id, **options)
+
+    async def publish(self, **params: Unpack[SDKAxonPublishParams]) -> PublishResultView:
+        """[Beta] Publish an event to this axon."""
+        return await self._client.axons.publish(self._id, **params)
+
+    async def subscribe_sse(self, **options: Unpack[BaseRequestOptions]) -> AsyncStream[AxonEventView]:
+        """[Beta] Subscribe to this axon's event stream via SSE."""
+        return await self._client.axons.subscribe_sse(self._id, **options)
diff --git a/src/runloop_api_client/sdk/axon.py b/src/runloop_api_client/sdk/axon.py
new file mode 100644
index 000000000..eb310a88f
--- /dev/null
+++ b/src/runloop_api_client/sdk/axon.py
@@ -0,0 +1,56 @@
+"""Axon resource class for synchronous operations."""
+
+from __future__ import annotations
+
+from typing_extensions import Unpack, override
+
+from ._types import (
+    BaseRequestOptions,
+    SDKAxonPublishParams,
+)
+from .._client import Runloop
+from .._streaming import Stream
+from ..types.axon_view import AxonView
+from ..types.axon_event_view import AxonEventView
+from ..types.publish_result_view import PublishResultView
+
+
+class Axon:
+    """[Beta] Wrapper around synchronous axon operations.
+
+    Axons are event communication channels that support publishing events
+    and subscribing to event streams via server-sent events (SSE).
+    Obtain instances via ``runloop.axon.create()`` or ``runloop.axon.from_id()``.
+
+    Example:
+        >>> runloop = RunloopSDK()
+        >>> axon = runloop.axon.create()
+        >>> axon.publish(event_type="task_done", origin="AGENT_EVENT", payload="{}", source="my-agent")
+        >>> with axon.subscribe_sse() as stream:
+        ...     for event in stream:
+        ...         print(event.event_type, event.payload)
+    """
+
+    def __init__(self, client: Runloop, axon_id: str) -> None:
+        self._client = client
+        self._id = axon_id
+
+    @override
+    def __repr__(self) -> str:
+        return f"<Axon id={self._id!r}>"
+
+    @property
+    def id(self) -> str:
+        return self._id
+
+    def get_info(self, **options: Unpack[BaseRequestOptions]) -> AxonView:
+        """[Beta] Retrieve the latest axon information."""
+        return self._client.axons.retrieve(self._id, **options)
+
+    def publish(self, **params: Unpack[SDKAxonPublishParams]) -> PublishResultView:
+        """[Beta] Publish an event to this axon."""
+        return self._client.axons.publish(self._id, **params)
+
+    def subscribe_sse(self, **options: Unpack[BaseRequestOptions]) -> Stream[AxonEventView]:
+        """[Beta] Subscribe to this axon's event stream via SSE."""
+        return self._client.axons.subscribe_sse(self._id, **options)
diff --git a/src/runloop_api_client/sdk/sync.py b/src/runloop_api_client/sdk/sync.py
index 07dd5e8fe..0fa9f8cea 100644
--- a/src/runloop_api_client/sdk/sync.py
+++ b/src/runloop_api_client/sdk/sync.py
@@ -9,11 +9,13 @@
 
 import httpx
 
+from .axon import Axon
 from .agent import Agent
 from ._types import (
     BaseRequestOptions,
     LongRequestOptions,
     SDKAgentListParams,
+    SDKAxonCreateParams,
     SDKDevboxListParams,
     SDKObjectListParams,
     SDKScorerListParams,
@@ -516,6 +518,33 @@ def upload_from_bytes(
         return obj
 
 
+class AxonOps:
+    """[Beta] Create and manage axons. Access via ``runloop.axon``.
+
+    Example:
+        >>> runloop = RunloopSDK()
+        >>> axon = runloop.axon.create()
+        >>> axon.publish(event_type="test", origin="USER_EVENT", payload="{}", source="sdk")
+    """
+
+    def __init__(self, client: Runloop) -> None:
+        self._client = client
+
+    def create(self, **params: Unpack[SDKAxonCreateParams]) -> Axon:
+        """[Beta] Create a new axon."""
+        response = self._client.axons.create(**params)
+        return Axon(self._client, response.id)
+
+    def from_id(self, axon_id: str) -> Axon:
+        """Get an Axon instance for an existing axon ID."""
+        return Axon(self._client, axon_id)
+
+    def list(self, **options: Unpack[BaseRequestOptions]) -> list[Axon]:
+        """[Beta] List all active axons."""
+        result = self._client.axons.list(**options)
+        return [Axon(self._client, axon.id) for axon in result.axons]
+
+
 class ScorerOps:
     """Create and manage custom scorers. Access via ``runloop.scorer``.
 
@@ -1230,6 +1259,8 @@ class RunloopSDK:
     :vartype api: Runloop
     :ivar agent: High-level interface for agent management.
     :vartype agent: AgentOps
+    :ivar axon: [Beta] High-level interface for axon management
+    :vartype axon: AxonOps
     :ivar benchmark: High-level interface for benchmark management
     :vartype benchmark: BenchmarkOps
     :ivar devbox: High-level interface for devbox management
@@ -1263,6 +1294,7 @@ class RunloopSDK:
 
     api: Runloop
     agent: AgentOps
+    axon: AxonOps
     benchmark: BenchmarkOps
     devbox: DevboxOps
     blueprint: BlueprintOps
@@ -1314,6 +1346,7 @@ def __init__(
         )
 
         self.agent = AgentOps(self.api)
+        self.axon = AxonOps(self.api)
         self.benchmark = BenchmarkOps(self.api)
         self.devbox = DevboxOps(self.api)
         self.blueprint = BlueprintOps(self.api)
diff --git a/tests/sdk/conftest.py b/tests/sdk/conftest.py
index a67f25b27..ab8d0d048 100644
--- a/tests/sdk/conftest.py
+++ b/tests/sdk/conftest.py
@@ -23,6 +23,7 @@
     "object": "obj_123",
     "scorer": "sco_123",
     "agent": "agt_123",
+    "axon": "axn_123",
     "scenario": "scn_123",
     "scenario_run": "scr_123",
     "benchmark": "bmd_123",
@@ -115,6 +116,23 @@ class MockAgentView:
     source: Any = None
 
 
+@dataclass
+class MockAxonView:
+    """Mock AxonView for testing."""
+
+    id: str = TEST_IDS["axon"]
+    created_at_ms: int = 1234567890000
+    name: str = "test-axon"
+
+
+@dataclass
+class MockPublishResultView:
+    """Mock PublishResultView for testing."""
+
+    sequence: int = 1
+    timestamp_ms: int = 1234567890000
+
+
 @dataclass
 class MockScenarioView:
     """Mock ScenarioView for testing."""
@@ -307,6 +325,12 @@ def agent_view() -> MockAgentView:
     return MockAgentView()
 
 
+@pytest.fixture
+def axon_view() -> MockAxonView:
+    """Create a mock AxonView."""
+    return MockAxonView()
+
+
 @pytest.fixture
 def scenario_view() -> MockScenarioView:
     """Create a mock ScenarioView."""
diff --git a/tests/sdk/test_async_axon.py b/tests/sdk/test_async_axon.py
new file mode 100644
index 000000000..f5ce9761d
--- /dev/null
+++ b/tests/sdk/test_async_axon.py
@@ -0,0 +1,77 @@
+"""Comprehensive tests for async Axon class."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+from tests.sdk.conftest import MockAxonView, MockPublishResultView
+from runloop_api_client.sdk import AsyncAxon
+
+
+class TestAsyncAxon:
+    """Tests for AsyncAxon class."""
+
+    def test_init(self, mock_async_client: AsyncMock) -> None:
+        """Test AsyncAxon initialization."""
+        axon = AsyncAxon(mock_async_client, "axn_123")
+        assert axon.id == "axn_123"
+
+    def test_repr(self, mock_async_client: AsyncMock) -> None:
+        """Test AsyncAxon string representation."""
+        axon = AsyncAxon(mock_async_client, "axn_123")
+        assert repr(axon) == "<AsyncAxon id='axn_123'>"
+
+    @pytest.mark.asyncio
+    async def test_get_info(self, mock_async_client: AsyncMock, axon_view: MockAxonView) -> None:
+        """Test get_info method."""
+        mock_async_client.axons.retrieve = AsyncMock(return_value=axon_view)
+
+        axon = AsyncAxon(mock_async_client, "axn_123")
+        result = await axon.get_info(
+            extra_headers={"X-Custom": "value"},
+            timeout=30.0,
+        )
+
+        assert result == axon_view
+        mock_async_client.axons.retrieve.assert_awaited_once_with(
+            "axn_123",
+            extra_headers={"X-Custom": "value"},
+            timeout=30.0,
+        )
+
+    @pytest.mark.asyncio
+    async def test_publish(self, mock_async_client: AsyncMock) -> None:
+        """Test publish method."""
+        mock_result = MockPublishResultView()
+        mock_async_client.axons.publish = AsyncMock(return_value=mock_result)
+
+        axon = AsyncAxon(mock_async_client, "axn_123")
+        result = await axon.publish(
+            event_type="test",
+            origin="USER_EVENT",
+            payload="{}",
+            source="sdk",
+        )
+
+        assert result == mock_result
+        mock_async_client.axons.publish.assert_awaited_once_with(
+            "axn_123",
+            event_type="test",
+            origin="USER_EVENT",
+            payload="{}",
+            source="sdk",
+        )
+
+    @pytest.mark.asyncio
+    async def test_subscribe_sse(self, mock_async_client: AsyncMock) -> None:
+        """Test subscribe_sse method."""
+        mock_stream = AsyncMock()
+        mock_async_client.axons.subscribe_sse = AsyncMock(return_value=mock_stream)
+
+        axon = AsyncAxon(mock_async_client, "axn_123")
+        result = await axon.subscribe_sse()
+
+        assert result == mock_stream
+        mock_async_client.axons.subscribe_sse.assert_awaited_once_with("axn_123")
diff --git a/tests/sdk/test_async_ops.py b/tests/sdk/test_async_ops.py
index b3f02ab5c..82a9535ce 100644
--- a/tests/sdk/test_async_ops.py
+++ b/tests/sdk/test_async_ops.py
@@ -11,6 +11,7 @@
 import pytest
 
 from tests.sdk.conftest import (
+    MockAxonView,
     MockAgentView,
     MockDevboxView,
     MockObjectView,
@@ -24,9 +25,11 @@
     create_mock_httpx_response,
 )
 from runloop_api_client.sdk import (
+    AsyncAxon,
     AsyncAgent,
     AsyncDevbox,
     AsyncScorer,
+    AsyncAxonOps,
     AsyncAgentOps,
     AsyncScenario,
     AsyncSnapshot,
@@ -780,6 +783,74 @@ async def test_list_multiple(self, mock_async_client: AsyncMock) -> None:
         mock_async_client.scenarios.scorers.list.assert_awaited_once()
 
 
+class TestAsyncAxonOps:
+    """Tests for AsyncAxonOps class."""
+
+    @pytest.mark.asyncio
+    async def test_create(self, mock_async_client: AsyncMock, axon_view: MockAxonView) -> None:
+        """Test create method."""
+        mock_async_client.axons.create = AsyncMock(return_value=axon_view)
+
+        ops = AsyncAxonOps(mock_async_client)
+        axon = await ops.create(name="test-axon")
+
+        assert isinstance(axon, AsyncAxon)
+        assert axon.id == "axn_123"
+        mock_async_client.axons.create.assert_awaited_once()
+
+    def test_from_id(self, mock_async_client: AsyncMock) -> None:
+        """Test from_id method."""
+        ops = AsyncAxonOps(mock_async_client)
+        axon = ops.from_id("axn_123")
+
+        assert isinstance(axon, AsyncAxon)
+        assert axon.id == "axn_123"
+
+    @pytest.mark.asyncio
+    async def test_list_empty(self, mock_async_client: AsyncMock) -> None:
+        """Test list method with empty results."""
+        page = SimpleNamespace(axons=[])
+        mock_async_client.axons.list = AsyncMock(return_value=page)
+
+        ops = AsyncAxonOps(mock_async_client)
+        axons = await ops.list()
+
+        assert len(axons) == 0
+        mock_async_client.axons.list.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_list_single(self, mock_async_client: AsyncMock, axon_view: MockAxonView) -> None:
+        """Test list method with single result."""
+        page = SimpleNamespace(axons=[axon_view])
+        mock_async_client.axons.list = AsyncMock(return_value=page)
+
+        ops = AsyncAxonOps(mock_async_client)
+        axons = await ops.list()
+
+        assert len(axons) == 1
+        assert isinstance(axons[0], AsyncAxon)
+        assert axons[0].id == "axn_123"
+        mock_async_client.axons.list.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_list_multiple(self, mock_async_client: AsyncMock) -> None:
+        """Test list method with multiple results."""
+        axon_view1 = MockAxonView(id="axn_001", name="axon-1")
+        axon_view2 = MockAxonView(id="axn_002", name="axon-2")
+        page = SimpleNamespace(axons=[axon_view1, axon_view2])
+        mock_async_client.axons.list = AsyncMock(return_value=page)
+
+        ops = AsyncAxonOps(mock_async_client)
+        axons = await ops.list()
+
+        assert len(axons) == 2
+        assert isinstance(axons[0], AsyncAxon)
+        assert isinstance(axons[1], AsyncAxon)
+        assert axons[0].id == "axn_001"
+        assert axons[1].id == "axn_002"
+        mock_async_client.axons.list.assert_awaited_once()
+
+
 class TestAsyncAgentClient:
     """Tests for AsyncAgentClient class."""
 
diff --git a/tests/sdk/test_axon.py b/tests/sdk/test_axon.py
new file mode 100644
index 000000000..58871d339
--- /dev/null
+++ b/tests/sdk/test_axon.py
@@ -0,0 +1,72 @@
+"""Comprehensive tests for sync Axon class."""
+
+from __future__ import annotations
+
+from unittest.mock import Mock
+
+from tests.sdk.conftest import MockAxonView, MockPublishResultView
+from runloop_api_client.sdk import Axon
+
+
+class TestAxon:
+    """Tests for Axon class."""
+
+    def test_init(self, mock_client: Mock) -> None:
+        """Test Axon initialization."""
+        axon = Axon(mock_client, "axn_123")
+        assert axon.id == "axn_123"
+
+    def test_repr(self, mock_client: Mock) -> None:
+        """Test Axon string representation."""
+        axon = Axon(mock_client, "axn_123")
+        assert repr(axon) == "<Axon id='axn_123'>"
+
+    def test_get_info(self, mock_client: Mock, axon_view: MockAxonView) -> None:
+        """Test get_info method."""
+        mock_client.axons.retrieve.return_value = axon_view
+
+        axon = Axon(mock_client, "axn_123")
+        result = axon.get_info(
+            extra_headers={"X-Custom": "value"},
+            timeout=30.0,
+        )
+
+        assert result == axon_view
+        mock_client.axons.retrieve.assert_called_once_with(
+            "axn_123",
+            extra_headers={"X-Custom": "value"},
+            timeout=30.0,
+        )
+
+    def test_publish(self, mock_client: Mock) -> None:
+        """Test publish method."""
+        mock_result = MockPublishResultView()
+        mock_client.axons.publish.return_value = mock_result
+
+        axon = Axon(mock_client, "axn_123")
+        result = axon.publish(
+            event_type="test",
+            origin="USER_EVENT",
+            payload="{}",
+            source="sdk",
+        )
+
+        assert result == mock_result
+        mock_client.axons.publish.assert_called_once_with(
+            "axn_123",
+            event_type="test",
+            origin="USER_EVENT",
+            payload="{}",
+            source="sdk",
+        )
+
+    def test_subscribe_sse(self, mock_client: Mock) -> None:
+        """Test subscribe_sse method."""
+        mock_stream = Mock()
+        mock_client.axons.subscribe_sse.return_value = mock_stream
+
+        axon = Axon(mock_client, "axn_123")
+        result = axon.subscribe_sse()
+
+        assert result == mock_stream
+        mock_client.axons.subscribe_sse.assert_called_once_with("axn_123")
diff --git a/tests/sdk/test_ops.py b/tests/sdk/test_ops.py
index 3ff648adc..974c47e3d 100644
--- a/tests/sdk/test_ops.py
+++ b/tests/sdk/test_ops.py
@@ -11,6 +11,7 @@
 import pytest
 
 from tests.sdk.conftest import (
+    MockAxonView,
     MockAgentView,
     MockDevboxView,
     MockObjectView,
@@ -24,9 +25,11 @@
     create_mock_httpx_response,
 )
 from runloop_api_client.sdk import (
+    Axon,
     Agent,
     Devbox,
     Scorer,
+    AxonOps,
     AgentOps,
     Scenario,
     Snapshot,
@@ -725,6 +728,70 @@ def test_list_multiple(self, mock_client: Mock) -> None:
         mock_client.scenarios.scorers.list.assert_called_once()
 
 
+class TestAxonOps:
+    """Tests for AxonOps class."""
+
+    def test_create(self, mock_client: Mock, axon_view: MockAxonView) -> None:
+        """Test create method."""
+        mock_client.axons.create.return_value = axon_view
+
+        ops = AxonOps(mock_client)
+        axon = ops.create(name="test-axon")
+
+        assert isinstance(axon, Axon)
+        assert axon.id == "axn_123"
+        mock_client.axons.create.assert_called_once()
+
+    def test_from_id(self, mock_client: Mock) -> None:
+        """Test from_id method."""
+        ops = AxonOps(mock_client)
+        axon = ops.from_id("axn_123")
+
+        assert isinstance(axon, Axon)
+        assert axon.id == "axn_123"
+
+    def test_list_empty(self, mock_client: Mock) -> None:
+        """Test list method with empty results."""
+        page = SimpleNamespace(axons=[])
+        mock_client.axons.list.return_value = page
+
+        ops = AxonOps(mock_client)
+        axons = ops.list()
+
+        assert len(axons) == 0
+        mock_client.axons.list.assert_called_once()
+
+    def test_list_single(self, mock_client: Mock, axon_view: MockAxonView) -> None:
+        """Test list method with single result."""
+        page = SimpleNamespace(axons=[axon_view])
+        mock_client.axons.list.return_value = page
+
+        ops = AxonOps(mock_client)
+        axons = ops.list()
+
+        assert len(axons) == 1
+        assert isinstance(axons[0], Axon)
+        assert axons[0].id == "axn_123"
+        mock_client.axons.list.assert_called_once()
+
+    def test_list_multiple(self, mock_client: Mock) -> None:
+        """Test list method with multiple results."""
+        axon_view1 = MockAxonView(id="axn_001", name="axon-1")
+        axon_view2 = MockAxonView(id="axn_002", name="axon-2")
+        page = SimpleNamespace(axons=[axon_view1, axon_view2])
+        mock_client.axons.list.return_value = page
+
+        ops = AxonOps(mock_client)
+        axons = ops.list()
+
+        assert len(axons) == 2
+        assert isinstance(axons[0], Axon)
+        assert isinstance(axons[1], Axon)
+        assert axons[0].id == "axn_001"
+        assert axons[1].id == "axn_002"
+        mock_client.axons.list.assert_called_once()
+
+
 class TestAgentClient:
     """Tests for AgentClient class."""
 
diff --git a/tests/smoketests/sdk/test_async_axon.py b/tests/smoketests/sdk/test_async_axon.py
new file mode 100644
index 000000000..ced85820d
--- /dev/null
+++ b/tests/smoketests/sdk/test_async_axon.py
@@ -0,0 +1,81 @@
+"""Asynchronous SDK smoke tests for Axon operations."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from runloop_api_client.sdk import AsyncRunloopSDK
+
+pytestmark = [pytest.mark.smoketest, pytest.mark.asyncio]
+
+THIRTY_SECOND_TIMEOUT = 30
+
+
+class TestAsyncAxonLifecycle:
+    """Test basic async axon lifecycle operations."""
+
+    @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
+    async def test_axon_create(self, async_sdk_client: AsyncRunloopSDK) -> None:
+        """Test creating an axon."""
+        axon = await async_sdk_client.axon.create()
+
+        try:
+            assert axon is not None
+            assert axon.id is not None
+            assert len(axon.id) > 0
+
+            info = await axon.get_info()
+            assert info.id == axon.id
+            assert info.created_at_ms > 0
+        finally:
+            # TODO: Add axon cleanup once delete endpoint is implemented
+            pass
+
+    @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
+    async def test_axon_from_id(self, async_sdk_client: AsyncRunloopSDK) -> None:
+        """Test retrieving axon by ID."""
+        created = await async_sdk_client.axon.create()
+
+        try:
+            retrieved = async_sdk_client.axon.from_id(created.id)
+            assert retrieved.id == created.id
+
+            info = await retrieved.get_info()
+            assert info.id == created.id
+        finally:
+            # TODO: Add axon cleanup once delete endpoint is implemented
+            pass
+
+    @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
+    async def test_axon_publish(self, async_sdk_client: AsyncRunloopSDK) -> None:
+        """Test publishing events to an axon."""
+        axon = await async_sdk_client.axon.create()
+
+        try:
+            result = await axon.publish(
+                event_type="test_event",
+                origin="USER_EVENT",
+                payload=json.dumps({"message": "hello"}),
+                source="sdk-smoke-test",
+            )
+
+            assert result is not None
+            assert result.sequence >= 0
+            assert result.timestamp_ms > 0
+        finally:
+            # TODO: Add axon cleanup once delete endpoint is implemented
+            pass
+
+
+class TestAsyncAxonListing:
+    """Test axon listing operations."""
+
+    @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
+    async def test_list_axons(self, async_sdk_client: AsyncRunloopSDK) -> None:
+        """Test listing axons."""
+        axons = await async_sdk_client.axon.list()
+
+        assert isinstance(axons, list)
+        assert len(axons) >= 0
diff --git a/tests/smoketests/sdk/test_axon.py b/tests/smoketests/sdk/test_axon.py
new file mode 100644
index 000000000..c8947d006
--- /dev/null
+++ b/tests/smoketests/sdk/test_axon.py
@@ -0,0 +1,81 @@
+"""Synchronous SDK smoke tests for Axon operations."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from runloop_api_client.sdk import RunloopSDK
+
+pytestmark = [pytest.mark.smoketest]
+
+THIRTY_SECOND_TIMEOUT = 30
+
+
+class TestAxonLifecycle:
+    """Test basic axon lifecycle operations."""
+
+    @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
+    def test_axon_create(self, sdk_client: RunloopSDK) -> None:
+        """Test creating an axon."""
+        axon = sdk_client.axon.create()
+
+        try:
+            assert axon is not None
+            assert axon.id is not None
+            assert len(axon.id) > 0
+
+            info = axon.get_info()
+            assert info.id == axon.id
+            assert info.created_at_ms > 0
+        finally:
+            # TODO: Add axon cleanup once delete endpoint is implemented
+            pass
+
+    @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
+    def test_axon_from_id(self, sdk_client: RunloopSDK) -> None:
+        """Test retrieving axon by ID."""
+        created = sdk_client.axon.create()
+
+        try:
+            retrieved = sdk_client.axon.from_id(created.id)
+            assert retrieved.id == created.id
+
+            info = retrieved.get_info()
+            assert info.id == created.id
+        finally:
+            # TODO: Add axon cleanup once delete endpoint is implemented
+            pass
+
+    @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
+    def test_axon_publish(self, sdk_client: RunloopSDK) -> None:
+        """Test publishing events to an axon."""
+        axon = sdk_client.axon.create()
+
+        try:
+            result = axon.publish(
+                event_type="test_event",
+                origin="USER_EVENT",
+                payload=json.dumps({"message": "hello"}),
+                source="sdk-smoke-test",
+            )
+
+            assert result is not None
+            assert result.sequence >= 0
+            assert result.timestamp_ms > 0
+        finally:
+            # TODO: Add axon cleanup once delete endpoint is implemented
+            pass
+
+
+class TestAxonListing:
+    """Test axon listing operations."""
+
+    @pytest.mark.timeout(THIRTY_SECOND_TIMEOUT)
+    def test_list_axons(self, sdk_client: RunloopSDK) -> None:
+        """Test listing axons."""
+        axons = sdk_client.axon.list()
+
+        assert isinstance(axons, list)
+        assert len(axons) >= 0
diff --git a/uv.lock b/uv.lock
index 79832c62a..a43d415ef 100644
--- a/uv.lock
+++ b/uv.lock
@@ -2386,7 +2386,7 @@ wheels = [
 
 [[package]]
 name = "runloop-api-client"
-version = "1.12.1"
+version = "1.13.0"
 source = { editable = "." }
 dependencies = [
     { name = "anyio" },