Merge pull request #237 from jlowin/nits

small typing fixes
This commit is contained in:
Jeremiah Lowin 2025-04-26 07:58:05 -04:00 committed by GitHub
commit ee68d1429d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 44 additions and 25 deletions

View file

@ -11,15 +11,28 @@ from smart_home.lights.hue_utils import _get_bridge, handle_phue_error
class HueAttributes(TypedDict, total=False):
"""TypedDict for optional light attributes."""
on: NotRequired[bool]
bri: NotRequired[Annotated[int, Field(ge=0, le=254)]]
hue: NotRequired[Annotated[int, Field(ge=0, le=65535)]]
sat: NotRequired[Annotated[int, Field(ge=0, le=254)]]
xy: NotRequired[list[float]]
ct: NotRequired[Annotated[int, Field(ge=153, le=500)]]
on: NotRequired[Annotated[bool, Field(description="on/off state")]]
bri: NotRequired[Annotated[int, Field(ge=0, le=254, description="brightness")]]
hue: NotRequired[
Annotated[
int,
Field(
ge=0,
le=254,
description="saturation",
),
]
]
xy: NotRequired[Annotated[list[float], Field(description="xy color coordinates")]]
ct: NotRequired[
Annotated[
int,
Field(ge=153, le=500, description="color temperature"),
]
]
alert: NotRequired[Literal["none", "select", "lselect"]]
effect: NotRequired[Literal["none", "colorloop"]]
transitiontime: NotRequired[int] # deciseconds
transitiontime: NotRequired[Annotated[int, Field(description="deciseconds")]]
lights_mcp = FastMCP(
@ -161,7 +174,7 @@ def activate_scene(group_name: str, scene_name: str) -> dict[str, Any]:
scenes_data = bridge.get_scene()
scene_found = False
scene_in_correct_group = False
for sid, sinfo in scenes_data.items():
for sinfo in scenes_data.values():
if sinfo.get("name") == scene_name:
scene_found = True
# Check if this scene is associated with the target group ID
@ -217,7 +230,7 @@ def set_light_attributes(light_name: str, attributes: HueAttributes) -> dict[str
}
try:
result = bridge.set_light(light_name, attributes)
result = bridge.set_light(light_name, dict(attributes))
return {
"light": light_name,
"set_attributes": attributes,
@ -243,7 +256,7 @@ def set_group_attributes(group_name: str, attributes: HueAttributes) -> dict[str
}
try:
result = bridge.set_group(group_name, attributes)
result = bridge.set_group(group_name, dict(attributes))
return {
"group": group_name,
"set_attributes": attributes,
@ -269,7 +282,7 @@ def list_lights_by_group() -> dict[str, list[str]] | list[str]:
lights_data = bridge.get_light_objects("id") # dict {light_id: {details}}
lights_by_group: dict[str, list[str]] = {}
for group_id, group_details in groups_data.items():
for group_details in groups_data.values():
group_name = group_details.get("name")
light_ids = group_details.get("lights", [])
if group_name and light_ids:
@ -282,11 +295,10 @@ def list_lights_by_group() -> dict[str, list[str]] | list[str]:
if light_name:
light_names.append(light_name)
if light_names:
light_names.sort() # Keep light list sorted
light_names.sort()
lights_by_group[group_name] = light_names
return lights_by_group
except (PhueException, Exception) as e:
# Return error as list
return [f"Error listing lights by group: {e}"]

View file

@ -14,6 +14,7 @@ __all__ = [
"FastMCP",
"Context",
"client",
"Client",
"settings",
"Image",
]

View file

@ -1,5 +1,7 @@
"""FastMCP - A more ergonomic interface for MCP servers."""
from __future__ import annotations
import datetime
from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import (
@ -63,7 +65,7 @@ class MountedServer:
def __init__(
self,
prefix: str,
server: "FastMCP",
server: FastMCP,
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
@ -149,7 +151,7 @@ class TimedCache:
@asynccontextmanager
async def default_lifespan(server: "FastMCP") -> AsyncIterator[Any]:
async def default_lifespan(server: FastMCP) -> AsyncIterator[Any]:
"""Default lifespan context manager that does nothing.
Args:
@ -162,8 +164,8 @@ async def default_lifespan(server: "FastMCP") -> AsyncIterator[Any]:
def _lifespan_wrapper(
app: "FastMCP",
lifespan: Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]],
app: FastMCP,
lifespan: Callable[[FastMCP], AbstractAsyncContextManager[LifespanResultT]],
) -> Callable[
[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]
]:
@ -182,7 +184,11 @@ class FastMCP(Generic[LifespanResultT]):
name: str | None = None,
instructions: str | None = None,
lifespan: (
Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]] | None
Callable[
[FastMCP[LifespanResultT]],
AbstractAsyncContextManager[LifespanResultT],
]
| None
) = None,
tags: set[str] | None = None,
**settings: Any,
@ -273,7 +279,7 @@ class FastMCP(Generic[LifespanResultT]):
self._mcp_server.get_prompt()(self._mcp_get_prompt)
self._mcp_server.list_resource_templates()(self._mcp_list_resource_templates)
def get_context(self) -> "Context[ServerSession, LifespanResultT]":
def get_context(self) -> Context[ServerSession, LifespanResultT]:
"""
Returns a Context object. Note that the context will only be valid
during a request; outside a request, most methods will error.
@ -766,7 +772,7 @@ class FastMCP(Generic[LifespanResultT]):
def mount(
self,
prefix: str,
server: "FastMCP",
server: FastMCP[LifespanResultT],
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
@ -791,7 +797,7 @@ class FastMCP(Generic[LifespanResultT]):
async def import_server(
self,
prefix: str,
server: "FastMCP",
server: FastMCP[LifespanResultT],
tool_separator: str | None = None,
resource_separator: str | None = None,
prompt_separator: str | None = None,
@ -865,7 +871,7 @@ class FastMCP(Generic[LifespanResultT]):
@classmethod
def from_openapi(
cls, openapi_spec: dict[str, Any], client: httpx.AsyncClient, **settings: Any
) -> "FastMCPOpenAPI":
) -> FastMCPOpenAPI:
"""
Create a FastMCP server from an OpenAPI specification.
"""
@ -875,8 +881,8 @@ class FastMCP(Generic[LifespanResultT]):
@classmethod
def from_fastapi(
cls, app: "Any", name: str | None = None, **settings: Any
) -> "FastMCPOpenAPI":
cls, app: Any, name: str | None = None, **settings: Any
) -> FastMCPOpenAPI:
"""
Create a FastMCP server from a FastAPI application.
"""
@ -894,7 +900,7 @@ class FastMCP(Generic[LifespanResultT]):
)
@classmethod
def from_client(cls, client: "Client", **settings: Any) -> "FastMCPProxy":
def from_client(cls, client: Client, **settings: Any) -> FastMCPProxy:
"""
Create a FastMCP proxy server from a FastMCP client.
"""