Как исправить ошибку python-a2a «Failed to communicate with agent»

Разбираем ошибку python-a2a «Failed to communicate with agent»: почему клиент получает error вместо function_response и как починить базовый URL и порт.

При интеграции легковесного агента для вызова функций с python-a2a распространённая проблема выглядит как вполне корректный запрос, который вместо function_response возвращает оболочку с ошибкой. Консольный вывод говорит прямо: клиент попытался обратиться к агенту, перебрал несколько вариантов конечной точки и всё равно получил ответ с типом error.

Неожиданный тип ответа от A2A-агента: error. Полное содержимое ответа: ErrorContent(message='Failed to communicate with agent ... Tried multiple endpoint variations.')

Пример кода, который демонстрирует проблему

Ниже приведены минимальные агент и клиент, показывающие конфигурацию. Обработчик выполняет базовые арифметические операции и возвращает результат через function_response. Логика повторяет рабочий шаблон.

import math
from python_a2a import (
    A2AServer, Message, TextContent, FunctionCallContent,
    FunctionResponseContent, FunctionParameter, MessageRole, run_server,
    Task
)
class MathOpsNode(A2AServer):
    def handle_message(self, msg):
        if msg.content.type == "text":
            return Message(
                content=TextContent(
                    text="I'm a [Python A2A](python-a2a.html) calculator agent. You can call my functions:\n"
                         "- calculate: Basic arithmetic (operation, a, b)\n"
                         "- sqrt: Square root (value)"
                ),
                role=MessageRole.AGENT,
                parent_message_id=msg.message_id,
                conversation_id=msg.conversation_id
            )
        elif msg.content.type == "function_call":
            fname = msg.content.name
            args_map = {p.name: p.value for p in msg.content.parameters}
            try:
                if fname == "calculate":
                    op = args_map.get("operation", "add")
                    x = float(args_map.get("a", 0))
                    y = float(args_map.get("b", 0))
                    if op == "add":
                        out_val = x + y
                    elif op == "subtract":
                        out_val = x - y
                    elif op == "multiply":
                        out_val = x * y
                    elif op == "divide":
                        if y == 0:
                            raise ValueError("Cannot divide by zero")
                        out_val = x / y
                    else:
                        raise ValueError(f"Unknown operation: {op}")
                    return Message(
                        content=FunctionResponseContent(
                            name="calculate",
                            response={"result": out_val}
                        ),
                        role=MessageRole.AGENT,
                        parent_message_id=msg.message_id,
                        conversation_id=msg.conversation_id
                    )
                elif fname == "sqrt":
                    val = float(args_map.get("value", 0))
                    if val < 0:
                        raise ValueError("Cannot calculate square root of negative number")
                    out_val = math.sqrt(val)
                    return Message(
                        content=FunctionResponseContent(
                            name="sqrt",
                            response={"result": out_val}
                        ),
                        role=MessageRole.AGENT,
                        parent_message_id=msg.message_id,
                        conversation_id=msg.conversation_id
                    )
            except Exception as err:
                return Message(
                    content=FunctionResponseContent(
                        name=fname,
                        response={"error": str(err)}
                    ),
                    role=MessageRole.AGENT,
                    parent_message_id=msg.message_id,
                    conversation_id=msg.conversation_id
                )
if __name__ == "__main__":
    svc = MathOpsNode()
    run_server(svc, host="0.0.0.0", port=5001)

import python_a2a
from python_a2a import (
    A2AClient, Message, FunctionCallContent,
    FunctionParameter, MessageRole
)
class CalcA2AInvoker:
    """
    Encapsulates interaction with a Python A2A agent.
    """
    def __init__(self, base_url: str):
        """
        Args:
            base_url: The A2A agent base URL (e.g., "http://127.0.0.1:5001").
        """
        self.conn = A2AClient(base_url)
        print(f"CalcA2AInvoker initialized for URL: {base_url}")
    def request_compute(self, operation: str, a: int, b: int) -> int | None:
        print(f"Sending calculation request: {operation}(a={a}, b={b})")
        call_payload = FunctionCallContent(
            name="calculate",
            parameters=[
                FunctionParameter(name="operation", value=operation),
                FunctionParameter(name="a", value=a),
                FunctionParameter(name="b", value=b)
            ]
        )
        outbound_msg = Message(
            content=call_payload,
            role=MessageRole.USER
        )
        try:
            reply = self.conn.send_message(outbound_msg)
            if reply.content.type == "function_response":
                result = reply.content.response.get("result")
                if result is not None:
                    print(f"Received result from A2A agent: {result}")
                    return result
                else:
                    print("Function response received, but 'result' key is missing.")
                    return None
            else:
                print(f"Unexpected response type from A2A agent: {reply.content.type}")
                print(f"Full response content: {reply.content}")
                return None
        except Exception as ex:
            print(f"An error occurred while communicating with the A2A agent: {ex}")
            return None
if __name__ == "__main__":
    endpoint = "http://127.0.0.1:5001"
    client = CalcA2AInvoker(endpoint)
    res = client.request_compute(operation="add", a=5, b=3)
    if res is not None:
        print(f"Addition Result: {res}")
    print("-" * 30)

Что на самом деле происходит

Клиент сообщает, что отправил запрос calculate, а затем фиксирует тип ответа error. В полезной нагрузке ошибки явно указано, что связаться с агентом не удалось, и приведены конкретные пути вроде /tasks/send или добавленного /a2a/a2a. Пакет за кадром пробует несколько вариантов конечных точек; если ни один не срабатывает, возвращается ErrorContent с типом error.

Локальный запуск агента и клиента, как показано выше, даёт корректный function_response с ожидаемым числовым результатом. Это указывает на то, что сам код в порядке. Такой сбой соответствует проблеме доступности по базовому URL и порту или использованию неподходящего базового URL, который не указывает на запущенный процесс агента.

Есть и косвенные подтверждения из окружений. В одной конфигурации тот же подход успешно отработал на Linux Mint с Python 3.13 и python-a2a 0.5.9. В другой — на Ubuntu при отключённом файрволе — ошибка сохранилась, что подчёркивает: дело не в логике примера, а, скорее всего, в достижимости сервиса и фактическом адресе назначения.

Решение

Используйте именно базовый URL агента и не добавляйте лишних сегментов пути. В примере агент запущен на порту 5001, и рабочий запуск использовал http://127.0.0.1:5001 в качестве базового URL клиента. Если указываете другой хост или нелокальный адрес, убедитесь, что адрес корректен и доступен, а трафик по порту 5001 разрешён в обоих направлениях. Полезно также проверить в браузере, что агент действительно отвечает по нужному адресу. Если работаете между разными машинами, проверьте, что конечная машина — именно та, где запущен сервис агента, и что маршрутизация посылает трафик туда.

Ниже тот же шаблон вызова, что и выше, но с базовым URL, указывающим на адрес, где слушает агент.

if __name__ == "__main__":
    endpoint = "http://127.0.0.1:5001"  # используйте базовый URL без дополнительного пути
    client = CalcA2AInvoker(endpoint)
    res = client.request_compute(operation="add", a=5, b=3)
    if res is not None:
        print(f"Addition Result: {res}")
    print("-" * 30)

Почему это важно

Когда клиент возвращает ErrorContent со строкой «Failed to communicate with agent … Tried multiple endpoint variations», легко заподозрить проблему пакета или версий. На практике экономит время простая проверка: базовый URL, точный используемый адрес и доступность порта. Это помогает не уходить в дебри отладки зависимостей, когда настоящим узким местом является транспорт.

Что запомнить

Начните с простого пути: запустите оба компонента локально с http://127.0.0.1:5001 и убедитесь, что получаете function_response. Если нужен нелокальный адрес, проверьте, что агент действительно поднят по этому адресу, что базовый URL соответствует способу его публикации и что порт 5001 разрешён. Есть по крайней мере одна проверенная связка — Linux Mint, Python 3.13, python-a2a 0.5.9 — где пример работает как ожидается, что лишний раз подчёркивает важность проверки соединения прежде, чем менять код. Если отключение файрвола не помогает в вашем окружении, вернитесь к проверкам URL и маршрутизации: вывод ошибки уже подсказывает, что клиент не может завершить запрос к целевому эндпоинту.