MAGG

Instructions

Implementation Reference

• magg/server/server.py:252-343 (handler)

  The core handler function implementing the magg_add_server tool logic: validates parameters, creates ServerConfig, mounts the server if enabled, saves configuration, and returns structured response.
  async def add_server( self, name: Annotated[str, Field(description="Unique server name")], source: Annotated[str, Field(description="URL of the server package/repository")], prefix: Annotated[str | None, Field(description="Tool prefix (defaults to conformed server name)")] = None, command: Annotated[str | None, Field( description="Full command to run (e.g., 'python server.py', 'npx @playwright/mcp@latest')" )] = None, uri: Annotated[AnyUrl | None, Field(description="URI for HTTP servers")] = None, env: Annotated[dict[str, str] | str | None, Field(description="Environment variables (dict or JSON string)")] = None, cwd: Annotated[str | None, Field(description="Working directory (for commands)")] = None, notes: Annotated[str | None, Field(description="Setup notes")] = None, enable: Annotated[bool | None, Field(description="Whether to enable the server immediately (default: True)")] = True, transport: Annotated[dict[str, Any] | str | None, Field( description=f"Transport-specific configuration (dict or JSON string){TRANSPORT_DOCS}" )] = None, ) -> MaggResponse: """Add a new MCP server.""" try: config = self.config if name in config.servers: return MaggResponse.error(f"Server '{name}' already exists") actual_command = command actual_args = None if command: parts = shlex.split(command) if len(parts) > 1: actual_command = parts[0] actual_args = parts[1:] elif len(parts) == 1: actual_command = parts[0] actual_args = None if cwd: validated_dir, error = validate_working_directory(cwd, source) if error: return MaggResponse.error(error) cwd = validated_dir try: server = ServerConfig( name=name, source=source, prefix=prefix, command=actual_command, args=actual_args, uri=uri, env=json_to_dict(env), cwd=cwd, notes=notes, enabled=enable if enable is not None else True, transport=json_to_dict(transport), ) except ValueError as e: return MaggResponse.error(str(e)) mount_success = None if server.enabled: mount_success = await self.server_manager.mount_server(server) if not mount_success: return MaggResponse.error(f"Failed to mount server '{name}'") config.add_server(server) if not self.save_config(config): return MaggResponse.error(f"Failed to save configuration for added server '{name}'") return MaggResponse.success({ "action": "server_added", "server": { "name": server.name, "source": server.source, "prefix": server.prefix, "command": ( f"{server.command} {' '.join(server.args) if server.args else ''}".strip() if server.command else None ), "uri": server.uri, "cwd": server.cwd, "notes": server.notes, "enabled": server.enabled, "mounted": mount_success } }) except Exception as e: return MaggResponse.error(f"Failed to add server: {str(e)}")
• magg/server/server.py:43-82 (registration)

  Registers the add_server method as the 'magg_add_server' tool using FastMCP's tool decorator within the MaggServer class initialization.
  def _register_tools(self): """Register all Magg management tools programmatically. """ self_prefix_ = self.self_prefix_ tools = [ (self.add_server, f"{self_prefix_}add_server", None), (self.remove_server, f"{self_prefix_}remove_server", None), (self.list_servers, f"{self_prefix_}list_servers", None), (self.enable_server, f"{self_prefix_}enable_server", None), (self.disable_server, f"{self_prefix_}disable_server", None), (self.search_servers, f"{self_prefix_}search_servers", None), (self.smart_configure, f"{self_prefix_}smart_configure", None), (self.analyze_servers, f"{self_prefix_}analyze_servers", None), (self.status, f"{self_prefix_}status", None), (self.check, f"{self_prefix_}check", None), (self.reload_config_tool, f"{self_prefix_}reload_config", None), (self.load_kit, f"{self_prefix_}load_kit", None), (self.unload_kit, f"{self_prefix_}unload_kit", None), (self.list_kits, f"{self_prefix_}list_kits", None), (self.kit_info, f"{self_prefix_}kit_info", None), ] def call_tool_wrapper(func): @wraps(func) async def wrapper(*args, **kwds): result = await func(*args, **kwds) if isinstance(result, MaggResponse): return result.as_json_text_content return result return wrapper for method, tool_name, options in tools: self.mcp.tool(name=tool_name, **(options or {}))(call_tool_wrapper(method)) self._register_resources() self._register_prompts()
• magg/server/defaults.py:17-99 (schema)

  Defines the detailed tool schema, parameter descriptions, types, and usage examples for magg_add_server, used in tool documentation and LLM prompts.
  MAGG_ADD_SERVER_DOC = """ Tool: magg_add_server Description: Add a new MCP server. Parameters: name (string) (required) Unique server name source (string) (required) URL of the server package/repository prefix (string | null) (optional) Tool prefix (defaults to conformed server name) command (string | null) (optional) Full command to run (e.g., 'python server.py', 'npx @playwright/mcp@latest') NOTE: This should include the full command, not just the executable name. Arguments will be split automatically. uri (string | null) (optional) URI for HTTP servers env (object | string | null) (optional) Environment variables (dict or JSON string) cwd (string | null) (optional) Working directory (for commands) notes (string | null) (optional) Setup notes enable (boolean | null) (optional) Whether to enable the server immediately (default: True) transport (object | string | null) (optional) Transport-specific configuration (dict or JSON string) Common options for all command-based servers: - `keep_alive` (boolean): Keep the process alive between requests (default: true) Python servers (command="python ..."): - `python_cmd` (string): Python executable path (default: sys.executable) Node.js servers (command="node ..."): - `node_cmd` (string): Node executable path (default: "node") NPX servers (command="npx ..."): - `use_package_lock` (boolean): Use package-lock.json if present (default: true) UVX servers (command="uvx ..."): - `python_version` (string): Python version to use (e.g., "3.11") - `with_packages` (array): Additional packages to install - `from_package` (string): Install tool from specific package HTTP/SSE servers (uri-based): - `headers` (object): HTTP headers to include - `auth` (string): Authentication method ("oauth" or bearer token) - `sse_read_timeout` (number): Timeout for SSE reads in seconds Examples: - Python: `{"keep_alive": false, "python_cmd": "/usr/bin/python3"}` - UVX: `{"python_version": "3.11", "with_packages": ["requests", "pandas"]}` - HTTP: `{"headers": {"Authorization": "Bearer token123"}, "sse_read_timeout": 30}` Example configurations arguments: [ "calc": { "name": "Calculator MCP", "source": "https://github.com/wrtnlabs/calculator-mcp", "prefix": "calc", "command": "npx -y @wrtnlabs/calculator-mcp@latest" }, "playwright": { "name": "Playwright MCP", "source": "https://github.com/microsoft/playwright-mcp", "prefix": "playwright", "notes": "Browser automation MCP server using Playwright.", "command": "npx @playwright/mcp@latest" }, "test": { "name": "test", "source": "play", "command": "python play/test_server.py" }, "hello": { "name": "hello", "source": "https://www.npmjs.com/package/mcp-hello-world", "command": "npx mcp-hello-world@latest" } ] """