feat(cli): add --tool-parser option for custom tool parsers

Add CLI option to specify tool parser by name (e.g., 'hermes', 'qwen_xml')
or path to a hook file exporting a custom ToolParser.

Also refactor CLI utils:
- Extract _load_hook_module() helper to reduce duplication
- Move ToolParser import to TYPE_CHECKING to avoid unnecessary deps
- Split build_model_factory() into backend-specific helpers
- Add tool_parser field to ModelConfig for proper serialization

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: Lawhy

src/strands_env/cli/__init__.py

@@ -110,6 +110,12 @@ def list_cmd():
     default=None,
     help="AWS role ARN for Bedrock (optional).",
 )
+@click.option(
+    "--tool-parser",
+    type=str,
+    default=None,
+    help="Tool parser: name (e.g., 'hermes', 'qwen_xml') or path to hook file.",
+)
 # Sampling params
 @click.option(
     "--temperature",
@@ -199,6 +205,7 @@ def eval_cmd(
     region: str | None,
     profile_name: str | None,
     role_arn: str | None,
+    tool_parser: str | None,
     # Sampling
     temperature: float,
     max_tokens: int,
@@ -260,6 +267,7 @@ def eval_cmd(
         base_url=base_url,
         model_id=model_id,
         tokenizer_path=tokenizer_path,
+        tool_parser=tool_parser,
         region=region,
         profile_name=profile_name,
         role_arn=role_arn,

src/strands_env/cli/config.py

@@ -51,6 +51,7 @@ class ModelConfig:
     # SGLang
     base_url: str = "http://localhost:30000"
     tokenizer_path: str | None = None  # Auto-detected if None
+    tool_parser: str | None = None  # Parser name or path to hook file
 
     # Bedrock
     model_id: str | None = None
@@ -67,6 +68,7 @@ def to_dict(self) -> dict:
             "backend": self.backend,
             "base_url": self.base_url,
             "tokenizer_path": self.tokenizer_path,
+            "tool_parser": self.tool_parser,
             "model_id": self.model_id,
             "region": self.region,
             "profile_name": self.profile_name,

src/strands_env/cli/utils.py

@@ -18,6 +18,7 @@
 
 import importlib.util
 from pathlib import Path
+from types import ModuleType
 from typing import TYPE_CHECKING, Callable
 
 import click
@@ -27,6 +28,8 @@
 from .config import EnvConfig, ModelConfig
 
 if TYPE_CHECKING:
+    from strands_sglang.tool_parsers import ToolParser
+
     from strands_env.eval import AsyncEnvFactory, Evaluator
 
 #: Type for the create_env_factory function exported by hook files.
@@ -36,82 +39,50 @@
 EvaluatorClass = type["Evaluator"]
 
 
-def build_model_factory(config: ModelConfig, max_concurrency: int) -> ModelFactory:
-    """Build a ModelFactory from ModelConfig.
+# ---------------------------------------------------------------------------
+# Hook Loading
+# ---------------------------------------------------------------------------
 
-    Args:
-        config: Model configuration.
-        max_concurrency: Max concurrent connections (for SGLang client pooling).
 
-    Returns:
-        ModelFactory callable.
-    """
-    sampling = config.sampling.to_dict()
+def _load_hook_module(path: Path, hook_name: str) -> ModuleType:
+    """Load a Python module from a file path.
 
-    if config.backend == "sglang":
-        from strands_env.utils.sglang import (
-            check_server_health,
-            get_cached_client,
-            get_cached_tokenizer,
-            get_model_id,
-        )
-
-        # Check server health before proceeding
-        try:
-            check_server_health(config.base_url)
-        except ConnectionError as e:
-            raise click.ClickException(str(e))
-
-        client = get_cached_client(config.base_url, max_concurrency)
-
-        # Resolve and backfill model_id/tokenizer_path for reproducibility
-        if not config.model_id:
-            config.model_id = get_model_id(config.base_url)
-        if not config.tokenizer_path:
-            config.tokenizer_path = config.model_id
-
-        tokenizer = get_cached_tokenizer(config.tokenizer_path)
-        return sglang_model_factory(
-            client=client, model_id=config.model_id, tokenizer=tokenizer, sampling_params=sampling
-        )
+    Args:
+        path: Path to the Python file.
+        hook_name: Name for the module (used in error messages).
 
-    elif config.backend == "bedrock":
-        from strands_env.utils.aws import get_assumed_role_session, get_boto3_session
+    Returns:
+        The loaded module.
 
-        if not config.model_id:
-            raise click.ClickException("--model-id is required for Bedrock backend")
-        if config.role_arn:
-            boto_session = get_assumed_role_session(config.role_arn, config.region)
-        else:
-            boto_session = get_boto3_session(config.region, config.profile_name)
-        return bedrock_model_factory(model_id=config.model_id, boto_session=boto_session, sampling_params=sampling)
+    Raises:
+        click.ClickException: If the file cannot be loaded.
+    """
+    spec = importlib.util.spec_from_file_location(hook_name, path)
+    if spec is None or spec.loader is None:
+        raise click.ClickException(f"Could not load {hook_name} file: {path}")
 
-    else:
-        raise click.ClickException(f"Unknown backend: {config.backend}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
 
 
-def load_env_hook(env_path: Path) -> EnvFactoryCreator:
+def load_env_hook(path: Path) -> EnvFactoryCreator:
     """Load environment hook file and return create_env_factory function.
 
     The hook file must export a `create_env_factory(model_factory, env_config)` function.
 
     Args:
-        env_path: Path to the Python hook file.
+        path: Path to the Python hook file.
 
     Returns:
         The create_env_factory function from the hook file.
 
     Raises:
         click.ClickException: If the file cannot be loaded or doesn't export the function.
     """
-    spec = importlib.util.spec_from_file_location("env_hook", env_path)
-    if spec is None or spec.loader is None:
-        raise click.ClickException(f"Could not load hook file: {env_path}")
+    module = _load_hook_module(path, "env_hook")
 
-    hook = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(hook)
-
-    if not hasattr(hook, "create_env_factory"):
+    if not hasattr(module, "create_env_factory"):
         raise click.ClickException(
             "Hook file must export 'create_env_factory(model_factory, env_config)' function.\n"
             "Example:\n"
@@ -125,16 +96,16 @@ def load_env_hook(env_path: Path) -> EnvFactoryCreator:
             "      return env_factory"
         )
 
-    return hook.create_env_factory
+    return module.create_env_factory
 
 
-def load_evaluator_hook(evaluator_path: Path) -> EvaluatorClass:
+def load_evaluator_hook(path: Path) -> EvaluatorClass:
     """Load evaluator hook file and return the Evaluator class.
 
     The hook file must export an `EvaluatorClass` that extends `Evaluator`.
 
     Args:
-        evaluator_path: Path to the Python hook file.
+        path: Path to the Python hook file.
 
     Returns:
         The Evaluator subclass from the hook file.
@@ -144,14 +115,9 @@ def load_evaluator_hook(evaluator_path: Path) -> EvaluatorClass:
     """
     from strands_env.eval import Evaluator
 
-    spec = importlib.util.spec_from_file_location("evaluator_hook", evaluator_path)
-    if spec is None or spec.loader is None:
-        raise click.ClickException(f"Could not load evaluator hook file: {evaluator_path}")
-
-    hook = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(hook)
+    module = _load_hook_module(path, "evaluator_hook")
 
-    if not hasattr(hook, "EvaluatorClass"):
+    if not hasattr(module, "EvaluatorClass"):
         raise click.ClickException(
             "Evaluator hook file must export 'EvaluatorClass' (an Evaluator subclass).\n"
             "Example:\n"
@@ -166,8 +132,161 @@ def load_evaluator_hook(evaluator_path: Path) -> EvaluatorClass:
             "  EvaluatorClass = MyEvaluator"
         )
 
-    evaluator_cls = hook.EvaluatorClass
+    evaluator_cls = module.EvaluatorClass
     if not isinstance(evaluator_cls, type) or not issubclass(evaluator_cls, Evaluator):
         raise click.ClickException("EvaluatorClass must be a subclass of Evaluator")
 
     return evaluator_cls
+
+
+def load_tool_parser(tool_parser_arg: str | None) -> ToolParser | None:
+    """Load tool parser from name or hook file path.
+
+    Args:
+        tool_parser_arg: Either a parser name (e.g., "hermes", "qwen_xml") or path to hook file.
+
+    Returns:
+        ToolParser instance, or None if not specified.
+
+    Raises:
+        click.ClickException: If the parser name is unknown or hook file is invalid.
+    """
+    if tool_parser_arg is None:
+        return None
+
+    # Check if it's a file path
+    path = Path(tool_parser_arg)
+    if path.exists() and path.is_file():
+        return _load_tool_parser_hook(path)
+
+    # Otherwise treat as parser name
+    from strands_sglang.tool_parsers import get_tool_parser
+
+    try:
+        return get_tool_parser(tool_parser_arg)
+    except KeyError as e:
+        raise click.ClickException(str(e))
+
+
+def _load_tool_parser_hook(path: Path) -> ToolParser:
+    """Load tool parser from hook file.
+
+    The hook file must export either:
+    - `tool_parser`: A ToolParser instance
+    - `ToolParserClass`: A ToolParser subclass (will be instantiated)
+
+    Args:
+        path: Path to the Python hook file.
+
+    Returns:
+        ToolParser instance from the hook file.
+
+    Raises:
+        click.ClickException: If the file cannot be loaded or doesn't export the parser.
+    """
+    from strands_sglang.tool_parsers import ToolParser
+
+    module = _load_hook_module(path, "tool_parser_hook")
+
+    # Check for tool_parser instance first
+    if hasattr(module, "tool_parser"):
+        parser = module.tool_parser
+        if not isinstance(parser, ToolParser):
+            raise click.ClickException("'tool_parser' must be a ToolParser instance")
+        return parser
+
+    # Check for ToolParserClass
+    if hasattr(module, "ToolParserClass"):
+        parser_cls = module.ToolParserClass
+        if not isinstance(parser_cls, type) or not issubclass(parser_cls, ToolParser):
+            raise click.ClickException("'ToolParserClass' must be a ToolParser subclass")
+        return parser_cls()
+
+    raise click.ClickException(
+        "Tool parser hook file must export 'tool_parser' (instance) or 'ToolParserClass' (subclass).\n"
+        "Example:\n"
+        "  from strands_sglang.tool_parsers import ToolParser, ToolParseResult\n"
+        "\n"
+        "  class MyToolParser(ToolParser):\n"
+        "      def parse(self, text: str) -> list[ToolParseResult]:\n"
+        "          ...\n"
+        "\n"
+        "  tool_parser = MyToolParser()\n"
+        "  # OR\n"
+        "  ToolParserClass = MyToolParser"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Model Factory
+# ---------------------------------------------------------------------------
+
+
+def build_model_factory(config: ModelConfig, max_concurrency: int) -> ModelFactory:
+    """Build a ModelFactory from ModelConfig.
+
+    Args:
+        config: Model configuration.
+        max_concurrency: Max concurrent connections (for SGLang client pooling).
+
+    Returns:
+        ModelFactory callable.
+    """
+    sampling = config.sampling.to_dict()
+
+    if config.backend == "sglang":
+        return _build_sglang_model_factory(config, max_concurrency, sampling)
+    elif config.backend == "bedrock":
+        return _build_bedrock_model_factory(config, sampling)
+    else:
+        raise click.ClickException(f"Unknown backend: {config.backend}")
+
+
+def _build_sglang_model_factory(config: ModelConfig, max_concurrency: int, sampling: dict) -> ModelFactory:
+    """Build SGLang model factory."""
+    from strands_env.utils.sglang import (
+        check_server_health,
+        get_cached_client,
+        get_cached_tokenizer,
+        get_model_id,
+    )
+
+    # Check server health before proceeding
+    try:
+        check_server_health(config.base_url)
+    except ConnectionError as e:
+        raise click.ClickException(str(e))
+
+    client = get_cached_client(config.base_url, max_concurrency)
+
+    # Resolve and backfill model_id/tokenizer_path for reproducibility
+    if not config.model_id:
+        config.model_id = get_model_id(config.base_url)
+    if not config.tokenizer_path:
+        config.tokenizer_path = config.model_id
+
+    tokenizer = get_cached_tokenizer(config.tokenizer_path)
+    tool_parser = load_tool_parser(config.tool_parser)
+
+    return sglang_model_factory(
+        client=client,
+        model_id=config.model_id,
+        tokenizer=tokenizer,
+        tool_parser=tool_parser,
+        sampling_params=sampling,
+    )
+
+
+def _build_bedrock_model_factory(config: ModelConfig, sampling: dict) -> ModelFactory:
+    """Build Bedrock model factory."""
+    from strands_env.utils.aws import get_assumed_role_session, get_boto3_session
+
+    if not config.model_id:
+        raise click.ClickException("--model-id is required for Bedrock backend")
+
+    if config.role_arn:
+        boto_session = get_assumed_role_session(config.role_arn, config.region)
+    else:
+        boto_session = get_boto3_session(config.region, config.profile_name)
+
+    return bedrock_model_factory(model_id=config.model_id, boto_session=boto_session, sampling_params=sampling)

src/strands_env/core/models.py

@@ -67,7 +67,7 @@ def sglang_model_factory(
     model_id: str,
     tokenizer: PreTrainedTokenizerBase,
     client: SGLangClient,
-    tool_parser: ToolParser = HermesToolParser(),
+    tool_parser: ToolParser | None = None,
     sampling_params: dict[str, Any] = DEFAULT_SAMPLING_PARAMS,
     enable_thinking: bool | None = None,
 ) -> ModelFactory:
@@ -77,9 +77,13 @@ def sglang_model_factory(
         model_id: SGLang model identifier.
         tokenizer: HuggingFace tokenizer for chat template and tokenization.
         client: `SGLangClient` for HTTP communication with the SGLang server.
+        tool_parser: Tool parser for extracting tool calls from model output. Defaults to `HermesToolParser`.
         sampling_params: Sampling parameters for the model (e.g. `{"max_new_tokens": 4096}`).
         enable_thinking: Enable thinking mode for Qwen3 hybrid models.
     """
+    if tool_parser is None:
+        tool_parser = HermesToolParser()
+
     return lambda: SGLangModel(
         tokenizer=tokenizer,
         client=client,