<< All versions
Skill v1.0.0
currentTrusted Publisher100/100microsoft/amplifier-module-tool-skills/module-development
──Details
PublishedOctober 14, 2026 at 07:32 PM
Content Hashsha256:d0dece4373169ed0...
Git SHA
──Files
Files (1 file, 6.0 KB)
SKILL.md6.0 KBactive
SKILL.md · 228 lines · 6.0 KB
name: module-development description: Guide for creating new Amplifier modules including protocol implementation, entry points, mount functions, and testing patterns. Use when creating new modules or understanding module architecture. version: 1.0.0 license: MIT metadata: category: development complexity: medium
Module Development Guide
Creating a New Module
Step 1: Choose Protocol
Determine which protocol your module implements:
- Tool: Agent capability (filesystem, bash, web, database)
- Provider: LLM backend (Anthropic, OpenAI, Ollama)
- Context: Conversation state management (simple, persistent)
- Orchestrator: Execution loop strategy (basic, streaming, events)
- Hook: Lifecycle observer (logging, approval, redaction)
Step 2: Implement Protocol
python
from typing import Anyfrom amplifier_core import ModuleCoordinator, ToolResultclass MyTool:"""Tool for doing something useful."""name = "my-tool"description = "Does something useful"def __init__(self: "MyTool", config: dict[str, Any]) -> None:"""Initialize tool with configuration."""self.config = configself.timeout = config.get("timeout", 30)@propertydef input_schema(self: "MyTool") -> dict:"""Return JSON schema for tool parameters."""return {"type": "object","properties": {"param": {"type": "string", "description": "Parameter description"}},"required": ["param"]}async def execute(self: "MyTool", input: dict[str, Any]) -> ToolResult:"""Execute tool operation."""param = input.get("param")if not param:return ToolResult(success=False,error={"message": "param is required"})# Implementation hereresult = f"Processed: {param}"return ToolResult(success=True,output={"result": result})
Step 3: Create Mount Function
python
async def mount(coordinator: ModuleCoordinator, config: dict[str, Any] | None = None) -> None:"""Mount the tool module.Args:coordinator: Module coordinator providing infrastructureconfig: Module configuration from profileReturns:Optional cleanup function"""config = config or {}tool = MyTool(config)await coordinator.mount("tools", tool, name=tool.name)logger.info("Mounted MyTool")return
Step 4: Register Entry Point
toml
# pyproject.toml[project]name = "amplifier-module-tool-mytool"version = "0.1.0"requires-python = ">=3.11"dependencies = ["amplifier-core",][project.entry-points."amplifier.modules"]tool-mytool = "amplifier_module_tool_mytool:mount"[tool.uv.sources]amplifier-core = { path = "../amplifier-core", editable = true }
Module Structure
Required files:
amplifier-module-tool-mytool/├── amplifier_module_tool_mytool/│ ├── __init__.py # mount() + tool class│ └── (optional modules)├── tests/│ └── test_mytool.py├── pyproject.toml├── Makefile└── README.md
Using Coordinator Infrastructure
The coordinator provides infrastructure context:
python
async def mount(coordinator: ModuleCoordinator, config: dict[str, Any] | None = None) -> None:# Access infrastructuresession_id = coordinator.session_id # Current sessionparent_id = coordinator.parent_id # Parent session (if child)session_config = coordinator.config # Full session configurationloader = coordinator.loader # Dynamic module loading# Emit eventsawait coordinator.hooks.emit("module:mounted", {"module_type": "tool","module_name": "my-tool"})# Register capabilities (optional)coordinator.register_capability("my-tool.version", "1.0.0")# Register cleanup (optional)async def cleanup():logger.info("Cleaning up MyTool")coordinator.register_cleanup(cleanup)
Testing Modules
Use TestCoordinator:
python
import pytestfrom amplifier_core.testing import TestCoordinatorfrom amplifier_module_tool_mytool import mount@pytest.mark.asyncioasync def test_tool_execution():coordinator = TestCoordinator()# Mount moduleawait mount(coordinator, {"timeout": 10})# Get tooltool = coordinator.get("tools", "my-tool")assert tool is not None# Executeresult = await tool.execute({"param": "test"})assert result.successassert "Processed: test" in result.output["result"]
Common Patterns
Configuration Handling
python
class MyTool:def __init__(self: "MyTool", config: dict[str, Any]) -> None:# Required configself.required = config.get("required_param")if not self.required:raise ValueError("required_param must be provided")# Optional config with defaultsself.timeout = config.get("timeout", 30)self.retries = config.get("retries", 3)
Event Emission
python
async def execute(self: "MyTool", input: dict[str, Any]) -> ToolResult:# Events emitted by orchestrator via hooks, not by tools directly# Tools just return resultsresult = await self._process(input)return ToolResult(success=True, output=result)
Error Patterns
python
# Return error, don't raisereturn ToolResult(success=False,error={"message": "Clear error message", "code": "ERROR_CODE"})# Log for debugginglogger.error(f"Failed to process {input}: {e}")return ToolResult(success=False, error={"message": str(e)})
Remember
- One responsibility per module - Don't create mega-modules
- Config via mount() - No hard-coded defaults in modules
- Async everywhere - All I/O must be async
- Protocols not inheritance - Implement interface, don't inherit
- Test at protocol level - Test behavior, not internals