Разобрался. Причина — в семантике @agent.tool в pydantic-ai, и в том, что AMS даёт в step один и тот же объект агента на каждой итерации.
Почему падает
@agent.tool → FunctionToolset.add_function() → add_tool(), а там жёсткая проверка:
def add_tool(self, tool):
    if tool.name in self.tools:
        raise UserError(f'Tool name conflicts with existing tool: {tool.name!r}')
    ...
    self.tools[tool.name] = tool
step вызывается много раз, но agent переиспользуется (именно поэтому agent.history и set_system_prompt имеют смысл — история копится в одном объекте). Значит на второй итерации регистрация read_file_lines с тем же именем бросает UserError: Tool name conflicts with existing tool: 'read_file_lines' (или молча затирается — в зависимости от версии, в 2.33 именно исключение).
Плюс AMS-обёртка StatefulAgent (это Cython-класс в /opt/ams/app/processors/tools/stateful_agent.so) проксирует .tool на внутренний pydantic_ai.Agent через __getattr__ — так что «re-register» бьёт прямо по тулсету внутреннего агента.
Как правильно (рекомендую вариант 1)
Вариант 1 — вообще не трогать агент, передавать тулсет на каждый run
Тулсет создаётся один раз при импорте модуля, а в ран передаётся через toolsets=. Ничего не накапливается на агенте → конфликта нет по построению:
import os
from datetime import datetime
from pydantic_ai import ModelSettings, RunContext, ModelRetry, FunctionToolset
from pydantic_ai.workspaces import LocalWorkspaceBackend
from pydantic_ai_harness.filesystem import FileSystem

CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
system_prompt = open(os.path.join(CURRENT_DIR, 'system_prompt.txt')).read()

# инструмент объявлен на уровне модуля (один раз)
async def read_file_lines(ctx: RunContext, path: str,
                          start_line: int = 1, max_lines: int = 50) -> str:
    """Read a specific range of lines from a text file..."""
    if start_line < 1:
        raise ModelRetry('start_line must be 1 or greater.')
    try:
        content = await ctx.workspace.read_text(path)
    except FileNotFoundError:
        raise ModelRetry(f"File '{path}' does not exist.")
    lines = content.splitlines()
    start_idx = start_line - 1
    end_idx = start_idx + max_lines
    chunk = lines[start_idx:end_idx]
    if not chunk:
        return f"EOF: No lines found in range {start_line}-{end_idx}."
    return f"[Lines {start_line}-{min(end_idx, len(lines))} of {len(lines)}]:\n" + "\n".join(chunk)

# создаётся ОДИН раз
TOOLSET = FunctionToolset(tools=[read_file_lines])
WORKSPACE = LocalWorkspaceBackend('./logs')

def step(agent, state, params):
    dt = datetime.now().strftime("%d.%m.%Y %H:%M")
    agent.set_system_prompt(system_prompt)

    model_settings = ModelSettings(**params['sampling_params'])
    prompt = params['prompt']
    # ... ваш блок с params['file'] ...

    if not prompt:
        return

    with agent.run_stream_sync(
        prompt,
        model_settings=model_settings,
        toolsets=[TOOLSET],          # <-- тул приходит на этот ран, не мутируя агента
        capabilities=[FileSystem()],
        workspace=WORKSPACE,
    ) as stream:
        for chunk in stream.stream_text(delta=True):
            yield {'type': 'chunk', 'data': chunk}
Вариант 2 — оставить @agent.tool, но регистрировать идемпотентно
Если хочется именно закрепить тул на агенте — вынесите функцию из step и защититесь флагом на объекте агента (не глобальным bool — в процессе может жить несколько агентов):
def _register_tools_once(agent):
    if getattr(agent, "_file_tools_registered", False):
        return
    agent.tool(read_file_lines)
    agent._file_tools_registered = True
и в начале step: _register_tools_once(agent).
Отличие: вариант 1 не зависит от того, переиспользуется ли объект агента/обёртки; вариант 2 надёжен, только пока AMS передаёт тот же самый объект agent. Поэтому вариант 1 безопаснее.
Ещё два момента
- capabilities=[FileSystem()] уже даёт файловые тулы (read_file, edit_file, list_files, grep, с параметрами max_read_lines/max_read_chars и т.п.). Так что ваш read_file_lines может быть избыточен — если нужен именно построчный диапазон + ModelRetry, оставляйте, но дублирующиеся имена (read_file) не создавайте.
- Проверьте окружение на сервере. На 192.168.102.120 в вене AMS стоит pydantic_ai 2.33.0, а в нём нет pydantic_ai.workspaces, не установлен pydantic_ai_harness, и параметра workspace= у run_stream_sync нет (есть deps, toolsets, capabilities). Актуальные версии — pydantic-ai 2.56.0 / pydantic-ai-harness 0.56.0. То есть этот handler.py на текущем серверном AMS не заимпортится — нужен либо апгрейд/установка harness, либо адаптация под 2.33 (capabilities=[FileSystem()] там есть, а workspace-механику придётся заменить).
То же самое с сигнатурой: в хелпе AMS (/opt/ams/app/help/agents/index.html) контракт — step(params, state, agent), но «имена аргументов являются частью контракта», поэтому ваш порядок (agent, state, params) сработает, если AMS вызывает по ключевым словам.