Python Coding Standards

Type Hints

ALL functions must have complete type hints:

python

from typing import Any

async def process_data(items: list[str], config: dict[str, Any]) -> dict[str, Any]:
    """Process data items with configuration."""
    results = {}
    for item in items:
        results[item] = await transform(item, config)
    return results

Include type hints for self:

python

class MyClass:
    def __init__(self: "MyClass", name: str) -> None:
        self.name = name

    async def process(self: "MyClass") -> str:
        return f"Processing {self.name}"

Async Patterns

All I/O operations must be async:

python

# Good
async def read_file(path: Path) -> str:
    content = path.read_text()  # For now, sync is OK
    return content

# Better (when using async libraries)
async def read_file(path: Path) -> str:
    async with aiofiles.open(path) as f:
        return await f.read()

Use asyncio.gather for parallel operations:

python

async def process_files(files: list[Path]) -> list[dict]:
    tasks = [process_file(f) for f in files]
    return await asyncio.gather(*tasks)

Error Handling

Return errors, don't raise:

python

from amplifier_core import ToolResult

async def execute(self: "MyTool", input: dict[str, Any]) -> ToolResult:
    """Execute tool operation."""
    try:
        result = await self._process(input)
        return ToolResult(success=True, output=result)
    except ValueError as e:
        logger.error(f"Validation error: {e}")
        return ToolResult(success=False, error={"message": str(e)})

Provide clear error messages:

python

# Good
return ToolResult(
    success=False,
    error={"message": f"File not found: {path}"}
)

# Bad
return ToolResult(
    success=False,
    error={"message": "Error"}
)

Formatting

Line length: 120 characters

Import organization:

python

# Standard library
import asyncio
import logging
from pathlib import Path
from typing import Any

# Third-party
import yaml
from pydantic import BaseModel

# Local/Amplifier
from amplifier_core import ModuleCoordinator, ToolResult

Files must end with newline - Add blank line at EOF

Use ruff for formatting:

bash

uv run ruff format .
uv run ruff check . --fix

Dependencies

Use uv for dependency management:

bash

# Add dependency
cd amplifier-module-tool-mytool
uv add package-name

# Add dev dependency
uv add --dev pytest ruff pyright

Never manually edit pyproject.toml dependencies - Use uv add

Testing

Test behavior at protocol level:

python

import pytest
from amplifier_core.testing import TestCoordinator

@pytest.mark.asyncio
async def test_tool_basic():
    """Test basic tool functionality."""
    coordinator = TestCoordinator()

    # Mount module
    await mount(coordinator, {"timeout": 10})

    # Get and test
    tool = coordinator.get("tools", "my-tool")
    result = await tool.execute({"param": "value"})

    assert result.success
    assert "expected" in result.output

Test pyramid: 60% unit, 30% integration, 10% end-to-end

Common Pitfalls

Don't use blocking I/O

python

# Bad
content = requests.get(url).text

# Good
async with aiohttp.ClientSession() as session:
    async with session.get(url) as response:
        content = await response.text()

Don't modify module internals from outside

python

# Bad
tool._internal_state = new_value

# Good
await tool.execute({"operation": "update", "value": new_value})

Don't put logic in mount()

python

# Bad
async def mount(coordinator, config):
    tool = MyTool()
    await tool.initialize_database()  # Heavy logic
    await coordinator.mount("tools", tool)

# Good
async def mount(coordinator, config):
    tool = MyTool(config)  # Light initialization only
    await coordinator.mount("tools", tool)
    # Heavy logic happens in execute(), not mount()

Remember

•Run make check before committing
•Keep modules focused and simple
•Document public interfaces
•Test at the protocol level