# Directory Structure
```
├── .gitignore
├── canon_camera.py
├── README.md
├── requirements.txt
└── server.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | .venv
2 | .idea
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Canon Camera MCP
2 |
3 | A minimal server for controlling Canon cameras via the Canon Camera Control API (CCAPI), using FastMCP for streamable HTTP transport.
4 |
5 | #### Demo 🎥
6 | [](http://www.youtube.com/watch?v=59icGndauho "Canon MCP Server Demo")
7 |
8 | [LinkedIn Post](https://www.linkedin.com/posts/ishanjoshi99_claude-ai-canon-camera-ive-been-activity-7333390072735535104-Sl0b?utm_source=share&utm_medium=member_desktop&rcm=ACoAACE9cFEBBFrka0tZ6SOykeuUIa1qgqTv7WE)
9 |
10 | ## Features
11 |
12 | - Control Canon cameras remotely via CCAPI.
13 | - Expose camera functions over HTTP using FastMCP.
14 | - Image compression and streaming support.
15 |
16 | ## Requirements
17 |
18 | - Python 3.10+
19 | - Canon camera with CCAPI enabled ([CCAPI activation guide](https://www.canon.com.au/apps/eos-digital-software-development-kit))
20 | - See `requirements.txt` for Python dependencies.
21 |
22 | ## Setup
23 |
24 | 1. **Install dependencies:**
25 | ```bash
26 | pip install -r requirements.txt
27 | ```
28 |
29 | 2. **Activate CCAPI on your Canon camera:**
30 | - Follow the official [Canon CCAPI activation instructions](https://www.canon.com.au/apps/eos-digital-software-development-kit).
31 |
32 | 3. **Configure camera IP:**
33 | - Set the `CANON_IP` environment variable to your camera’s IP address, or pass it as an argument.
34 |
35 | ## Usage
36 |
37 | To run the server with Claude Desktop Client
38 |
39 | ```json
40 | {
41 | "mcpServers": {
42 | "Canon Camera Controller": {
43 | "command": "uv",
44 | "args": [
45 | "--directory",
46 | "/path/to/dir",
47 | "run",
48 | "server.py"
49 | ],
50 | "env": {
51 | "CANON_IP": "192.168.0.111"
52 | }
53 | }
54 | }
55 | }
56 | ```
57 |
58 | Or with plain Python:
59 |
60 | ```bash
61 | python server.py
62 | ```
63 |
64 | ## References
65 |
66 | - Based on [laszewsk/canon-r7-ccapi](https://github.com/laszewsk/canon-r7-ccapi)
67 |
68 | ## Project Structure
69 |
70 | - `canon_camera.py`: Canon camera CCAPI interface.
71 | - `server.py`: FastMCP HTTP server exposing camera controls.
72 | - `requirements.txt`: Python dependencies.
73 |
74 |
75 | ## Extending the project
76 | The license terms of CCAPI access do not permit sharing the API reference.
77 |
78 | Once you have access, it's quite straightforward to get it working.
79 |
80 | You may also refer to the Canon CCAPI Feature [list](https://developercommunity.usa.canon.com/s/article/CCAPI-Function-List)
81 |
82 |
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
1 | annotated-types==0.7.0
2 | anyio==4.9.0
3 | certifi==2025.4.26
4 | charset-normalizer==3.4.2
5 | click==8.2.1
6 | h11==0.16.0
7 | httpcore==1.0.9
8 | httpx==0.28.1
9 | httpx-sse==0.4.0
10 | idna==3.10
11 | markdown-it-py==3.0.0
12 | mcp==1.9.1
13 | mdurl==0.1.2
14 | pillow==11.2.1
15 | pydantic==2.11.5
16 | pydantic-settings==2.9.1
17 | pydantic_core==2.33.2
18 | Pygments==2.19.1
19 | python-dotenv==1.1.0
20 | python-multipart==0.0.20
21 | requests==2.32.3
22 | rich==14.0.0
23 | shellingham==1.5.4
24 | sniffio==1.3.1
25 | sse-starlette==2.3.5
26 | starlette==0.46.2
27 | typer==0.16.0
28 | typer-cli==0.16.0
29 | typing-inspection==0.4.1
30 | typing_extensions==4.13.2
31 | urllib3==2.4.0
32 | uvicorn==0.34.2
33 |
```
--------------------------------------------------------------------------------
/canon_camera.py:
--------------------------------------------------------------------------------
```python
1 | import base64
2 | import os
3 |
4 | import requests
5 | import logging
6 |
7 | logging.basicConfig(level=logging.INFO)
8 | logger = logging.getLogger("canon-camera")
9 |
10 |
11 | class CanonCamera:
12 | """Canon Camera CCAPI interface"""
13 |
14 | def __init__(self, ip: str = None, port: int = 8080):
15 | self.ip = ip or os.environ.get("CANON_IP", None)
16 | self.port = port
17 | self.base_url = f"http://{self.ip}:{self.port}"
18 |
19 | def _get(self, path: str) -> requests.Response:
20 | """Execute GET request"""
21 | url = f"{self.base_url}{path}"
22 | logger.info(f"GET: {url}")
23 | response = requests.get(url, timeout=10)
24 | response.raise_for_status()
25 | return response
26 |
27 | def _put(self, path: str, data: dict) -> requests.Response:
28 | """Execute PUT request"""
29 | url = f"{self.base_url}{path}"
30 | logger.info(f"PUT: {url} <- {data}")
31 | response = requests.put(url, json=data, timeout=10)
32 | response.raise_for_status()
33 | return response
34 |
35 | def get_all_settings(self) -> dict:
36 | """Get all shooting settings"""
37 | response = self._get("/ccapi/ver100/shooting/settings")
38 | return response.json()
39 |
40 | def get_setting(self, setting_name: str) -> dict:
41 | """Get specific shooting setting"""
42 | response = self._get(f"/ccapi/ver100/shooting/settings/{setting_name}")
43 | return response.json()
44 |
45 | def set_setting(self, setting_name: str, value: str) -> dict:
46 | """Set specific shooting setting"""
47 | # First get current setting to validate
48 | current = self.get_setting(setting_name)
49 |
50 | if "ability" in current and value not in current["ability"]:
51 | raise ValueError(f"Invalid value '{value}' for {setting_name}. "
52 | f"Available options: {current['ability']}")
53 |
54 | response = self._put(f"/ccapi/ver100/shooting/settings/{setting_name}",
55 | {"value": value})
56 | response.raise_for_status()
57 |
58 | # Return updated setting info
59 | return {
60 | "setting": setting_name,
61 | "previous_value": current.get("value"),
62 | "new_value": value,
63 | "success": True
64 | }
65 |
66 | def init_live_view(self, liveviewsize="small", cameradisplay="keep"):
67 | try:
68 | url = f"{self.base_url}/ccapi/ver100/shooting/liveview/"
69 | body = {"liveviewsize": liveviewsize, "cameradisplay": cameradisplay}
70 | res = requests.post(url, json=body)
71 | return res.status_code
72 | except Exception as e:
73 | logger.error(f"Failed to init live view {e}")
74 | raise
75 |
76 | def get_liveview_image(self) -> str:
77 | """Get live view image as base64 string"""
78 | try:
79 | url = f"{self.base_url}/ccapi/ver100/shooting/liveview/flip"
80 | logger.info(f"Getting live view: {url}")
81 | response = requests.get(url, stream=True, timeout=15)
82 | response.raise_for_status()
83 |
84 | # Convert to base64
85 | image_data = f"{base64.b64encode(response.content).decode('utf-8')}\n"
86 | return image_data
87 | except Exception as e:
88 | logger.error(f"Failed to get live view: {e}")
89 | raise
90 |
```
--------------------------------------------------------------------------------
/server.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | Canon Camera MCP Server with FastMCP Streamable HTTP Transport
4 | A minimal MCP server for controlling Canon cameras via CCAPI using FastMCP
5 | """
6 | import io
7 | import json
8 | import os
9 | import base64
10 | import typing
11 |
12 | import requests
13 | import logging
14 | from typing import Literal
15 |
16 | from mcp.server.fastmcp import FastMCP, Image
17 |
18 | from PIL import Image as PILImage
19 |
20 | from canon_camera import CanonCamera
21 |
22 | # Configure logging
23 | logging.basicConfig(level=logging.INFO)
24 | logger = logging.getLogger("canon-camera-mcp")
25 |
26 |
27 |
28 |
29 |
30 | def compress_image_to_target_size(pil_img, target_size_mb=1, format="JPEG"):
31 | """
32 | Compress a PIL image to approximately the target size in MB.
33 | < Written by ChatGPT >
34 |
35 | Args:
36 | pil_img: PIL Image object
37 | target_size_mb: Target size in megabytes (default: 1)
38 | format: Image format ("JPEG" or "PNG")
39 |
40 | Returns:
41 | bytes: Compressed image data
42 | """
43 | target_size_bytes = target_size_mb * 1024 * 1024 # Convert MB to bytes
44 |
45 | # Convert to RGB if saving as JPEG (JPEG doesn't support transparency)
46 | if format.upper() == "JPEG" and pil_img.mode in ("RGBA", "P"):
47 | pil_img = pil_img.convert("RGB")
48 |
49 | # Start with high quality and reduce if needed
50 | quality = 95
51 | min_quality = 10
52 |
53 | while quality >= min_quality:
54 | img_byte_arr = io.BytesIO()
55 |
56 | if format.upper() == "JPEG":
57 | pil_img.save(img_byte_arr, format="JPEG", quality=quality, optimize=True)
58 | else:
59 | # For PNG, use optimization
60 | pil_img.save(img_byte_arr, format="PNG", optimize=True)
61 |
62 | img_size = img_byte_arr.tell()
63 |
64 | if img_size <= target_size_bytes:
65 | return img_byte_arr.getvalue()
66 |
67 | # Reduce quality for next iteration
68 | quality -= 5
69 |
70 | # If still too large, resize the image
71 | return resize_and_compress_image(pil_img, target_size_bytes, format)
72 |
73 |
74 | def resize_and_compress_image(pil_img, target_size_bytes, format="JPEG"):
75 | """
76 | Resize and compress image if quality reduction alone isn't enough.
77 | """
78 | original_width, original_height = pil_img.size
79 | scale_factor = 0.9 # Start by reducing size by 10%
80 | img_byte_arr = None
81 |
82 | while scale_factor > 0.1: # Don't go below 10% of original size
83 | new_width = int(original_width * scale_factor)
84 | new_height = int(original_height * scale_factor)
85 |
86 | resized_img = pil_img.resize((new_width, new_height), PILImage.Resampling.LANCZOS)
87 |
88 | # Try with medium quality after resizing
89 | img_byte_arr = io.BytesIO()
90 |
91 | if format.upper() == "JPEG":
92 | resized_img.save(img_byte_arr, format="JPEG", quality=75, optimize=True)
93 | else:
94 | resized_img.save(img_byte_arr, format="PNG", optimize=True)
95 |
96 | img_size = img_byte_arr.tell()
97 |
98 | if img_size <= target_size_bytes:
99 | return img_byte_arr.getvalue()
100 |
101 | scale_factor -= 0.1
102 |
103 | # If still too large, return the smallest version
104 | return img_byte_arr.getvalue()
105 |
106 |
107 |
108 |
109 |
110 | # Initialize camera and FastMCP server
111 | camera = CanonCamera()
112 |
113 | mcp = FastMCP("Canon Camera Controller")
114 |
115 |
116 | @mcp.tool()
117 | def get_camera_settings(setting: Literal["all", "av", "tv", "iso"]) -> str:
118 | """
119 | Get all camera shooting settings or a specific setting
120 |
121 | Args:
122 | setting: Specific setting to get (av, tv, iso, shootingmodedial) or 'all' for all settings
123 | """
124 | try:
125 | if setting == "all":
126 | result = camera.get_all_settings()
127 | # Filter the result to only the keys
128 | keys_to_keep = ["av", "tv", "iso", "shootingmodedial"]
129 | result = {key: result[key] for key in keys_to_keep if key in result}
130 |
131 | else:
132 | valid_settings = ["av", "tv", "iso", "shootingmodedial"]
133 | if setting not in valid_settings:
134 | raise ValueError(f"Invalid setting '{setting}'. Valid options: {valid_settings}")
135 |
136 | result = camera.get_setting(setting)
137 | result["setting_name"] = setting
138 |
139 | res = json.dumps(result, indent=2)
140 | return res
141 |
142 | except requests.exceptions.RequestException as e:
143 | logger.error(f"Camera communication error: {e}")
144 | return json.dumps({
145 | "success": False,
146 | "error": "camera_communication_error",
147 | "message": f"Failed to communicate with camera: {str(e)}"
148 | }, indent=2)
149 |
150 | except ValueError as e:
151 | logger.error(f"Invalid parameter: {e}")
152 | return json.dumps({
153 | "success": False,
154 | "error": "invalid_parameter",
155 | "message": str(e)
156 | }, indent=2)
157 |
158 | except Exception as e:
159 | logger.error(f"Unexpected error: {e}")
160 | return json.dumps({
161 | "success": False,
162 | "error": "internal_error",
163 | "message": f"Unexpected error: {str(e)}"
164 | }, indent=2)
165 |
166 |
167 | @mcp.tool()
168 | def set_camera_setting(setting: str, value: str) -> str:
169 | """
170 | Set a camera shooting setting (aperture, shutter speed, or ISO)
171 |
172 | Args:
173 | setting: Setting to change (av, tv, iso)
174 | value: Value to set (must be from the setting's ability list)
175 | """
176 | try:
177 | valid_settings = ["av", "tv", "iso"]
178 | if setting not in valid_settings:
179 | raise ValueError(f"Invalid setting '{setting}'. Valid options: {valid_settings}")
180 |
181 | if not value:
182 | raise ValueError("Value is required")
183 |
184 | result = camera.set_setting(setting, value)
185 | return json.dumps(result, indent=2)
186 |
187 | except requests.exceptions.RequestException as e:
188 | logger.error(f"Camera communication error: {e}")
189 | return json.dumps({
190 | "success": False,
191 | "error": "camera_communication_error",
192 | "message": f"Failed to communicate with camera: {str(e)}"
193 | }, indent=2)
194 |
195 | except ValueError as e:
196 | logger.error(f"Invalid parameter: {e}")
197 | return json.dumps({
198 | "success": False,
199 | "error": "invalid_parameter",
200 | "message": str(e)
201 | }, indent=2)
202 |
203 | except Exception as e:
204 | logger.error(f"Unexpected error: {e}")
205 | return json.dumps({
206 | "success": False,
207 | "error": "internal_error",
208 | "message": f"Unexpected error: {str(e)}"
209 | }, indent=2)
210 |
211 |
212 | @mcp.tool()
213 | def get_liveview() -> typing.Union[Image, str]:
214 | """
215 | Get current live view image from camera. Use this to cross check if the new settings have actually worked.
216 |
217 | Returns an image.
218 | """
219 | try:
220 | image_data_b64 = camera.get_liveview_image()
221 | image_data_bytes = base64.b64decode(image_data_b64)
222 | pil_img = PILImage.open(io.BytesIO(image_data_bytes))
223 |
224 | # Compress the image to ~1MB
225 | compressed_img_bytes = compress_image_to_target_size(pil_img, target_size_mb=1, format="JPEG")
226 | img = Image(data=compressed_img_bytes, format="jpeg")
227 | img_content = img.to_image_content()
228 | logger.info(f"Image content: {img_content}")
229 | logger.info(f"Compressed image size: {len(compressed_img_bytes) / (1024 * 1024):.2f} MB")
230 |
231 | return img
232 |
233 | except requests.exceptions.RequestException as e:
234 | logger.error(f"Camera communication error: {e}")
235 | return json.dumps({
236 | "success": False,
237 | "error": "camera_communication_error",
238 | "message": f"Failed to communicate with camera: {str(e)}"
239 | }, indent=2)
240 |
241 | except Exception as e:
242 | logger.error(f"Unexpected error: {e}")
243 | return json.dumps({
244 | "success": False,
245 | "error": "internal_error",
246 | "message": f"Unexpected error: {str(e)}"
247 | }, indent=2)
248 |
249 |
250 | def main():
251 | """Main entry point"""
252 | # Configuration
253 | host = os.environ.get("MCP_HOST", "localhost")
254 | port = int(os.environ.get("MCP_PORT", "3001"))
255 |
256 | logger.info(f"Starting Canon Camera MCP Server on {host}:{port}")
257 | logger.info(f"Canon Camera IP: {camera.ip}:{camera.port}")
258 | logger.info(f"MCP endpoint: http://{host}:{port}/")
259 |
260 | # Test camera connection on startup
261 | try:
262 | camera.get_all_settings()
263 | camera.init_live_view()
264 | logger.info("✓ Camera connection successful")
265 | except Exception as e:
266 | logger.warning(f"⚠ Camera connection failed: {e}")
267 | logger.info("Server will start anyway - camera connection will be retried on requests")
268 |
269 | # Run server with streamable HTTP transport
270 | # mcp.run(transport="streamable-http")
271 | mcp.run(transport="stdio")
272 |
273 |
274 | if __name__ == "__main__":
275 | main()
276 |
277 |
278 |
```