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"
      }
  ]
  """