macOS Defaults MCP Server

import asyncio import subprocess from mcp.server.models import InitializationOptions import mcp.types as types from mcp.server import NotificationOptions, Server from pydantic import AnyUrl import mcp.server.stdio # Store notes as a simple key-value dict to demonstrate state management notes: dict[str, str] = {} server = Server("mcp-server-macos-defaults") # @server.list_resources() # async def handle_list_resources() -> list[types.Resource]: # """ # List available note resources. # Each note is exposed as a resource with a custom note:// URI scheme. # """ # return [ # types.Resource( # uri=AnyUrl(f"note://internal/{name}"), # name=f"Note: {name}", # description=f"A simple note named {name}", # mimeType="text/plain", # ) # for name in notes # ] # # @server.read_resource() # async def handle_read_resource(uri: AnyUrl) -> str: # """ # Read a specific note's content by its URI. # The note name is extracted from the URI host component. # """ # if uri.scheme != "note": # raise ValueError(f"Unsupported URI scheme: {uri.scheme}") # # name = uri.path # if name is not None: # name = name.lstrip("/") # return notes[name] # raise ValueError(f"Note not found: {name}") # # @server.list_prompts() # async def handle_list_prompts() -> list[types.Prompt]: # """ # List available prompts. # Each prompt can have optional arguments to customize its behavior. # """ # return [ # types.Prompt( # name="summarize-notes", # description="Creates a summary of all notes", # arguments=[ # types.PromptArgument( # name="style", # description="Style of the summary (brief/detailed)", # required=False, # ) # ], # ) # ] # # @server.get_prompt() # async def handle_get_prompt( # name: str, arguments: dict[str, str] | None # ) -> types.GetPromptResult: # """ # Generate a prompt by combining arguments with server state. # The prompt includes all current notes and can be customized via arguments. # """ # if name != "summarize-notes": # raise ValueError(f"Unknown prompt: {name}") # # style = (arguments or {}).get("style", "brief") # detail_prompt = " Give extensive details." if style == "detailed" else "" # # return types.GetPromptResult( # description="Summarize the current notes", # messages=[ # types.PromptMessage( # role="user", # content=types.TextContent( # type="text", # text=f"Here are the current notes to summarize:{detail_prompt}\n\n" # + "\n".join( # f"- {name}: {content}" # for name, content in notes.items() # ), # ), # ) # ], # ) @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """ List available tools. Each tool specifies its arguments using JSON Schema validation. """ return [ types.Tool( name="list-domains", description="List all available macOS domains, same as `defaults domains`", inputSchema= { "type": "object", "properties": {}, }, # TODO filter domains? ), types.Tool( name="find", description="Find entries container given word", inputSchema= { "type": "object", "properties": { "word": { "type": "string", "description": "Word to search for", }, }, }, ), # defaults read <domain> <key> # key is optional, domain required types.Tool( name="defaults-read", description = "use the `defaults read <domain> <key>` command", inputSchema = { "type": "object", "properties": { "domain": { "type": "string", "description": "Domain to read from", }, "key": { "type": "string", "description": "Key to read from", }, }, "required": ["domain"], }, ), # defaults write <domain> <key> <value> types.Tool( name="defaults-write", description = "use the `defaults write <domain> <key> <value>` command", inputSchema = { "type": "object", "properties": { "domain": { "type": "string", "description": "Domain to write to", }, "key": { "type": "string", "description": "Key to write to", }, "value": { "type": "string", "description": "Value to write", }, }, "required": ["domain", "key", "value"], }, ), # TODO dictionary values? # defaults read-type <domain> <key> # defaults delete <domain> <key> # key is optional, domain required ] def defaults_read(arguments: dict | None) -> list[types.TextContent]: if arguments is None: return [] domain = arguments["domain"] key = arguments.get("key") if key is None: result = subprocess.run(["defaults", "read", domain], capture_output=True) return [types.TextContent(type="text", text=result.stdout.decode("utf-8"))] result = subprocess.run(["defaults", "read", domain, key], capture_output=True) value = result.stdout.decode("utf-8").strip() return [types.TextContent(type="text", text=f"{key}: {value}")] def defaults_write(arguments: dict | None) -> list[types.TextContent]: if arguments is None: return [] domain = arguments["domain"] key = arguments["key"] value = arguments["value"] # TODO do I need to notify client that resource changed? can't it just ask for it again? result = subprocess.run(["defaults", "write", domain, key, value], capture_output=True) return [types.TextContent(type="text", text=result.stdout.decode("utf-8"))] def list_domains() -> list[types.TextContent]: # get array of domains: # run command `defaults domains` result = subprocess.run(["defaults", "domains"], capture_output=True) domains = result.stdout.decode("utf-8").split(",") domains = [domain.strip() for domain in domains] return [types.TextContent(type="text", text=f"Domains: {domains}")] def find(arguments: dict | None) -> list[types.TextContent]: # ask for help to find a setting, that alone is possibly very useful if arguments is None: raise ValueError("Arguments are required") word = arguments["word"] result = subprocess.run(["defaults", "find", word], capture_output=True) return [types.TextContent(type="text", text=f"Found: {result.stdout.decode('utf-8')}")] @server.call_tool() async def handle_call_tool(name: str, arguments: dict | None) -> list[types.TextContent]: # TODO => other types? | types.ImageContent | types.EmbeddedResource]: # # Notify clients that resources have changed (if server changes resources) # await server.request_context.session.send_resource_list_changed() if name == "list-domains": return list_domains() elif name == "find": return find(arguments) elif name == "defaults-read": return defaults_read(arguments) elif name == "defaults-write": return defaults_write(arguments) else: raise ValueError(f"Unknown tool: {name}") async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="mcp-server-macos-defaults", server_version="0.1.2", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), )