# 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:
--------------------------------------------------------------------------------
```
1 | GOVEE_API_KEY=<API_KEY_FROM_GOVEE_APP>
2 | GOVEE_DEVICE_ID=<MAC_ADDRESS_OF_THE_DEVICE>
3 | GOVEE_SKU=<SKU_OF_THE_DEVICE>
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | local_docs/
2 | .DS_Store
3 | Thumbs.db
4 |
5 | # Byte-compiled / optimized / DLL files
6 | __pycache__/
7 | *.py[cod]
8 | *$py.class
9 |
10 | # C extensions
11 | *.so
12 |
13 | # Distribution / packaging
14 | .Python
15 | env/
16 | venv/
17 | ENV/
18 | env.bak/
19 | venv.bak/
20 | *.egg-info/
21 | dist/
22 | build/
23 |
24 | # PyInstaller
25 | # Usually these files are written by a python script from a template
26 | # before PyInstaller builds the exe, so as to inject date/other infos into it.
27 | *.manifest
28 | *.spec
29 |
30 | # Installer logs
31 | pip-log.txt
32 | pip-delete-this-directory.txt
33 |
34 | # Unit test / coverage reports
35 | htmlcov/
36 | .tox/
37 | .nox/
38 | .coverage
39 | .coverage.*
40 | .cache
41 | nosetests.xml
42 | coverage.xml
43 | *.cover
44 | *.py,cover
45 | .hypothesis/
46 |
47 | # Translations
48 | *.mo
49 | *.pot
50 |
51 | # Django stuff:
52 | *.log
53 | local_settings.py
54 | db.sqlite3
55 |
56 | # Flask stuff:
57 | instance/
58 | .webassets-cache
59 |
60 | # Scrapy stuff:
61 | .scrapy
62 |
63 | # Sphinx documentation
64 | docs/_build/
65 |
66 | # PyBuilder
67 | target/
68 |
69 | # Jupyter Notebook
70 | .ipynb_checkpoints
71 |
72 | # IPython
73 | profile_default/
74 | ipython_config.py
75 |
76 | # pyenv
77 | .python-version
78 |
79 | # celery beat schedule file
80 | celerybeat-schedule
81 |
82 | # dotenv
83 | .env
84 |
85 | # virtualenv
86 | .venv/
87 | venv/
88 | ENV/
89 | env.bak/
90 | venv.bak/
91 |
92 | # Spyder project settings
93 | .spyderproject
94 | .spyproject
95 |
96 | # Rope project settings
97 | .ropeproject
98 |
99 | # mkdocs documentation
100 | /site
101 |
102 | # mypy
103 | .mypy_cache/
104 | .dmypy.json
105 | dmypy.json
106 |
107 | # Pyre type checker
108 | .pyre/
109 |
110 | # pytype static type analyzer
111 | .pytype/
112 |
113 | # Cython debug symbols
114 | cython_debug/
115 |
116 | # VSCode settings
117 | .vscode/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Govee MCP Server
2 |
3 | [](https://smithery.ai/server/@mathd/govee_mcp_server)
4 |
5 | An MCP server for controlling Govee LED devices through the Govee API.
6 |
7 | ## Setup
8 |
9 | ### Environment Variables
10 |
11 | Create a `.env` file in the root directory with the following variables:
12 |
13 | ```bash
14 | GOVEE_API_KEY=your_api_key_here
15 | GOVEE_DEVICE_ID=your_device_id_here
16 | GOVEE_SKU=your_device_sku_here
17 | ```
18 |
19 | To get these values:
20 | 1. Get your API key from the Govee Developer Portal
21 | 2. Use the Govee Home app to find your device ID and SKU
22 |
23 | ## Installation
24 |
25 | ### Installing via Smithery
26 |
27 | To install Govee MCP Server for Claude Desktop automatically via [Smithery](https://smithery.ai/server/@mathd/govee_mcp_server):
28 |
29 | ```bash
30 | npx -y @smithery/cli install @mathd/govee_mcp_server --client claude
31 | ```
32 |
33 | ### Manual Installation
34 |
35 | ```bash
36 | # Install with pip
37 | pip install .
38 |
39 | # For development (includes test dependencies)
40 | pip install -e ".[test]"
41 | ```
42 |
43 | ## Usage
44 |
45 | ### MCP Server
46 |
47 | The MCP server provides tools for controlling Govee devices through the Model Context Protocol. It can be used with Cline or other MCP clients.
48 |
49 | Available tools:
50 | - `turn_on_off`: Turn the LED on or off
51 | - `set_color`: Set the LED color using RGB values
52 | - `set_brightness`: Set the LED brightness level
53 |
54 | ### Command Line Interface
55 |
56 | A CLI is provided for direct control of Govee devices:
57 |
58 | ```bash
59 | # Turn device on/off
60 | govee-cli power on
61 | govee-cli power off
62 |
63 | # Set color using RGB values (0-255)
64 | govee-cli color 255 0 0 # Red
65 | govee-cli color 0 255 0 # Green
66 | govee-cli color 0 0 255 # Blue
67 |
68 | # Set brightness (0-100)
69 | govee-cli brightness 50
70 | ```
71 |
72 | Run `govee-cli --help` for full command documentation.
73 |
74 | ## Development
75 |
76 | ### Running Tests
77 |
78 | To run the test suite:
79 |
80 | ```bash
81 | # Install test dependencies
82 | pip install -e ".[test]"
83 |
84 | # Run all tests
85 | pytest tests/
86 |
87 | # Run specific test files
88 | pytest tests/test_server.py # Server tests (mocked API calls)
89 | pytest tests/test_cli.py # CLI tests (real API calls)
90 |
91 | # Run tests with verbose output
92 | pytest tests/ -v
93 | ```
94 |
95 | 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.
96 |
97 | ### Project Structure
98 |
99 | ```
100 | .
101 | ├── src/govee_mcp_server/
102 | │ ├── __init__.py
103 | │ ├── server.py # MCP server implementation
104 | │ └── cli.py # Command-line interface
105 | ├── tests/
106 | │ ├── test_server.py # Server tests (with mocked API)
107 | │ └── test_cli.py # CLI tests (real API calls)
108 | └── pyproject.toml # Project configuration
109 | ```
110 |
111 | ### Test Coverage
112 |
113 | - Server tests cover:
114 | - Environment initialization
115 | - Govee API client methods
116 | - Server tools and utilities
117 | - Error handling
118 |
119 | - CLI tests perform real-world integration testing by executing actual API calls to control your Govee device.
120 |
```
--------------------------------------------------------------------------------
/inspector.bat:
--------------------------------------------------------------------------------
```
1 | npx @modelcontextprotocol/inspector uv --directory . run python src\govee_mcp_server\server.py
```
--------------------------------------------------------------------------------
/inspector.sh:
--------------------------------------------------------------------------------
```bash
1 | npx @modelcontextprotocol/inspector uv --directory . run python src/govee_mcp_server/server.py
2 |
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/exceptions.py:
--------------------------------------------------------------------------------
```python
1 | class GoveeError(Exception):
2 | """Base exception for Govee-related errors."""
3 | pass
4 |
5 | class GoveeAPIError(GoveeError):
6 | """Raised when API communication fails."""
7 | pass
8 |
9 | class GoveeConfigError(GoveeError):
10 | """Raised when there are configuration-related errors."""
11 | pass
12 |
13 | class GoveeValidationError(GoveeError):
14 | """Raised when input validation fails."""
15 | pass
16 |
17 | class GoveeConnectionError(GoveeError):
18 | """Raised when network connection issues occur."""
19 | pass
20 |
21 | class GoveeTimeoutError(GoveeError):
22 | """Raised when requests timeout."""
23 | pass
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "govee-mcp-server"
3 | version = "0.1.0"
4 | description = "MCP server to control Govee LED devices"
5 | authors = []
6 | dependencies = [
7 | "mcp[cli]",
8 | "govee-api-laggat>=0.2.2",
9 | "python-dotenv"
10 | ]
11 | requires-python = ">=3.10"
12 |
13 | [project.scripts]
14 | govee-cli = "govee_mcp_server.cli:cli_main"
15 |
16 | [project.optional-dependencies]
17 | test = [
18 | "pytest>=7.0",
19 | "pytest-asyncio>=0.21.0",
20 | "pytest-mock>=3.10.0"
21 | ]
22 |
23 | [build-system]
24 | requires = ["hatchling"]
25 | build-backend = "hatchling.build"
26 |
27 | [tool.hatch.build.targets.wheel]
28 | packages = ["src/govee_mcp_server"]
29 |
30 | [tool.hatch.metadata]
31 | allow-direct-references = true
32 |
33 | [tool.pytest.ini_options]
34 | asyncio_mode = "auto"
35 | testpaths = ["tests"]
36 | pythonpath = ["src"]
37 | asyncio_default_fixture_loop_scope = "function"
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
1 | # Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
2 |
3 | startCommand:
4 | type: stdio
5 | configSchema:
6 | # JSON Schema defining the configuration options for the MCP.
7 | type: object
8 | required:
9 | - goveeApiKey
10 | - goveeDeviceId
11 | - goveeSku
12 | properties:
13 | goveeApiKey:
14 | type: string
15 | description: The API key for the Govee API.
16 | goveeDeviceId:
17 | type: string
18 | description: The device ID for the Govee device.
19 | goveeSku:
20 | type: string
21 | description: The SKU for the Govee device.
22 | commandFunction:
23 | # A function that produces the CLI command to start the MCP on stdio.
24 | |-
25 | (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
1 | """
2 | Govee MCP Server - A Model Context Protocol server for controlling Govee LED devices.
3 |
4 | This package provides a complete interface for controlling Govee LED devices through both
5 | an MCP server implementation and a CLI interface.
6 | """
7 |
8 | from .api import GoveeAPI
9 | from .config import GoveeConfig, load_config
10 | from .exceptions import (
11 | GoveeError,
12 | GoveeAPIError,
13 | GoveeConfigError,
14 | GoveeValidationError,
15 | GoveeConnectionError,
16 | GoveeTimeoutError
17 | )
18 | from .transformers import ColorTransformer
19 | from .interfaces import PowerControl, ColorControl, BrightnessControl
20 |
21 | __version__ = "0.1.0"
22 | __all__ = [
23 | 'GoveeAPI',
24 | 'GoveeConfig',
25 | 'load_config',
26 | 'GoveeError',
27 | 'GoveeAPIError',
28 | 'GoveeConfigError',
29 | 'GoveeValidationError',
30 | 'GoveeConnectionError',
31 | 'GoveeTimeoutError',
32 | 'ColorTransformer',
33 | 'PowerControl',
34 | 'ColorControl',
35 | 'BrightnessControl'
36 | ]
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
1 | # Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
2 | # Use a Python image with uv pre-installed
3 | FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim AS uv
4 |
5 | # Set the working directory
6 | WORKDIR /app
7 |
8 | # Enable bytecode compilation
9 | ENV UV_COMPILE_BYTECODE=1
10 |
11 | # Copy from the cache instead of linking since it's a mounted volume
12 | ENV UV_LINK_MODE=copy
13 |
14 | # Install the project's dependencies using the lockfile and settings
15 | RUN --mount=type=cache,target=/root/.cache/uv \
16 | --mount=type=bind,source=uv.lock,target=uv.lock \
17 | --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
18 | uv sync --frozen --no-install-project --no-dev --no-editable
19 |
20 | # Copy the project source code
21 | COPY . /app
22 |
23 | # Install the project
24 | RUN --mount=type=cache,target=/root/.cache/uv \
25 | uv sync --frozen --no-dev --no-editable
26 |
27 | # Final image
28 | FROM python:3.12-slim-bookworm
29 |
30 | # Set the working directory
31 | WORKDIR /app
32 |
33 | # Copy the environment setup from the uv image
34 | COPY --from=uv --chown=app:app /app/.venv /app/.venv
35 |
36 | # Place executables in the environment at the front of the path
37 | ENV PATH="/app/.venv/bin:$PATH"
38 |
39 | # when running the container, add --db-path and a bind mount to the host's db file
40 | ENTRYPOINT ["python", "src/govee_mcp_server/server.py"]
```
--------------------------------------------------------------------------------
/tests/test_cli.py:
--------------------------------------------------------------------------------
```python
1 | import pytest
2 | import sys
3 | import asyncio
4 | from govee_mcp_server.cli import main
5 | from govee_mcp_server.config import load_config, GoveeConfigError
6 |
7 | # Delay between commands (in seconds)
8 | DELAY = 1
9 |
10 | @pytest.mark.asyncio
11 | async def test_cli_interface():
12 | """Test CLI interface with real API calls"""
13 | try:
14 | # Load actual config from environment
15 | config = load_config()
16 |
17 | # Power on
18 | sys.argv = ['cli.py', 'power', 'on']
19 | await main()
20 | await asyncio.sleep(DELAY)
21 |
22 | # Red color
23 | sys.argv = ['cli.py', 'color', '255', '0', '0']
24 | await main()
25 | await asyncio.sleep(DELAY)
26 |
27 | # Green color
28 | sys.argv = ['cli.py', 'color', '0', '255', '0']
29 | await main()
30 | await asyncio.sleep(DELAY)
31 |
32 | # Blue color
33 | sys.argv = ['cli.py', 'color', '0', '0', '255']
34 | await main()
35 | await asyncio.sleep(DELAY)
36 |
37 | # Power off
38 | sys.argv = ['cli.py', 'power', 'off']
39 | await main()
40 |
41 | except GoveeConfigError as e:
42 | pytest.skip(f"Skipping test: {str(e)}")
43 |
44 | except Exception as e:
45 | # If we hit rate limits or other API errors, fail with clear message
46 | pytest.fail(f"API Error: {str(e)}")
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/config.py:
--------------------------------------------------------------------------------
```python
1 | from pathlib import Path
2 | from dataclasses import dataclass
3 | import os
4 | from dotenv import load_dotenv
5 | from typing import Optional
6 |
7 | @dataclass
8 | class GoveeConfig:
9 | """Configuration class for Govee API settings."""
10 | api_key: str
11 | device_id: str
12 | sku: str
13 |
14 | class GoveeConfigError(Exception):
15 | """Configuration-related errors for Govee MCP server."""
16 | pass
17 |
18 | def load_config() -> GoveeConfig:
19 | """
20 | Load and validate configuration from environment variables.
21 |
22 | Returns:
23 | GoveeConfig: Configuration object with API settings
24 |
25 | Raises:
26 | GoveeConfigError: If required environment variables are missing
27 | """
28 | env_path = Path(__file__).resolve().parent.parent.parent / '.env'
29 | load_dotenv(env_path)
30 |
31 | api_key = os.getenv('GOVEE_API_KEY')
32 | device_id = os.getenv('GOVEE_DEVICE_ID')
33 | sku = os.getenv('GOVEE_SKU')
34 |
35 | missing = []
36 | if not api_key:
37 | missing.append('GOVEE_API_KEY')
38 | if not device_id:
39 | missing.append('GOVEE_DEVICE_ID')
40 | if not sku:
41 | missing.append('GOVEE_SKU')
42 |
43 | if missing:
44 | raise GoveeConfigError(f"Missing required environment variables: {', '.join(missing)}")
45 |
46 | return GoveeConfig(
47 | api_key=api_key,
48 | device_id=device_id,
49 | sku=sku
50 | )
```
--------------------------------------------------------------------------------
/documentation/index.md:
--------------------------------------------------------------------------------
```markdown
1 | # Tutorial: govee_mcp_server
2 |
3 | This project lets you control your **Govee LED lights** over the internet.
4 | It acts as a bridge, translating commands into the *Govee API language*.
5 | You can interact with it either through an **AI assistant** (using the *MCP server*) or directly using simple **command-line** instructions.
6 | It needs your Govee *API key* and *device details* to work, which it reads from configuration settings.
7 |
8 |
9 | **Source Repository:** [https://github.com/mathd/govee_mcp_server](https://github.com/mathd/govee_mcp_server)
10 |
11 | ```mermaid
12 | flowchart TD
13 | A0["Govee API Client"]
14 | A1["MCP Server Implementation"]
15 | A2["Configuration Management"]
16 | A3["Device Control Interfaces"]
17 | A4["Command Line Interface (CLI)"]
18 | A5["Custom Error Handling"]
19 | A1 -- "Executes commands via" --> A0
20 | A4 -- "Executes commands via" --> A0
21 | A0 -- "Reads settings from" --> A2
22 | A0 -- "Implements" --> A3
23 | A0 -- "Raises API/Network Errors" --> A5
24 | A1 -- "Handles Errors From" --> A5
25 | A4 -- "Handles Errors From" --> A5
26 | A2 -- "Raises Configuration Errors" --> A5
27 | A3 -- "Raises Validation Errors" --> A5
28 | ```
29 |
30 | ## Chapters
31 |
32 | 1. [Configuration Management](01_configuration_management.md)
33 | 2. [Device Control Interfaces](02_device_control_interfaces.md)
34 | 3. [Govee API Client](03_govee_api_client.md)
35 | 4. [Command Line Interface (CLI)](04_command_line_interface__cli_.md)
36 | 5. [MCP Server Implementation](05_mcp_server_implementation.md)
37 | 6. [Custom Error Handling](06_custom_error_handling.md)
38 |
39 |
40 | ---
41 |
42 | Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/interfaces.py:
--------------------------------------------------------------------------------
```python
1 | from abc import ABC, abstractmethod
2 | from typing import Tuple
3 | from functools import wraps
4 | from .exceptions import GoveeValidationError
5 |
6 | def validate_rgb(func):
7 | """Decorator to validate RGB color values."""
8 | @wraps(func)
9 | async def wrapper(self, r: int, g: int, b: int, *args, **kwargs):
10 | for name, value in [('red', r), ('green', g), ('blue', b)]:
11 | if not isinstance(value, int):
12 | raise GoveeValidationError(f"{name} value must be an integer")
13 | if not 0 <= value <= 255:
14 | raise GoveeValidationError(f"{name} value must be between 0-255")
15 | return await func(self, r, g, b, *args, **kwargs)
16 | return wrapper
17 |
18 | class PowerControl(ABC):
19 | """Interface for power control capabilities."""
20 | @abstractmethod
21 | async def set_power(self, state: bool) -> Tuple[bool, str]:
22 | """
23 | Set device power state.
24 |
25 | Args:
26 | state: True for on, False for off
27 |
28 | Returns:
29 | Tuple of (success: bool, message: str)
30 | """
31 | pass
32 |
33 | @abstractmethod
34 | async def get_power_state(self) -> Tuple[bool, str]:
35 | """
36 | Get current power state.
37 |
38 | Returns:
39 | Tuple of (is_on: bool, message: str)
40 | """
41 | pass
42 |
43 | class ColorControl(ABC):
44 | """Interface for color control capabilities."""
45 | @abstractmethod
46 | @validate_rgb
47 | async def set_color(self, r: int, g: int, b: int) -> Tuple[bool, str]:
48 | """
49 | Set device color using RGB values.
50 |
51 | Args:
52 | r: Red value (0-255)
53 | g: Green value (0-255)
54 | b: Blue value (0-255)
55 |
56 | Returns:
57 | Tuple of (success: bool, message: str)
58 | """
59 | pass
60 |
61 | @abstractmethod
62 | async def get_color(self) -> Tuple[Tuple[int, int, int], str]:
63 | """
64 | Get current color values.
65 |
66 | Returns:
67 | Tuple of ((r, g, b): Tuple[int, int, int], message: str)
68 | """
69 | pass
70 |
71 | class BrightnessControl(ABC):
72 | """Interface for brightness control capabilities."""
73 | @abstractmethod
74 | async def set_brightness(self, level: int) -> Tuple[bool, str]:
75 | """
76 | Set device brightness level.
77 |
78 | Args:
79 | level: Brightness level (0-100)
80 |
81 | Returns:
82 | Tuple of (success: bool, message: str)
83 | """
84 | pass
85 |
86 | @abstractmethod
87 | async def get_brightness(self) -> Tuple[int, str]:
88 | """
89 | Get current brightness level.
90 |
91 | Returns:
92 | Tuple of (level: int, message: str)
93 | """
94 | pass
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/transformers.py:
--------------------------------------------------------------------------------
```python
1 | from typing import Tuple, Dict, Any
2 | from .exceptions import GoveeValidationError
3 |
4 | class ColorTransformer:
5 | """Handle color transformations and validations."""
6 |
7 | @staticmethod
8 | def validate_rgb(r: int, g: int, b: int) -> None:
9 | """
10 | Validate RGB color values.
11 |
12 | Args:
13 | r: Red value (0-255)
14 | g: Green value (0-255)
15 | b: Blue value (0-255)
16 |
17 | Raises:
18 | GoveeValidationError: If values are invalid
19 | """
20 | for name, value in [('red', r), ('green', g), ('blue', b)]:
21 | if not isinstance(value, int):
22 | raise GoveeValidationError(f"{name} value must be an integer")
23 | if not 0 <= value <= 255:
24 | raise GoveeValidationError(f"{name} value must be between 0-255")
25 |
26 | @staticmethod
27 | def rgb_to_hex(r: int, g: int, b: int) -> str:
28 | """
29 | Convert RGB values to hexadecimal color code.
30 |
31 | Args:
32 | r: Red value (0-255)
33 | g: Green value (0-255)
34 | b: Blue value (0-255)
35 |
36 | Returns:
37 | str: Hexadecimal color code
38 | """
39 | ColorTransformer.validate_rgb(r, g, b)
40 | return f"#{r:02x}{g:02x}{b:02x}"
41 |
42 | @staticmethod
43 | def hex_to_rgb(hex_color: str) -> Tuple[int, int, int]:
44 | """
45 | Convert hexadecimal color code to RGB values.
46 |
47 | Args:
48 | hex_color: Hexadecimal color code (e.g., '#ff00ff' or 'ff00ff')
49 |
50 | Returns:
51 | Tuple[int, int, int]: RGB values
52 |
53 | Raises:
54 | GoveeValidationError: If hex color format is invalid
55 | """
56 | # Remove '#' if present
57 | hex_color = hex_color.lstrip('#')
58 |
59 | if len(hex_color) != 6:
60 | raise GoveeValidationError("Invalid hex color format")
61 |
62 | try:
63 | r = int(hex_color[:2], 16)
64 | g = int(hex_color[2:4], 16)
65 | b = int(hex_color[4:], 16)
66 | return (r, g, b)
67 | except ValueError:
68 | raise GoveeValidationError("Invalid hex color format")
69 |
70 | @staticmethod
71 | def to_api_payload(r: int, g: int, b: int) -> Dict[str, Any]:
72 | """
73 | Convert RGB values to API payload format.
74 |
75 | Args:
76 | r: Red value (0-255)
77 | g: Green value (0-255)
78 | b: Blue value (0-255)
79 |
80 | Returns:
81 | Dict[str, Any]: API payload
82 | """
83 | ColorTransformer.validate_rgb(r, g, b)
84 | return {
85 | "color": {
86 | "r": r,
87 | "g": g,
88 | "b": b
89 | }
90 | }
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
1 | import pytest
2 | import asyncio
3 | import aiohttp
4 | from unittest.mock import patch, MagicMock
5 | from govee_mcp_server.server import (
6 | init_env,
7 | GoveeDirectAPI,
8 | fix_json,
9 | rgb_to_int,
10 | )
11 |
12 | @pytest.fixture
13 | def mock_env_vars():
14 | with patch.dict('os.environ', {
15 | 'GOVEE_API_KEY': 'test-api-key',
16 | 'GOVEE_DEVICE_ID': 'test-device-id',
17 | 'GOVEE_SKU': 'test-sku'
18 | }):
19 | yield
20 |
21 | def test_init_env(mock_env_vars):
22 | api_key, device_id, sku = init_env()
23 | assert api_key == 'test-api-key'
24 | assert device_id == 'test-device-id'
25 | assert sku == 'test-sku'
26 |
27 | @patch('os.getenv', return_value=None)
28 | def test_init_env_missing_vars(_):
29 | with pytest.raises(SystemExit):
30 | init_env()
31 |
32 | def test_fix_json():
33 | malformed = '{"key1""value1"}{"key2""value2"}'
34 | expected = '{"key1","value1"},{"key2","value2"}'
35 | assert fix_json(malformed) == expected
36 |
37 | def test_rgb_to_int():
38 | assert rgb_to_int(255, 0, 0) == 0xFF0000
39 | assert rgb_to_int(0, 255, 0) == 0x00FF00
40 | assert rgb_to_int(0, 0, 255) == 0x0000FF
41 | assert rgb_to_int(255, 255, 255) == 0xFFFFFF
42 |
43 | class TestGoveeDirectAPI:
44 | @pytest.fixture
45 | def api(self):
46 | return GoveeDirectAPI('test-api-key')
47 |
48 | def test_init(self, api):
49 | assert api.api_key == 'test-api-key'
50 | assert api.headers['Govee-API-Key'] == 'test-api-key'
51 | assert api.headers['Content-Type'] == 'application/json'
52 |
53 | @pytest.mark.asyncio
54 | async def test_get_devices_success(self, api):
55 | mock_response = MagicMock()
56 | mock_response.status = 200
57 | async def mock_text():
58 | return '{"data":[{"device":"test"}]}'
59 | mock_response.text = mock_text
60 |
61 | with patch('aiohttp.ClientSession.get') as mock_get:
62 | mock_get.return_value.__aenter__.return_value = mock_response
63 | devices, error = await api.get_devices()
64 | assert devices == [{"device": "test"}]
65 | assert error is None
66 |
67 | @pytest.mark.asyncio
68 | async def test_get_devices_error(self, api):
69 | mock_response = MagicMock()
70 | mock_response.status = 401
71 | async def mock_text():
72 | return '{"message":"Unauthorized"}'
73 | mock_response.text = mock_text
74 |
75 | with patch('aiohttp.ClientSession.get') as mock_get:
76 | mock_get.return_value.__aenter__.return_value = mock_response
77 | devices, error = await api.get_devices()
78 | assert devices is None
79 | assert error == "Unauthorized"
80 |
81 | @pytest.mark.asyncio
82 | async def test_control_device_success(self, api):
83 | mock_response = MagicMock()
84 | mock_response.status = 200
85 | async def mock_text():
86 | return '{"message":"Success"}'
87 | mock_response.text = mock_text
88 |
89 | with patch('aiohttp.ClientSession.post') as mock_post:
90 | mock_post.return_value.__aenter__.return_value = mock_response
91 | success, error = await api.control_device(
92 | "sku123",
93 | "device123",
94 | "devices.capabilities.on_off",
95 | "powerSwitch",
96 | 1
97 | )
98 | assert success is True
99 | assert error == "Success"
100 |
101 | @pytest.mark.asyncio
102 | async def test_control_device_error(self, api):
103 | mock_response = MagicMock()
104 | mock_response.status = 400
105 | async def mock_text():
106 | return '{"message":"Bad Request"}'
107 | mock_response.text = mock_text
108 |
109 | with patch('aiohttp.ClientSession.post') as mock_post:
110 | mock_post.return_value.__aenter__.return_value = mock_response
111 | success, error = await api.control_device(
112 | "sku123",
113 | "device123",
114 | "devices.capabilities.on_off",
115 | "powerSwitch",
116 | 1
117 | )
118 | assert success is False
119 | assert error == "Bad Request"
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/cli.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | import sys
3 | import argparse
4 | import asyncio
5 | from .config import load_config
6 | from .api import GoveeAPI
7 | from .exceptions import GoveeError, GoveeValidationError
8 |
9 | def create_parser() -> argparse.ArgumentParser:
10 | """Create and configure argument parser."""
11 | parser = argparse.ArgumentParser(description='Control Govee LED device')
12 | subparsers = parser.add_subparsers(dest='command', help='Commands')
13 |
14 | # Power command
15 | power_parser = subparsers.add_parser('power', help='Turn device on/off')
16 | power_parser.add_argument('state', choices=['on', 'off'], help='Power state')
17 |
18 | # Color command
19 | color_parser = subparsers.add_parser('color', help='Set device color')
20 | color_parser.add_argument('red', type=int, help='Red value (0-255)')
21 | color_parser.add_argument('green', type=int, help='Green value (0-255)')
22 | color_parser.add_argument('blue', type=int, help='Blue value (0-255)')
23 |
24 | # Brightness command
25 | brightness_parser = subparsers.add_parser('brightness', help='Set device brightness')
26 | brightness_parser.add_argument('level', type=int, help='Brightness level (0-100)')
27 |
28 | # Status command
29 | subparsers.add_parser('status', help='Show device status')
30 |
31 | return parser
32 |
33 | async def handle_power(api: GoveeAPI, state: str) -> None:
34 | """Handle power command."""
35 | success, message = await api.set_power(state == 'on')
36 | if not success:
37 | raise GoveeError(message)
38 | print(message)
39 |
40 | async def handle_color(api: GoveeAPI, red: int, green: int, blue: int) -> None:
41 | """Handle color command."""
42 | try:
43 | success, message = await api.set_color(red, green, blue)
44 | if not success:
45 | raise GoveeError(message)
46 | print(message)
47 | except GoveeValidationError as e:
48 | print(f"Error: {e}")
49 | sys.exit(1)
50 |
51 | async def handle_brightness(api: GoveeAPI, level: int) -> None:
52 | """Handle brightness command."""
53 | if not 0 <= level <= 100:
54 | print("Error: Brightness level must be between 0 and 100")
55 | sys.exit(1)
56 |
57 | success, message = await api.set_brightness(level)
58 | if not success:
59 | raise GoveeError(message)
60 | print(message)
61 |
62 | async def handle_status(api: GoveeAPI) -> None:
63 | """Handle status command."""
64 | # Get power state
65 | power_state, power_msg = await api.get_power_state()
66 | print(f"Power: {'ON' if power_state else 'OFF'}")
67 |
68 | # Get color
69 | color, color_msg = await api.get_color()
70 | print(f"Color: RGB({color[0]}, {color[1]}, {color[2]})")
71 |
72 | # Get brightness
73 | brightness, bright_msg = await api.get_brightness()
74 | print(f"Brightness: {brightness}%")
75 |
76 | async def main() -> None:
77 | """Main CLI entrypoint."""
78 | try:
79 | # Load configuration
80 | config = load_config()
81 | api = GoveeAPI(config)
82 |
83 | # Parse arguments
84 | parser = create_parser()
85 | args = parser.parse_args()
86 |
87 | if args.command == 'power':
88 | await handle_power(api, args.state)
89 | elif args.command == 'color':
90 | await handle_color(api, args.red, args.green, args.blue)
91 | elif args.command == 'brightness':
92 | await handle_brightness(api, args.level)
93 | elif args.command == 'status':
94 | await handle_status(api)
95 | else:
96 | parser.print_help()
97 | sys.exit(1)
98 |
99 | except GoveeError as e:
100 | print(f"Error: {str(e)}")
101 | sys.exit(1)
102 | except Exception as e:
103 | print(f"Unexpected error: {str(e)}")
104 | sys.exit(1)
105 | finally:
106 | # Always close the API session
107 | if 'api' in locals():
108 | await api.close()
109 |
110 | def cli_main():
111 | """CLI entry point that handles running the async main."""
112 | try:
113 | asyncio.run(main())
114 | except KeyboardInterrupt:
115 | print("\nOperation cancelled by user")
116 | sys.exit(1)
117 |
118 | if __name__ == "__main__":
119 | cli_main()
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/server.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | import sys
3 | from mcp.server.fastmcp import FastMCP
4 | from govee_mcp_server.config import load_config
5 | from govee_mcp_server.api import GoveeAPI
6 | from govee_mcp_server.exceptions import GoveeError
7 |
8 | # Initialize FastMCP server with WARNING log level
9 | mcp = FastMCP(
10 | "govee",
11 | capabilities={
12 | "server_info": {
13 | "name": "govee-mcp",
14 | "version": "0.1.0",
15 | "description": "MCP server for controlling Govee LED devices"
16 | }
17 | },
18 | log_level='WARNING'
19 | )
20 |
21 | print("Loading configuration...", file=sys.stderr, flush=True)
22 | try:
23 | config = load_config()
24 | except GoveeError as e:
25 | print(f"Configuration error: {e}", file=sys.stderr)
26 | sys.exit(1)
27 |
28 | print("Setting up tools...", file=sys.stderr, flush=True)
29 |
30 | @mcp.tool("turn_on_off")
31 | async def turn_on_off(power: bool) -> str:
32 | """
33 | Turn the LED on or off.
34 |
35 | Args:
36 | power: True for on, False for off
37 | """
38 | api = GoveeAPI(config)
39 | try:
40 | success, message = await api.set_power(power)
41 | await api.close() # Clean up the session
42 | return message if success else f"Failed: {message}"
43 | except GoveeError as e:
44 | await api.close()
45 | return f"Error: {str(e)}"
46 | except Exception as e:
47 | await api.close()
48 | return f"Unexpected error: {str(e)}"
49 |
50 | @mcp.tool("set_color")
51 | async def set_color(red: int, green: int, blue: int) -> str:
52 | """
53 | Set the LED color using RGB values.
54 |
55 | Args:
56 | red: Red value (0-255)
57 | green: Green value (0-255)
58 | blue: Blue value (0-255)
59 | """
60 | api = GoveeAPI(config)
61 | try:
62 | success, message = await api.set_color(red, green, blue)
63 | await api.close()
64 | return message if success else f"Failed: {message}"
65 | except GoveeError as e:
66 | await api.close()
67 | return f"Error: {str(e)}"
68 | except Exception as e:
69 | await api.close()
70 | return f"Unexpected error: {str(e)}"
71 |
72 | @mcp.tool("set_brightness")
73 | async def set_brightness(brightness: int) -> str:
74 | """
75 | Set the LED brightness.
76 |
77 | Args:
78 | brightness: Brightness level (0-100)
79 | """
80 | api = GoveeAPI(config)
81 | try:
82 | success, message = await api.set_brightness(brightness)
83 | await api.close()
84 | return message if success else f"Failed: {message}"
85 | except GoveeError as e:
86 | await api.close()
87 | return f"Error: {str(e)}"
88 | except Exception as e:
89 | await api.close()
90 | return f"Unexpected error: {str(e)}"
91 |
92 | @mcp.tool("get_status")
93 | async def get_status() -> dict:
94 | """Get the current status of the LED device."""
95 | api = GoveeAPI(config)
96 | try:
97 | power_state, power_msg = await api.get_power_state()
98 | color, color_msg = await api.get_color()
99 | brightness, bright_msg = await api.get_brightness()
100 | await api.close()
101 |
102 | return {
103 | "power": {
104 | "state": "on" if power_state else "off",
105 | "message": power_msg
106 | },
107 | "color": {
108 | "r": color[0],
109 | "g": color[1],
110 | "b": color[2],
111 | "message": color_msg
112 | },
113 | "brightness": {
114 | "level": brightness,
115 | "message": bright_msg
116 | }
117 | }
118 | except GoveeError as e:
119 | await api.close()
120 | return {"error": str(e)}
121 | except Exception as e:
122 | await api.close()
123 | return {"error": f"Unexpected error: {str(e)}"}
124 |
125 | async def handle_initialize(params):
126 | """Handle initialize request"""
127 | return {
128 | "protocolVersion": "0.1.0",
129 | "capabilities": mcp.capabilities
130 | }
131 |
132 | mcp.on_initialize = handle_initialize
133 |
134 | if __name__ == "__main__":
135 | try:
136 | import asyncio
137 | asyncio.run(mcp.run(transport='stdio'))
138 | except KeyboardInterrupt:
139 | print("\nServer stopped by user", file=sys.stderr)
140 | except Exception as e:
141 | print(f"Server error: {e}", file=sys.stderr)
142 | sys.exit(1)
```
--------------------------------------------------------------------------------
/src/govee_mcp_server/api.py:
--------------------------------------------------------------------------------
```python
1 | import aiohttp
2 | from typing import Optional, Dict, Any, Tuple
3 | import asyncio
4 | from time import time
5 | from .exceptions import (
6 | GoveeError,
7 | GoveeAPIError,
8 | GoveeConnectionError,
9 | GoveeTimeoutError
10 | )
11 | from .interfaces import PowerControl, ColorControl, BrightnessControl
12 | from .transformers import ColorTransformer
13 | from .config import GoveeConfig
14 |
15 | class GoveeAPI(PowerControl, ColorControl, BrightnessControl):
16 | """
17 | Govee API client implementing device control interfaces.
18 |
19 | Includes connection pooling, request timeouts, and retries.
20 | """
21 |
22 | BASE_URL = "https://openapi.api.govee.com"
23 | MAX_RETRIES = 3
24 | RETRY_DELAY = 1 # seconds
25 | REQUEST_TIMEOUT = 10 # seconds
26 |
27 | def __init__(self, config: GoveeConfig):
28 | """
29 | Initialize API client with configuration.
30 |
31 | Args:
32 | config: GoveeConfig instance with API credentials
33 | """
34 | self.config = config
35 | self.session: Optional[aiohttp.ClientSession] = None
36 | self._transformer = ColorTransformer()
37 |
38 | async def _ensure_session(self) -> None:
39 | """Ensure aiohttp session exists or create a new one."""
40 | if self.session is None or self.session.closed:
41 | self.session = aiohttp.ClientSession(
42 | headers={
43 | "Govee-API-Key": self.config.api_key,
44 | "Content-Type": "application/json"
45 | },
46 | timeout=aiohttp.ClientTimeout(total=self.REQUEST_TIMEOUT)
47 | )
48 |
49 | async def close(self) -> None:
50 | """Close the API session."""
51 | if self.session and not self.session.closed:
52 | await self.session.close()
53 |
54 | async def _make_request(
55 | self,
56 | method: str,
57 | endpoint: str,
58 | **kwargs
59 | ) -> Tuple[Dict[str, Any], str]:
60 | """
61 | Make HTTP request with retries and error handling.
62 |
63 | Args:
64 | method: HTTP method
65 | endpoint: API endpoint
66 | **kwargs: Additional request arguments
67 |
68 | Returns:
69 | Tuple[Dict[str, Any], str]: API response data and message
70 |
71 | Raises:
72 | GoveeAPIError: On API errors
73 | GoveeConnectionError: On connection issues
74 | GoveeTimeoutError: On request timeout
75 | """
76 | await self._ensure_session()
77 |
78 | for attempt in range(self.MAX_RETRIES):
79 | try:
80 | async with self.session.request(
81 | method,
82 | f"{self.BASE_URL}/{endpoint}",
83 | **kwargs
84 | ) as response:
85 | data = await response.json()
86 |
87 | if response.status != 200:
88 | raise GoveeAPIError(
89 | f"API error: {response.status} - {data.get('message', 'Unknown error')}"
90 | )
91 |
92 | return data, data.get('message', 'Success')
93 |
94 | except asyncio.TimeoutError:
95 | if attempt == self.MAX_RETRIES - 1:
96 | raise GoveeTimeoutError(f"Request timed out after {self.REQUEST_TIMEOUT}s")
97 | except aiohttp.ClientError as e:
98 | if attempt == self.MAX_RETRIES - 1:
99 | raise GoveeConnectionError(f"Connection error: {str(e)}")
100 |
101 | await asyncio.sleep(self.RETRY_DELAY * (attempt + 1))
102 |
103 | raise GoveeAPIError("Max retries exceeded")
104 |
105 | async def set_power(self, state: bool) -> Tuple[bool, str]:
106 | """Implement PowerControl.set_power"""
107 | try:
108 | _, message = await self._make_request(
109 | "POST",
110 | "router/api/v1/device/control",
111 | json={
112 | "requestId": str(int(time())), # Using timestamp as requestId
113 | "payload": {
114 | "sku": self.config.sku,
115 | "device": self.config.device_id,
116 | "capability": {
117 | "type": "devices.capabilities.on_off",
118 | "instance": "powerSwitch",
119 | "value": 1 if state else 0
120 | }
121 | }
122 | }
123 | )
124 | return True, message
125 | except GoveeError as e:
126 | return False, str(e)
127 |
128 | async def get_power_state(self) -> Tuple[bool, str]:
129 | """Implement PowerControl.get_power_state"""
130 | try:
131 | data, message = await self._make_request(
132 | "GET",
133 | f"devices/state",
134 | params={
135 | "device": self.config.device_id,
136 | "model": self.config.sku
137 | }
138 | )
139 | return data.get('powerState') == 'on', message
140 | except GoveeError as e:
141 | return False, str(e)
142 |
143 | async def set_color(self, r: int, g: int, b: int) -> Tuple[bool, str]:
144 | """Implement ColorControl.set_color"""
145 | try:
146 | color_value = ((r & 0xFF) << 16) | ((g & 0xFF) << 8) | (b & 0xFF)
147 |
148 | _, message = await self._make_request(
149 | "POST",
150 | "router/api/v1/device/control",
151 | json={
152 | "requestId": str(int(time())),
153 | "payload": {
154 | "sku": self.config.sku,
155 | "device": self.config.device_id,
156 | "capability": {
157 | "type": "devices.capabilities.color_setting",
158 | "instance": "colorRgb",
159 | "value": color_value
160 | }
161 | }
162 | }
163 | )
164 | return True, message
165 | except GoveeError as e:
166 | return False, str(e)
167 |
168 | async def get_color(self) -> Tuple[Tuple[int, int, int], str]:
169 | """Implement ColorControl.get_color"""
170 | try:
171 | data, message = await self._make_request(
172 | "GET",
173 | f"devices/state",
174 | params={
175 | "device": self.config.device_id,
176 | "model": self.config.sku
177 | }
178 | )
179 | color = data.get('color', {})
180 | return (
181 | color.get('r', 0),
182 | color.get('g', 0),
183 | color.get('b', 0)
184 | ), message
185 | except GoveeError as e:
186 | return (0, 0, 0), str(e)
187 |
188 | async def set_brightness(self, level: int) -> Tuple[bool, str]:
189 | """Implement BrightnessControl.set_brightness"""
190 | if not 0 <= level <= 100:
191 | return False, "Brightness must be between 0-100"
192 |
193 | try:
194 | _, message = await self._make_request(
195 | "POST",
196 | "router/api/v1/device/control",
197 | json={
198 | "requestId": str(int(time())),
199 | "payload": {
200 | "sku": self.config.sku,
201 | "device": self.config.device_id,
202 | "capability": {
203 | "type": "devices.capabilities.range",
204 | "instance": "brightness",
205 | "value": level
206 | }
207 | }
208 | }
209 | )
210 | return True, message
211 | except GoveeError as e:
212 | return False, str(e)
213 |
214 | async def get_brightness(self) -> Tuple[int, str]:
215 | """Implement BrightnessControl.get_brightness"""
216 | try:
217 | data, message = await self._make_request(
218 | "GET",
219 | f"devices/state",
220 | params={
221 | "device": self.config.device_id,
222 | "model": self.config.sku
223 | }
224 | )
225 | return data.get('brightness', 0), message
226 | except GoveeError as e:
227 | return 0, str(e)
```
--------------------------------------------------------------------------------
/documentation/01_configuration_management.md:
--------------------------------------------------------------------------------
```markdown
1 | # Chapter 1: Configuration Management - Giving Your App Its Credentials
2 |
3 | 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.
4 |
5 | ## What's the Big Idea?
6 |
7 | Imagine you want to send a letter. You need a few key things:
8 | 1. The **address** of the recipient (who are you sending it to?).
9 | 2. A **stamp** (proof you're allowed to send mail).
10 | 3. Maybe the **type of envelope** needed for that specific address.
11 |
12 | Our `govee_mcp_server` application is similar. To talk to your Govee lights, it needs to know:
13 | 1. **Which specific device** it should control (like the address).
14 | 2. Your **Govee API Key** (like a secret password or stamp proving it's allowed to talk to Govee).
15 | 3. The **device model (SKU)** (like knowing the type of envelope needed).
16 |
17 | "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.
18 |
19 | ## Storing Secrets: Environment Variables and `.env`
20 |
21 | 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**.
22 |
23 | 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.
24 |
25 | For convenience, especially during development, we often store these environment variables in a special file named `.env` located in the project's main folder.
26 |
27 | **Example `.env` file:**
28 |
29 | ```bash
30 | # This is a comment - lines starting with # are ignored
31 | GOVEE_API_KEY=abcdef12-3456-7890-abcd-ef1234567890
32 | GOVEE_DEVICE_ID=AB:CD:EF:12:34:56:78:90
33 | GOVEE_SKU=H6159
34 | ```
35 |
36 | * `GOVEE_API_KEY`: Your personal key from Govee. Keep this secret!
37 | * `GOVEE_DEVICE_ID`: The unique identifier for your specific Govee light.
38 | * `GOVEE_SKU`: The model number of your Govee light.
39 |
40 | **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).
41 |
42 | ## Loading the Settings: The `load_config` Function
43 |
44 | 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`.
45 |
46 | Let's look at the heart of this process in `src/govee_mcp_server/config.py`:
47 |
48 | ```python
49 | # Simplified snippet from src/govee_mcp_server/config.py
50 |
51 | import os
52 | from dotenv import load_dotenv
53 | from pathlib import Path
54 |
55 | # ... (GoveeConfig class defined elsewhere) ...
56 | # ... (GoveeConfigError class defined elsewhere) ...
57 |
58 | def load_config():
59 | # Find and load the .env file
60 | env_path = Path(__file__).resolve().parent.parent.parent / '.env'
61 | load_dotenv(dotenv_path=env_path) # Reads .env into environment
62 |
63 | # Read values from the environment
64 | api_key = os.getenv('GOVEE_API_KEY')
65 | device_id = os.getenv('GOVEE_DEVICE_ID')
66 | sku = os.getenv('GOVEE_SKU')
67 |
68 | # Check if any are missing
69 | if not api_key or not device_id or not sku:
70 | # If something is missing, raise an error!
71 | raise GoveeConfigError("Missing required environment variables!")
72 |
73 | # If all good, package them up and return
74 | return GoveeConfig(api_key=api_key, device_id=device_id, sku=sku)
75 | ```
76 |
77 | **Explanation:**
78 | 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.
79 | 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`.
80 | 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.
81 | 4. `return GoveeConfig(...)`: If all values are found, they are bundled neatly into a `GoveeConfig` object (we'll see this next) and returned.
82 |
83 | ## A Tidy Package: The `GoveeConfig` Object
84 |
85 | 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`.
86 |
87 | ```python
88 | # Simplified snippet from src/govee_mcp_server/config.py
89 | from dataclasses import dataclass
90 |
91 | @dataclass
92 | class GoveeConfig:
93 | """Configuration class for Govee API settings."""
94 | api_key: str
95 | device_id: str
96 | sku: str
97 | ```
98 |
99 | **Explanation:**
100 | * `@dataclass`: This is a shortcut in Python to create simple classes that mostly just hold data.
101 | * `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.
102 |
103 | **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.
104 |
105 | ## Using the Configuration
106 |
107 | 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.
108 |
109 | Here's a simplified idea of how it's used in `src/govee_mcp_server/api.py`:
110 |
111 | ```python
112 | # Simplified concept from src/govee_mcp_server/api.py
113 |
114 | from .config import GoveeConfig # Import the config class
115 |
116 | class GoveeAPI:
117 | # The __init__ method runs when a GoveeAPI object is created
118 | def __init__(self, config: GoveeConfig):
119 | """Initialize API client with configuration."""
120 | self.config = config # Store the passed-in config object
121 | # Now, self.config.api_key, self.config.device_id,
122 | # and self.config.sku are available inside this class.
123 |
124 | async def _make_request(self, ...):
125 | # When making a real request to Govee...
126 | api_key = self.config.api_key # <-- Access the stored key
127 | headers = {"Govee-API-Key": api_key, ...}
128 | # ... use the key in the request headers ...
129 |
130 | async def set_power(self, state: bool):
131 | # When controlling the device...
132 | device = self.config.device_id # <-- Access the stored ID
133 | sku = self.config.sku # <-- Access the stored SKU
134 | payload = {
135 | "sku": sku,
136 | "device": device,
137 | # ... other control details ...
138 | }
139 | # ... use the ID and SKU in the request body ...
140 | ```
141 |
142 | **Explanation:**
143 | * The `GoveeAPI` class takes the `GoveeConfig` object when it's created.
144 | * It stores this `config` object internally.
145 | * 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`).
146 |
147 | ## Under the Hood: How Loading Works
148 |
149 | Let's trace the steps when the application starts and needs its configuration:
150 |
151 | ```mermaid
152 | sequenceDiagram
153 | participant App as Application (CLI/Server)
154 | participant LC as load_config()
155 | participant DotEnv as .env File
156 | participant OS as Operating System Env
157 | participant GC as GoveeConfig Object
158 |
159 | App->>LC: Start! Need config. Call load_config()
160 | LC->>DotEnv: Look for .env file in project root
161 | Note over DotEnv: Contains GOVEE_API_KEY=... etc.
162 | DotEnv-->>LC: Load variables found into OS Env
163 | LC->>OS: Read 'GOVEE_API_KEY' variable
164 | OS-->>LC: Return 'abcdef...' (value)
165 | LC->>OS: Read 'GOVEE_DEVICE_ID' variable
166 | OS-->>LC: Return 'AB:CD:...' (value)
167 | LC->>OS: Read 'GOVEE_SKU' variable
168 | OS-->>LC: Return 'H6159' (value)
169 | LC->>LC: Check: Are all values present? (Yes)
170 | LC->>GC: Create GoveeConfig object with these values
171 | GC-->>LC: Here's the new object
172 | LC-->>App: Return the filled GoveeConfig object
173 | ```
174 |
175 | This sequence shows how the application reliably gets its necessary startup information from the environment, validates it, and packages it for easy use.
176 |
177 | ## Conclusion
178 |
179 | You've learned about the first crucial step: **Configuration Management**. You now know:
180 |
181 | * Why applications need configuration (like API keys and device IDs).
182 | * How `govee_mcp_server` uses environment variables and `.env` files to store these settings securely.
183 | * How the `load_config` function reads and validates these settings.
184 | * How the `GoveeConfig` object provides a tidy way to access these settings throughout the application.
185 |
186 | 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.
187 |
188 | 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.
189 |
190 | ---
191 |
192 | Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/02_device_control_interfaces.md:
--------------------------------------------------------------------------------
```markdown
1 | # Chapter 2: Device Control Interfaces - The Standard Sockets for Lights
2 |
3 | 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.
4 |
5 | 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.
6 |
7 | ## What's the Big Idea? Standardizing Actions
8 |
9 | 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!
10 |
11 | This is where **Device Control Interfaces** come in. They act like **standard electrical sockets** or **blueprints** for device abilities.
12 |
13 | * **Power Socket:** Defines *how* to plug something in to get power (e.g., Turn On/Off).
14 | * **Color Socket:** Defines *how* to tell a device which color to display.
15 | * **Brightness Socket:** Defines *how* to adjust the brightness level.
16 |
17 | 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.
18 |
19 | ## The Blueprints: `PowerControl`, `ColorControl`, `BrightnessControl`
20 |
21 | 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`:
22 |
23 | ```python
24 | # Simplified from src/govee_mcp_server/interfaces.py
25 | from abc import ABC, abstractmethod
26 | from typing import Tuple # Used to describe the function's output type
27 |
28 | class PowerControl(ABC): # ABC means "This is a blueprint/interface"
29 | """Blueprint for turning things on/off."""
30 |
31 | @abstractmethod # Means: Any class using this blueprint MUST provide this method
32 | async def set_power(self, state: bool) -> Tuple[bool, str]:
33 | """
34 | Blueprint for a function to turn the device on (True) or off (False).
35 |
36 | It MUST take 'state' (True/False) as input.
37 | It MUST return a pair: (was it successful?, a message).
38 | """
39 | pass # The blueprint only defines *what*, not *how*. Implementations fill this.
40 |
41 | @abstractmethod
42 | async def get_power_state(self) -> Tuple[bool, str]:
43 | """
44 | Blueprint for checking if the device is currently on.
45 |
46 | It MUST return a pair: (is it on?, a message).
47 | """
48 | pass
49 | ```
50 |
51 | **Explanation:**
52 | * `class PowerControl(ABC):`: Defines a blueprint named `PowerControl`. `ABC` marks it as an abstract blueprint, not a real, usable object on its own.
53 | * `@abstractmethod`: This marker says "Any real class that claims to be a `PowerControl` *must* provide its own version of the function below."
54 | * `async def set_power(...)`: Defines the *signature* of the `set_power` function:
55 | * `async`: It's designed to work asynchronously (we'll see why in later chapters).
56 | * `self`: A standard reference to the object itself.
57 | * `state: bool`: It must accept one input argument named `state`, which must be a boolean (`True` or `False`).
58 | * `-> Tuple[bool, str]`: It must return a "tuple" (an ordered pair) containing a boolean (success/failure) and a string (a status message).
59 | * `pass`: In the blueprint, the methods don't actually *do* anything. They just define the requirements.
60 |
61 | Similarly, we have `ColorControl` (requiring `set_color`, `get_color`) and `BrightnessControl` (requiring `set_brightness`, `get_brightness`).
62 |
63 | ## Using the Blueprints: The `GoveeAPI` Class
64 |
65 | 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`).
66 |
67 | Look at the very first line defining this class:
68 |
69 | ```python
70 | # Simplified from src/govee_mcp_server/api.py
71 | from .interfaces import PowerControl, ColorControl, BrightnessControl
72 | # ... other imports ...
73 | from .config import GoveeConfig
74 |
75 | class GoveeAPI(PowerControl, ColorControl, BrightnessControl):
76 | """
77 | Govee API client that PROMISES to follow the rules
78 | of PowerControl, ColorControl, and BrightnessControl.
79 | """
80 | def __init__(self, config: GoveeConfig):
81 | """Gets the API key and device details when created."""
82 | self.config = config
83 | # ... other setup ...
84 |
85 | # --- Implementing PowerControl ---
86 | async def set_power(self, state: bool) -> Tuple[bool, str]:
87 | """Turns the actual Govee device on/off using its API."""
88 | print(f"Actually telling Govee API to set power: {state}")
89 | # ... code to make the real web request to Govee ...
90 | # (We'll look inside this in Chapter 3!)
91 | success = True # Let's assume it worked for now
92 | message = f"Device power set to {state} via API"
93 | return success, message
94 |
95 | async def get_power_state(self) -> Tuple[bool, str]:
96 | """Asks the Govee API if the device is on."""
97 | print("Actually asking Govee API for power state")
98 | # ... code to make the real web request to Govee ...
99 | is_on = True # Let's pretend it's on
100 | message = "Device is currently ON (from API)"
101 | return is_on, message
102 |
103 | # --- Implementing ColorControl (methods like set_color) ---
104 | # ... implementations for set_color, get_color ...
105 |
106 | # --- Implementing BrightnessControl (methods like set_brightness) ---
107 | # ... implementations for set_brightness, get_brightness ...
108 |
109 | ```
110 |
111 | **Explanation:**
112 | * `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.
113 | * 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!
114 | * 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.
115 |
116 | **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.
117 |
118 | ## Benefits: Why Is This Structure Useful?
119 |
120 | 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.
121 | 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.
122 | 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.
123 |
124 | ## Safety First: Input Validation with Decorators
125 |
126 | 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.
127 |
128 | 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.
129 |
130 | In `src/govee_mcp_server/interfaces.py`, we have a `validate_rgb` decorator:
131 |
132 | ```python
133 | # Simplified from src/govee_mcp_server/interfaces.py
134 | from .exceptions import GoveeValidationError # Our custom error for bad input
135 |
136 | def validate_rgb(func):
137 | """A function that wraps another function to check RGB values first."""
138 | async def wrapper(self, r: int, g: int, b: int, *args, **kwargs):
139 | print(f"VALIDATOR: Checking RGB ({r}, {g}, {b})") # Show the check
140 | is_valid = True
141 | for name, value in [('Red', r), ('Green', g), ('Blue', b)]:
142 | if not (isinstance(value, int) and 0 <= value <= 255):
143 | print(f"VALIDATOR: Invalid value for {name}: {value}")
144 | is_valid = False
145 | # Stop! Raise an error instead of calling the real function.
146 | raise GoveeValidationError(f"{name} value must be an integer 0-255")
147 |
148 | if is_valid:
149 | # If all checks pass, call the original function (like set_color)
150 | print("VALIDATOR: Looks good! Proceeding.")
151 | return await func(self, r, g, b, *args, **kwargs)
152 | return wrapper
153 | ```
154 |
155 | And we apply this decorator to the `set_color` blueprint in `ColorControl`:
156 |
157 | ```python
158 | # Simplified from src/govee_mcp_server/interfaces.py
159 | # ... imports ...
160 |
161 | class ColorControl(ABC):
162 | @abstractmethod
163 | @validate_rgb # <-- Apply the validator HERE
164 | async def set_color(self, r: int, g: int, b: int) -> Tuple[bool, str]:
165 | """
166 | Set device color. The @validate_rgb ensures values are checked
167 | BEFORE the actual implementation in GoveeAPI is even called.
168 | """
169 | pass
170 |
171 | # ... get_color method ...
172 | ```
173 |
174 | **Explanation:**
175 | * `@validate_rgb`: This line attaches the `validate_rgb` wrapper to the `set_color` method definition in the blueprint.
176 | * Now, whenever *any* implementation of `set_color` (like the one in `GoveeAPI`) is called, the `validate_rgb` code runs *first*.
177 | * 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.
178 |
179 | **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.
180 |
181 | ## How it Works Under the Hood
182 |
183 | Let's trace what happens when some user code tries to turn the light on using our `GoveeAPI` object, which implements the `PowerControl` interface:
184 |
185 | ```mermaid
186 | sequenceDiagram
187 | participant UserCode as Your Code
188 | participant GoveeObj as GoveeAPI Object (implements PowerControl)
189 | participant PCInterface as PowerControl Blueprint
190 | participant GoveeCloud as Govee Cloud Service
191 |
192 | UserCode->>GoveeObj: Call `set_power(True)`
193 | Note over GoveeObj,PCInterface: GoveeAPI promised to have `set_power` because it implements PowerControl.
194 | GoveeObj->>GoveeObj: Execute its own `set_power` code.
195 | GoveeObj->>GoveeCloud: Send command: { device: '...', sku: '...', capability: 'power', value: 1 }
196 | GoveeCloud-->>GoveeObj: Respond: { status: 200, message: 'Success' }
197 | GoveeObj->>GoveeObj: Process the response.
198 | GoveeObj-->>UserCode: Return `(True, "Device power set to True via API")`
199 | ```
200 |
201 | 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`.
202 |
203 | ## Conclusion
204 |
205 | 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.
206 |
207 | * Interfaces (`PowerControl`, `ColorControl`, etc.) act as **contracts** or **standard sockets**.
208 | * They use `ABC` and `@abstractmethod` to define required functions (like `set_power`, `get_color`).
209 | * Classes like `GoveeAPI` **implement** these interfaces, promising to provide the required functions.
210 | * This brings **consistency**, **flexibility**, and **testability** to our code.
211 | * **Decorators** like `@validate_rgb` can be attached to interface methods to add checks (like input validation) automatically.
212 |
213 | 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.
214 |
215 | Get ready to explore the engine room in [Chapter 3: Govee API Client](03_govee_api_client.md)!
216 |
217 | ---
218 |
219 | Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/05_mcp_server_implementation.md:
--------------------------------------------------------------------------------
```markdown
1 | # Chapter 5: MCP Server Implementation - The Universal Remote Receiver
2 |
3 | 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.
4 |
5 | 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.
6 |
7 | This is where the **MCP Server** comes in. It's the core of our `govee_mcp_server` project!
8 |
9 | ## What's the Big Idea? A Standard Interface for Programs
10 |
11 | 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!
12 |
13 | 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:
14 | 1. Ask another program (our **MCP Server**) what "tools" it offers.
15 | 2. Ask the server to use one of those tools with specific instructions.
16 |
17 | 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.
18 |
19 | **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.
20 |
21 | ## How MCP Works: Tools and Communication
22 |
23 | 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`."
24 | 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`."
25 | 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.
26 | 4. **Response:** The server sends a structured message back to the client, like, "Okay, I executed `set_color`. The result was: 'Success'."
27 |
28 | 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.
29 |
30 | ## Implementing the Server (`src/govee_mcp_server/server.py`)
31 |
32 | The main logic for our MCP server lives in the `src/govee_mcp_server/server.py` file. Let's look at the key parts.
33 |
34 | **1. Setting Up the Server**
35 |
36 | First, we import the necessary pieces and create the `FastMCP` server instance.
37 |
38 | ```python
39 | # Simplified from src/govee_mcp_server/server.py
40 | import sys
41 | from mcp.server.fastmcp import FastMCP # The MCP library
42 | from govee_mcp_server.config import load_config # To get API key etc.
43 | from govee_mcp_server.api import GoveeAPI # To control the light
44 | from govee_mcp_server.exceptions import GoveeError # Our custom errors
45 |
46 | # Initialize the MCP server
47 | mcp = FastMCP(
48 | "govee", # A simple name for our server
49 | # Define some basic info about the server
50 | capabilities={
51 | "server_info": { "name": "govee-mcp", "version": "0.1.0" }
52 | },
53 | log_level='WARNING' # Keep logs tidy
54 | )
55 | ```
56 |
57 | **Explanation:**
58 | * We import `FastMCP`, our configuration loader, and the `GoveeAPI` client.
59 | * `mcp = FastMCP(...)` creates the server object. We give it a name (`"govee"`) and some basic information that clients might ask for.
60 |
61 | **2. Loading Configuration**
62 |
63 | The server needs the API key and device details, just like the CLI did.
64 |
65 | ```python
66 | # Simplified from src/govee_mcp_server/server.py
67 |
68 | print("Loading configuration...", file=sys.stderr)
69 | try:
70 | # Load API Key, Device ID, SKU from .env or environment
71 | config = load_config() # Uses the function from Chapter 1
72 | except GoveeError as e:
73 | print(f"Configuration error: {e}", file=sys.stderr)
74 | sys.exit(1) # Stop if config fails
75 |
76 | # 'config' now holds our GoveeConfig object
77 | ```
78 |
79 | **Explanation:**
80 | * This uses the `load_config` function from [Chapter 1: Configuration Management](01_configuration_management.md) to get the necessary credentials.
81 | * If loading fails (e.g., missing `.env` file), it prints an error and stops.
82 |
83 | **3. Defining the Tools**
84 |
85 | This is where we define the actions the server can perform. We use the `@mcp.tool()` decorator to register functions as MCP tools.
86 |
87 | ```python
88 | # Simplified from src/govee_mcp_server/server.py
89 |
90 | print("Setting up tools...", file=sys.stderr)
91 |
92 | # --- Tool: turn_on_off ---
93 | @mcp.tool("turn_on_off") # Register this function as an MCP tool
94 | async def turn_on_off(power: bool) -> str:
95 | """Turn the LED on (True) or off (False)."""
96 | api = GoveeAPI(config) # Create API client *for this request*
97 | try:
98 | # Use the API client to send the command
99 | success, message = await api.set_power(power)
100 | # Return a simple status message
101 | return message if success else f"Failed: {message}"
102 | except GoveeError as e:
103 | return f"Error: {str(e)}" # Handle expected errors
104 | finally:
105 | await api.close() # IMPORTANT: Clean up the connection
106 |
107 | # --- Tool: set_color ---
108 | @mcp.tool("set_color")
109 | async def set_color(red: int, green: int, blue: int) -> str:
110 | """Set the LED color using RGB (0-255)."""
111 | api = GoveeAPI(config)
112 | try:
113 | # Use the API client to set the color
114 | success, message = await api.set_color(red, green, blue)
115 | return message if success else f"Failed: {message}"
116 | except GoveeError as e:
117 | return f"Error: {str(e)}"
118 | finally:
119 | await api.close()
120 |
121 | # --- Other tools (set_brightness, get_status) defined similarly ---
122 | # @mcp.tool("set_brightness")
123 | # async def set_brightness(brightness: int) -> str: ...
124 | # @mcp.tool("get_status")
125 | # async def get_status() -> dict: ...
126 | ```
127 |
128 | **Explanation:**
129 | * `@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(...)`)."
130 | * `async def turn_on_off(power: bool) -> str:`: Defines the function that handles the `turn_on_off` tool.
131 | * It takes `power` (a boolean: `True` or `False`) as input, just as the client specified.
132 | * It's declared `async` because it uses `await` to call the asynchronous `GoveeAPI` methods.
133 | * It's defined to return a `str` (a string message indicating success or failure).
134 | * **Inside the Tool:**
135 | 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.
136 | 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.
137 | 3. `return ...`: It returns a simple string message back to the MCP framework.
138 | 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.
139 | * 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.
140 |
141 | **4. Running the Server**
142 |
143 | Finally, we need code to actually start the server and make it listen for incoming MCP requests.
144 |
145 | ```python
146 | # Simplified from src/govee_mcp_server/server.py
147 | import asyncio
148 |
149 | # ... other code above ...
150 |
151 | if __name__ == "__main__": # Standard Python way to run code directly
152 | try:
153 | print("Starting MCP server...", file=sys.stderr)
154 | # Tell FastMCP to run, listening via standard input/output
155 | asyncio.run(mcp.run(transport='stdio'))
156 | except KeyboardInterrupt:
157 | # Allow stopping the server with Ctrl+C
158 | print("\nServer stopped by user", file=sys.stderr)
159 | except Exception as e:
160 | # Catch unexpected server errors
161 | print(f"Server error: {e}", file=sys.stderr)
162 | sys.exit(1)
163 | ```
164 |
165 | **Explanation:**
166 | * `if __name__ == "__main__":`: This code runs only when you execute the `server.py` script directly (e.g., `python -m govee_mcp_server.server`).
167 | * `asyncio.run(mcp.run(transport='stdio'))`: This is the command that starts the `FastMCP` server.
168 | * `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).
169 | * `asyncio.run(...)` starts the asynchronous event loop needed for `async`/`await` functions.
170 | * The `try...except` blocks handle stopping the server cleanly (Ctrl+C) or catching unexpected crashes.
171 |
172 | ## Under the Hood: A Client Calls `set_brightness(50)`
173 |
174 | Let's trace what happens when an MCP client sends a request:
175 |
176 | 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`."
177 | 2. **FastMCP Receives:** The `mcp.run()` loop in our server reads this message from its standard input.
178 | 3. **FastMCP Parses:** The `FastMCP` library parses the message, identifies the target tool (`set_brightness`), and extracts the arguments (`brightness=50`).
179 | 4. **FastMCP Dispatches:** It finds the Python function registered with `@mcp.tool("set_brightness")`, which is our `async def set_brightness(brightness: int)` function.
180 | 5. **Tool Function Executes:** `FastMCP` calls `await set_brightness(brightness=50)`.
181 | * Inside `set_brightness`:
182 | * `api = GoveeAPI(config)`: Creates a new Govee API client.
183 | * `await api.set_brightness(50)`: Calls the API client method.
184 | * The `GoveeAPI` client builds the JSON request, adds the API key, sends it to the Govee cloud, gets the response (e.g., "Success").
185 | * The API client returns `(True, "Success")` to the `set_brightness` function.
186 | * The function prepares the return string: `"Success"`.
187 | * `finally: await api.close()`: The API client connection is closed.
188 | * The function `return`s the string `"Success"`.
189 | 6. **FastMCP Receives Result:** `FastMCP` gets the `"Success"` string back from the tool function.
190 | 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.
191 | 8. **Client Receives Response:** The MCP client reads the response from its standard input and knows the command was successful.
192 |
193 | Here's a diagram of that flow:
194 |
195 | ```mermaid
196 | sequenceDiagram
197 | participant Client as MCP Client
198 | participant FastMCP as FastMCP Library (in server.py)
199 | participant ToolFunc as set_brightness() Tool
200 | participant APIClient as GoveeAPI Client
201 | participant GoveeCloud as Govee Cloud
202 |
203 | Client->>+FastMCP: MCP Request: execute 'set_brightness', brightness=50
204 | FastMCP->>+ToolFunc: await set_brightness(brightness=50)
205 | ToolFunc->>APIClient: Create GoveeAPI(config)
206 | ToolFunc->>+APIClient: await set_brightness(50)
207 | APIClient->>GoveeCloud: Send 'Set Brightness=50' request
208 | GoveeCloud-->>APIClient: Respond ('Success')
209 | APIClient-->>-ToolFunc: Return (True, "Success")
210 | ToolFunc->>APIClient: await close()
211 | ToolFunc-->>-FastMCP: Return "Success"
212 | FastMCP-->>-Client: MCP Response: result="Success"
213 |
214 | ```
215 |
216 | ## Conclusion
217 |
218 | You've now learned how the **MCP Server Implementation** works in `govee_mcp_server`:
219 |
220 | * 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).
221 | * It uses the **Mission Control Protocol (MCP)** standard for communication.
222 | * It leverages the `FastMCP` library to handle incoming requests and outgoing responses.
223 | * It defines specific **tools** (`turn_on_off`, `set_color`, etc.) using the `@mcp.tool` decorator.
224 | * 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.
225 | * Proper cleanup (closing the API client) is essential within each tool.
226 |
227 | 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.
228 |
229 | Let's explore how we manage errors gracefully in [Chapter 6: Custom Error Handling](06_custom_error_handling.md).
230 |
231 | ---
232 |
233 | Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/04_command_line_interface__cli_.md:
--------------------------------------------------------------------------------
```markdown
1 | # Chapter 4: Command Line Interface (CLI) - Your Simple Remote Control
2 |
3 | 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).
4 |
5 | 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**.
6 |
7 | ## What's the Big Idea? Direct Control from Your Terminal
8 |
9 | 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.
10 |
11 | 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).
12 |
13 | Instead of needing a full graphical application or starting the main server, you can just type something like:
14 |
15 | ```bash
16 | govee-cli power on
17 | ```
18 |
19 | or
20 |
21 | ```bash
22 | govee-cli color 255 0 0 # Set color to Red
23 | ```
24 |
25 | This provides a fast and straightforward way to manually control your light or even use it in simple scripts.
26 |
27 | ## How Does It Work? Typing Commands
28 |
29 | When you use a CLI tool, here's the basic flow:
30 | 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).
31 | * `govee-cli` is the program name.
32 | * `power` or `color` is the command (the action).
33 | * `on` or `255 0 0` are the arguments (the details for the action).
34 | 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.
35 | 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.
36 | 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.
37 |
38 | ## Using the `govee-cli`
39 |
40 | 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).
41 |
42 | **(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.)**
43 |
44 | 1. **Get Help:** See what commands are available.
45 | ```bash
46 | govee-cli --help
47 | ```
48 | * **Output:** Shows a list of commands like `power`, `color`, `brightness`, `status` and how to use them.
49 |
50 | 2. **Turn Power On:**
51 | ```bash
52 | govee-cli power on
53 | ```
54 | * **Output (if successful):** `Success` (or a similar message from the Govee API). Your light should turn on!
55 |
56 | 3. **Turn Power Off:**
57 | ```bash
58 | govee-cli power off
59 | ```
60 | * **Output (if successful):** `Success`. Your light should turn off.
61 |
62 | 4. **Set Color (Red):**
63 | ```bash
64 | govee-cli color 255 0 0
65 | ```
66 | * **Output (if successful):** `Success`. Your light should turn red.
67 | * **Note:** The arguments are Red, Green, Blue values (0-255).
68 |
69 | 5. **Set Brightness (50%):**
70 | ```bash
71 | govee-cli brightness 50
72 | ```
73 | * **Output (if successful):** `Success`. Your light's brightness should change.
74 | * **Note:** The argument is the brightness percentage (0-100).
75 |
76 | 6. **Check Status:** Get the current state of the light.
77 | ```bash
78 | govee-cli status
79 | ```
80 | * **Output (example):**
81 | ```
82 | Power: ON
83 | Color: RGB(255, 0, 0)
84 | Brightness: 50%
85 | ```
86 |
87 | 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.
88 |
89 | ## Under the Hood: What Happens When You Type `govee-cli power on`?
90 |
91 | Let's trace the steps behind the scenes:
92 |
93 | 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.
94 | 2. **CLI Script Starts:** The Python script `src/govee_mcp_server/cli.py` begins running.
95 | 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`."
96 | 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).
97 | 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=...)`
98 | 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`).
99 | 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.
100 | 8. **Get Result:** The `set_power` method returns a result (e.g., `(True, "Success")`) back to the CLI script.
101 | 9. **Print Output:** The CLI script takes the result message ("Success") and prints it to your terminal.
102 | 10. **Cleanup:** The script makes sure to close the network connection used by the `GoveeAPI` client (`await api_client.close()`).
103 |
104 | Here's a diagram showing this flow:
105 |
106 | ```mermaid
107 | sequenceDiagram
108 | participant User as You (Terminal)
109 | participant CLI as govee-cli Script
110 | participant Parser as Argument Parser (argparse)
111 | participant Config as Config Loader
112 | participant APIClient as GoveeAPI Client
113 | participant GoveeCloud as Govee Cloud
114 |
115 | User->>CLI: Run `govee-cli power on`
116 | CLI->>Parser: Parse ["power", "on"]
117 | Parser-->>CLI: Command='power', State='on'
118 | CLI->>Config: Load config()
119 | Config-->>CLI: Return GoveeConfig (API Key, etc.)
120 | CLI->>APIClient: Create GoveeAPI(config)
121 | CLI->>APIClient: Call `await set_power(True)`
122 | APIClient->>GoveeCloud: Send 'Power On' request (with API Key)
123 | GoveeCloud-->>APIClient: Respond ('Success')
124 | APIClient-->>CLI: Return (True, "Success")
125 | CLI->>User: Print "Success"
126 | CLI->>APIClient: Call `await close()`
127 | APIClient->>APIClient: Close network connection
128 | ```
129 |
130 | ## Inside the Code: `src/govee_mcp_server/cli.py`
131 |
132 | Let's look at simplified parts of the code that make this work.
133 |
134 | **1. Defining the Commands (`create_parser`)**
135 |
136 | This function sets up `argparse` to understand the commands and arguments.
137 |
138 | ```python
139 | # Simplified from src/govee_mcp_server/cli.py
140 | import argparse
141 |
142 | def create_parser() -> argparse.ArgumentParser:
143 | """Create and configure argument parser."""
144 | parser = argparse.ArgumentParser(description='Control Govee LED device')
145 | # Create categories for commands (like 'power', 'color')
146 | subparsers = parser.add_subparsers(dest='command', help='Commands', required=True)
147 |
148 | # --- Define the 'power' command ---
149 | power_parser = subparsers.add_parser('power', help='Turn device on/off')
150 | # It needs one argument: 'state', which must be 'on' or 'off'
151 | power_parser.add_argument('state', choices=['on', 'off'], help='Power state')
152 |
153 | # --- Define the 'color' command ---
154 | color_parser = subparsers.add_parser('color', help='Set device color')
155 | # It needs three integer arguments: red, green, blue
156 | color_parser.add_argument('red', type=int, help='Red value (0-255)')
157 | color_parser.add_argument('green', type=int, help='Green value (0-255)')
158 | color_parser.add_argument('blue', type=int, help='Blue value (0-255)')
159 |
160 | # --- Define other commands similarly (brightness, status) ---
161 | # ... parser setup for brightness ...
162 | # ... parser setup for status ...
163 |
164 | return parser
165 | ```
166 |
167 | **Explanation:**
168 | * `argparse.ArgumentParser`: Creates the main parser object.
169 | * `parser.add_subparsers()`: Allows us to define distinct commands (like `git commit`, `git push`). `dest='command'` stores which command was used.
170 | * `subparsers.add_parser('power', ...)`: Defines the `power` command.
171 | * `power_parser.add_argument('state', ...)`: Specifies that the `power` command requires an argument named `state`, which must be either the text `"on"` or `"off"`.
172 | * Similarly, it defines the `color` command and its required `red`, `green`, `blue` integer arguments.
173 |
174 | **2. Handling a Specific Command (`handle_power`)**
175 |
176 | Each command has a small function to handle its logic.
177 |
178 | ```python
179 | # Simplified from src/govee_mcp_server/cli.py
180 | from .api import GoveeAPI # We need the API client class
181 | from .exceptions import GoveeError # We need our custom error
182 |
183 | async def handle_power(api: GoveeAPI, state: str) -> None:
184 | """Handle the 'power' command."""
185 | # Convert 'on'/'off' string to True/False boolean
186 | is_on = (state == 'on')
187 |
188 | print(f"Sending command: Power {'ON' if is_on else 'OFF'}...")
189 | # Call the actual API client method
190 | success, message = await api.set_power(is_on)
191 |
192 | # Check the result from the API client
193 | if success:
194 | print(f"Result: {message}")
195 | else:
196 | # If the API client reported failure, raise an error to stop the script
197 | print(f"Error reported by API: {message}")
198 | raise GoveeError(f"Failed to set power: {message}")
199 | ```
200 |
201 | **Explanation:**
202 | * Takes the `api` client object and the parsed `state` ("on" or "off") as input.
203 | * Converts the string `state` into a boolean `is_on` (`True` or `False`).
204 | * 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).
205 | * Prints the result or raises an error if the API call failed.
206 |
207 | **3. The Main Script Logic (`main` function)**
208 |
209 | This function ties everything together: load config, parse args, call the right handler.
210 |
211 | ```python
212 | # Simplified from src/govee_mcp_server/cli.py
213 | import sys
214 | import asyncio
215 | from .config import load_config
216 | from .api import GoveeAPI
217 | from .exceptions import GoveeError
218 |
219 | # ... create_parser() defined above ...
220 | # ... handle_power(), handle_color(), etc. defined above ...
221 |
222 | async def main() -> None:
223 | """Main CLI entrypoint."""
224 | api = None # Initialize api variable
225 | try:
226 | # 1. Load configuration
227 | config = load_config()
228 | print(f"Loaded config for device: {config.device_id}")
229 |
230 | # 2. Create the Govee API Client
231 | api = GoveeAPI(config)
232 | print("GoveeAPI client ready.")
233 |
234 | # 3. Parse command line arguments
235 | parser = create_parser()
236 | args = parser.parse_args() # Reads sys.argv (what you typed)
237 | print(f"Executing command: {args.command}")
238 |
239 | # 4. Call the appropriate handler based on the command
240 | if args.command == 'power':
241 | await handle_power(api, args.state)
242 | elif args.command == 'color':
243 | await handle_color(api, args.red, args.green, args.blue)
244 | # ... elif for brightness ...
245 | # ... elif for status ...
246 | else:
247 | # Should not happen if parser is configured correctly
248 | print(f"Unknown command: {args.command}")
249 | parser.print_help()
250 | sys.exit(1) # Exit with an error code
251 |
252 | except GoveeError as e:
253 | # Catch errors from our API client or handlers
254 | print(f"\nOperation Failed: {str(e)}")
255 | sys.exit(1) # Exit with an error code
256 | except Exception as e:
257 | # Catch any other unexpected errors
258 | print(f"\nAn Unexpected Error Occurred: {str(e)}")
259 | sys.exit(1) # Exit with an error code
260 | finally:
261 | # 5. ALWAYS try to clean up the API client connection
262 | if api:
263 | print("Closing API connection...")
264 | await api.close()
265 | print("Connection closed.")
266 |
267 | # --- This part runs the main async function ---
268 | # def cli_main():
269 | # asyncio.run(main())
270 | # if __name__ == "__main__":
271 | # cli_main()
272 | ```
273 |
274 | **Explanation:**
275 | 1. Calls `load_config()` to get settings.
276 | 2. Creates the `GoveeAPI` instance.
277 | 3. Calls `parser.parse_args()` which uses `argparse` to figure out which command and arguments were given.
278 | 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`).
279 | 5. Includes `try...except` blocks to catch potential errors (like configuration errors, API errors, or validation errors) and print user-friendly messages before exiting.
280 | 6. The `finally` block ensures that `await api.close()` is called to release network resources, even if an error occurred.
281 |
282 | ## Conclusion
283 |
284 | You've learned about the **Command Line Interface (CLI)**, a simple way to directly control your Govee device from the terminal.
285 |
286 | * It acts like a **basic remote control**.
287 | * You use it by typing commands like `govee-cli power on`.
288 | * It **parses** your command and arguments using `argparse`.
289 | * Crucially, it uses the **same `GoveeAPI` client** from [Chapter 3: Govee API Client](03_govee_api_client.md) to interact with the Govee Cloud.
290 | * It provides immediate feedback in your terminal.
291 |
292 | 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?
293 |
294 | 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).
295 |
296 | ---
297 |
298 | Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/06_custom_error_handling.md:
--------------------------------------------------------------------------------
```markdown
1 | # Chapter 6: Custom Error Handling - Specific Warning Lights
2 |
3 | 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.
4 |
5 | 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.
6 |
7 | This chapter is about how `govee_mcp_server` uses **Custom Error Handling** to provide specific, helpful information when problems occur.
8 |
9 | ## What's the Big Idea? Knowing *Why* Things Broke
10 |
11 | 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.
12 |
13 | 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.
14 |
15 | **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:
16 | * "Error talking to the Govee API!" (`GoveeAPIError`)
17 | * "Error: Your API key is missing or wrong!" (`GoveeConfigError`)
18 | * "Error: That's not a valid color value!" (`GoveeValidationError`)
19 | * "Error: Couldn't connect to the Govee servers!" (`GoveeConnectionError`)
20 | * "Error: Govee took too long to reply!" (`GoveeTimeoutError`)
21 |
22 | 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.
23 |
24 | ## Our Custom Warning Lights: The Exception Classes
25 |
26 | 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.
27 |
28 | 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.
29 |
30 | ```python
31 | # Simplified from src/govee_mcp_server/exceptions.py
32 |
33 | class GoveeError(Exception):
34 | """Base exception for all Govee-related errors in this project."""
35 | pass # It's just a label, inheriting basic Exception features
36 |
37 | class GoveeAPIError(GoveeError):
38 | """Problem communicating with the Govee API itself
39 | (e.g., Govee returned an error message like 'Invalid API Key')."""
40 | pass
41 |
42 | class GoveeConfigError(GoveeError):
43 | """Problem with the configuration settings
44 | (e.g., missing GOVEE_API_KEY in the .env file)."""
45 | pass
46 |
47 | class GoveeValidationError(GoveeError):
48 | """Problem with the data provided by the user/client
49 | (e.g., brightness=150, or invalid RGB values)."""
50 | pass
51 |
52 | class GoveeConnectionError(GoveeError):
53 | """Problem with the network connection to Govee servers."""
54 | pass
55 |
56 | class GoveeTimeoutError(GoveeError):
57 | """Govee's servers took too long to respond to our request."""
58 | pass
59 | ```
60 |
61 | **Explanation:**
62 | * `class GoveeError(Exception):`: This defines our main category. Anything that's a `GoveeConfigError` is *also* a `GoveeError`.
63 | * 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.
64 |
65 | ## Turning On the Warning Lights: Raising Exceptions
66 |
67 | Different parts of our code are responsible for detecting specific problems and `raise`-ing the appropriate custom exception.
68 |
69 | **Example 1: Missing Configuration (`load_config`)**
70 |
71 | 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`.
72 |
73 | ```python
74 | # Simplified from src/govee_mcp_server/config.py
75 | from .exceptions import GoveeConfigError # Import the specific error
76 | # ... other imports ...
77 |
78 | def load_config():
79 | # ... code to load api_key, device_id, sku ...
80 | api_key = None # Let's pretend API key wasn't found
81 |
82 | if not api_key: # Check if it's missing
83 | # Raise the specific configuration error!
84 | raise GoveeConfigError("Missing required environment variable: GOVEE_API_KEY")
85 |
86 | # ... return config object if all is well ...
87 | ```
88 |
89 | **Explanation:**
90 | * If `api_key` is not found, `raise GoveeConfigError(...)` immediately stops the `load_config` function and signals this specific problem.
91 |
92 | **Example 2: Invalid Color Input (`validate_rgb`)**
93 |
94 | 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.
95 |
96 | ```python
97 | # Simplified from src/govee_mcp_server/interfaces.py
98 | from .exceptions import GoveeValidationError # Import the specific error
99 | # ... other imports ...
100 |
101 | def validate_rgb(func):
102 | async def wrapper(self, r: int, g: int, b: int, *args, **kwargs):
103 | if not (0 <= r <= 255): # Check if Red value is valid
104 | # Raise the specific validation error!
105 | raise GoveeValidationError("red value must be between 0-255")
106 | # ... check g and b ...
107 | # If all good, call the original function (like set_color)
108 | return await func(self, r, g, b, *args, **kwargs)
109 | return wrapper
110 |
111 | class ColorControl(ABC):
112 | @abstractmethod
113 | @validate_rgb # Apply the validator
114 | async def set_color(self, r: int, g: int, b: int) -> Tuple[bool, str]:
115 | pass
116 | ```
117 |
118 | **Explanation:**
119 | * If `r` is outside the 0-255 range, `raise GoveeValidationError(...)` stops the process *before* the actual `set_color` logic in `GoveeAPI` runs.
120 |
121 | **Example 3: API Communication Problems (`_make_request`)**
122 |
123 | The core `_make_request` function inside the [Govee API Client](03_govee_api_client.md) handles errors from the Govee servers or network issues.
124 |
125 | ```python
126 | # Simplified from src/govee_mcp_server/api.py
127 | import aiohttp
128 | import asyncio
129 | from .exceptions import GoveeAPIError, GoveeConnectionError, GoveeTimeoutError
130 |
131 | class GoveeAPI:
132 | # ... other methods ...
133 |
134 | async def _make_request(self, method: str, endpoint: str, **kwargs):
135 | # ... setup ...
136 | for attempt in range(self.MAX_RETRIES):
137 | try:
138 | async with self.session.request(...) as response:
139 | data = await response.json()
140 | if response.status == 401: # Example: Unauthorized
141 | # Raise specific API error!
142 | raise GoveeAPIError(f"API error: 401 - Invalid API Key")
143 | elif response.status != 200:
144 | # Raise general API error for other bad statuses
145 | raise GoveeAPIError(f"API error: {response.status} - {data.get('message')}")
146 | # ... success path ...
147 | return data, data.get('message', 'Success')
148 |
149 | except asyncio.TimeoutError:
150 | if attempt == self.MAX_RETRIES - 1:
151 | # Raise specific timeout error!
152 | raise GoveeTimeoutError("Request timed out")
153 | except aiohttp.ClientError as e:
154 | if attempt == self.MAX_RETRIES - 1:
155 | # Raise specific connection error!
156 | raise GoveeConnectionError(f"Connection error: {e}")
157 | # ... retry logic ...
158 | raise GoveeAPIError("Max retries exceeded") # If all retries fail
159 | ```
160 |
161 | **Explanation:**
162 | * Based on the HTTP status code from Govee (like 401) or network exceptions (`TimeoutError`, `ClientError`), this code raises the corresponding specific `GoveeError`.
163 |
164 | ## Reacting to Warning Lights: Catching Exceptions
165 |
166 | 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.
167 |
168 | **Example: Handling Errors in the CLI**
169 |
170 | 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.
171 |
172 | ```python
173 | # Simplified from src/govee_mcp_server/cli.py
174 | import sys
175 | from .exceptions import GoveeError, GoveeConfigError, GoveeValidationError
176 | # ... other imports ...
177 |
178 | async def main() -> None:
179 | api = None
180 | try:
181 | # --- Operations that might raise GoveeErrors ---
182 | config = load_config() # Might raise GoveeConfigError
183 | api = GoveeAPI(config)
184 | args = parser.parse_args()
185 |
186 | if args.command == 'color':
187 | # Might raise GoveeValidationError (from decorator)
188 | # or GoveeAPIError, etc. (from api.set_color)
189 | await handle_color(api, args.red, args.green, args.blue)
190 | # ... other command handlers ...
191 | # --- End of operations ---
192 |
193 | # --- Catching specific errors ---
194 | except GoveeConfigError as e:
195 | print(f"\nConfiguration Problem: {str(e)}")
196 | print("Please check your .env file or environment variables.")
197 | sys.exit(1) # Exit with an error status
198 | except GoveeValidationError as e:
199 | print(f"\nInvalid Input: {str(e)}")
200 | sys.exit(1)
201 | except GoveeError as e: # Catch any other Govee-specific error
202 | print(f"\nOperation Failed: {str(e)}")
203 | sys.exit(1)
204 | except Exception as e: # Catch any totally unexpected error
205 | print(f"\nAn Unexpected Error Occurred: {str(e)}")
206 | sys.exit(1)
207 | finally:
208 | # Cleanup runs whether there was an error or not
209 | if api:
210 | await api.close()
211 |
212 | # ... function to run main ...
213 | ```
214 |
215 | **Explanation:**
216 | * The code inside the `try` block is executed.
217 | * 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.
218 | * If `handle_color` (or the underlying `api.set_color`) raises a `GoveeValidationError`, it jumps to the `except GoveeValidationError as e:` block.
219 | * If any *other* type of `GoveeError` occurs (like `GoveeAPIError`, `GoveeTimeoutError`), it's caught by the more general `except GoveeError as e:` block.
220 | * This allows the CLI to give tailored feedback based on *what kind* of error happened.
221 |
222 | 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.
223 |
224 | ## Under the Hood: Tracing an "Invalid API Key" Error
225 |
226 | Let's follow the journey of an error when the user tries to run the CLI with a wrong API key:
227 |
228 | 1. **User Runs:** `govee-cli power on`
229 | 2. **CLI Starts:** `cli.py` `main()` function begins.
230 | 3. **Load Config:** `load_config()` runs successfully (it finds *an* API key, just the wrong one).
231 | 4. **Create API Client:** `api = GoveeAPI(config)` is created.
232 | 5. **Call Handler:** `handle_power(api, 'on')` is called.
233 | 6. **Call API Method:** `await api.set_power(True)` is called.
234 | 7. **Make Request:** Inside `set_power`, `await self._make_request(...)` is called.
235 | 8. **Send to Govee:** `_make_request` sends the command to the Govee Cloud API, including the *invalid* API key in the headers.
236 | 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"}`).
237 | 10. **Detect Error:** `_make_request` receives the 401 status.
238 | 11. **Raise Specific Error:** It executes `raise GoveeAPIError("API error: 401 - Invalid API Key")`. This stops `_make_request`.
239 | 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.
240 | 13. **Catch in Handler:** The `try...except` block in `handle_power` catches the `GoveeAPIError`.
241 | 14. **Raise from Handler:** The handler re-raises the error: `raise GoveeError(message)` (or simply lets the original `GoveeAPIError` propagate).
242 | 15. **Catch in Main:** The main `try...except` block in `cli.py` `main()` catches the `GoveeError` (since `GoveeAPIError` is a type of `GoveeError`).
243 | 16. **Print Message:** The code inside `except GoveeError as e:` runs, printing: `Operation Failed: API error: 401 - Invalid API Key`
244 | 17. **Exit:** The CLI exits with an error code (`sys.exit(1)`).
245 |
246 | ```mermaid
247 | sequenceDiagram
248 | participant User
249 | participant CLI as cli.py main()
250 | participant Handler as handle_power()
251 | participant API as GoveeAPI.set_power()
252 | participant Request as GoveeAPI._make_request()
253 | participant GoveeCloud
254 |
255 | User->>CLI: Run `govee-cli power on`
256 | CLI->>Handler: Call handle_power()
257 | Handler->>API: Call api.set_power(True)
258 | API->>Request: Call _make_request()
259 | Request->>GoveeCloud: Send request (with bad API key)
260 | GoveeCloud-->>Request: Respond: HTTP 401 Unauthorized
261 | Request-->>Request: Detect 401 status
262 | Request-->>API: raise GoveeAPIError("Invalid API Key")
263 | API-->>Handler: Error propagates up
264 | Handler-->>CLI: Error propagates up
265 | CLI-->>CLI: Catch GoveeAPIError (as GoveeError)
266 | CLI->>User: Print "Operation Failed: API error: 401 - Invalid API Key"
267 | CLI->>User: Exit with error code
268 | ```
269 |
270 | 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.
271 |
272 | ## Conclusion
273 |
274 | You've reached the end of the `govee_mcp_server` tutorial! In this chapter, we learned about **Custom Error Handling**:
275 |
276 | * It's like having specific **warning lights** instead of one generic "Check Engine" light.
277 | * We defined our own exception classes (`GoveeAPIError`, `GoveeConfigError`, `GoveeValidationError`, etc.) inheriting from a base `GoveeError`.
278 | * Different parts of the code `raise` these specific errors when problems are detected (e.g., bad config, invalid input, API failure).
279 | * 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.
280 | * This makes debugging easier and the application more user-friendly.
281 |
282 | 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.
283 |
284 | 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!
285 |
286 | ---
287 |
288 | Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```
--------------------------------------------------------------------------------
/documentation/03_govee_api_client.md:
--------------------------------------------------------------------------------
```markdown
1 | # Chapter 3: Govee API Client - The Messenger to Govee
2 |
3 | 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.
4 |
5 | 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.
6 |
7 | This chapter introduces the **Govee API Client** – the heart of our communication system.
8 |
9 | ## What's the Big Idea? Talking the Govee Talk
10 |
11 | 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:
12 |
13 | 1. **The Language:** Exactly how to phrase the request (the structure, the vocabulary).
14 | 2. **The Protocol:** How to properly address the message and where to send it (the specific web address or "endpoint").
15 | 3. **The Credentials:** How to prove they are allowed to speak on your behalf (using your API Key).
16 | 4. **Handling Issues:** What to do if the message doesn't get through (like trying again or reporting an error).
17 |
18 | 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.
19 |
20 | ## Key Roles of the Govee API Client
21 |
22 | Let's break down the main jobs of this crucial component:
23 |
24 | 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.
25 | 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 ...}}`.
26 | 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/...`).
27 | 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).
28 | 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.
29 |
30 | ## Using the Govee API Client
31 |
32 | Let's see how simple it looks from the outside, thanks to the groundwork laid in previous chapters.
33 |
34 | First, we need the configuration and then we create an instance of the `GoveeAPI` client:
35 |
36 | ```python
37 | # --- Conceptual Example ---
38 | from govee_mcp_server.config import load_config
39 | from govee_mcp_server.api import GoveeAPI
40 | import asyncio # We need this for 'async' functions
41 |
42 | async def main():
43 | # 1. Load the configuration (API Key, Device ID, SKU)
44 | try:
45 | config = load_config()
46 | print(f"Loaded config for device: {config.device_id}")
47 | except Exception as e:
48 | print(f"Error loading config: {e}")
49 | return
50 |
51 | # 2. Create the Govee API Client instance, giving it the config
52 | api_client = GoveeAPI(config=config)
53 | print("GoveeAPI client created.")
54 |
55 | # 3. Now we can use the client to control the device!
56 | # (Example in the next section)
57 |
58 | # Don't forget to close the connection when done
59 | await api_client.close()
60 | print("API client connection closed.")
61 |
62 | # Run the example function
63 | # asyncio.run(main()) # You'd typically run this in a real script
64 | ```
65 |
66 | **Explanation:**
67 | 1. We use `load_config()` from [Chapter 1: Configuration Management](01_configuration_management.md) to get our settings.
68 | 2. We create `GoveeAPI` by passing the `config` object to it. The client now knows *which* device to talk to and *how* to authenticate.
69 | 3. The `api_client` object is now ready to send commands.
70 | 4. `await api_client.close()` is important to clean up network connections gracefully when we're finished.
71 |
72 | 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:
73 |
74 | ```python
75 | # --- Continuing the conceptual example ---
76 |
77 | # Assume 'api_client' was created as shown above
78 |
79 | # Let's turn the light ON
80 | print("Attempting to turn power ON...")
81 | success, message = await api_client.set_power(True)
82 |
83 | if success:
84 | print(f"Success! Govee replied: '{message}'")
85 | else:
86 | print(f"Failed. Govee error: '{message}'")
87 |
88 | # Example Output (if successful):
89 | # Attempting to turn power ON...
90 | # Success! Govee replied: 'Success'
91 |
92 | # Example Output (if failed, e.g., wrong API key):
93 | # Attempting to turn power ON...
94 | # Failed. Govee error: 'API error: 401 - Invalid API Key'
95 | ```
96 |
97 | **Explanation:**
98 | * 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.
99 | * 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`).
100 |
101 | ## Under the Hood: How a Command is Sent
102 |
103 | Let's trace the journey of a `set_power(True)` command:
104 |
105 | 1. **You Call:** Your code calls `await api_client.set_power(True)`.
106 | 2. **Client Receives:** The `set_power` method inside the `GoveeAPI` object starts running.
107 | 3. **Get Credentials:** It accesses the stored `config` object to retrieve `api_key`, `device_id`, and `sku`.
108 | 4. **Build the Message:** It constructs the specific JSON payload Govee expects for power control. This might look something like:
109 | ```json
110 | {
111 | "requestId": "some_unique_id", // Often based on time
112 | "payload": {
113 | "sku": "H6159", // Your device model
114 | "device": "AB:CD:EF...", // Your device ID
115 | "capability": {
116 | "type": "devices.capabilities.on_off",
117 | "instance": "powerSwitch",
118 | "value": 1 // 1 means ON, 0 means OFF
119 | }
120 | }
121 | }
122 | ```
123 | 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.
124 | 6. **Send:** It sends the request over the internet using an HTTP library (`aiohttp` in our case).
125 | 7. **Wait & Listen:** It waits for Govee's servers to respond.
126 | 8. **Check Reply:** Govee sends back a response (e.g., HTTP status code 200 OK and some JSON data like `{"message": "Success"}`).
127 | 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).
128 | 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.
129 | 11. **Return:** It returns the `(success, message)` tuple back to your code.
130 |
131 | Here's a simplified diagram of this flow:
132 |
133 | ```mermaid
134 | sequenceDiagram
135 | participant YourCode as Your Code
136 | participant APIClient as GoveeAPI Object
137 | participant Config as GoveeConfig
138 | participant Formatter as Request Formatter
139 | participant HTTP as HTTP Sender (aiohttp)
140 | participant GoveeCloud as Govee Cloud API
141 |
142 | YourCode->>APIClient: await set_power(True)
143 | APIClient->>Config: Get api_key, device_id, sku
144 | Config-->>APIClient: Return values
145 | APIClient->>Formatter: Build power ON JSON payload
146 | Formatter-->>APIClient: Return formatted JSON
147 | APIClient->>HTTP: Prepare POST request (URL, JSON, Headers with API Key)
148 | loop Retries (if needed)
149 | HTTP->>GoveeCloud: Send HTTP POST request
150 | GoveeCloud-->>HTTP: Send HTTP Response (e.g., 200 OK, {'message':'Success'})
151 | alt Successful Response
152 | HTTP-->>APIClient: Forward response
153 | Note over APIClient: Success! Exit loop.
154 | else Network Error or Bad Response
155 | HTTP-->>APIClient: Report error
156 | Note over APIClient: Wait and retry (if attempts remain)
157 | end
158 | end
159 | APIClient->>YourCode: return (True, "Success") or (False, "Error message")
160 |
161 | ```
162 |
163 | ## Inside the Code: Key Parts of `GoveeAPI`
164 |
165 | Let's peek at simplified snippets from `src/govee_mcp_server/api.py` to see how this is implemented.
166 |
167 | **1. Initialization (`__init__`)**
168 |
169 | ```python
170 | # Simplified from src/govee_mcp_server/api.py
171 | import aiohttp # Library for making async web requests
172 | from .config import GoveeConfig
173 | # ... other imports ...
174 |
175 | class GoveeAPI(PowerControl, ColorControl, BrightnessControl):
176 | # ... constants like BASE_URL, MAX_RETRIES ...
177 |
178 | def __init__(self, config: GoveeConfig):
179 | """Initialize API client with configuration."""
180 | self.config = config # Store the config object
181 | self.session: Optional[aiohttp.ClientSession] = None # Network session setup later
182 | # ... maybe other setup ...
183 | print(f"GoveeAPI initialized for device {self.config.device_id}")
184 | ```
185 |
186 | **Explanation:**
187 | * The `__init__` method runs when you create `GoveeAPI(config=...)`.
188 | * 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`.
189 | * It also prepares a variable `self.session` to hold the network connection details, which will be created when needed.
190 |
191 | **2. Making the Request (`_make_request`)**
192 |
193 | This is the core helper method that handles sending *any* request, including retries and basic error handling.
194 |
195 | ```python
196 | # Simplified from src/govee_mcp_server/api.py
197 | import asyncio
198 | # ... other imports ...
199 |
200 | class GoveeAPI:
201 | # ... __init__ and constants ...
202 |
203 | async def _ensure_session(self) -> None:
204 | """Create network session if needed."""
205 | if self.session is None or self.session.closed:
206 | print("Creating new network session...")
207 | self.session = aiohttp.ClientSession(
208 | headers={"Govee-API-Key": self.config.api_key}, # Set API key for ALL requests!
209 | timeout=aiohttp.ClientTimeout(total=10) # 10-second timeout
210 | )
211 |
212 | async def _make_request(self, method: str, endpoint: str, **kwargs):
213 | """Make HTTP request with retries."""
214 | await self._ensure_session() # Make sure we have a network session
215 |
216 | last_error = None
217 | for attempt in range(self.MAX_RETRIES): # Try up to MAX_RETRIES times
218 | try:
219 | print(f"Attempt {attempt+1}: Sending {method} to {endpoint}")
220 | async with self.session.request(method, f"{self.BASE_URL}/{endpoint}", **kwargs) as response:
221 | data = await response.json()
222 | print(f"Received status: {response.status}")
223 | if response.status == 200: # HTTP 200 means OK!
224 | return data, data.get('message', 'Success') # Return data and message
225 | else:
226 | # Govee returned an error status
227 | last_error = GoveeAPIError(f"API error: {response.status} - {data.get('message', 'Unknown')}")
228 |
229 | except asyncio.TimeoutError:
230 | last_error = GoveeTimeoutError("Request timed out")
231 | except aiohttp.ClientError as e:
232 | last_error = GoveeConnectionError(f"Connection error: {e}")
233 |
234 | # If error occurred, wait before retrying (increasing delay)
235 | if attempt < self.MAX_RETRIES - 1:
236 | delay = self.RETRY_DELAY * (attempt + 1)
237 | print(f"Request failed. Retrying in {delay}s...")
238 | await asyncio.sleep(delay)
239 |
240 | # If all retries failed, raise the last encountered error
241 | print("Max retries exceeded.")
242 | raise last_error
243 |
244 | ```
245 |
246 | **Explanation:**
247 | * `_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.
248 | * The main loop runs `MAX_RETRIES` times (e.g., 3 times).
249 | * `try...except`: It tries to make the request using `self.session.request(...)`. It catches specific network errors like `TimeoutError` or general `ClientError`.
250 | * `if response.status == 200`: Checks if Govee reported success. If yes, it returns the data.
251 | * `else`: If Govee returns an error code (like 401 Unauthorized or 400 Bad Request), it creates a `GoveeAPIError`.
252 | * Retry Logic: If an error occurs, it waits (`asyncio.sleep`) before the next attempt. The delay increases with each retry.
253 | * If all retries fail, it `raise`s the last error it encountered, which can then be caught by the calling method (like `set_power`).
254 |
255 | **3. Implementing an Interface Method (`set_power`)**
256 |
257 | This method uses `_make_request` to perform a specific action.
258 |
259 | ```python
260 | # Simplified from src/govee_mcp_server/api.py
261 | from time import time # To generate a unique request ID
262 |
263 | class GoveeAPI:
264 | # ... __init__, _make_request ...
265 |
266 | async def set_power(self, state: bool) -> Tuple[bool, str]:
267 | """Implement PowerControl.set_power"""
268 | try:
269 | # Prepare the specific JSON payload for the power command
270 | payload = {
271 | "requestId": str(int(time())), # Unique ID for the request
272 | "payload": {
273 | "sku": self.config.sku,
274 | "device": self.config.device_id,
275 | "capability": {
276 | "type": "devices.capabilities.on_off",
277 | "instance": "powerSwitch",
278 | "value": 1 if state else 0 # Convert True/False to 1/0
279 | }
280 | }
281 | }
282 | # Call the generic request handler
283 | _, message = await self._make_request(
284 | method="POST",
285 | endpoint="router/api/v1/device/control", # Govee's endpoint for control
286 | json=payload # Pass the JSON data
287 | )
288 | return True, message # If _make_request succeeded, return True
289 | except GoveeError as e:
290 | # If _make_request raised an error after retries
291 | print(f"set_power failed: {e}")
292 | return False, str(e) # Return False and the error message
293 | ```
294 |
295 | **Explanation:**
296 | * It builds the `payload` dictionary exactly as Govee requires for the `on_off` capability, using the `device_id` and `sku` from `self.config`.
297 | * It calls `self._make_request`, telling it to use the `POST` method, the specific `/device/control` endpoint, and the `payload` as JSON data.
298 | * `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)`.
299 |
300 | The other methods (`set_color`, `get_brightness`, etc.) follow a very similar pattern: build the specific payload for that capability and call `_make_request`.
301 |
302 | ## Conclusion
303 |
304 | You've now explored the **Govee API Client (`GoveeAPI`)**, the component that acts as our application's diplomat to the Govee Cloud.
305 |
306 | You learned:
307 | * Its role is to translate standard commands (from our interfaces) into Govee's specific API language.
308 | * It uses the configuration ([Chapter 1: Configuration Management](01_configuration_management.md)) for authentication and targeting.
309 | * It handles formatting requests, sending them via HTTP, and processing responses.
310 | * It includes essential features like automatic retries and handling network errors (`TimeoutError`, `ConnectionError`).
311 | * Internally, it uses a helper (`_make_request`) to manage the common parts of API communication.
312 | * Methods like `set_power` build the specific command payload and rely on `_make_request` to send it.
313 |
314 | 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?
315 |
316 | 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)!
317 |
318 | ---
319 |
320 | Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
```