feat: add ha_fire_event tool for firing events on the HA event bus (#996)

- New ha_fire_event tool in ServiceTools; accepts event_type + optional data
  (dict or JSON string). List data rejected with ToolError.
- Adds fire to approved tool verb list in AGENTS.md.
- Unit tests (7): no-data, dict, JSON-string, list-rejected, fallback-message,
  connection-error, forwarding.
- E2E tests (5): no-data, dict-data, JSON-string, list-rejected, built-in event.

Author: Kan-nguyen

AGENTS.md

@@ -508,6 +508,7 @@ These feed the picker tiles in the markup section AND the wizard `<script>` bloc
 - `delete` — delete dashboards, config entries, or files (`ha_config_delete_dashboard`, `ha_delete_file`)
 - `remove` — remove registry items (`ha_remove_entity`, `ha_remove_area_or_floor`)
 - `call` — execute (`ha_call_service`)
+- `fire` — dispatch a one-shot event onto the HA event bus (`ha_fire_event`)
 - `manage` — multi-modal tools combining several operations behind one interface (`ha_manage_addon`)
 
 **Namespace prefixes**: An optional `<namespace>_` prefix between `ha_` and the verb is allowed for grouped tool families that share a domain. The full shape becomes `ha_<namespace>_<verb>_<noun>`:

src/ha_mcp/tools/tools_service.py

@@ -431,6 +431,62 @@ async def ha_bulk_control(
         )
         return cast(dict[str, Any], result)
 
+    @tool(
+        name="ha_fire_event",
+        tags={"Service & Device Control"},
+        annotations={
+            "destructiveHint": False,
+            "idempotentHint": False,
+            "title": "Fire Event",
+        },
+    )
+    @log_tool_usage
+    async def ha_fire_event(
+        self,
+        event_type: str,
+        data: str | dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """Fire a Home Assistant event on the event bus.
+
+        When NOT to use: for controlling entities (lights, switches, climate) — use
+        ha_call_service instead. For triggering automations by name, use
+        ha_call_service("automation", "trigger").
+
+        Use this to fire custom event types consumed by event-triggered automations,
+        Node-RED flows, or custom integrations that subscribe to specific event types.
+        """
+        try:
+            parsed_data: dict[str, Any] | None = None
+            if data is not None:
+                raw = parse_json_param(data, "data")
+                if raw is not None and not isinstance(raw, dict):
+                    raise_tool_error(
+                        create_validation_error(
+                            "Event data must be a JSON object (dict), not a list",
+                            parameter="data",
+                        )
+                    )
+                parsed_data = cast(dict[str, Any], raw)
+
+            response = await self._client.fire_event(event_type, parsed_data)
+
+            return {
+                "success": True,
+                "event_type": event_type,
+                "message": response.get("message", f"Event {event_type} fired."),
+            }
+        except ToolError:
+            raise
+        except Exception as e:
+            exception_to_structured_error(
+                e,
+                context={"event_type": event_type},
+                suggestions=[
+                    "Check Home Assistant connection",
+                    "Verify event_type is a valid identifier",
+                ],
+            )
+
 
 def register_service_tools(mcp: Any, client: Any, **kwargs: Any) -> None:
     """Register service call and operation monitoring tools with the MCP server."""

tests/src/e2e/tools/test_ha_fire_event.py

@@ -0,0 +1,75 @@
+"""
+E2E tests for ha_fire_event tool - fire events onto the HA event bus.
+"""
+
+import logging
+
+import pytest
+from fastmcp.exceptions import ToolError
+
+from ..utilities.assertions import assert_mcp_success, safe_call_tool
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.asyncio
+async def test_fire_event_no_data(mcp_client):
+    """Fire a custom event without event data."""
+    result = await mcp_client.call_tool(
+        "ha_fire_event",
+        {"event_type": "test_mcp_event_no_data"},
+    )
+    data = assert_mcp_success(result, "fire event without data")
+    assert data["success"] is True
+    assert data["event_type"] == "test_mcp_event_no_data"
+    assert "message" in data
+    logger.info("Fired event without data: %s", data)
+
+
+@pytest.mark.asyncio
+async def test_fire_event_with_dict_data(mcp_client):
+    """Fire a custom event with dict event data."""
+    result = await mcp_client.call_tool(
+        "ha_fire_event",
+        {"event_type": "test_mcp_event_with_data", "data": {"key": "value", "count": 1}},
+    )
+    data = assert_mcp_success(result, "fire event with dict data")
+    assert data["success"] is True
+    assert data["event_type"] == "test_mcp_event_with_data"
+    logger.info("Fired event with dict data: %s", data)
+
+
+@pytest.mark.asyncio
+async def test_fire_event_with_json_string_data(mcp_client):
+    """Fire a custom event with JSON-string event data (auto-parsed)."""
+    result = await mcp_client.call_tool(
+        "ha_fire_event",
+        {"event_type": "test_mcp_event_json_str", "data": '{"source": "e2e_test"}'},
+    )
+    data = assert_mcp_success(result, "fire event with JSON string data")
+    assert data["success"] is True
+    assert data["event_type"] == "test_mcp_event_json_str"
+    logger.info("Fired event with JSON string data: %s", data)
+
+
+@pytest.mark.asyncio
+async def test_fire_event_list_data_rejected(mcp_client):
+    """Event data must be a dict — a JSON array is rejected with ToolError."""
+    with pytest.raises(ToolError):
+        await mcp_client.call_tool(
+            "ha_fire_event",
+            {"event_type": "test_mcp_event_bad_data", "data": "[1, 2, 3]"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_fire_builtin_event_type(mcp_client):
+    """Fire a built-in HA event type to verify the API accepts it."""
+    result = await mcp_client.call_tool(
+        "ha_fire_event",
+        {"event_type": "homeassistant_start"},
+    )
+    data = assert_mcp_success(result, "fire homeassistant_start event")
+    assert data["success"] is True
+    assert data["event_type"] == "homeassistant_start"
+    logger.info("Fired homeassistant_start: %s", data)

tests/src/unit/test_ha_fire_event.py

@@ -0,0 +1,64 @@
+"""Unit tests for ha_fire_event tool."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from fastmcp.exceptions import ToolError
+
+from ha_mcp.tools.tools_service import ServiceTools
+
+
+def _make_tools(fire_event_return: dict | Exception) -> ServiceTools:
+    client = MagicMock()
+    if isinstance(fire_event_return, Exception):
+        client.fire_event = AsyncMock(side_effect=fire_event_return)
+    else:
+        client.fire_event = AsyncMock(return_value=fire_event_return)
+    tools = ServiceTools.__new__(ServiceTools)
+    tools._client = client
+    tools._device_tools = MagicMock()
+    return tools
+
+
+class TestHaFireEvent:
+    async def test_fires_event_with_no_data(self):
+        tools = _make_tools({"message": "Event my_event fired."})
+        result = await tools.ha_fire_event("my_event")
+        assert result["success"] is True
+        assert result["event_type"] == "my_event"
+        assert "fired" in result["message"]
+        tools._client.fire_event.assert_called_once_with("my_event", None)
+
+    async def test_fires_event_with_dict_data(self):
+        tools = _make_tools({"message": "Event custom_event fired."})
+        result = await tools.ha_fire_event("custom_event", {"key": "value"})
+        assert result["success"] is True
+        tools._client.fire_event.assert_called_once_with("custom_event", {"key": "value"})
+
+    async def test_fires_event_with_json_string_data(self):
+        tools = _make_tools({"message": "Event my_event fired."})
+        result = await tools.ha_fire_event("my_event", '{"temperature": 22}')
+        assert result["success"] is True
+        tools._client.fire_event.assert_called_once_with("my_event", {"temperature": 22})
+
+    async def test_raises_tool_error_on_list_data(self):
+        tools = _make_tools({"message": "ok"})
+        with pytest.raises(ToolError):
+            await tools.ha_fire_event("my_event", "[1, 2, 3]")
+
+    async def test_returns_fallback_message_when_response_empty(self):
+        tools = _make_tools({})
+        result = await tools.ha_fire_event("my_event")
+        assert "my_event" in result["message"]
+
+    async def test_propagates_connection_error_as_tool_error(self):
+        tools = _make_tools(ConnectionError("HA unreachable"))
+        with pytest.raises(ToolError):
+            await tools.ha_fire_event("my_event")
+
+    async def test_event_type_passed_to_client(self):
+        tools = _make_tools({"message": "Event zone_entered fired."})
+        await tools.ha_fire_event("zone_entered", {"zone": "home"})
+        call_args = tools._client.fire_event.call_args
+        assert call_args[0][0] == "zone_entered"
+        assert call_args[0][1] == {"zone": "home"}