New in version:
{version}
);
};
## /docs/style.css
```css path="/docs/style.css"
/* Code highlighting -- target only inline code elements, not code blocks */
p code:not(pre code),
table code:not(pre code),
li code:not(pre code),
h1 code:not(pre code),
h2 code:not(pre code),
h3 code:not(pre code),
h4 code:not(pre code),
h5 code:not(pre code),
h6 code:not(pre code) {
color: #f72585 !important;
background-color: rgba(247, 37, 133, 0.09);
}
/* Version badge -- display a badge with the current version of the documentation */
.version-badge {
display: inline-block;
align-items: center;
gap: 0.3em;
padding: 0.2em 0.8em;
font-size: 1.1em;
font-weight: 400;
font-family: "Inter", sans-serif;
letter-spacing: 0.025em;
color: #ff5400;
background: #ffeee6;
border: 1px solid rgb(255, 84, 0, 0.5);
border-radius: 6px;
box-shadow: none;
vertical-align: middle;
position: relative;
transition: box-shadow 0.2s, transform 0.15s;
}
.version-badge-container {
margin: 0;
padding: 0;
}
.version-badge:hover {
box-shadow: 0 2px 8px 0 rgba(160, 132, 252, 0.1);
transform: translateY(-1px) scale(1.03);
}
.dark .version-badge {
color: #fff;
background: #312e81;
border: 1.5px solid #a78bfa;
}
```
## /examples/complex_inputs.py
```py path="/examples/complex_inputs.py"
"""
FastMCP Complex inputs Example
Demonstrates validation via pydantic with complex models.
"""
from typing import Annotated
from pydantic import BaseModel, Field
from fastmcp.server import FastMCP
mcp = FastMCP("Shrimp Tank")
class ShrimpTank(BaseModel):
class Shrimp(BaseModel):
name: Annotated[str, Field(max_length=10)]
shrimp: list[Shrimp]
@mcp.tool()
def name_shrimp(
tank: ShrimpTank,
# You can use pydantic Field in function signatures for validation.
extra_names: Annotated[list[str], Field(max_length=10)],
) -> list[str]:
"""List all shrimp names in the tank"""
return [shrimp.name for shrimp in tank.shrimp] + extra_names
```
## /examples/desktop.py
```py path="/examples/desktop.py"
"""
FastMCP Desktop Example
A simple example that exposes the desktop directory as a resource.
"""
from pathlib import Path
from fastmcp.server import FastMCP
# Create server
mcp = FastMCP("Demo")
@mcp.resource("dir://desktop")
def desktop() -> list[str]:
"""List the files in the user's desktop"""
desktop = Path.home() / "Desktop"
return [str(f) for f in desktop.iterdir()]
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
```
## /examples/echo.py
```py path="/examples/echo.py"
"""
FastMCP Echo Server
"""
from fastmcp import FastMCP
# Create server
mcp = FastMCP("Echo Server")
@mcp.tool()
def echo_tool(text: str) -> str:
"""Echo the input text"""
return text
@mcp.resource("echo://static")
def echo_resource() -> str:
return "Echo!"
@mcp.resource("echo://{text}")
def echo_template(text: str) -> str:
"""Echo the input text"""
return f"Echo: {text}"
@mcp.prompt("echo")
def echo_prompt(text: str) -> str:
return text
```
## /examples/memory.py
```py path="/examples/memory.py"
# /// script
# dependencies = ["pydantic-ai-slim[openai]", "asyncpg", "numpy", "pgvector", "fastmcp"]
# ///
# uv pip install 'pydantic-ai-slim[openai]' asyncpg numpy pgvector fastmcp
"""
Recursive memory system inspired by the human brain's clustering of memories.
Uses OpenAI's 'text-embedding-3-small' model and pgvector for efficient similarity search.
"""
import asyncio
import math
import os
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Annotated, Self
import asyncpg
import numpy as np
from openai import AsyncOpenAI
from pgvector.asyncpg import register_vector # Import register_vector
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from fastmcp import FastMCP
MAX_DEPTH = 5
SIMILARITY_THRESHOLD = 0.7
DECAY_FACTOR = 0.99
REINFORCEMENT_FACTOR = 1.1
DEFAULT_LLM_MODEL = "openai:gpt-4o"
DEFAULT_EMBEDDING_MODEL = "text-embedding-3-small"
mcp = FastMCP(
"memory",
dependencies=[
"pydantic-ai-slim[openai]",
"asyncpg",
"numpy",
"pgvector",
],
)
DB_DSN = "postgresql://postgres:postgres@localhost:54320/memory_db"
# reset memory with rm ~/.fastmcp/{USER}/memory/*
PROFILE_DIR = (
Path.home() / ".fastmcp" / os.environ.get("USER", "anon") / "memory"
).resolve()
PROFILE_DIR.mkdir(parents=True, exist_ok=True)
def cosine_similarity(a: list[float], b: list[float]) -> float:
a_array = np.array(a, dtype=np.float64)
b_array = np.array(b, dtype=np.float64)
return np.dot(a_array, b_array) / (
np.linalg.norm(a_array) * np.linalg.norm(b_array)
)
async def do_ai[T](
user_prompt: str,
system_prompt: str,
result_type: type[T] | Annotated,
deps=None,
) -> T:
agent = Agent(
DEFAULT_LLM_MODEL,
system_prompt=system_prompt,
result_type=result_type,
)
result = await agent.run(user_prompt, deps=deps)
return result.data
@dataclass
class Deps:
openai: AsyncOpenAI
pool: asyncpg.Pool
async def get_db_pool() -> asyncpg.Pool:
async def init(conn):
await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
await register_vector(conn)
pool = await asyncpg.create_pool(DB_DSN, init=init)
return pool
class MemoryNode(BaseModel):
id: int | None = None
content: str
summary: str = ""
importance: float = 1.0
access_count: int = 0
timestamp: float = Field(
default_factory=lambda: datetime.now(timezone.utc).timestamp()
)
embedding: list[float]
@classmethod
async def from_content(cls, content: str, deps: Deps):
embedding = await get_embedding(content, deps)
return cls(content=content, embedding=embedding)
async def save(self, deps: Deps):
async with deps.pool.acquire() as conn:
if self.id is None:
result = await conn.fetchrow(
"""
INSERT INTO memories (content, summary, importance, access_count, timestamp, embedding)
VALUES ($1, $2, $3, $4, $5, $6)
RETURNING id
""",
self.content,
self.summary,
self.importance,
self.access_count,
self.timestamp,
self.embedding,
)
self.id = result["id"]
else:
await conn.execute(
"""
UPDATE memories
SET content = $1, summary = $2, importance = $3,
access_count = $4, timestamp = $5, embedding = $6
WHERE id = $7
""",
self.content,
self.summary,
self.importance,
self.access_count,
self.timestamp,
self.embedding,
self.id,
)
async def merge_with(self, other: Self, deps: Deps):
self.content = await do_ai(
f"{self.content}\n\n{other.content}",
"Combine the following two texts into a single, coherent text.",
str,
deps,
)
self.importance += other.importance
self.access_count += other.access_count
self.embedding = [(a + b) / 2 for a, b in zip(self.embedding, other.embedding)]
self.summary = await do_ai(
self.content, "Summarize the following text concisely.", str, deps
)
await self.save(deps)
# Delete the merged node from the database
if other.id is not None:
await delete_memory(other.id, deps)
def get_effective_importance(self):
return self.importance * (1 + math.log(self.access_count + 1))
async def get_embedding(text: str, deps: Deps) -> list[float]:
embedding_response = await deps.openai.embeddings.create(
input=text,
model=DEFAULT_EMBEDDING_MODEL,
)
return embedding_response.data[0].embedding
async def delete_memory(memory_id: int, deps: Deps):
async with deps.pool.acquire() as conn:
await conn.execute("DELETE FROM memories WHERE id = $1", memory_id)
async def add_memory(content: str, deps: Deps):
new_memory = await MemoryNode.from_content(content, deps)
await new_memory.save(deps)
similar_memories = await find_similar_memories(new_memory.embedding, deps)
for memory in similar_memories:
if memory.id != new_memory.id:
await new_memory.merge_with(memory, deps)
await update_importance(new_memory.embedding, deps)
await prune_memories(deps)
return f"Remembered: {content}"
async def find_similar_memories(embedding: list[float], deps: Deps) -> list[MemoryNode]:
async with deps.pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT id, content, summary, importance, access_count, timestamp, embedding
FROM memories
ORDER BY embedding <-> $1
LIMIT 5
""",
embedding,
)
memories = [
MemoryNode(
id=row["id"],
content=row["content"],
summary=row["summary"],
importance=row["importance"],
access_count=row["access_count"],
timestamp=row["timestamp"],
embedding=row["embedding"],
)
for row in rows
]
return memories
async def update_importance(user_embedding: list[float], deps: Deps):
async with deps.pool.acquire() as conn:
rows = await conn.fetch(
"SELECT id, importance, access_count, embedding FROM memories"
)
for row in rows:
memory_embedding = row["embedding"]
similarity = cosine_similarity(user_embedding, memory_embedding)
if similarity > SIMILARITY_THRESHOLD:
new_importance = row["importance"] * REINFORCEMENT_FACTOR
new_access_count = row["access_count"] + 1
else:
new_importance = row["importance"] * DECAY_FACTOR
new_access_count = row["access_count"]
await conn.execute(
"""
UPDATE memories
SET importance = $1, access_count = $2
WHERE id = $3
""",
new_importance,
new_access_count,
row["id"],
)
async def prune_memories(deps: Deps):
async with deps.pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT id, importance, access_count
FROM memories
ORDER BY importance DESC
OFFSET $1
""",
MAX_DEPTH,
)
for row in rows:
await conn.execute("DELETE FROM memories WHERE id = $1", row["id"])
async def display_memory_tree(deps: Deps) -> str:
async with deps.pool.acquire() as conn:
rows = await conn.fetch(
"""
SELECT content, summary, importance, access_count
FROM memories
ORDER BY importance DESC
LIMIT $1
""",
MAX_DEPTH,
)
result = ""
for row in rows:
effective_importance = row["importance"] * (
1 + math.log(row["access_count"] + 1)
)
summary = row["summary"] or row["content"]
result += f"- {summary} (Importance: {effective_importance:.2f})\n"
return result
@mcp.tool()
async def remember(
contents: list[str] = Field(
description="List of observations or memories to store"
),
):
deps = Deps(openai=AsyncOpenAI(), pool=await get_db_pool())
try:
return "\n".join(
await asyncio.gather(*[add_memory(content, deps) for content in contents])
)
finally:
await deps.pool.close()
@mcp.tool()
async def read_profile() -> str:
deps = Deps(openai=AsyncOpenAI(), pool=await get_db_pool())
profile = await display_memory_tree(deps)
await deps.pool.close()
return profile
async def initialize_database():
pool = await asyncpg.create_pool(
"postgresql://postgres:postgres@localhost:54320/postgres"
)
try:
async with pool.acquire() as conn:
await conn.execute("""
SELECT pg_terminate_backend(pg_stat_activity.pid)
FROM pg_stat_activity
WHERE pg_stat_activity.datname = 'memory_db'
AND pid <> pg_backend_pid();
""")
await conn.execute("DROP DATABASE IF EXISTS memory_db;")
await conn.execute("CREATE DATABASE memory_db;")
finally:
await pool.close()
pool = await asyncpg.create_pool(DB_DSN)
try:
async with pool.acquire() as conn:
await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
await register_vector(conn)
await conn.execute("""
CREATE TABLE IF NOT EXISTS memories (
id SERIAL PRIMARY KEY,
content TEXT NOT NULL,
summary TEXT,
importance REAL NOT NULL,
access_count INT NOT NULL,
timestamp DOUBLE PRECISION NOT NULL,
embedding vector(1536) NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_memories_embedding ON memories USING hnsw (embedding vector_l2_ops);
""")
finally:
await pool.close()
if __name__ == "__main__":
asyncio.run(initialize_database())
```
## /examples/mount_example.py
```py path="/examples/mount_example.py"
"""Example of mounting FastMCP apps together.
This example demonstrates how to mount FastMCP apps together using
the ToolManager's import_tools functionality. It shows how to:
1. Create sub-applications for different domains
2. Mount those sub-applications to a main application
3. Access tools with prefixed names and resources with prefixed URIs
"""
import asyncio
from fastmcp import FastMCP
# Weather sub-application
weather_app = FastMCP("Weather App")
@weather_app.tool()
def get_weather_forecast(location: str) -> str:
"""Get the weather forecast for a location."""
return f"Sunny skies for {location} today!"
@weather_app.resource(uri="weather://forecast")
async def weather_data():
"""Return current weather data."""
return {"temperature": 72, "conditions": "sunny", "humidity": 45, "wind_speed": 5}
# News sub-application
news_app = FastMCP("News App")
@news_app.tool()
def get_news_headlines() -> list[str]:
"""Get the latest news headlines."""
return [
"Tech company launches new product",
"Local team wins championship",
"Scientists make breakthrough discovery",
]
@news_app.resource(uri="news://headlines")
async def news_data():
"""Return latest news data."""
return {
"top_story": "Breaking news: Important event happened",
"categories": ["politics", "sports", "technology"],
"sources": ["AP", "Reuters", "Local Sources"],
}
# Main application
app = FastMCP(
"Main App", dependencies=["fastmcp@git+https://github.com/jlowin/fastmcp.git"]
)
@app.tool()
def check_app_status() -> dict[str, str]:
"""Check the status of the main application."""
return {"status": "running", "version": "1.0.0", "uptime": "3h 24m"}
# Mount sub-applications
app.mount("weather", weather_app)
app.mount("news", news_app)
async def get_server_details():
"""Print information about mounted resources."""
# Print available tools
tools = app._tool_manager.list_tools()
print(f"\nAvailable tools ({len(tools)}):")
for tool in tools:
print(f" - {tool.name}: {tool.description}")
# Print available resources
print("\nAvailable resources:")
# Distinguish between native and imported resources
# Native resources would be those directly in the main app (not prefixed)
native_resources = [
uri
for uri in app._resource_manager._resources
if not (uri.startswith("weather+") or uri.startswith("news+"))
]
# Imported resources - categorized by source app
weather_resources = [
uri for uri in app._resource_manager._resources if uri.startswith("weather+")
]
news_resources = [
uri for uri in app._resource_manager._resources if uri.startswith("news+")
]
print(f" - Native app resources: {native_resources}")
print(f" - Imported from weather app: {weather_resources}")
print(f" - Imported from news app: {news_resources}")
# Let's try to access resources using the prefixed URI
weather_data = await app.read_resource("weather+weather://forecast")
print(f"\nWeather data from prefixed URI: {weather_data}")
if __name__ == "__main__":
# First run our async function to display info
asyncio.run(get_server_details())
# Then start the server (uncomment to run the server)
app.run()
```
## /examples/readme-quickstart.py
```py path="/examples/readme-quickstart.py"
from fastmcp import FastMCP
# Create an MCP server
mcp = FastMCP("Demo")
# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
```
## /examples/sampling.py
```py path="/examples/sampling.py"
"""
Example of using sampling to request an LLM completion via Marvin
"""
import asyncio
import marvin
from mcp.types import TextContent
from fastmcp import Client, Context, FastMCP
from fastmcp.client.sampling import RequestContext, SamplingMessage, SamplingParams
# -- Create a server that sends a sampling request to the LLM
mcp = FastMCP("Sampling Example")
@mcp.tool()
async def example_tool(prompt: str, context: Context) -> str:
"""Sample a completion from the LLM."""
response = await context.sample(
"What is your favorite programming language?",
system_prompt="You love languages named after snakes.",
)
assert isinstance(response, TextContent)
return response.text
# -- Create a client that can handle the sampling request
async def sampling_fn(
messages: list[SamplingMessage],
params: SamplingParams,
ctx: RequestContext,
) -> str:
return await marvin.say_async(
message=[m.content.text for m in messages],
instructions=params.systemPrompt,
)
async def run():
async with Client(mcp, sampling_handler=sampling_fn) as client:
result = await client.call_tool(
"example_tool", {"prompt": "What is the best programming language?"}
)
print(result)
if __name__ == "__main__":
asyncio.run(run())
```
## /examples/screenshot.py
```py path="/examples/screenshot.py"
"""
FastMCP Screenshot Example
Give Claude a tool to capture and view screenshots.
"""
import io
from fastmcp import FastMCP, Image
# Create server
mcp = FastMCP("Screenshot Demo", dependencies=["pyautogui", "Pillow"])
@mcp.tool()
def take_screenshot() -> Image:
"""
Take a screenshot of the user's screen and return it as an image. Use
this tool anytime the user wants you to look at something they're doing.
"""
import pyautogui
buffer = io.BytesIO()
# if the file exceeds ~1MB, it will be rejected by Claude
screenshot = pyautogui.screenshot()
screenshot.convert("RGB").save(buffer, format="JPEG", quality=60, optimize=True)
return Image(data=buffer.getvalue(), format="jpeg")
```
## /examples/simple_echo.py
```py path="/examples/simple_echo.py"
"""
FastMCP Echo Server
"""
from fastmcp import FastMCP
# Create server
mcp = FastMCP("Echo Server")
@mcp.tool()
def echo(text: str) -> str:
"""Echo the input text"""
return text
```
## /examples/smart_home/README.md
## /examples/smart_home/pyproject.toml
```toml path="/examples/smart_home/pyproject.toml"
[project]
name = "smart-home"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
authors = [{ name = "zzstoatzz", email = "thrast36@gmail.com" }]
requires-python = ">=3.12"
dependencies = ["fastmcp@git+https://github.com/jlowin/fastmcp.git", "phue2"]
[project.scripts]
smart-home = "smart_home.__main__:main"
[dependency-groups]
dev = ["ruff", "ipython"]
[build-system]
requires = ["uv_build"]
build-backend = "uv_build"
```
## /examples/smart_home/src/smart_home/__init__.py
```py path="/examples/smart_home/src/smart_home/__init__.py"
from smart_home.settings import settings
__all__ = ["settings"]
```
## /examples/smart_home/src/smart_home/__main__.py
```py path="/examples/smart_home/src/smart_home/__main__.py"
from smart_home.hub import hub_mcp
def main():
hub_mcp.run()
if __name__ == "__main__":
main()
```
## /examples/smart_home/src/smart_home/hub.py
```py path="/examples/smart_home/src/smart_home/hub.py"
from phue2 import Bridge
from fastmcp import FastMCP
from smart_home.lights.server import lights_mcp
from smart_home.settings import settings
hub_mcp = FastMCP(
"Smart Home Hub (phue2)",
dependencies=[
"smart_home@git+https://github.com/jlowin/fastmcp.git#subdirectory=examples/smart_home",
],
)
# Mount the lights service under the 'hue' prefix
hub_mcp.mount("hue", lights_mcp)
# Add a status check for the hub
@hub_mcp.tool()
def hub_status() -> str:
"""Checks the status of the main hub and connections."""
try:
bridge = Bridge(
ip=str(settings.hue_bridge_ip),
username=settings.hue_bridge_username,
save_config=False,
)
bridge.connect()
return "Hub OK. Hue Bridge Connected (via phue2)."
except Exception as e:
return f"Hub Warning: Hue Bridge connection failed or not attempted: {e}"
# Add mounting points for other services later
# hub_mcp.mount("thermo", thermostat_mcp)
```
## /examples/smart_home/src/smart_home/lights/__init__.py
```py path="/examples/smart_home/src/smart_home/lights/__init__.py"
```
## /examples/smart_home/src/smart_home/lights/hue_utils.py
```py path="/examples/smart_home/src/smart_home/lights/hue_utils.py"
from typing import Any
from phue2 import Bridge
from phue2.exceptions import PhueException
from smart_home.settings import settings
def _get_bridge() -> Bridge | None:
"""Attempts to connect to the Hue bridge using settings."""
try:
return Bridge(
ip=str(settings.hue_bridge_ip),
username=settings.hue_bridge_username,
save_config=False,
)
except Exception:
# Broad exception to catch potential connection issues
# TODO: Add more specific logging or error handling
return None
def handle_phue_error(
light_or_group: str, operation: str, error: Exception
) -> dict[str, Any]:
"""Creates a standardized error response for phue2 operations."""
base_info = {"target": light_or_group, "operation": operation, "success": False}
if isinstance(error, KeyError):
base_info["error"] = f"Target '{light_or_group}' not found"
elif isinstance(error, PhueException):
base_info["error"] = f"phue2 error during {operation}: {error}"
else:
base_info["error"] = f"Unexpected error during {operation}: {error}"
return base_info
```
## /examples/smart_home/src/smart_home/lights/server.py
```py path="/examples/smart_home/src/smart_home/lights/server.py"
from typing import Annotated, Any, Literal, TypedDict
from phue2.exceptions import PhueException
from pydantic import Field
from typing_extensions import NotRequired
from fastmcp import FastMCP
from smart_home.lights.hue_utils import _get_bridge, handle_phue_error
class HueAttributes(TypedDict, total=False):
"""TypedDict for optional light attributes."""
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[Annotated[int, Field(description="deciseconds")]]
lights_mcp = FastMCP(
"Hue Lights Service (phue2)",
dependencies=[
"smart_home@git+https://github.com/jlowin/fastmcp.git#subdirectory=examples/smart_home",
],
)
@lights_mcp.tool()
def read_all_lights() -> list[str]:
"""Lists the names of all available Hue lights using phue2."""
if not (bridge := _get_bridge()):
return ["Error: Bridge not connected"]
try:
light_dict = bridge.get_light_objects("list")
return [light.name for light in light_dict]
except (PhueException, Exception) as e:
# Simplified error handling for list return type
return [f"Error listing lights: {e}"]
# --- Tools ---
@lights_mcp.tool()
def toggle_light(light_name: str, state: bool) -> dict[str, Any]:
"""Turns a specific light on (true) or off (false) using phue2."""
if not (bridge := _get_bridge()):
return {"error": "Bridge not connected", "success": False}
try:
result = bridge.set_light(light_name, "on", state)
return {
"light": light_name,
"set_on_state": state,
"success": True,
"phue2_result": result,
}
except (KeyError, PhueException, Exception) as e:
return handle_phue_error(light_name, "toggle_light", e)
@lights_mcp.tool()
def set_brightness(light_name: str, brightness: int) -> dict[str, Any]:
"""Sets the brightness of a specific light (0-254) using phue2."""
if not (bridge := _get_bridge()):
return {"error": "Bridge not connected", "success": False}
if not 0 <= brightness <= 254:
# Keep specific input validation error here
return {
"light": light_name,
"error": "Brightness must be between 0 and 254",
"success": False,
}
try:
result = bridge.set_light(light_name, "bri", brightness)
return {
"light": light_name,
"set_brightness": brightness,
"success": True,
"phue2_result": result,
}
except (KeyError, PhueException, Exception) as e:
return handle_phue_error(light_name, "set_brightness", e)
@lights_mcp.tool()
def list_groups() -> list[str]:
"""Lists the names of all available Hue light groups."""
if not (bridge := _get_bridge()):
return ["Error: Bridge not connected"]
try:
# phue2 get_group() returns a dict {id: {details}} including name
groups = bridge.get_group()
return [group_details["name"] for group_details in groups.values()]
except (PhueException, Exception) as e:
return [f"Error listing groups: {e}"]
@lights_mcp.tool()
def list_scenes() -> dict[str, list[str]] | list[str]:
"""Lists Hue scenes, grouped by the light group they belong to.
Returns:
dict[str, list[str]]: A dictionary mapping group names to a list of scene names within that group.
list[str]: An error message list if the bridge connection fails or an error occurs.
"""
if not (bridge := _get_bridge()):
return ["Error: Bridge not connected"]
try:
scenes_data = bridge.get_scene() # Returns dict {scene_id: {details...}}
groups_data = bridge.get_group() # Returns dict {group_id: {details...}}
# Create a lookup for group name by group ID
group_id_to_name = {gid: ginfo["name"] for gid, ginfo in groups_data.items()}
scenes_by_group: dict[str, list[str]] = {}
for scene_id, scene_details in scenes_data.items():
scene_name = scene_details.get("name")
# Scenes might be associated with a group via 'group' key or lights
# Using 'group' key if available is more direct for group scenes
group_id = scene_details.get("group")
if scene_name and group_id and group_id in group_id_to_name:
group_name = group_id_to_name[group_id]
if group_name not in scenes_by_group:
scenes_by_group[group_name] = []
# Avoid duplicate scene names within a group listing (though unlikely)
if scene_name not in scenes_by_group[group_name]:
scenes_by_group[group_name].append(scene_name)
# Sort scenes within each group for consistent output
for group_name in scenes_by_group:
scenes_by_group[group_name].sort()
return scenes_by_group
except (PhueException, Exception) as e:
# Return error as list to match other list-returning tools on error
return [f"Error listing scenes by group: {e}"]
@lights_mcp.tool()
def activate_scene(group_name: str, scene_name: str) -> dict[str, Any]:
"""Activates a specific scene within a specified light group, verifying the scene belongs to the group."""
if not (bridge := _get_bridge()):
return {"error": "Bridge not connected", "success": False}
try:
# 1. Find the target group ID
groups_data = bridge.get_group()
target_group_id = None
for gid, ginfo in groups_data.items():
if ginfo.get("name") == group_name:
target_group_id = gid
break
if not target_group_id:
return {"error": f"Group '{group_name}' not found", "success": False}
# 2. Find the target scene and check its group association
scenes_data = bridge.get_scene()
scene_found = False
scene_in_correct_group = False
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
if sinfo.get("group") == target_group_id:
scene_in_correct_group = True
break # Found the scene in the correct group
if not scene_found:
return {"error": f"Scene '{scene_name}' not found", "success": False}
if not scene_in_correct_group:
return {
"error": f"Scene '{scene_name}' does not belong to group '{group_name}'",
"success": False,
}
# 3. Activate the scene (now that we've verified it)
result = bridge.run_scene(group_name=group_name, scene_name=scene_name)
if result:
return {
"group": group_name,
"activated_scene": scene_name,
"success": True,
"phue2_result": result,
}
else:
# This case might indicate the scene/group exists but activation failed internally
return {
"group": group_name,
"scene": scene_name,
"error": "Scene activation failed (phue2 returned False)",
"success": False,
}
except (KeyError, PhueException, Exception) as e:
# Handle potential errors during bridge communication or data parsing
return handle_phue_error(f"{group_name}/{scene_name}", "activate_scene", e)
@lights_mcp.tool()
def set_light_attributes(light_name: str, attributes: HueAttributes) -> dict[str, Any]:
"""Sets multiple attributes (e.g., hue, sat, bri, ct, xy, transitiontime) for a specific light."""
if not (bridge := _get_bridge()):
return {"error": "Bridge not connected", "success": False}
# Basic validation (more specific validation could be added)
if not isinstance(attributes, dict) or not attributes:
return {
"error": "Attributes must be a non-empty dictionary",
"success": False,
"light": light_name,
}
try:
result = bridge.set_light(light_name, dict(attributes))
return {
"light": light_name,
"set_attributes": attributes,
"success": True,
"phue2_result": result,
}
except (KeyError, PhueException, ValueError, Exception) as e:
# ValueError might occur for invalid attribute values
return handle_phue_error(light_name, "set_light_attributes", e)
@lights_mcp.tool()
def set_group_attributes(group_name: str, attributes: HueAttributes) -> dict[str, Any]:
"""Sets multiple attributes for all lights within a specific group."""
if not (bridge := _get_bridge()):
return {"error": "Bridge not connected", "success": False}
if not isinstance(attributes, dict) or not attributes:
return {
"error": "Attributes must be a non-empty dictionary",
"success": False,
"group": group_name,
}
try:
result = bridge.set_group(group_name, dict(attributes))
return {
"group": group_name,
"set_attributes": attributes,
"success": True,
"phue2_result": result,
}
except (KeyError, PhueException, ValueError, Exception) as e:
return handle_phue_error(group_name, "set_group_attributes", e)
@lights_mcp.tool()
def list_lights_by_group() -> dict[str, list[str]] | list[str]:
"""Lists Hue lights, grouped by the room/group they belong to.
Returns:
dict[str, list[str]]: A dictionary mapping group names to a list of light names within that group.
list[str]: An error message list if the bridge connection fails or an error occurs.
"""
if not (bridge := _get_bridge()):
return ["Error: Bridge not connected"]
try:
groups_data = bridge.get_group() # dict {group_id: {details}}
lights_data = bridge.get_light_objects("id") # dict {light_id: {details}}
lights_by_group: dict[str, list[str]] = {}
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:
light_names = []
for light_id in light_ids:
# phue uses string IDs for lights in group, but int IDs in get_light_objects
light_id_int = int(light_id)
if light_id_int in lights_data:
light_name = lights_data[light_id_int].name
if light_name:
light_names.append(light_name)
if light_names:
light_names.sort()
lights_by_group[group_name] = light_names
return lights_by_group
except (PhueException, Exception) as e:
return [f"Error listing lights by group: {e}"]
```
## /examples/smart_home/src/smart_home/py.typed
```typed path="/examples/smart_home/src/smart_home/py.typed"
```
## /examples/smart_home/src/smart_home/settings.py
```py path="/examples/smart_home/src/smart_home/settings.py"
from pydantic import Field, IPvAnyAddress
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
hue_bridge_ip: IPvAnyAddress = Field(default=...)
hue_bridge_username: str = Field(default=...)
settings = Settings()
```
The content has been capped at 50000 tokens, and files over NaN bytes have been omitted. The user could consider applying other filters to refine the result. The better and more specific the context, the better the LLM can follow instructions. If the context seems verbose, the user can refine the filter using uithub. Thank you for using https://uithub.com - Perfect LLM context for any GitHub repo.