Add Home Automation capability

pydantic_ai_harness/__init__.py

@@ -4,13 +4,18 @@
 
 if TYPE_CHECKING:
     from .code_mode import CodeMode
+    from .home_automation import HomeAutomation
 
-__all__ = ['CodeMode']
+__all__ = ['CodeMode', 'HomeAutomation']
 
 
 def __getattr__(name: str) -> object:
     if name == 'CodeMode':
         from .code_mode import CodeMode
 
         return CodeMode
+    if name == 'HomeAutomation':
+        from .home_automation import HomeAutomation
+
+        return HomeAutomation
     raise AttributeError(f'module {__name__!r} has no attribute {name!r}')

pydantic_ai_harness/home_automation/README.md

@@ -0,0 +1,63 @@
+# Home Automation
+
+Home Automation exposes smart home entities and service calls to a Pydantic AI
+agent. The current backend targets the Home Assistant REST API.
+
+```python
+from pydantic_ai import Agent
+from pydantic_ai_harness import HomeAutomation
+from pydantic_ai_harness.home_automation import HomeAssistantBackend
+
+backend = HomeAssistantBackend(
+    url='http://localhost:8123',
+    token='...',
+)
+
+agent = Agent(
+    'openai:gpt-5',
+    capabilities=[HomeAutomation(backend=backend)],
+)
+```
+
+## Tools
+
+The capability exposes the backend methods directly as tools:
+
+- `list_services(domain=None)`: list callable Home Assistant services and their normalized arguments.
+- `list_entities(domain=None)`: list entity summaries.
+- `get_state(entity_id)`: get the current state for one entity.
+- `list_states(domain=None)`: list current entity states.
+- `call_service(domain, entity_id, service_name, **data)`: call a service for one entity.
+
+Agents should usually discover entities with `list_entities`, inspect available
+services with `list_services`, then call `call_service` with the exact
+`domain`, `service_name`, and `entity_id`.
+
+## Home Assistant Backend
+
+`HomeAssistantBackend` validates `/api/services` and `/api/states` responses
+with Pydantic models, then converts them into backend-neutral dataclasses for
+the agent-facing tools.
+
+Service calls return a `ServiceCallResult` with:
+
+- `changed_states`: states Home Assistant reports as changed during execution.
+- `service_response`: response data for services that support or require it.
+- `verified_state`: a best-effort post-call state for the target entity.
+
+If Home Assistant returns no changed states or service response data,
+`HomeAssistantBackend` may poll the target entity state to populate
+`verified_state`. Configure this with `verification_poll_attempts` and
+`verification_poll_interval`.
+
+```python
+backend = HomeAssistantBackend(
+    url='http://localhost:8123',
+    token='...',
+    verification_poll_attempts=3,
+    verification_poll_interval=0.5,
+)
+```
+
+Injected `httpx.AsyncClient` instances are not closed by `aclose`; clients
+created by the backend are closed by `aclose`.

pydantic_ai_harness/home_automation/__init__.py

@@ -0,0 +1,7 @@
+"""Home automation capability and backend exports."""
+
+from pydantic_ai_harness.home_automation._capability import HomeAutomation
+from pydantic_ai_harness.home_automation._toolset import HomeAutomationToolset
+from pydantic_ai_harness.home_automation.backends.home_assistant._backend import HomeAssistantBackend
+
+__all__ = ['HomeAutomation', 'HomeAutomationToolset', 'HomeAssistantBackend']

pydantic_ai_harness/home_automation/_capability.py

@@ -0,0 +1,34 @@
+from dataclasses import dataclass
+
+from pydantic_ai._run_context import AgentDepsT
+from pydantic_ai.capabilities import AbstractCapability
+
+from pydantic_ai_harness.home_automation._toolset import HomeAutomationToolset
+from pydantic_ai_harness.home_automation.backends import HomeBackend
+
+
+@dataclass
+class HomeAutomation(AbstractCapability[AgentDepsT]):
+    """Capability exposing Home Assistant-style service discovery."""
+
+    backend: HomeBackend
+
+    def get_instructions(self) -> str:
+        """Return guidance for agents using the home automation capability."""
+        return (
+            'You have access to tools that let you inspect and control smart home '
+            'entities such as lights, switches, and climate devices through Home '
+            'Assistant. Use `list_entities` or `list_states` to discover available '
+            'entities, and `get_state` when you need the current state of one entity. '
+            'Use `list_services` to discover valid services and their arguments before '
+            'calling them. When using `call_service`, pass the exact `domain`, '
+            '`service_name`, and `entity_id` returned by Home Assistant. After a '
+            'service call, prefer `verified_state` as the strongest confirmation of '
+            'what happened, then `changed_states`, and finally `service_response`. '
+            '`call_service` may perform follow-up state reads when Home Assistant does '
+            'not return changed states or response data.'
+        )
+
+    def get_toolset(self) -> HomeAutomationToolset[AgentDepsT]:
+        """Expose the home automation tools to the agent runtime."""
+        return HomeAutomationToolset(self.backend)

pydantic_ai_harness/home_automation/_toolset.py

@@ -0,0 +1,24 @@
+"""Toolset for exposing home automation backends to agents."""
+
+from pydantic_ai import FunctionToolset
+from pydantic_ai._run_context import AgentDepsT
+
+from pydantic_ai_harness.home_automation.backends import HomeBackend
+
+
+class HomeAutomationToolset(FunctionToolset[AgentDepsT]):
+    """Function toolset backed by a `HomeBackend` implementation."""
+
+    backend: HomeBackend
+
+    def __init__(self, backend: HomeBackend) -> None:
+        self.backend = backend
+        super().__init__(
+            tools=[
+                backend.list_services,
+                backend.list_entities,
+                backend.get_state,
+                backend.list_states,
+                backend.call_service,
+            ]
+        )

pydantic_ai_harness/home_automation/backends/__init__.py

@@ -0,0 +1,12 @@
+"""Backend contracts and shared home automation models."""
+
+from pydantic_ai_harness.home_automation.backends._base_backend import (
+    Entity,
+    EntityState,
+    HomeBackend,
+    Service,
+    ServiceArgument,
+    ServiceCallResult,
+)
+
+__all__ = ['Entity', 'EntityState', 'HomeBackend', 'Service', 'ServiceArgument', 'ServiceCallResult']

pydantic_ai_harness/home_automation/backends/_base_backend.py

@@ -0,0 +1,78 @@
+from dataclasses import dataclass, field
+from typing import Any, Protocol
+
+
+@dataclass
+class ServiceArgument:
+    name: str
+    value_type: str
+    required: bool = False
+    description: str | None = None
+    enum: tuple[str, ...] = ()
+    minimum: float | int | None = None
+    maximum: float | int | None = None
+
+
+@dataclass
+class Service:
+    domain: str
+    name: str
+    args: tuple[ServiceArgument, ...] = ()
+
+
+@dataclass
+class Entity:
+    entity_id: str
+    domain: str
+    name: str | None
+    current_state: str
+
+
+@dataclass
+class EntityState:
+    entity_id: str
+    last_updated: str
+    state: str
+
+
+def _empty_entity_states() -> list[EntityState]:
+    return []
+
+
+@dataclass
+class ServiceCallResult:
+    changed_states: list[EntityState] = field(default_factory=_empty_entity_states)
+    service_response: dict[str, Any] | None = None
+    verified_state: EntityState | None = None
+
+
+class HomeBackend(Protocol):
+    """Backend contract for home automation service discovery."""
+
+    async def list_services(self, domain: str | None = None) -> list[Service]:  # pragma: no cover
+        """Return the services that can be called, optionally filtered by domain."""
+        ...
+
+    async def list_entities(self, domain: str | None = None) -> list[Entity]:  # pragma: no cover
+        """Return known entities, optionally filtered by domain."""
+        ...
+
+    async def get_state(self, entity_id: str) -> EntityState:  # pragma: no cover
+        """Return the current state for a single entity."""
+        ...
+
+    async def list_states(self, domain: str | None = None) -> list[EntityState]:  # pragma: no cover
+        """Return current states, optionally filtered by domain."""
+        ...
+
+    async def call_service(
+        self,
+        domain: str,
+        entity_id: str,
+        service_name: str,
+        *,
+        want_response: bool = False,
+        **data: Any,
+    ) -> ServiceCallResult:  # pragma: no cover
+        """Call a service for one entity and return the observed outcome."""
+        ...

pydantic_ai_harness/home_automation/backends/home_assistant/__init__.py

@@ -0,0 +1,5 @@
+"""Home Assistant backend exports."""
+
+from ._backend import HomeAssistantBackend as HomeAssistantBackend
+
+__all__ = ['HomeAssistantBackend']