# Directory Structure
```
├── .env.example
├── .gitignore
├── Dockerfile
├── documentation
│ ├── 01_configuration_management.md
│ ├── 02_device_control_interfaces.md
│ ├── 03_govee_api_client.md
│ ├── 04_command_line_interface__cli_.md
│ ├── 05_mcp_server_implementation.md
│ ├── 06_custom_error_handling.md
│ └── index.md
├── inspector.bat
├── inspector.sh
├── LICENSE
├── pyproject.toml
├── README.md
├── slides
│ └── Meetup 30 janvier.pptx
├── smithery.yaml
├── src
│ └── govee_mcp_server
│ ├── __init__.py
│ ├── api.py
│ ├── cli.py
│ ├── config.py
│ ├── exceptions.py
│ ├── interfaces.py
│ ├── server.py
│ └── transformers.py
├── tests
│ ├── test_cli.py
│ └── test_server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
GOVEE_API_KEY=<API_KEY_FROM_GOVEE_APP>
GOVEE_DEVICE_ID=<MAC_ADDRESS_OF_THE_DEVICE>
GOVEE_SKU=<SKU_OF_THE_DEVICE>
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
local_docs/
.DS_Store
Thumbs.db
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
env/
venv/
ENV/
env.bak/
venv.bak/
*.egg-info/
dist/
build/
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# dotenv
.env
# virtualenv
.venv/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# VSCode settings
.vscode/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Govee MCP Server
[](https://smithery.ai/server/@mathd/govee_mcp_server)
An MCP server for controlling Govee LED devices through the Govee API.
## Setup
### Environment Variables
Create a `.env` file in the root directory with the following variables:
```bash
GOVEE_API_KEY=your_api_key_here
GOVEE_DEVICE_ID=your_device_id_here
GOVEE_SKU=your_device_sku_here
```
To get these values:
1. Get your API key from the Govee Developer Portal
2. Use the Govee Home app to find your device ID and SKU
## Installation
### Installing via Smithery
To install Govee MCP Server for Claude Desktop automatically via [Smithery](https://smithery.ai/server/@mathd/govee_mcp_server):
```bash
npx -y @smithery/cli install @mathd/govee_mcp_server --client claude
```
### Manual Installation
```bash
# Install with pip
pip install .
# For development (includes test dependencies)
pip install -e ".[test]"
```
## Usage
### MCP Server
The MCP server provides tools for controlling Govee devices through the Model Context Protocol. It can be used with Cline or other MCP clients.
Available tools:
- `turn_on_off`: Turn the LED on or off
- `set_color`: Set the LED color using RGB values
- `set_brightness`: Set the LED brightness level
### Command Line Interface
A CLI is provided for direct control of Govee devices:
```bash
# Turn device on/off
govee-cli power on
govee-cli power off
# Set color using RGB values (0-255)
govee-cli color 255 0 0 # Red
govee-cli color 0 255 0 # Green
govee-cli color 0 0 255 # Blue
# Set brightness (0-100)
govee-cli brightness 50
```
Run `govee-cli --help` for full command documentation.
## Development
### Running Tests
To run the test suite:
```bash
# Install test dependencies
pip install -e ".[test]"
# Run all tests
pytest tests/
# Run specific test files
pytest tests/test_server.py # Server tests (mocked API calls)
pytest tests/test_cli.py # CLI tests (real API calls)
# Run tests with verbose output
pytest tests/ -v
```
Note: The CLI tests make real API calls to your Govee device and will actually control it. Make sure your device is powered and connected before running these tests.
### Project Structure
```
.
├── src/govee_mcp_server/
│ ├── __init__.py
│ ├── server.py # MCP server implementation
│ └── cli.py # Command-line interface
├── tests/
│ ├── test_server.py # Server tests (with mocked API)
│ └── test_cli.py # CLI tests (real API calls)
└── pyproject.toml # Project configuration
```
### Test Coverage
- Server tests cover:
- Environment initialization
- Govee API client methods
- Server tools and utilities
- Error handling
- CLI tests perform real-world integration testing by executing actual API calls to control your Govee device.
```
--------------------------------------------------------------------------------
/inspector.bat:
--------------------------------------------------------------------------------
```
npx @modelcontextprotocol/inspector uv --directory . run python src\govee_mcp_server\server.py
```
--------------------------------------------------------------------------------
/inspector.sh:
--------------------------------------------------------------------------------
```bash
npx @modelcontextprotocol/inspector uv --directory . run python src/govee_mcp_server/server.py
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/exceptions.py:
--------------------------------------------------------------------------------
```python
class GoveeError(Exception):
"""Base exception for Govee-related errors."""
pass
class GoveeAPIError(GoveeError):
"""Raised when API communication fails."""
pass
class GoveeConfigError(GoveeError):
"""Raised when there are configuration-related errors."""
pass
class GoveeValidationError(GoveeError):
"""Raised when input validation fails."""
pass
class GoveeConnectionError(GoveeError):
"""Raised when network connection issues occur."""
pass
class GoveeTimeoutError(GoveeError):
"""Raised when requests timeout."""
pass
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "govee-mcp-server"
version = "0.1.0"
description = "MCP server to control Govee LED devices"
authors = []
dependencies = [
"mcp[cli]",
"govee-api-laggat>=0.2.2",
"python-dotenv"
]
requires-python = ">=3.10"
[project.scripts]
govee-cli = "govee_mcp_server.cli:cli_main"
[project.optional-dependencies]
test = [
"pytest>=7.0",
"pytest-asyncio>=0.21.0",
"pytest-mock>=3.10.0"
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/govee_mcp_server"]
[tool.hatch.metadata]
allow-direct-references = true
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
pythonpath = ["src"]
asyncio_default_fixture_loop_scope = "function"
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
# Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
startCommand:
type: stdio
configSchema:
# JSON Schema defining the configuration options for the MCP.
type: object
required:
- goveeApiKey
- goveeDeviceId
- goveeSku
properties:
goveeApiKey:
type: string
description: The API key for the Govee API.
goveeDeviceId:
type: string
description: The device ID for the Govee device.
goveeSku:
type: string
description: The SKU for the Govee device.
commandFunction:
# A function that produces the CLI command to start the MCP on stdio.
|-
(config) => ({command: 'python', args: ['src/govee_mcp_server/server.py'], env: {GOVEE_API_KEY: config.goveeApiKey, GOVEE_DEVICE_ID: config.goveeDeviceId, GOVEE_SKU: config.goveeSku}})
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Govee MCP Server - A Model Context Protocol server for controlling Govee LED devices.
This package provides a complete interface for controlling Govee LED devices through both
an MCP server implementation and a CLI interface.
"""
from .api import GoveeAPI
from .config import GoveeConfig, load_config
from .exceptions import (
GoveeError,
GoveeAPIError,
GoveeConfigError,
GoveeValidationError,
GoveeConnectionError,
GoveeTimeoutError
)
from .transformers import ColorTransformer
from .interfaces import PowerControl, ColorControl, BrightnessControl
__version__ = "0.1.0"
__all__ = [
'GoveeAPI',
'GoveeConfig',
'load_config',
'GoveeError',
'GoveeAPIError',
'GoveeConfigError',
'GoveeValidationError',
'GoveeConnectionError',
'GoveeTimeoutError',
'ColorTransformer',
'PowerControl',
'ColorControl',
'BrightnessControl'
]
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
# Use a Python image with uv pre-installed
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS uv
# Set the working directory
WORKDIR /app
# Enable bytecode compilation
ENV UV_COMPILE_BYTECODE=1
# Copy from the cache instead of linking since it's a mounted volume
ENV UV_LINK_MODE=copy
# Install the project's dependencies using the lockfile and settings
RUN --mount=type=cache,target=/root/.cache/uv \
--mount=type=bind,source=uv.lock,target=uv.lock \
--mount=type=bind,source=pyproject.toml,target=pyproject.toml \
uv sync --frozen --no-install-project --no-dev --no-editable
# Copy the project source code
COPY . /app
# Install the project
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --frozen --no-dev --no-editable
# Final image
FROM python:3.12-slim-bookworm
# Set the working directory
WORKDIR /app
# Copy the environment setup from the uv image
COPY --from=uv --chown=app:app /app/.venv /app/.venv
# Place executables in the environment at the front of the path
ENV PATH="/app/.venv/bin:$PATH"
# when running the container, add --db-path and a bind mount to the host's db file
ENTRYPOINT ["python", "src/govee_mcp_server/server.py"]
```
--------------------------------------------------------------------------------
/tests/test_cli.py:
--------------------------------------------------------------------------------
```python
import pytest
import sys
import asyncio
from govee_mcp_server.cli import main
from govee_mcp_server.config import load_config, GoveeConfigError
# Delay between commands (in seconds)
DELAY = 1
@pytest.mark.asyncio
async def test_cli_interface():
"""Test CLI interface with real API calls"""
try:
# Load actual config from environment
config = load_config()
# Power on
sys.argv = ['cli.py', 'power', 'on']
await main()
await asyncio.sleep(DELAY)
# Red color
sys.argv = ['cli.py', 'color', '255', '0', '0']
await main()
await asyncio.sleep(DELAY)
# Green color
sys.argv = ['cli.py', 'color', '0', '255', '0']
await main()
await asyncio.sleep(DELAY)
# Blue color
sys.argv = ['cli.py', 'color', '0', '0', '255']
await main()
await asyncio.sleep(DELAY)
# Power off
sys.argv = ['cli.py', 'power', 'off']
await main()
except GoveeConfigError as e:
pytest.skip(f"Skipping test: {str(e)}")
except Exception as e:
# If we hit rate limits or other API errors, fail with clear message
pytest.fail(f"API Error: {str(e)}")
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/config.py:
--------------------------------------------------------------------------------
```python
from pathlib import Path
from dataclasses import dataclass
import os
from dotenv import load_dotenv
from typing import Optional
@dataclass
class GoveeConfig:
"""Configuration class for Govee API settings."""
api_key: str
device_id: str
sku: str
class GoveeConfigError(Exception):
"""Configuration-related errors for Govee MCP server."""
pass
def load_config() -> GoveeConfig:
"""
Load and validate configuration from environment variables.
Returns:
GoveeConfig: Configuration object with API settings
Raises:
GoveeConfigError: If required environment variables are missing
"""
env_path = Path(__file__).resolve().parent.parent.parent / '.env'
load_dotenv(env_path)
api_key = os.getenv('GOVEE_API_KEY')
device_id = os.getenv('GOVEE_DEVICE_ID')
sku = os.getenv('GOVEE_SKU')
missing = []
if not api_key:
missing.append('GOVEE_API_KEY')
if not device_id:
missing.append('GOVEE_DEVICE_ID')
if not sku:
missing.append('GOVEE_SKU')
if missing:
raise GoveeConfigError(f"Missing required environment variables: {', '.join(missing)}")
return GoveeConfig(
api_key=api_key,
device_id=device_id,
sku=sku
)
```
--------------------------------------------------------------------------------
/documentation/index.md:
--------------------------------------------------------------------------------
```markdown
# Tutorial: govee_mcp_server
This project lets you control your **Govee LED lights** over the internet.
It acts as a bridge, translating commands into the *Govee API language*.
You can interact with it either through an **AI assistant** (using the *MCP server*) or directly using simple **command-line** instructions.
It needs your Govee *API key* and *device details* to work, which it reads from configuration settings.
**Source Repository:** [https://github.com/mathd/govee_mcp_server](https://github.com/mathd/govee_mcp_server)
```mermaid
flowchart TD
A0["Govee API Client"]
A1["MCP Server Implementation"]
A2["Configuration Management"]
A3["Device Control Interfaces"]
A4["Command Line Interface (CLI)"]
A5["Custom Error Handling"]
A1 -- "Executes commands via" --> A0
A4 -- "Executes commands via" --> A0
A0 -- "Reads settings from" --> A2
A0 -- "Implements" --> A3
A0 -- "Raises API/Network Errors" --> A5
A1 -- "Handles Errors From" --> A5
A4 -- "Handles Errors From" --> A5
A2 -- "Raises Configuration Errors" --> A5
A3 -- "Raises Validation Errors" --> A5
```
## Chapters
1. [Configuration Management](01_configuration_management.md)
2. [Device Control Interfaces](02_device_control_interfaces.md)
3. [Govee API Client](03_govee_api_client.md)
4. [Command Line Interface (CLI)](04_command_line_interface__cli_.md)
5. [MCP Server Implementation](05_mcp_server_implementation.md)
6. [Custom Error Handling](06_custom_error_handling.md)
---
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/interfaces.py:
--------------------------------------------------------------------------------
```python
from abc import ABC, abstractmethod
from typing import Tuple
from functools import wraps
from .exceptions import GoveeValidationError
def validate_rgb(func):
"""Decorator to validate RGB color values."""
@wraps(func)
async def wrapper(self, r: int, g: int, b: int, *args, **kwargs):
for name, value in [('red', r), ('green', g), ('blue', b)]:
if not isinstance(value, int):
raise GoveeValidationError(f"{name} value must be an integer")
if not 0 <= value <= 255:
raise GoveeValidationError(f"{name} value must be between 0-255")
return await func(self, r, g, b, *args, **kwargs)
return wrapper
class PowerControl(ABC):
"""Interface for power control capabilities."""
@abstractmethod
async def set_power(self, state: bool) -> Tuple[bool, str]:
"""
Set device power state.
Args:
state: True for on, False for off
Returns:
Tuple of (success: bool, message: str)
"""
pass
@abstractmethod
async def get_power_state(self) -> Tuple[bool, str]:
"""
Get current power state.
Returns:
Tuple of (is_on: bool, message: str)
"""
pass
class ColorControl(ABC):
"""Interface for color control capabilities."""
@abstractmethod
@validate_rgb
async def set_color(self, r: int, g: int, b: int) -> Tuple[bool, str]:
"""
Set device color using RGB values.
Args:
r: Red value (0-255)
g: Green value (0-255)
b: Blue value (0-255)
Returns:
Tuple of (success: bool, message: str)
"""
pass
@abstractmethod
async def get_color(self) -> Tuple[Tuple[int, int, int], str]:
"""
Get current color values.
Returns:
Tuple of ((r, g, b): Tuple[int, int, int], message: str)
"""
pass
class BrightnessControl(ABC):
"""Interface for brightness control capabilities."""
@abstractmethod
async def set_brightness(self, level: int) -> Tuple[bool, str]:
"""
Set device brightness level.
Args:
level: Brightness level (0-100)
Returns:
Tuple of (success: bool, message: str)
"""
pass
@abstractmethod
async def get_brightness(self) -> Tuple[int, str]:
"""
Get current brightness level.
Returns:
Tuple of (level: int, message: str)
"""
pass
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/transformers.py:
--------------------------------------------------------------------------------
```python
from typing import Tuple, Dict, Any
from .exceptions import GoveeValidationError
class ColorTransformer:
"""Handle color transformations and validations."""
@staticmethod
def validate_rgb(r: int, g: int, b: int) -> None:
"""
Validate RGB color values.
Args:
r: Red value (0-255)
g: Green value (0-255)
b: Blue value (0-255)
Raises:
GoveeValidationError: If values are invalid
"""
for name, value in [('red', r), ('green', g), ('blue', b)]:
if not isinstance(value, int):
raise GoveeValidationError(f"{name} value must be an integer")
if not 0 <= value <= 255:
raise GoveeValidationError(f"{name} value must be between 0-255")
@staticmethod
def rgb_to_hex(r: int, g: int, b: int) -> str:
"""
Convert RGB values to hexadecimal color code.
Args:
r: Red value (0-255)
g: Green value (0-255)
b: Blue value (0-255)
Returns:
str: Hexadecimal color code
"""
ColorTransformer.validate_rgb(r, g, b)
return f"#{r:02x}{g:02x}{b:02x}"
@staticmethod
def hex_to_rgb(hex_color: str) -> Tuple[int, int, int]:
"""
Convert hexadecimal color code to RGB values.
Args:
hex_color: Hexadecimal color code (e.g., '#ff00ff' or 'ff00ff')
Returns:
Tuple[int, int, int]: RGB values
Raises:
GoveeValidationError: If hex color format is invalid
"""
# Remove '#' if present
hex_color = hex_color.lstrip('#')
if len(hex_color) != 6:
raise GoveeValidationError("Invalid hex color format")
try:
r = int(hex_color[:2], 16)
g = int(hex_color[2:4], 16)
b = int(hex_color[4:], 16)
return (r, g, b)
except ValueError:
raise GoveeValidationError("Invalid hex color format")
@staticmethod
def to_api_payload(r: int, g: int, b: int) -> Dict[str, Any]:
"""
Convert RGB values to API payload format.
Args:
r: Red value (0-255)
g: Green value (0-255)
b: Blue value (0-255)
Returns:
Dict[str, Any]: API payload
"""
ColorTransformer.validate_rgb(r, g, b)
return {
"color": {
"r": r,
"g": g,
"b": b
}
}
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
import pytest
import asyncio
import aiohttp
from unittest.mock import patch, MagicMock
from govee_mcp_server.server import (
init_env,
GoveeDirectAPI,
fix_json,
rgb_to_int,
)
@pytest.fixture
def mock_env_vars():
with patch.dict('os.environ', {
'GOVEE_API_KEY': 'test-api-key',
'GOVEE_DEVICE_ID': 'test-device-id',
'GOVEE_SKU': 'test-sku'
}):
yield
def test_init_env(mock_env_vars):
api_key, device_id, sku = init_env()
assert api_key == 'test-api-key'
assert device_id == 'test-device-id'
assert sku == 'test-sku'
@patch('os.getenv', return_value=None)
def test_init_env_missing_vars(_):
with pytest.raises(SystemExit):
init_env()
def test_fix_json():
malformed = '{"key1""value1"}{"key2""value2"}'
expected = '{"key1","value1"},{"key2","value2"}'
assert fix_json(malformed) == expected
def test_rgb_to_int():
assert rgb_to_int(255, 0, 0) == 0xFF0000
assert rgb_to_int(0, 255, 0) == 0x00FF00
assert rgb_to_int(0, 0, 255) == 0x0000FF
assert rgb_to_int(255, 255, 255) == 0xFFFFFF
class TestGoveeDirectAPI:
@pytest.fixture
def api(self):
return GoveeDirectAPI('test-api-key')
def test_init(self, api):
assert api.api_key == 'test-api-key'
assert api.headers['Govee-API-Key'] == 'test-api-key'
assert api.headers['Content-Type'] == 'application/json'
@pytest.mark.asyncio
async def test_get_devices_success(self, api):
mock_response = MagicMock()
mock_response.status = 200
async def mock_text():
return '{"data":[{"device":"test"}]}'
mock_response.text = mock_text
with patch('aiohttp.ClientSession.get') as mock_get:
mock_get.return_value.__aenter__.return_value = mock_response
devices, error = await api.get_devices()
assert devices == [{"device": "test"}]
assert error is None
@pytest.mark.asyncio
async def test_get_devices_error(self, api):
mock_response = MagicMock()
mock_response.status = 401
async def mock_text():
return '{"message":"Unauthorized"}'
mock_response.text = mock_text
with patch('aiohttp.ClientSession.get') as mock_get:
mock_get.return_value.__aenter__.return_value = mock_response
devices, error = await api.get_devices()
assert devices is None
assert error == "Unauthorized"
@pytest.mark.asyncio
async def test_control_device_success(self, api):
mock_response = MagicMock()
mock_response.status = 200
async def mock_text():
return '{"message":"Success"}'
mock_response.text = mock_text
with patch('aiohttp.ClientSession.post') as mock_post:
mock_post.return_value.__aenter__.return_value = mock_response
success, error = await api.control_device(
"sku123",
"device123",
"devices.capabilities.on_off",
"powerSwitch",
1
)
assert success is True
assert error == "Success"
@pytest.mark.asyncio
async def test_control_device_error(self, api):
mock_response = MagicMock()
mock_response.status = 400
async def mock_text():
return '{"message":"Bad Request"}'
mock_response.text = mock_text
with patch('aiohttp.ClientSession.post') as mock_post:
mock_post.return_value.__aenter__.return_value = mock_response
success, error = await api.control_device(
"sku123",
"device123",
"devices.capabilities.on_off",
"powerSwitch",
1
)
assert success is False
assert error == "Bad Request"
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/cli.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
import sys
import argparse
import asyncio
from .config import load_config
from .api import GoveeAPI
from .exceptions import GoveeError, GoveeValidationError
def create_parser() -> argparse.ArgumentParser:
"""Create and configure argument parser."""
parser = argparse.ArgumentParser(description='Control Govee LED device')
subparsers = parser.add_subparsers(dest='command', help='Commands')
# Power command
power_parser = subparsers.add_parser('power', help='Turn device on/off')
power_parser.add_argument('state', choices=['on', 'off'], help='Power state')
# Color command
color_parser = subparsers.add_parser('color', help='Set device color')
color_parser.add_argument('red', type=int, help='Red value (0-255)')
color_parser.add_argument('green', type=int, help='Green value (0-255)')
color_parser.add_argument('blue', type=int, help='Blue value (0-255)')
# Brightness command
brightness_parser = subparsers.add_parser('brightness', help='Set device brightness')
brightness_parser.add_argument('level', type=int, help='Brightness level (0-100)')
# Status command
subparsers.add_parser('status', help='Show device status')
return parser
async def handle_power(api: GoveeAPI, state: str) -> None:
"""Handle power command."""
success, message = await api.set_power(state == 'on')
if not success:
raise GoveeError(message)
print(message)
async def handle_color(api: GoveeAPI, red: int, green: int, blue: int) -> None:
"""Handle color command."""
try:
success, message = await api.set_color(red, green, blue)
if not success:
raise GoveeError(message)
print(message)
except GoveeValidationError as e:
print(f"Error: {e}")
sys.exit(1)
async def handle_brightness(api: GoveeAPI, level: int) -> None:
"""Handle brightness command."""
if not 0 <= level <= 100:
print("Error: Brightness level must be between 0 and 100")
sys.exit(1)
success, message = await api.set_brightness(level)
if not success:
raise GoveeError(message)
print(message)
async def handle_status(api: GoveeAPI) -> None:
"""Handle status command."""
# Get power state
power_state, power_msg = await api.get_power_state()
print(f"Power: {'ON' if power_state else 'OFF'}")
# Get color
color, color_msg = await api.get_color()
print(f"Color: RGB({color[0]}, {color[1]}, {color[2]})")
# Get brightness
brightness, bright_msg = await api.get_brightness()
print(f"Brightness: {brightness}%")
async def main() -> None:
"""Main CLI entrypoint."""
try:
# Load configuration
config = load_config()
api = GoveeAPI(config)
# Parse arguments
parser = create_parser()
args = parser.parse_args()
if args.command == 'power':
await handle_power(api, args.state)
elif args.command == 'color':
await handle_color(api, args.red, args.green, args.blue)
elif args.command == 'brightness':
await handle_brightness(api, args.level)
elif args.command == 'status':
await handle_status(api)
else:
parser.print_help()
sys.exit(1)
except GoveeError as e:
print(f"Error: {str(e)}")
sys.exit(1)
except Exception as e:
print(f"Unexpected error: {str(e)}")
sys.exit(1)
finally:
# Always close the API session
if 'api' in locals():
await api.close()
def cli_main():
"""CLI entry point that handles running the async main."""
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nOperation cancelled by user")
sys.exit(1)
if __name__ == "__main__":
cli_main()
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/server.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
import sys
from mcp.server.fastmcp import FastMCP
from govee_mcp_server.config import load_config
from govee_mcp_server.api import GoveeAPI
from govee_mcp_server.exceptions import GoveeError
# Initialize FastMCP server with WARNING log level
mcp = FastMCP(
"govee",
capabilities={
"server_info": {
"name": "govee-mcp",
"version": "0.1.0",
"description": "MCP server for controlling Govee LED devices"
}
},
log_level='WARNING'
)
print("Loading configuration...", file=sys.stderr, flush=True)
try:
config = load_config()
except GoveeError as e:
print(f"Configuration error: {e}", file=sys.stderr)
sys.exit(1)
print("Setting up tools...", file=sys.stderr, flush=True)
@mcp.tool("turn_on_off")
async def turn_on_off(power: bool) -> str:
"""
Turn the LED on or off.
Args:
power: True for on, False for off
"""
api = GoveeAPI(config)
try:
success, message = await api.set_power(power)
await api.close() # Clean up the session
return message if success else f"Failed: {message}"
except GoveeError as e:
await api.close()
return f"Error: {str(e)}"
except Exception as e:
await api.close()
return f"Unexpected error: {str(e)}"
@mcp.tool("set_color")
async def set_color(red: int, green: int, blue: int) -> str:
"""
Set the LED color using RGB values.
Args:
red: Red value (0-255)
green: Green value (0-255)
blue: Blue value (0-255)
"""
api = GoveeAPI(config)
try:
success, message = await api.set_color(red, green, blue)
await api.close()
return message if success else f"Failed: {message}"
except GoveeError as e:
await api.close()
return f"Error: {str(e)}"
except Exception as e:
await api.close()
return f"Unexpected error: {str(e)}"
@mcp.tool("set_brightness")
async def set_brightness(brightness: int) -> str:
"""
Set the LED brightness.
Args:
brightness: Brightness level (0-100)
"""
api = GoveeAPI(config)
try:
success, message = await api.set_brightness(brightness)
await api.close()
return message if success else f"Failed: {message}"
except GoveeError as e:
await api.close()
return f"Error: {str(e)}"
except Exception as e:
await api.close()
return f"Unexpected error: {str(e)}"
@mcp.tool("get_status")
async def get_status() -> dict:
"""Get the current status of the LED device."""
api = GoveeAPI(config)
try:
power_state, power_msg = await api.get_power_state()
color, color_msg = await api.get_color()
brightness, bright_msg = await api.get_brightness()
await api.close()
return {
"power": {
"state": "on" if power_state else "off",
"message": power_msg
},
"color": {
"r": color[0],
"g": color[1],
"b": color[2],
"message": color_msg
},
"brightness": {
"level": brightness,
"message": bright_msg
}
}
except GoveeError as e:
await api.close()
return {"error": str(e)}
except Exception as e:
await api.close()
return {"error": f"Unexpected error: {str(e)}"}
async def handle_initialize(params):
"""Handle initialize request"""
return {
"protocolVersion": "0.1.0",
"capabilities": mcp.capabilities
}
mcp.on_initialize = handle_initialize
if __name__ == "__main__":
try:
import asyncio
asyncio.run(mcp.run(transport='stdio'))
except KeyboardInterrupt:
print("\nServer stopped by user", file=sys.stderr)
except Exception as e:
print(f"Server error: {e}", file=sys.stderr)
sys.exit(1)
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/api.py:
--------------------------------------------------------------------------------
```python
import aiohttp
from typing import Optional, Dict, Any, Tuple
import asyncio
from time import time
from .exceptions import (
GoveeError,
GoveeAPIError,
GoveeConnectionError,
GoveeTimeoutError
)
from .interfaces import PowerControl, ColorControl, BrightnessControl
from .transformers import ColorTransformer
from .config import GoveeConfig
class GoveeAPI(PowerControl, ColorControl, BrightnessControl):
"""
Govee API client implementing device control interfaces.
Includes connection pooling, request timeouts, and retries.
"""
BASE_URL = "https://openapi.api.govee.com"
MAX_RETRIES = 3
RETRY_DELAY = 1 # seconds
REQUEST_TIMEOUT = 10 # seconds
def __init__(self, config: GoveeConfig):
"""
Initialize API client with configuration.
Args:
config: GoveeConfig instance with API credentials
"""
self.config = config
self.session: Optional[aiohttp.ClientSession] = None
self._transformer = ColorTransformer()
async def _ensure_session(self) -> None:
"""Ensure aiohttp session exists or create a new one."""
if self.session is None or self.session.closed:
self.session = aiohttp.ClientSession(
headers={
"Govee-API-Key": self.config.api_key,
"Content-Type": "application/json"
},
timeout=aiohttp.ClientTimeout(total=self.REQUEST_TIMEOUT)
)
async def close(self) -> None:
"""Close the API session."""
if self.session and not self.session.closed:
await self.session.close()
async def _make_request(
self,
method: str,
endpoint: str,
**kwargs
) -> Tuple[Dict[str, Any], str]:
"""
Make HTTP request with retries and error handling.
Args:
method: HTTP method
endpoint: API endpoint
**kwargs: Additional request arguments
Returns:
Tuple[Dict[str, Any], str]: API response data and message
Raises:
GoveeAPIError: On API errors
GoveeConnectionError: On connection issues
GoveeTimeoutError: On request timeout
"""
await self._ensure_session()
for attempt in range(self.MAX_RETRIES):
try:
async with self.session.request(
method,
f"{self.BASE_URL}/{endpoint}",
**kwargs
) as response:
data = await response.json()
if response.status != 200:
raise GoveeAPIError(
f"API error: {response.status} - {data.get('message', 'Unknown error')}"
)
return data, data.get('message', 'Success')
except asyncio.TimeoutError:
if attempt == self.MAX_RETRIES - 1:
raise GoveeTimeoutError(f"Request timed out after {self.REQUEST_TIMEOUT}s")
except aiohttp.ClientError as e:
if attempt == self.MAX_RETRIES - 1:
raise GoveeConnectionError(f"Connection error: {str(e)}")
await asyncio.sleep(self.RETRY_DELAY * (attempt + 1))
raise GoveeAPIError("Max retries exceeded")
async def set_power(self, state: bool) -> Tuple[bool, str]:
"""Implement PowerControl.set_power"""
try:
_, message = await self._make_request(
"POST",
"router/api/v1/device/control",
json={
"requestId": str(int(time())), # Using timestamp as requestId
"payload": {
"sku": self.config.sku,
"device": self.config.device_id,
"capability": {
"type": "devices.capabilities.on_off",
"instance": "powerSwitch",
"value": 1 if state else 0
}
}
}
)
return True, message
except GoveeError as e:
return False, str(e)
async def get_power_state(self) -> Tuple[bool, str]:
"""Implement PowerControl.get_power_state"""
try:
data, message = await self._make_request(
"GET",
f"devices/state",
params={
"device": self.config.device_id,
"model": self.config.sku
}
)
return data.get('powerState') == 'on', message
except GoveeError as e:
return False, str(e)
async def set_color(self, r: int, g: int, b: int) -> Tuple[bool, str]:
"""Implement ColorControl.set_color"""
try:
color_value = ((r & 0xFF) << 16) | ((g & 0xFF) << 8) | (b & 0xFF)
_, message = await self._make_request(
"POST",
"router/api/v1/device/control",
json={
"requestId": str(int(time())),
"payload": {
"sku": self.config.sku,
"device": self.config.device_id,
"capability": {
"type": "devices.capabilities.color_setting",
"instance": "colorRgb",
"value": color_value
}
}
}
)
return True, message
except GoveeError as e:
return False, str(e)
async def get_color(self) -> Tuple[Tuple[int, int, int], str]:
"""Implement ColorControl.get_color"""
try:
data, message = await self._make_request(
"GET",
f"devices/state",
params={
"device": self.config.device_id,
"model": self.config.sku
}
)
color = data.get('color', {})
return (
color.get('r', 0),
color.get('g', 0),
color.get('b', 0)
), message
except GoveeError as e:
return (0, 0, 0), str(e)
async def set_brightness(self, level: int) -> Tuple[bool, str]:
"""Implement BrightnessControl.set_brightness"""
if not 0 <= level <= 100:
return False, "Brightness must be between 0-100"
try:
_, message = await self._make_request(
"POST",
"router/api/v1/device/control",
json={
"requestId": str(int(time())),
"payload": {
"sku": self.config.sku,
"device": self.config.device_id,
"capability": {
"type": "devices.capabilities.range",
"instance": "brightness",
"value": level
}
}
}
)
return True, message
except GoveeError as e:
return False, str(e)
async def get_brightness(self) -> Tuple[int, str]:
"""Implement BrightnessControl.get_brightness"""
try:
data, message = await self._make_request(
"GET",
f"devices/state",
params={
"device": self.config.device_id,
"model": self.config.sku
}
)
return data.get('brightness', 0), message
except GoveeError as e:
return 0, str(e)
```
--------------------------------------------------------------------------------
/documentation/01_configuration_management.md:
--------------------------------------------------------------------------------
```markdown
# Chapter 1: Configuration Management - Giving Your App Its Credentials
Welcome to the `govee_mcp_server` tutorial! We're excited to guide you through how this project works. Let's start with the very first step any application needs: figuring out its basic settings.
## What's the Big Idea?
Imagine you want to send a letter. You need a few key things:
1. The **address** of the recipient (who are you sending it to?).
2. A **stamp** (proof you're allowed to send mail).
3. Maybe the **type of envelope** needed for that specific address.
Our `govee_mcp_server` application is similar. To talk to your Govee lights, it needs to know:
1. **Which specific device** it should control (like the address).
2. Your **Govee API Key** (like a secret password or stamp proving it's allowed to talk to Govee).
3. The **device model (SKU)** (like knowing the type of envelope needed).
"Configuration Management" is just a fancy term for how the application finds and manages these essential pieces of information. It's like the application's **ID card and key storage** – it holds the credentials and details needed to operate.
## Storing Secrets: Environment Variables and `.env`
We need a place to store sensitive information like your Govee API Key. Putting it directly into the code isn't safe or flexible. Instead, we use **environment variables**.
Think of environment variables as settings that live *outside* the application code, within the computer's operating system environment. This keeps secrets separate from the main logic.
For convenience, especially during development, we often store these environment variables in a special file named `.env` located in the project's main folder.
**Example `.env` file:**
```bash
# This is a comment - lines starting with # are ignored
GOVEE_API_KEY=abcdef12-3456-7890-abcd-ef1234567890
GOVEE_DEVICE_ID=AB:CD:EF:12:34:56:78:90
GOVEE_SKU=H6159
```
* `GOVEE_API_KEY`: Your personal key from Govee. Keep this secret!
* `GOVEE_DEVICE_ID`: The unique identifier for your specific Govee light.
* `GOVEE_SKU`: The model number of your Govee light.
**Analogy:** Think of the `.env` file like a secure sticky note next to your computer with login details, rather than writing them directly in your public diary (the code).
## Loading the Settings: The `load_config` Function
Okay, so the details are in the `.env` file (or set directly as environment variables). How does our application actually *read* them? It uses a helper function called `load_config`.
Let's look at the heart of this process in `src/govee_mcp_server/config.py`:
```python
# Simplified snippet from src/govee_mcp_server/config.py
import os
from dotenv import load_dotenv
from pathlib import Path
# ... (GoveeConfig class defined elsewhere) ...
# ... (GoveeConfigError class defined elsewhere) ...
def load_config():
# Find and load the .env file
env_path = Path(__file__).resolve().parent.parent.parent / '.env'
load_dotenv(dotenv_path=env_path) # Reads .env into environment
# Read values from the environment
api_key = os.getenv('GOVEE_API_KEY')
device_id = os.getenv('GOVEE_DEVICE_ID')
sku = os.getenv('GOVEE_SKU')
# Check if any are missing
if not api_key or not device_id or not sku:
# If something is missing, raise an error!
raise GoveeConfigError("Missing required environment variables!")
# If all good, package them up and return
return GoveeConfig(api_key=api_key, device_id=device_id, sku=sku)
```
**Explanation:**
1. `load_dotenv(dotenv_path=env_path)`: This line uses a library (`python-dotenv`) to read your `.env` file and load its contents into the environment variables for the current run.
2. `os.getenv(...)`: This standard Python function reads the value of an environment variable. It checks for `GOVEE_API_KEY`, `GOVEE_DEVICE_ID`, and `GOVEE_SKU`.
3. **Validation:** The code checks if it successfully found all three required values. If any are missing (`None`), it raises a `GoveeConfigError` to stop the application with a helpful message. This prevents weird errors later on.
4. `return GoveeConfig(...)`: If all values are found, they are bundled neatly into a `GoveeConfig` object (we'll see this next) and returned.
## A Tidy Package: The `GoveeConfig` Object
Reading variables one by one is okay, but it's much cleaner to pass them around bundled together. That's where the `GoveeConfig` object comes in. It's defined using a Python feature called `dataclass`.
```python
# Simplified snippet from src/govee_mcp_server/config.py
from dataclasses import dataclass
@dataclass
class GoveeConfig:
"""Configuration class for Govee API settings."""
api_key: str
device_id: str
sku: str
```
**Explanation:**
* `@dataclass`: This is a shortcut in Python to create simple classes that mostly just hold data.
* `api_key: str`, `device_id: str`, `sku: str`: This defines the "slots" within our `GoveeConfig` object. It expects to hold three pieces of text (strings): the API key, device ID, and SKU.
**Analogy:** The `GoveeConfig` object is like a small, labelled box specifically designed to hold the application's ID card (`api_key`), address label (`device_id`), and envelope type (`sku`). The `load_config` function fills this box.
## Using the Configuration
Once `load_config()` creates the `GoveeConfig` object, it's passed to other parts of the application that need this information. The most important user is the [Govee API Client](03_govee_api_client.md), which handles the actual communication with Govee.
Here's a simplified idea of how it's used in `src/govee_mcp_server/api.py`:
```python
# Simplified concept from src/govee_mcp_server/api.py
from .config import GoveeConfig # Import the config class
class GoveeAPI:
# The __init__ method runs when a GoveeAPI object is created
def __init__(self, config: GoveeConfig):
"""Initialize API client with configuration."""
self.config = config # Store the passed-in config object
# Now, self.config.api_key, self.config.device_id,
# and self.config.sku are available inside this class.
async def _make_request(self, ...):
# When making a real request to Govee...
api_key = self.config.api_key # <-- Access the stored key
headers = {"Govee-API-Key": api_key, ...}
# ... use the key in the request headers ...
async def set_power(self, state: bool):
# When controlling the device...
device = self.config.device_id # <-- Access the stored ID
sku = self.config.sku # <-- Access the stored SKU
payload = {
"sku": sku,
"device": device,
# ... other control details ...
}
# ... use the ID and SKU in the request body ...
```
**Explanation:**
* The `GoveeAPI` class takes the `GoveeConfig` object when it's created.
* It stores this `config` object internally.
* Whenever it needs the API key, device ID, or SKU to talk to the Govee servers, it simply accesses them from its stored `self.config` object (e.g., `self.config.api_key`).
## Under the Hood: How Loading Works
Let's trace the steps when the application starts and needs its configuration:
```mermaid
sequenceDiagram
participant App as Application (CLI/Server)
participant LC as load_config()
participant DotEnv as .env File
participant OS as Operating System Env
participant GC as GoveeConfig Object
App->>LC: Start! Need config. Call load_config()
LC->>DotEnv: Look for .env file in project root
Note over DotEnv: Contains GOVEE_API_KEY=... etc.
DotEnv-->>LC: Load variables found into OS Env
LC->>OS: Read 'GOVEE_API_KEY' variable
OS-->>LC: Return 'abcdef...' (value)
LC->>OS: Read 'GOVEE_DEVICE_ID' variable
OS-->>LC: Return 'AB:CD:...' (value)
LC->>OS: Read 'GOVEE_SKU' variable
OS-->>LC: Return 'H6159' (value)
LC->>LC: Check: Are all values present? (Yes)
LC->>GC: Create GoveeConfig object with these values
GC-->>LC: Here's the new object
LC-->>App: Return the filled GoveeConfig object
```
This sequence shows how the application reliably gets its necessary startup information from the environment, validates it, and packages it for easy use.
## Conclusion
You've learned about the first crucial step: **Configuration Management**. You now know:
* Why applications need configuration (like API keys and device IDs).
* How `govee_mcp_server` uses environment variables and `.env` files to store these settings securely.
* How the `load_config` function reads and validates these settings.
* How the `GoveeConfig` object provides a tidy way to access these settings throughout the application.
With the configuration loaded, the application knows *which* device to talk to and *how* to authenticate. But what actions *can* it perform on that device? That leads us to our next topic: defining the capabilities of our device.
Let's move on to [Chapter 2: Device Control Interfaces](02_device_control_interfaces.md) to see how we define actions like turning the light on/off or changing its color.
---
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/02_device_control_interfaces.md:
--------------------------------------------------------------------------------
```markdown
# Chapter 2: Device Control Interfaces - The Standard Sockets for Lights
In [Chapter 1: Configuration Management](01_configuration_management.md), we learned how our application gets its essential settings, like which Govee device to talk to (`device_id`) and the secret key (`api_key`) needed to communicate. Think of it like knowing the address and having the key to a specific smart light bulb.
But just knowing *which* bulb doesn't tell us *what we can do* with it. Can we turn it on? Can we change its color? Can we dim it? This chapter is about defining a standard way to talk about these capabilities.
## What's the Big Idea? Standardizing Actions
Imagine you have different smart devices: a Govee light strip, maybe a smart plug from another brand, or even a simulated light for testing. You want a consistent way to tell them "turn on" or "set color to blue". You don't want to learn a completely different set of commands for each one!
This is where **Device Control Interfaces** come in. They act like **standard electrical sockets** or **blueprints** for device abilities.
* **Power Socket:** Defines *how* to plug something in to get power (e.g., Turn On/Off).
* **Color Socket:** Defines *how* to tell a device which color to display.
* **Brightness Socket:** Defines *how* to adjust the brightness level.
An "interface" in programming is a **contract**. It says: "If you claim to have power control, you *must* provide a way to `set_power` and `get_power_state` exactly like this." Any part of our code that knows how to control a Govee device (like the `GoveeAPI` client we'll see more of soon) has to promise to follow these contracts.
## The Blueprints: `PowerControl`, `ColorControl`, `BrightnessControl`
In our project, these blueprints are defined in `src/govee_mcp_server/interfaces.py` using Python's `abc` (Abstract Base Classes) module. Let's look at a simplified version of `PowerControl`:
```python
# Simplified from src/govee_mcp_server/interfaces.py
from abc import ABC, abstractmethod
from typing import Tuple # Used to describe the function's output type
class PowerControl(ABC): # ABC means "This is a blueprint/interface"
"""Blueprint for turning things on/off."""
@abstractmethod # Means: Any class using this blueprint MUST provide this method
async def set_power(self, state: bool) -> Tuple[bool, str]:
"""
Blueprint for a function to turn the device on (True) or off (False).
It MUST take 'state' (True/False) as input.
It MUST return a pair: (was it successful?, a message).
"""
pass # The blueprint only defines *what*, not *how*. Implementations fill this.
@abstractmethod
async def get_power_state(self) -> Tuple[bool, str]:
"""
Blueprint for checking if the device is currently on.
It MUST return a pair: (is it on?, a message).
"""
pass
```
**Explanation:**
* `class PowerControl(ABC):`: Defines a blueprint named `PowerControl`. `ABC` marks it as an abstract blueprint, not a real, usable object on its own.
* `@abstractmethod`: This marker says "Any real class that claims to be a `PowerControl` *must* provide its own version of the function below."
* `async def set_power(...)`: Defines the *signature* of the `set_power` function:
* `async`: It's designed to work asynchronously (we'll see why in later chapters).
* `self`: A standard reference to the object itself.
* `state: bool`: It must accept one input argument named `state`, which must be a boolean (`True` or `False`).
* `-> Tuple[bool, str]`: It must return a "tuple" (an ordered pair) containing a boolean (success/failure) and a string (a status message).
* `pass`: In the blueprint, the methods don't actually *do* anything. They just define the requirements.
Similarly, we have `ColorControl` (requiring `set_color`, `get_color`) and `BrightnessControl` (requiring `set_brightness`, `get_brightness`).
## Using the Blueprints: The `GoveeAPI` Class
Okay, we have blueprints. Now we need something concrete that *follows* these blueprints. In our project, the main class responsible for talking to the actual Govee Cloud service is `GoveeAPI` (found in `src/govee_mcp_server/api.py`).
Look at the very first line defining this class:
```python
# Simplified from src/govee_mcp_server/api.py
from .interfaces import PowerControl, ColorControl, BrightnessControl
# ... other imports ...
from .config import GoveeConfig
class GoveeAPI(PowerControl, ColorControl, BrightnessControl):
"""
Govee API client that PROMISES to follow the rules
of PowerControl, ColorControl, and BrightnessControl.
"""
def __init__(self, config: GoveeConfig):
"""Gets the API key and device details when created."""
self.config = config
# ... other setup ...
# --- Implementing PowerControl ---
async def set_power(self, state: bool) -> Tuple[bool, str]:
"""Turns the actual Govee device on/off using its API."""
print(f"Actually telling Govee API to set power: {state}")
# ... code to make the real web request to Govee ...
# (We'll look inside this in Chapter 3!)
success = True # Let's assume it worked for now
message = f"Device power set to {state} via API"
return success, message
async def get_power_state(self) -> Tuple[bool, str]:
"""Asks the Govee API if the device is on."""
print("Actually asking Govee API for power state")
# ... code to make the real web request to Govee ...
is_on = True # Let's pretend it's on
message = "Device is currently ON (from API)"
return is_on, message
# --- Implementing ColorControl (methods like set_color) ---
# ... implementations for set_color, get_color ...
# --- Implementing BrightnessControl (methods like set_brightness) ---
# ... implementations for set_brightness, get_brightness ...
```
**Explanation:**
* `class GoveeAPI(PowerControl, ColorControl, BrightnessControl):`: This line is the **promise**. It declares that the `GoveeAPI` class will provide *concrete implementations* for all the `@abstractmethod` functions defined in the `PowerControl`, `ColorControl`, and `BrightnessControl` blueprints.
* If `GoveeAPI` *forgot* to include, say, the `set_power` method, Python would give an error when you try to use it, because it broke its promise!
* Inside `GoveeAPI`, the `set_power` method now has real code (represented by the `print` statement and comments for now) that actually interacts with the Govee web service. Crucially, its signature (`async def set_power(self, state: bool) -> Tuple[bool, str]`) exactly matches the blueprint.
**Analogy:** If `PowerControl` is the blueprint for a standard wall socket, `GoveeAPI` is like a specific brand of smart plug that is built according to that blueprint and actually connects to the house wiring (the Govee Cloud API) when used.
## Benefits: Why Is This Structure Useful?
1. **Consistency:** Any code that needs to turn a light on/off can expect *any* object that follows the `PowerControl` blueprint to have a `.set_power(state)` method. It doesn't need to know if it's talking to a `GoveeAPI` object, a `GoveeBluetooth` object (if we added one later), or even a `FakeLightForTesting` object, as long as they all follow the blueprint.
2. **Flexibility (Swappability):** Imagine we want to add control via Bluetooth later. We could create a `GoveeBluetooth(PowerControl, ...)` class. Code that only cares about power control could use either `GoveeAPI` or `GoveeBluetooth` interchangeably without modification, because both fulfill the `PowerControl` contract.
3. **Testability:** It's much easier to write tests. We can create a simple `MockPowerControl` class that just pretends to turn things on/off without actually talking to Govee, and use it to test other parts of our application that rely on power control.
## Safety First: Input Validation with Decorators
What if someone tries to set the color to RGB values like `(300, -50, 1000)`? Those aren't valid! Red, Green, and Blue values must be between 0 and 255.
Interfaces can also help enforce rules like this. We use a feature called **decorators**. Think of a decorator as a wrapper function that adds extra behavior (like a check) before or after another function runs.
In `src/govee_mcp_server/interfaces.py`, we have a `validate_rgb` decorator:
```python
# Simplified from src/govee_mcp_server/interfaces.py
from .exceptions import GoveeValidationError # Our custom error for bad input
def validate_rgb(func):
"""A function that wraps another function to check RGB values first."""
async def wrapper(self, r: int, g: int, b: int, *args, **kwargs):
print(f"VALIDATOR: Checking RGB ({r}, {g}, {b})") # Show the check
is_valid = True
for name, value in [('Red', r), ('Green', g), ('Blue', b)]:
if not (isinstance(value, int) and 0 <= value <= 255):
print(f"VALIDATOR: Invalid value for {name}: {value}")
is_valid = False
# Stop! Raise an error instead of calling the real function.
raise GoveeValidationError(f"{name} value must be an integer 0-255")
if is_valid:
# If all checks pass, call the original function (like set_color)
print("VALIDATOR: Looks good! Proceeding.")
return await func(self, r, g, b, *args, **kwargs)
return wrapper
```
And we apply this decorator to the `set_color` blueprint in `ColorControl`:
```python
# Simplified from src/govee_mcp_server/interfaces.py
# ... imports ...
class ColorControl(ABC):
@abstractmethod
@validate_rgb # <-- Apply the validator HERE
async def set_color(self, r: int, g: int, b: int) -> Tuple[bool, str]:
"""
Set device color. The @validate_rgb ensures values are checked
BEFORE the actual implementation in GoveeAPI is even called.
"""
pass
# ... get_color method ...
```
**Explanation:**
* `@validate_rgb`: This line attaches the `validate_rgb` wrapper to the `set_color` method definition in the blueprint.
* Now, whenever *any* implementation of `set_color` (like the one in `GoveeAPI`) is called, the `validate_rgb` code runs *first*.
* If the RGB values are invalid (e.g., `r=300`), `validate_rgb` raises a `GoveeValidationError` immediately, preventing the invalid values from ever reaching the code that talks to the Govee API.
**Analogy:** The `@validate_rgb` decorator is like a safety fuse built into the "Color Socket" blueprint. It checks the "electrical current" (the RGB values) before letting it flow to the actual device, preventing damage or errors.
## How it Works Under the Hood
Let's trace what happens when some user code tries to turn the light on using our `GoveeAPI` object, which implements the `PowerControl` interface:
```mermaid
sequenceDiagram
participant UserCode as Your Code
participant GoveeObj as GoveeAPI Object (implements PowerControl)
participant PCInterface as PowerControl Blueprint
participant GoveeCloud as Govee Cloud Service
UserCode->>GoveeObj: Call `set_power(True)`
Note over GoveeObj,PCInterface: GoveeAPI promised to have `set_power` because it implements PowerControl.
GoveeObj->>GoveeObj: Execute its own `set_power` code.
GoveeObj->>GoveeCloud: Send command: { device: '...', sku: '...', capability: 'power', value: 1 }
GoveeCloud-->>GoveeObj: Respond: { status: 200, message: 'Success' }
GoveeObj->>GoveeObj: Process the response.
GoveeObj-->>UserCode: Return `(True, "Device power set to True via API")`
```
If we called `set_color` instead, the flow would be similar, but the `@validate_rgb` check would happen right after `UserCode` calls `set_color` on `GoveeObj`, before the object tries to talk to the `GoveeCloud`.
## Conclusion
You've now learned about **Device Control Interfaces**, the crucial blueprints that define *what* actions can be performed on a device in a standard way.
* Interfaces (`PowerControl`, `ColorControl`, etc.) act as **contracts** or **standard sockets**.
* They use `ABC` and `@abstractmethod` to define required functions (like `set_power`, `get_color`).
* Classes like `GoveeAPI` **implement** these interfaces, promising to provide the required functions.
* This brings **consistency**, **flexibility**, and **testability** to our code.
* **Decorators** like `@validate_rgb` can be attached to interface methods to add checks (like input validation) automatically.
We have the configuration (Chapter 1) and the blueprints for control (Chapter 2). Now, let's dive into the specific class that follows these blueprints and actually communicates with the Govee servers.
Get ready to explore the engine room in [Chapter 3: Govee API Client](03_govee_api_client.md)!
---
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/05_mcp_server_implementation.md:
--------------------------------------------------------------------------------
```markdown
# Chapter 5: MCP Server Implementation - The Universal Remote Receiver
In [Chapter 4: Command Line Interface (CLI)](04_command_line_interface__cli_.md), we created a simple way for *you* to control your Govee light by typing commands directly into your terminal. It's like having a basic remote with just a few buttons – useful, but limited.
What if you want something more powerful? What if you want *other programs*, like an AI assistant or a smart home dashboard, to be able to control the light? They can't easily "type" commands into your terminal. They need a standardized way to send instructions over a network or through standard program communication channels.
This is where the **MCP Server** comes in. It's the core of our `govee_mcp_server` project!
## What's the Big Idea? A Standard Interface for Programs
Imagine you have various smart devices (lights, plugs, maybe a fan). You want your fancy AI assistant to control them all. It would be messy if the AI needed to learn a completely different communication method for each device!
The **Mission Control Protocol (MCP)** is designed to solve this. It's a standard way for a program (like our AI, called an **MCP Client**) to:
1. Ask another program (our **MCP Server**) what "tools" it offers.
2. Ask the server to use one of those tools with specific instructions.
Our `govee_mcp_server` acts as an MCP server specifically for your Govee light. It advertises "tools" like `turn_on_off`, `set_color`, `set_brightness`, and `get_status`. When an MCP client calls one of these tools, our server receives the request, uses the [Govee API Client](03_govee_api_client.md) to perform the action on the real light, and sends a standard response back.
**Analogy:** Think of the MCP server like a **universal remote control receiver** attached to your Govee light. Any device that knows the standard "universal remote language" (MCP) can send signals to this receiver, which then translates them into the specific commands your Govee light understands.
## How MCP Works: Tools and Communication
1. **Discovery (Optional but Common):** An MCP client might first ask the server, "What capabilities or tools do you have?" Our server would respond, "I have `turn_on_off` (takes a power state: true/false), `set_color` (takes red, green, blue numbers), `set_brightness` (takes a number 0-100), and `get_status`."
2. **Execution:** The client then sends a structured message like, "Please execute the tool named `set_color` with arguments `red=0`, `green=255`, `blue=0`."
3. **Action:** Our server receives this, finds its `set_color` function, calls it with the provided arguments (0, 255, 0). Inside that function, it uses the `GoveeAPI` client to actually tell the Govee cloud to set the light to green.
4. **Response:** The server sends a structured message back to the client, like, "Okay, I executed `set_color`. The result was: 'Success'."
MCP communication often happens over standard input/output (stdio) between processes or sometimes over network sockets (like TCP). Our server uses a library called `FastMCP` which handles the details of listening for these requests and sending responses in the correct MCP format.
## Implementing the Server (`src/govee_mcp_server/server.py`)
The main logic for our MCP server lives in the `src/govee_mcp_server/server.py` file. Let's look at the key parts.
**1. Setting Up the Server**
First, we import the necessary pieces and create the `FastMCP` server instance.
```python
# Simplified from src/govee_mcp_server/server.py
import sys
from mcp.server.fastmcp import FastMCP # The MCP library
from govee_mcp_server.config import load_config # To get API key etc.
from govee_mcp_server.api import GoveeAPI # To control the light
from govee_mcp_server.exceptions import GoveeError # Our custom errors
# Initialize the MCP server
mcp = FastMCP(
"govee", # A simple name for our server
# Define some basic info about the server
capabilities={
"server_info": { "name": "govee-mcp", "version": "0.1.0" }
},
log_level='WARNING' # Keep logs tidy
)
```
**Explanation:**
* We import `FastMCP`, our configuration loader, and the `GoveeAPI` client.
* `mcp = FastMCP(...)` creates the server object. We give it a name (`"govee"`) and some basic information that clients might ask for.
**2. Loading Configuration**
The server needs the API key and device details, just like the CLI did.
```python
# Simplified from src/govee_mcp_server/server.py
print("Loading configuration...", file=sys.stderr)
try:
# Load API Key, Device ID, SKU from .env or environment
config = load_config() # Uses the function from Chapter 1
except GoveeError as e:
print(f"Configuration error: {e}", file=sys.stderr)
sys.exit(1) # Stop if config fails
# 'config' now holds our GoveeConfig object
```
**Explanation:**
* This uses the `load_config` function from [Chapter 1: Configuration Management](01_configuration_management.md) to get the necessary credentials.
* If loading fails (e.g., missing `.env` file), it prints an error and stops.
**3. Defining the Tools**
This is where we define the actions the server can perform. We use the `@mcp.tool()` decorator to register functions as MCP tools.
```python
# Simplified from src/govee_mcp_server/server.py
print("Setting up tools...", file=sys.stderr)
# --- Tool: turn_on_off ---
@mcp.tool("turn_on_off") # Register this function as an MCP tool
async def turn_on_off(power: bool) -> str:
"""Turn the LED on (True) or off (False)."""
api = GoveeAPI(config) # Create API client *for this request*
try:
# Use the API client to send the command
success, message = await api.set_power(power)
# Return a simple status message
return message if success else f"Failed: {message}"
except GoveeError as e:
return f"Error: {str(e)}" # Handle expected errors
finally:
await api.close() # IMPORTANT: Clean up the connection
# --- Tool: set_color ---
@mcp.tool("set_color")
async def set_color(red: int, green: int, blue: int) -> str:
"""Set the LED color using RGB (0-255)."""
api = GoveeAPI(config)
try:
# Use the API client to set the color
success, message = await api.set_color(red, green, blue)
return message if success else f"Failed: {message}"
except GoveeError as e:
return f"Error: {str(e)}"
finally:
await api.close()
# --- Other tools (set_brightness, get_status) defined similarly ---
# @mcp.tool("set_brightness")
# async def set_brightness(brightness: int) -> str: ...
# @mcp.tool("get_status")
# async def get_status() -> dict: ...
```
**Explanation:**
* `@mcp.tool("tool_name")`: This magic line tells the `FastMCP` library: "When a client asks to run the tool named `turn_on_off`, execute the following function (`async def turn_on_off(...)`)."
* `async def turn_on_off(power: bool) -> str:`: Defines the function that handles the `turn_on_off` tool.
* It takes `power` (a boolean: `True` or `False`) as input, just as the client specified.
* It's declared `async` because it uses `await` to call the asynchronous `GoveeAPI` methods.
* It's defined to return a `str` (a string message indicating success or failure).
* **Inside the Tool:**
1. `api = GoveeAPI(config)`: A *new* instance of the [Govee API Client](03_govee_api_client.md) is created for each tool request, using the loaded configuration.
2. `await api.set_power(power)`: This is where the actual work happens! It calls the API client method to communicate with the Govee cloud.
3. `return ...`: It returns a simple string message back to the MCP framework.
4. `finally: await api.close()`: **Crucially**, after the tool logic is done (whether it succeeded or failed), it closes the network connection associated with the `GoveeAPI` client instance to release resources. This happens in the `finally` block to ensure cleanup.
* The `set_color` tool follows the same pattern: decorate, define function with correct arguments, create API client, call the relevant API method (`set_color`), handle errors, return a message, and clean up the client.
**4. Running the Server**
Finally, we need code to actually start the server and make it listen for incoming MCP requests.
```python
# Simplified from src/govee_mcp_server/server.py
import asyncio
# ... other code above ...
if __name__ == "__main__": # Standard Python way to run code directly
try:
print("Starting MCP server...", file=sys.stderr)
# Tell FastMCP to run, listening via standard input/output
asyncio.run(mcp.run(transport='stdio'))
except KeyboardInterrupt:
# Allow stopping the server with Ctrl+C
print("\nServer stopped by user", file=sys.stderr)
except Exception as e:
# Catch unexpected server errors
print(f"Server error: {e}", file=sys.stderr)
sys.exit(1)
```
**Explanation:**
* `if __name__ == "__main__":`: This code runs only when you execute the `server.py` script directly (e.g., `python -m govee_mcp_server.server`).
* `asyncio.run(mcp.run(transport='stdio'))`: This is the command that starts the `FastMCP` server.
* `transport='stdio'` tells it to listen for MCP requests coming in via standard input and send responses via standard output. This is a common way for programs to talk to each other locally. (MCP can also run over network sockets).
* `asyncio.run(...)` starts the asynchronous event loop needed for `async`/`await` functions.
* The `try...except` blocks handle stopping the server cleanly (Ctrl+C) or catching unexpected crashes.
## Under the Hood: A Client Calls `set_brightness(50)`
Let's trace what happens when an MCP client sends a request:
1. **Client Request:** The MCP client program sends a specially formatted message (likely JSON) via standard input to our running `server.py` process. The message essentially says: "Execute tool `set_brightness` with argument `brightness=50`."
2. **FastMCP Receives:** The `mcp.run()` loop in our server reads this message from its standard input.
3. **FastMCP Parses:** The `FastMCP` library parses the message, identifies the target tool (`set_brightness`), and extracts the arguments (`brightness=50`).
4. **FastMCP Dispatches:** It finds the Python function registered with `@mcp.tool("set_brightness")`, which is our `async def set_brightness(brightness: int)` function.
5. **Tool Function Executes:** `FastMCP` calls `await set_brightness(brightness=50)`.
* Inside `set_brightness`:
* `api = GoveeAPI(config)`: Creates a new Govee API client.
* `await api.set_brightness(50)`: Calls the API client method.
* The `GoveeAPI` client builds the JSON request, adds the API key, sends it to the Govee cloud, gets the response (e.g., "Success").
* The API client returns `(True, "Success")` to the `set_brightness` function.
* The function prepares the return string: `"Success"`.
* `finally: await api.close()`: The API client connection is closed.
* The function `return`s the string `"Success"`.
6. **FastMCP Receives Result:** `FastMCP` gets the `"Success"` string back from the tool function.
7. **FastMCP Responds:** `FastMCP` formats an MCP response message (likely JSON) containing the result `"Success"` and sends it back to the client program via its standard output.
8. **Client Receives Response:** The MCP client reads the response from its standard input and knows the command was successful.
Here's a diagram of that flow:
```mermaid
sequenceDiagram
participant Client as MCP Client
participant FastMCP as FastMCP Library (in server.py)
participant ToolFunc as set_brightness() Tool
participant APIClient as GoveeAPI Client
participant GoveeCloud as Govee Cloud
Client->>+FastMCP: MCP Request: execute 'set_brightness', brightness=50
FastMCP->>+ToolFunc: await set_brightness(brightness=50)
ToolFunc->>APIClient: Create GoveeAPI(config)
ToolFunc->>+APIClient: await set_brightness(50)
APIClient->>GoveeCloud: Send 'Set Brightness=50' request
GoveeCloud-->>APIClient: Respond ('Success')
APIClient-->>-ToolFunc: Return (True, "Success")
ToolFunc->>APIClient: await close()
ToolFunc-->>-FastMCP: Return "Success"
FastMCP-->>-Client: MCP Response: result="Success"
```
## Conclusion
You've now learned how the **MCP Server Implementation** works in `govee_mcp_server`:
* It acts as a **standardized interface** (like a universal remote receiver) for controlling the Govee device, designed to be used by *other programs* (MCP clients).
* It uses the **Mission Control Protocol (MCP)** standard for communication.
* It leverages the `FastMCP` library to handle incoming requests and outgoing responses.
* It defines specific **tools** (`turn_on_off`, `set_color`, etc.) using the `@mcp.tool` decorator.
* Each tool function uses the [Govee API Client](03_govee_api_client.md) (created per-request) to interact with the actual Govee device via the cloud.
* Proper cleanup (closing the API client) is essential within each tool.
This server provides a powerful and flexible way to integrate your Govee light control into larger systems. However, real-world interactions aren't always perfect. What happens if the Govee API returns an error, or the network connection fails, or the client sends invalid input? We need robust ways to handle these situations.
Let's explore how we manage errors gracefully in [Chapter 6: Custom Error Handling](06_custom_error_handling.md).
---
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/04_command_line_interface__cli_.md:
--------------------------------------------------------------------------------
```markdown
# Chapter 4: Command Line Interface (CLI) - Your Simple Remote Control
In the [Chapter 3: Govee API Client](03_govee_api_client.md), we built the "translator" – the `GoveeAPI` client that knows how to talk directly to the Govee Cloud service using your API key and device details. It can turn the light on, change its color, and more, based on the blueprints from [Chapter 2: Device Control Interfaces](02_device_control_interfaces.md).
But how do *you*, the user, tell this `GoveeAPI` client what to do? You need a way to interact with it. One of the simplest ways is through a **Command Line Interface**, or **CLI**.
## What's the Big Idea? Direct Control from Your Terminal
Imagine you have a fancy TV system with a complex universal remote (that will be our [MCP Server Implementation](05_mcp_server_implementation.md) later). But sometimes, you just want a really simple, basic remote with only a few buttons: Power On/Off, maybe Volume Up/Down. It's quick, easy, and gets the job done for simple tasks.
The **Command Line Interface (CLI)** for our project is like that simple remote control. It lets you interact with your Govee light directly by typing commands into your computer's **terminal** (also called a command prompt or shell).
Instead of needing a full graphical application or starting the main server, you can just type something like:
```bash
govee-cli power on
```
or
```bash
govee-cli color 255 0 0 # Set color to Red
```
This provides a fast and straightforward way to manually control your light or even use it in simple scripts.
## How Does It Work? Typing Commands
When you use a CLI tool, here's the basic flow:
1. **You Type:** You open your terminal and type the name of the program, followed by the action you want to perform, and any necessary details (arguments).
* `govee-cli` is the program name.
* `power` or `color` is the command (the action).
* `on` or `255 0 0` are the arguments (the details for the action).
2. **Program Reads:** The `govee-cli` program reads the command and arguments you typed. It uses a helper library (in Python, often `argparse`) to understand what you're asking it to do.
3. **Program Acts:** Based on your command, it uses the **same `GoveeAPI` client** from [Chapter 3: Govee API Client](03_govee_api_client.md) to send the instruction to your Govee light via the Govee Cloud.
4. **Program Reports:** It usually prints a message back to your terminal telling you if the command was successful or if there was an error.
## Using the `govee-cli`
Let's try some commands! (Make sure you have your `.env` file set up as described in [Chapter 1: Configuration Management](01_configuration_management.md) so the CLI knows your API key and device).
**(Note: You might need to install the project first or run it using `python -m govee_mcp_server.cli ...` depending on your setup. For simplicity, we'll assume `govee-cli` is available.)**
1. **Get Help:** See what commands are available.
```bash
govee-cli --help
```
* **Output:** Shows a list of commands like `power`, `color`, `brightness`, `status` and how to use them.
2. **Turn Power On:**
```bash
govee-cli power on
```
* **Output (if successful):** `Success` (or a similar message from the Govee API). Your light should turn on!
3. **Turn Power Off:**
```bash
govee-cli power off
```
* **Output (if successful):** `Success`. Your light should turn off.
4. **Set Color (Red):**
```bash
govee-cli color 255 0 0
```
* **Output (if successful):** `Success`. Your light should turn red.
* **Note:** The arguments are Red, Green, Blue values (0-255).
5. **Set Brightness (50%):**
```bash
govee-cli brightness 50
```
* **Output (if successful):** `Success`. Your light's brightness should change.
* **Note:** The argument is the brightness percentage (0-100).
6. **Check Status:** Get the current state of the light.
```bash
govee-cli status
```
* **Output (example):**
```
Power: ON
Color: RGB(255, 0, 0)
Brightness: 50%
```
If something goes wrong (like a typo in the command, invalid color values, or a problem talking to Govee), the CLI will usually print an error message.
## Under the Hood: What Happens When You Type `govee-cli power on`?
Let's trace the steps behind the scenes:
1. **Terminal:** You hit Enter after typing `govee-cli power on`. The terminal finds the `govee-cli` program and runs it, passing `power` and `on` as inputs.
2. **CLI Script Starts:** The Python script `src/govee_mcp_server/cli.py` begins running.
3. **Parse Arguments:** The script uses the `argparse` library to figure out: "The user wants the `power` command with the `state` argument set to `on`."
4. **Load Config:** It calls `load_config()` from [Chapter 1: Configuration Management](01_configuration_management.md) to read your `GOVEE_API_KEY`, `GOVEE_DEVICE_ID`, and `GOVEE_SKU` from the environment (or `.env` file).
5. **Create API Client:** It creates an instance of the `GoveeAPI` client (from [Chapter 3: Govee API Client](03_govee_api_client.md)), giving it the configuration details it just loaded. `api_client = GoveeAPI(config=...)`
6. **Call API Method:** Since the command was `power on`, the script calls the corresponding method on the API client: `await api_client.set_power(True)`. (It converts "on" to the boolean `True`).
7. **API Client Works:** The `GoveeAPI` client does its job (as described in Chapter 3): formats the request, sends it to the Govee Cloud with the API key, gets the response, handles retries if needed.
8. **Get Result:** The `set_power` method returns a result (e.g., `(True, "Success")`) back to the CLI script.
9. **Print Output:** The CLI script takes the result message ("Success") and prints it to your terminal.
10. **Cleanup:** The script makes sure to close the network connection used by the `GoveeAPI` client (`await api_client.close()`).
Here's a diagram showing this flow:
```mermaid
sequenceDiagram
participant User as You (Terminal)
participant CLI as govee-cli Script
participant Parser as Argument Parser (argparse)
participant Config as Config Loader
participant APIClient as GoveeAPI Client
participant GoveeCloud as Govee Cloud
User->>CLI: Run `govee-cli power on`
CLI->>Parser: Parse ["power", "on"]
Parser-->>CLI: Command='power', State='on'
CLI->>Config: Load config()
Config-->>CLI: Return GoveeConfig (API Key, etc.)
CLI->>APIClient: Create GoveeAPI(config)
CLI->>APIClient: Call `await set_power(True)`
APIClient->>GoveeCloud: Send 'Power On' request (with API Key)
GoveeCloud-->>APIClient: Respond ('Success')
APIClient-->>CLI: Return (True, "Success")
CLI->>User: Print "Success"
CLI->>APIClient: Call `await close()`
APIClient->>APIClient: Close network connection
```
## Inside the Code: `src/govee_mcp_server/cli.py`
Let's look at simplified parts of the code that make this work.
**1. Defining the Commands (`create_parser`)**
This function sets up `argparse` to understand the commands and arguments.
```python
# Simplified from src/govee_mcp_server/cli.py
import argparse
def create_parser() -> argparse.ArgumentParser:
"""Create and configure argument parser."""
parser = argparse.ArgumentParser(description='Control Govee LED device')
# Create categories for commands (like 'power', 'color')
subparsers = parser.add_subparsers(dest='command', help='Commands', required=True)
# --- Define the 'power' command ---
power_parser = subparsers.add_parser('power', help='Turn device on/off')
# It needs one argument: 'state', which must be 'on' or 'off'
power_parser.add_argument('state', choices=['on', 'off'], help='Power state')
# --- Define the 'color' command ---
color_parser = subparsers.add_parser('color', help='Set device color')
# It needs three integer arguments: red, green, blue
color_parser.add_argument('red', type=int, help='Red value (0-255)')
color_parser.add_argument('green', type=int, help='Green value (0-255)')
color_parser.add_argument('blue', type=int, help='Blue value (0-255)')
# --- Define other commands similarly (brightness, status) ---
# ... parser setup for brightness ...
# ... parser setup for status ...
return parser
```
**Explanation:**
* `argparse.ArgumentParser`: Creates the main parser object.
* `parser.add_subparsers()`: Allows us to define distinct commands (like `git commit`, `git push`). `dest='command'` stores which command was used.
* `subparsers.add_parser('power', ...)`: Defines the `power` command.
* `power_parser.add_argument('state', ...)`: Specifies that the `power` command requires an argument named `state`, which must be either the text `"on"` or `"off"`.
* Similarly, it defines the `color` command and its required `red`, `green`, `blue` integer arguments.
**2. Handling a Specific Command (`handle_power`)**
Each command has a small function to handle its logic.
```python
# Simplified from src/govee_mcp_server/cli.py
from .api import GoveeAPI # We need the API client class
from .exceptions import GoveeError # We need our custom error
async def handle_power(api: GoveeAPI, state: str) -> None:
"""Handle the 'power' command."""
# Convert 'on'/'off' string to True/False boolean
is_on = (state == 'on')
print(f"Sending command: Power {'ON' if is_on else 'OFF'}...")
# Call the actual API client method
success, message = await api.set_power(is_on)
# Check the result from the API client
if success:
print(f"Result: {message}")
else:
# If the API client reported failure, raise an error to stop the script
print(f"Error reported by API: {message}")
raise GoveeError(f"Failed to set power: {message}")
```
**Explanation:**
* Takes the `api` client object and the parsed `state` ("on" or "off") as input.
* Converts the string `state` into a boolean `is_on` (`True` or `False`).
* Calls the corresponding method on the API client: `await api.set_power(is_on)`. This is where the communication with Govee happens (via the API client).
* Prints the result or raises an error if the API call failed.
**3. The Main Script Logic (`main` function)**
This function ties everything together: load config, parse args, call the right handler.
```python
# Simplified from src/govee_mcp_server/cli.py
import sys
import asyncio
from .config import load_config
from .api import GoveeAPI
from .exceptions import GoveeError
# ... create_parser() defined above ...
# ... handle_power(), handle_color(), etc. defined above ...
async def main() -> None:
"""Main CLI entrypoint."""
api = None # Initialize api variable
try:
# 1. Load configuration
config = load_config()
print(f"Loaded config for device: {config.device_id}")
# 2. Create the Govee API Client
api = GoveeAPI(config)
print("GoveeAPI client ready.")
# 3. Parse command line arguments
parser = create_parser()
args = parser.parse_args() # Reads sys.argv (what you typed)
print(f"Executing command: {args.command}")
# 4. Call the appropriate handler based on the command
if args.command == 'power':
await handle_power(api, args.state)
elif args.command == 'color':
await handle_color(api, args.red, args.green, args.blue)
# ... elif for brightness ...
# ... elif for status ...
else:
# Should not happen if parser is configured correctly
print(f"Unknown command: {args.command}")
parser.print_help()
sys.exit(1) # Exit with an error code
except GoveeError as e:
# Catch errors from our API client or handlers
print(f"\nOperation Failed: {str(e)}")
sys.exit(1) # Exit with an error code
except Exception as e:
# Catch any other unexpected errors
print(f"\nAn Unexpected Error Occurred: {str(e)}")
sys.exit(1) # Exit with an error code
finally:
# 5. ALWAYS try to clean up the API client connection
if api:
print("Closing API connection...")
await api.close()
print("Connection closed.")
# --- This part runs the main async function ---
# def cli_main():
# asyncio.run(main())
# if __name__ == "__main__":
# cli_main()
```
**Explanation:**
1. Calls `load_config()` to get settings.
2. Creates the `GoveeAPI` instance.
3. Calls `parser.parse_args()` which uses `argparse` to figure out which command and arguments were given.
4. Uses an `if/elif` block to check `args.command` and call the correct handler function (like `handle_power` or `handle_color`), passing the `api` client and the specific arguments needed (like `args.state` or `args.red`).
5. Includes `try...except` blocks to catch potential errors (like configuration errors, API errors, or validation errors) and print user-friendly messages before exiting.
6. The `finally` block ensures that `await api.close()` is called to release network resources, even if an error occurred.
## Conclusion
You've learned about the **Command Line Interface (CLI)**, a simple way to directly control your Govee device from the terminal.
* It acts like a **basic remote control**.
* You use it by typing commands like `govee-cli power on`.
* It **parses** your command and arguments using `argparse`.
* Crucially, it uses the **same `GoveeAPI` client** from [Chapter 3: Govee API Client](03_govee_api_client.md) to interact with the Govee Cloud.
* It provides immediate feedback in your terminal.
The CLI is great for quick tests, manual control, or simple automation scripts. However, it requires you to be at your computer typing commands. What if we want other applications or systems to control the light over a network, perhaps using a standard protocol?
That's where the main part of our project comes in: the server. Let's move on to [Chapter 5: MCP Server Implementation](05_mcp_server_implementation.md) to see how we build a server that listens for commands using the Mission Control Protocol (MCP).
---
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/06_custom_error_handling.md:
--------------------------------------------------------------------------------
```markdown
# Chapter 6: Custom Error Handling - Specific Warning Lights
Welcome to the final chapter of our `govee_mcp_server` tutorial! In [Chapter 5: MCP Server Implementation](05_mcp_server_implementation.md), we saw how our server listens for commands from other programs using the Mission Control Protocol (MCP) and uses the [Govee API Client](03_govee_api_client.md) to control the light.
But what happens when things don't go as planned? Maybe the internet connection drops, the Govee service is temporarily down, your API key is wrong, or a client sends an invalid command like trying to set the color to "purple" instead of RGB values. If the program just crashes with a generic error, it's hard to figure out what went wrong.
This chapter is about how `govee_mcp_server` uses **Custom Error Handling** to provide specific, helpful information when problems occur.
## What's the Big Idea? Knowing *Why* Things Broke
Imagine driving a car. If something goes wrong, you wouldn't want just one single "Check Engine" light for *every* possible problem, would you? Is it low oil? Are the brakes failing? Is a tire flat? A good dashboard has specific lights: "Low Fuel," "Oil Pressure Low," "Brake System Alert," etc. This tells you exactly what needs attention.
Standard Python errors (like `ValueError`, `TypeError`, `Exception`) are like that generic "Check Engine" light. They tell you *something* is wrong, but not always the specific context within *our* application.
**Custom Error Handling** in our project means we've created our *own* types of errors, specific to talking with Govee devices. Instead of just saying "Error!", we can say:
* "Error talking to the Govee API!" (`GoveeAPIError`)
* "Error: Your API key is missing or wrong!" (`GoveeConfigError`)
* "Error: That's not a valid color value!" (`GoveeValidationError`)
* "Error: Couldn't connect to the Govee servers!" (`GoveeConnectionError`)
* "Error: Govee took too long to reply!" (`GoveeTimeoutError`)
This makes it much easier for both the user and the developer to understand and fix the problem. It's like having those specific warning lights on our Govee control dashboard.
## Our Custom Warning Lights: The Exception Classes
In programming, errors are often handled using "Exceptions". When something goes wrong, the code can `raise` an exception, which stops the normal flow and signals that a problem occurred.
We've defined our custom exception types in the file `src/govee_mcp_server/exceptions.py`. They all inherit from a base `GoveeError`, making it easy to catch any Govee-specific problem if needed.
```python
# Simplified from src/govee_mcp_server/exceptions.py
class GoveeError(Exception):
"""Base exception for all Govee-related errors in this project."""
pass # It's just a label, inheriting basic Exception features
class GoveeAPIError(GoveeError):
"""Problem communicating with the Govee API itself
(e.g., Govee returned an error message like 'Invalid API Key')."""
pass
class GoveeConfigError(GoveeError):
"""Problem with the configuration settings
(e.g., missing GOVEE_API_KEY in the .env file)."""
pass
class GoveeValidationError(GoveeError):
"""Problem with the data provided by the user/client
(e.g., brightness=150, or invalid RGB values)."""
pass
class GoveeConnectionError(GoveeError):
"""Problem with the network connection to Govee servers."""
pass
class GoveeTimeoutError(GoveeError):
"""Govee's servers took too long to respond to our request."""
pass
```
**Explanation:**
* `class GoveeError(Exception):`: This defines our main category. Anything that's a `GoveeConfigError` is *also* a `GoveeError`.
* The other classes (`GoveeAPIError`, `GoveeConfigError`, etc.) inherit from `GoveeError`. They don't need any special code inside them; their *name* is what makes them specific. Think of them as different colored warning light bulbs.
## Turning On the Warning Lights: Raising Exceptions
Different parts of our code are responsible for detecting specific problems and `raise`-ing the appropriate custom exception.
**Example 1: Missing Configuration (`load_config`)**
Remember the `load_config` function from [Chapter 1: Configuration Management](01_configuration_management.md)? If it can't find your API key, it raises a `GoveeConfigError`.
```python
# Simplified from src/govee_mcp_server/config.py
from .exceptions import GoveeConfigError # Import the specific error
# ... other imports ...
def load_config():
# ... code to load api_key, device_id, sku ...
api_key = None # Let's pretend API key wasn't found
if not api_key: # Check if it's missing
# Raise the specific configuration error!
raise GoveeConfigError("Missing required environment variable: GOVEE_API_KEY")
# ... return config object if all is well ...
```
**Explanation:**
* If `api_key` is not found, `raise GoveeConfigError(...)` immediately stops the `load_config` function and signals this specific problem.
**Example 2: Invalid Color Input (`validate_rgb`)**
The `@validate_rgb` decorator we saw in [Chapter 2: Device Control Interfaces](02_device_control_interfaces.md) checks color values before they even reach the API client.
```python
# Simplified from src/govee_mcp_server/interfaces.py
from .exceptions import GoveeValidationError # Import the specific error
# ... other imports ...
def validate_rgb(func):
async def wrapper(self, r: int, g: int, b: int, *args, **kwargs):
if not (0 <= r <= 255): # Check if Red value is valid
# Raise the specific validation error!
raise GoveeValidationError("red value must be between 0-255")
# ... check g and b ...
# If all good, call the original function (like set_color)
return await func(self, r, g, b, *args, **kwargs)
return wrapper
class ColorControl(ABC):
@abstractmethod
@validate_rgb # Apply the validator
async def set_color(self, r: int, g: int, b: int) -> Tuple[bool, str]:
pass
```
**Explanation:**
* If `r` is outside the 0-255 range, `raise GoveeValidationError(...)` stops the process *before* the actual `set_color` logic in `GoveeAPI` runs.
**Example 3: API Communication Problems (`_make_request`)**
The core `_make_request` function inside the [Govee API Client](03_govee_api_client.md) handles errors from the Govee servers or network issues.
```python
# Simplified from src/govee_mcp_server/api.py
import aiohttp
import asyncio
from .exceptions import GoveeAPIError, GoveeConnectionError, GoveeTimeoutError
class GoveeAPI:
# ... other methods ...
async def _make_request(self, method: str, endpoint: str, **kwargs):
# ... setup ...
for attempt in range(self.MAX_RETRIES):
try:
async with self.session.request(...) as response:
data = await response.json()
if response.status == 401: # Example: Unauthorized
# Raise specific API error!
raise GoveeAPIError(f"API error: 401 - Invalid API Key")
elif response.status != 200:
# Raise general API error for other bad statuses
raise GoveeAPIError(f"API error: {response.status} - {data.get('message')}")
# ... success path ...
return data, data.get('message', 'Success')
except asyncio.TimeoutError:
if attempt == self.MAX_RETRIES - 1:
# Raise specific timeout error!
raise GoveeTimeoutError("Request timed out")
except aiohttp.ClientError as e:
if attempt == self.MAX_RETRIES - 1:
# Raise specific connection error!
raise GoveeConnectionError(f"Connection error: {e}")
# ... retry logic ...
raise GoveeAPIError("Max retries exceeded") # If all retries fail
```
**Explanation:**
* Based on the HTTP status code from Govee (like 401) or network exceptions (`TimeoutError`, `ClientError`), this code raises the corresponding specific `GoveeError`.
## Reacting to Warning Lights: Catching Exceptions
Raising an error is only half the story. Other parts of the code need to *catch* these specific errors and handle them gracefully, perhaps by showing a helpful message to the user. This is done using `try...except` blocks.
**Example: Handling Errors in the CLI**
The main function in our [Chapter 4: Command Line Interface (CLI)](04_command_line_interface__cli_.md) wraps the core logic in a `try...except` block.
```python
# Simplified from src/govee_mcp_server/cli.py
import sys
from .exceptions import GoveeError, GoveeConfigError, GoveeValidationError
# ... other imports ...
async def main() -> None:
api = None
try:
# --- Operations that might raise GoveeErrors ---
config = load_config() # Might raise GoveeConfigError
api = GoveeAPI(config)
args = parser.parse_args()
if args.command == 'color':
# Might raise GoveeValidationError (from decorator)
# or GoveeAPIError, etc. (from api.set_color)
await handle_color(api, args.red, args.green, args.blue)
# ... other command handlers ...
# --- End of operations ---
# --- Catching specific errors ---
except GoveeConfigError as e:
print(f"\nConfiguration Problem: {str(e)}")
print("Please check your .env file or environment variables.")
sys.exit(1) # Exit with an error status
except GoveeValidationError as e:
print(f"\nInvalid Input: {str(e)}")
sys.exit(1)
except GoveeError as e: # Catch any other Govee-specific error
print(f"\nOperation Failed: {str(e)}")
sys.exit(1)
except Exception as e: # Catch any totally unexpected error
print(f"\nAn Unexpected Error Occurred: {str(e)}")
sys.exit(1)
finally:
# Cleanup runs whether there was an error or not
if api:
await api.close()
# ... function to run main ...
```
**Explanation:**
* The code inside the `try` block is executed.
* If `load_config()` raises a `GoveeConfigError`, the code jumps directly to the `except GoveeConfigError as e:` block. It prints a specific message about configuration and exits.
* If `handle_color` (or the underlying `api.set_color`) raises a `GoveeValidationError`, it jumps to the `except GoveeValidationError as e:` block.
* If any *other* type of `GoveeError` occurs (like `GoveeAPIError`, `GoveeTimeoutError`), it's caught by the more general `except GoveeError as e:` block.
* This allows the CLI to give tailored feedback based on *what kind* of error happened.
The MCP server in [Chapter 5: MCP Server Implementation](05_mcp_server_implementation.md) uses a similar `try...except` structure within each `@mcp.tool` function to catch `GoveeError`s and return an informative error message to the MCP client instead of crashing.
## Under the Hood: Tracing an "Invalid API Key" Error
Let's follow the journey of an error when the user tries to run the CLI with a wrong API key:
1. **User Runs:** `govee-cli power on`
2. **CLI Starts:** `cli.py` `main()` function begins.
3. **Load Config:** `load_config()` runs successfully (it finds *an* API key, just the wrong one).
4. **Create API Client:** `api = GoveeAPI(config)` is created.
5. **Call Handler:** `handle_power(api, 'on')` is called.
6. **Call API Method:** `await api.set_power(True)` is called.
7. **Make Request:** Inside `set_power`, `await self._make_request(...)` is called.
8. **Send to Govee:** `_make_request` sends the command to the Govee Cloud API, including the *invalid* API key in the headers.
9. **Govee Responds:** The Govee API sees the invalid key and sends back an HTTP 401 Unauthorized error response (e.g., `{"message": "Invalid API Key"}`).
10. **Detect Error:** `_make_request` receives the 401 status.
11. **Raise Specific Error:** It executes `raise GoveeAPIError("API error: 401 - Invalid API Key")`. This stops `_make_request`.
12. **Propagate Error:** The `GoveeAPIError` travels up out of `_make_request` and then out of `api.set_power`. (Note: our simplified `set_power` in the previous chapter *caught* the error and returned `False`, but the real one might let it propagate or re-raise it). Let's assume for this trace it propagates up.
13. **Catch in Handler:** The `try...except` block in `handle_power` catches the `GoveeAPIError`.
14. **Raise from Handler:** The handler re-raises the error: `raise GoveeError(message)` (or simply lets the original `GoveeAPIError` propagate).
15. **Catch in Main:** The main `try...except` block in `cli.py` `main()` catches the `GoveeError` (since `GoveeAPIError` is a type of `GoveeError`).
16. **Print Message:** The code inside `except GoveeError as e:` runs, printing: `Operation Failed: API error: 401 - Invalid API Key`
17. **Exit:** The CLI exits with an error code (`sys.exit(1)`).
```mermaid
sequenceDiagram
participant User
participant CLI as cli.py main()
participant Handler as handle_power()
participant API as GoveeAPI.set_power()
participant Request as GoveeAPI._make_request()
participant GoveeCloud
User->>CLI: Run `govee-cli power on`
CLI->>Handler: Call handle_power()
Handler->>API: Call api.set_power(True)
API->>Request: Call _make_request()
Request->>GoveeCloud: Send request (with bad API key)
GoveeCloud-->>Request: Respond: HTTP 401 Unauthorized
Request-->>Request: Detect 401 status
Request-->>API: raise GoveeAPIError("Invalid API Key")
API-->>Handler: Error propagates up
Handler-->>CLI: Error propagates up
CLI-->>CLI: Catch GoveeAPIError (as GoveeError)
CLI->>User: Print "Operation Failed: API error: 401 - Invalid API Key"
CLI->>User: Exit with error code
```
This detailed flow shows how the specific error generated deep within the API client travels up and is eventually caught by the top-level error handler, allowing for a precise message to be shown to the user.
## Conclusion
You've reached the end of the `govee_mcp_server` tutorial! In this chapter, we learned about **Custom Error Handling**:
* It's like having specific **warning lights** instead of one generic "Check Engine" light.
* We defined our own exception classes (`GoveeAPIError`, `GoveeConfigError`, `GoveeValidationError`, etc.) inheriting from a base `GoveeError`.
* Different parts of the code `raise` these specific errors when problems are detected (e.g., bad config, invalid input, API failure).
* Other parts of the code (like the CLI and MCP server) use `try...except` blocks to `catch` these specific errors and provide informative feedback or take corrective action.
* This makes debugging easier and the application more user-friendly.
Throughout this tutorial, you've seen how `govee_mcp_server` manages configuration, defines device capabilities using interfaces, communicates with the Govee API, provides both a command-line interface and an MCP server, and handles errors gracefully.
We hope this journey has given you a clear understanding of how the project is structured and how its core components work together. Happy controlling!
---
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/03_govee_api_client.md:
--------------------------------------------------------------------------------
```markdown
# Chapter 3: Govee API Client - The Messenger to Govee
In [Chapter 1: Configuration Management](01_configuration_management.md), we learned how our application gets its essential details (like your API Key and Device ID). Then, in [Chapter 2: Device Control Interfaces](02_device_control_interfaces.md), we saw how we create standard blueprints (`PowerControl`, `ColorControl`, etc.) that define *what* actions we can perform, like turning a light on or changing its color.
But defining a blueprint for a "power socket" doesn't actually provide electricity! We need a component that takes our standard command (like "turn power on") and translates it into the specific language the Govee Cloud understands, then actually sends that message over the internet.
This chapter introduces the **Govee API Client** – the heart of our communication system.
## What's the Big Idea? Talking the Govee Talk
Imagine you need to send instructions to someone who only speaks a very specific, technical language (the "Govee language"). You wouldn't just shout your request in English! You'd hire a specialized **translator or diplomat** who knows:
1. **The Language:** Exactly how to phrase the request (the structure, the vocabulary).
2. **The Protocol:** How to properly address the message and where to send it (the specific web address or "endpoint").
3. **The Credentials:** How to prove they are allowed to speak on your behalf (using your API Key).
4. **Handling Issues:** What to do if the message doesn't get through (like trying again or reporting an error).
The **Govee API Client** (implemented in our project as the `GoveeAPI` class in `src/govee_mcp_server/api.py`) is exactly this translator. It takes simple commands defined by our interfaces (like `set_power(True)`) and handles all the complex details of communicating with the actual Govee servers.
## Key Roles of the Govee API Client
Let's break down the main jobs of this crucial component:
1. **Using Credentials:** It takes the `GoveeConfig` object (containing your API Key, Device ID, and SKU) that we loaded in [Chapter 1: Configuration Management](01_configuration_management.md). It needs these to authenticate itself and specify which device to control.
2. **Formatting Requests:** Govee's API expects commands in a very specific format, usually JSON (a standard way to structure data). The client builds these correctly formatted messages. For example, turning a light on isn't just sending "ON"; it's sending structured data like `{"device": "YOUR_ID", "sku": "YOUR_SKU", "capability": {... "value": 1 ...}}`.
3. **Sending Over the Internet:** It uses standard web protocols (HTTP) to send these formatted requests to the correct Govee API server address (like `https://openapi.api.govee.com/...`).
4. **Processing Responses:** Govee's servers send back replies, also usually in JSON. The client needs to understand these replies to know if the command worked and to get information (like the current brightness).
5. **Handling Network Hiccups:** The internet isn't always perfect. Connections can drop, or servers might be slow. The client includes logic to automatically **retry** sending a request a few times if it fails initially. It also handles specific errors like timeouts or connection failures gracefully.
## Using the Govee API Client
Let's see how simple it looks from the outside, thanks to the groundwork laid in previous chapters.
First, we need the configuration and then we create an instance of the `GoveeAPI` client:
```python
# --- Conceptual Example ---
from govee_mcp_server.config import load_config
from govee_mcp_server.api import GoveeAPI
import asyncio # We need this for 'async' functions
async def main():
# 1. Load the configuration (API Key, Device ID, SKU)
try:
config = load_config()
print(f"Loaded config for device: {config.device_id}")
except Exception as e:
print(f"Error loading config: {e}")
return
# 2. Create the Govee API Client instance, giving it the config
api_client = GoveeAPI(config=config)
print("GoveeAPI client created.")
# 3. Now we can use the client to control the device!
# (Example in the next section)
# Don't forget to close the connection when done
await api_client.close()
print("API client connection closed.")
# Run the example function
# asyncio.run(main()) # You'd typically run this in a real script
```
**Explanation:**
1. We use `load_config()` from [Chapter 1: Configuration Management](01_configuration_management.md) to get our settings.
2. We create `GoveeAPI` by passing the `config` object to it. The client now knows *which* device to talk to and *how* to authenticate.
3. The `api_client` object is now ready to send commands.
4. `await api_client.close()` is important to clean up network connections gracefully when we're finished.
Now, let's use the client. Remember how `GoveeAPI` implements the interfaces from [Chapter 2: Device Control Interfaces](02_device_control_interfaces.md)? We can call those interface methods directly on our `api_client` object:
```python
# --- Continuing the conceptual example ---
# Assume 'api_client' was created as shown above
# Let's turn the light ON
print("Attempting to turn power ON...")
success, message = await api_client.set_power(True)
if success:
print(f"Success! Govee replied: '{message}'")
else:
print(f"Failed. Govee error: '{message}'")
# Example Output (if successful):
# Attempting to turn power ON...
# Success! Govee replied: 'Success'
# Example Output (if failed, e.g., wrong API key):
# Attempting to turn power ON...
# Failed. Govee error: 'API error: 401 - Invalid API Key'
```
**Explanation:**
* We call `api_client.set_power(True)`. Even though this looks simple, behind the scenes, the `GoveeAPI` client is doing all the hard work: formatting the request, adding the API key, sending it, and interpreting the response.
* Because our interfaces (and thus the `GoveeAPI` implementation) are defined to return a `Tuple[bool, str]`, we get back whether the operation succeeded (`success`) and a message from the API (`message`).
## Under the Hood: How a Command is Sent
Let's trace the journey of a `set_power(True)` command:
1. **You Call:** Your code calls `await api_client.set_power(True)`.
2. **Client Receives:** The `set_power` method inside the `GoveeAPI` object starts running.
3. **Get Credentials:** It accesses the stored `config` object to retrieve `api_key`, `device_id`, and `sku`.
4. **Build the Message:** It constructs the specific JSON payload Govee expects for power control. This might look something like:
```json
{
"requestId": "some_unique_id", // Often based on time
"payload": {
"sku": "H6159", // Your device model
"device": "AB:CD:EF...", // Your device ID
"capability": {
"type": "devices.capabilities.on_off",
"instance": "powerSwitch",
"value": 1 // 1 means ON, 0 means OFF
}
}
}
```
5. **Prepare the Envelope:** It gets ready to make an HTTP POST request. It sets the destination URL (`https://openapi.api.govee.com/router/api/v1/device/control`) and adds required headers, most importantly the `Govee-API-Key` header with your key.
6. **Send:** It sends the request over the internet using an HTTP library (`aiohttp` in our case).
7. **Wait & Listen:** It waits for Govee's servers to respond.
8. **Check Reply:** Govee sends back a response (e.g., HTTP status code 200 OK and some JSON data like `{"message": "Success"}`).
9. **Handle Problems (Retry?):** If there was a network error (e.g., timeout) or a bad response from Govee (e.g., status 500 Server Error), the client might wait a second and try sending the request again (up to a certain limit).
10. **Translate Result:** Based on the final response (or lack thereof after retries), it determines if the command was successful (`True` or `False`) and extracts a relevant message.
11. **Return:** It returns the `(success, message)` tuple back to your code.
Here's a simplified diagram of this flow:
```mermaid
sequenceDiagram
participant YourCode as Your Code
participant APIClient as GoveeAPI Object
participant Config as GoveeConfig
participant Formatter as Request Formatter
participant HTTP as HTTP Sender (aiohttp)
participant GoveeCloud as Govee Cloud API
YourCode->>APIClient: await set_power(True)
APIClient->>Config: Get api_key, device_id, sku
Config-->>APIClient: Return values
APIClient->>Formatter: Build power ON JSON payload
Formatter-->>APIClient: Return formatted JSON
APIClient->>HTTP: Prepare POST request (URL, JSON, Headers with API Key)
loop Retries (if needed)
HTTP->>GoveeCloud: Send HTTP POST request
GoveeCloud-->>HTTP: Send HTTP Response (e.g., 200 OK, {'message':'Success'})
alt Successful Response
HTTP-->>APIClient: Forward response
Note over APIClient: Success! Exit loop.
else Network Error or Bad Response
HTTP-->>APIClient: Report error
Note over APIClient: Wait and retry (if attempts remain)
end
end
APIClient->>YourCode: return (True, "Success") or (False, "Error message")
```
## Inside the Code: Key Parts of `GoveeAPI`
Let's peek at simplified snippets from `src/govee_mcp_server/api.py` to see how this is implemented.
**1. Initialization (`__init__`)**
```python
# Simplified from src/govee_mcp_server/api.py
import aiohttp # Library for making async web requests
from .config import GoveeConfig
# ... other imports ...
class GoveeAPI(PowerControl, ColorControl, BrightnessControl):
# ... constants like BASE_URL, MAX_RETRIES ...
def __init__(self, config: GoveeConfig):
"""Initialize API client with configuration."""
self.config = config # Store the config object
self.session: Optional[aiohttp.ClientSession] = None # Network session setup later
# ... maybe other setup ...
print(f"GoveeAPI initialized for device {self.config.device_id}")
```
**Explanation:**
* The `__init__` method runs when you create `GoveeAPI(config=...)`.
* It simply stores the passed-in `config` object so other methods in the class can access the API key, device ID, etc., using `self.config.api_key`.
* It also prepares a variable `self.session` to hold the network connection details, which will be created when needed.
**2. Making the Request (`_make_request`)**
This is the core helper method that handles sending *any* request, including retries and basic error handling.
```python
# Simplified from src/govee_mcp_server/api.py
import asyncio
# ... other imports ...
class GoveeAPI:
# ... __init__ and constants ...
async def _ensure_session(self) -> None:
"""Create network session if needed."""
if self.session is None or self.session.closed:
print("Creating new network session...")
self.session = aiohttp.ClientSession(
headers={"Govee-API-Key": self.config.api_key}, # Set API key for ALL requests!
timeout=aiohttp.ClientTimeout(total=10) # 10-second timeout
)
async def _make_request(self, method: str, endpoint: str, **kwargs):
"""Make HTTP request with retries."""
await self._ensure_session() # Make sure we have a network session
last_error = None
for attempt in range(self.MAX_RETRIES): # Try up to MAX_RETRIES times
try:
print(f"Attempt {attempt+1}: Sending {method} to {endpoint}")
async with self.session.request(method, f"{self.BASE_URL}/{endpoint}", **kwargs) as response:
data = await response.json()
print(f"Received status: {response.status}")
if response.status == 200: # HTTP 200 means OK!
return data, data.get('message', 'Success') # Return data and message
else:
# Govee returned an error status
last_error = GoveeAPIError(f"API error: {response.status} - {data.get('message', 'Unknown')}")
except asyncio.TimeoutError:
last_error = GoveeTimeoutError("Request timed out")
except aiohttp.ClientError as e:
last_error = GoveeConnectionError(f"Connection error: {e}")
# If error occurred, wait before retrying (increasing delay)
if attempt < self.MAX_RETRIES - 1:
delay = self.RETRY_DELAY * (attempt + 1)
print(f"Request failed. Retrying in {delay}s...")
await asyncio.sleep(delay)
# If all retries failed, raise the last encountered error
print("Max retries exceeded.")
raise last_error
```
**Explanation:**
* `_ensure_session`: Checks if a network `session` exists; if not, it creates one using `aiohttp`. Crucially, it sets the `Govee-API-Key` header here, so it's automatically included in all requests made with this session. It also sets a timeout.
* The main loop runs `MAX_RETRIES` times (e.g., 3 times).
* `try...except`: It tries to make the request using `self.session.request(...)`. It catches specific network errors like `TimeoutError` or general `ClientError`.
* `if response.status == 200`: Checks if Govee reported success. If yes, it returns the data.
* `else`: If Govee returns an error code (like 401 Unauthorized or 400 Bad Request), it creates a `GoveeAPIError`.
* Retry Logic: If an error occurs, it waits (`asyncio.sleep`) before the next attempt. The delay increases with each retry.
* If all retries fail, it `raise`s the last error it encountered, which can then be caught by the calling method (like `set_power`).
**3. Implementing an Interface Method (`set_power`)**
This method uses `_make_request` to perform a specific action.
```python
# Simplified from src/govee_mcp_server/api.py
from time import time # To generate a unique request ID
class GoveeAPI:
# ... __init__, _make_request ...
async def set_power(self, state: bool) -> Tuple[bool, str]:
"""Implement PowerControl.set_power"""
try:
# Prepare the specific JSON payload for the power command
payload = {
"requestId": str(int(time())), # Unique ID for the request
"payload": {
"sku": self.config.sku,
"device": self.config.device_id,
"capability": {
"type": "devices.capabilities.on_off",
"instance": "powerSwitch",
"value": 1 if state else 0 # Convert True/False to 1/0
}
}
}
# Call the generic request handler
_, message = await self._make_request(
method="POST",
endpoint="router/api/v1/device/control", # Govee's endpoint for control
json=payload # Pass the JSON data
)
return True, message # If _make_request succeeded, return True
except GoveeError as e:
# If _make_request raised an error after retries
print(f"set_power failed: {e}")
return False, str(e) # Return False and the error message
```
**Explanation:**
* It builds the `payload` dictionary exactly as Govee requires for the `on_off` capability, using the `device_id` and `sku` from `self.config`.
* It calls `self._make_request`, telling it to use the `POST` method, the specific `/device/control` endpoint, and the `payload` as JSON data.
* `try...except GoveeError`: It wraps the call to `_make_request`. If `_make_request` succeeds, it returns `(True, message)`. If `_make_request` eventually fails (after retries) and raises a `GoveeError` (or one of its subtypes like `GoveeAPIError`, `GoveeTimeoutError`), the `except` block catches it and returns `(False, error_message)`.
The other methods (`set_color`, `get_brightness`, etc.) follow a very similar pattern: build the specific payload for that capability and call `_make_request`.
## Conclusion
You've now explored the **Govee API Client (`GoveeAPI`)**, the component that acts as our application's diplomat to the Govee Cloud.
You learned:
* Its role is to translate standard commands (from our interfaces) into Govee's specific API language.
* It uses the configuration ([Chapter 1: Configuration Management](01_configuration_management.md)) for authentication and targeting.
* It handles formatting requests, sending them via HTTP, and processing responses.
* It includes essential features like automatic retries and handling network errors (`TimeoutError`, `ConnectionError`).
* Internally, it uses a helper (`_make_request`) to manage the common parts of API communication.
* Methods like `set_power` build the specific command payload and rely on `_make_request` to send it.
With the configuration loaded, interfaces defined, and the API client ready to communicate, we have the core logic in place. But how does a user actually *trigger* these actions?
Next, we'll look at one way users can interact with our application: the command line. Let's move on to [Chapter 4: Command Line Interface (CLI)](04_command_line_interface__cli_.md)!
---
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```