Создание погодного агента на Python с использованием протокола Agent Communication Protocol (ACP)

Что такое Agent Communication Protocol (ACP)

ACP — это открытый стандарт, который позволяет ИИ-агентам, приложениям и людям эффективно взаимодействовать, несмотря на различия в инфраструктуре и фреймворках. Он объединяет разные способы коммуникации через RESTful API, поддерживая мультимодальную связь, синхронные и асинхронные сообщения, потоковую передачу данных в реальном времени, состояние взаимодействия агентов, их обнаружение и выполнение длительных задач.

Настройка окружения

Для начала установите необходимые библиотеки Python:

pip install acp acp-sdk beeai-framework httpx

Создание ACP сервера

Создайте файл agent.py и импортируйте нужные библиотеки, включая httpx для получения данных о погоде через API Open-Meteo.

import asyncio
from collections.abc import AsyncGenerator
import httpx                              
 
from acp_sdk.models import Message, MessagePart
from acp_sdk.server import Context, RunYield, RunYieldResume, Server
 
server = Server()

Определите асинхронную функцию get_london_weather, которая получает текущую погоду в Лондоне:

async def get_london_weather() -> str:
    """Получение текущей погоды в Лондоне через бесплатный API Open-Meteo."""
    params = {
        "latitude": 51.5072,          # координаты Лондона
        "longitude": -0.1276,
        "current_weather": True,
        "timezone": "Europe/London"
    }
    url = "https://api.open-meteo.com/v1/forecast"
 
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(url, params=params)
        resp.raise_for_status()
        cw = resp.json()["current_weather"]
 
    return (
        f"Погода в Лондоне: {cw['temperature']} °C, "
        f"ветер {cw['windspeed']} км/ч, код {cw['weathercode']}."
    )

Используйте декоратор @server.agent() для определения агента ACP, который обрабатывает входящие сообщения и возвращает погодные данные:

@server.agent()
async def london_weather_agent(
    input: list[Message], context: Context
) -> AsyncGenerator[RunYield, RunYieldResume]:
    """Возвращает текущую погоду в Лондоне."""
    for _ in input:
        yield {"thought": "Получение погоды в Лондоне..."}
        weather = await get_london_weather()
        yield Message(
            role="agent",
            parts=[MessagePart(content=weather, content_type="text/plain")]
        )
 
server.run()

Запуск ACP сервера

Запустите сервер командой:

python agent.py

Проверьте, что сервер работает, запросив список агентов:

curl http://localhost:8000/agents

Вы получите JSON с подтверждением, что агент london_weather_agent готов к работе.

Создание ACP клиента

Создайте скрипт client.py для взаимодействия с ACP сервером. Клиент отправляет синхронное сообщение с просьбой о погоде:

import asyncio
 
from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart
 
 
async def call_london_weather_agent() -> None:
    async with Client(base_url="http://localhost:8000") as client:
        run = await client.run_sync(
            agent="london_weather_agent",
            input=[
                Message(
                    parts=[MessagePart(content="Tell me the weather", content_type="text/plain")]
                )
            ],
        )
 
        print("Ответ от london_weather_agent:")
        for message in run.output:
            for part in message.parts:
                print("-", part.content)
 
 
if __name__ == "__main__":
    asyncio.run(call_london_weather_agent())

Запуск клиента

В другом терминале выполните:

python client.py

Вы увидите ответ с текущей погодой в Лондоне, например:

Ответ от london_weather_agent:
- Погода в Лондоне: 20.8 °C, ветер 10.1 км/ч, код 3.

Этот пример показывает, как ACP упрощает создание совместимых ИИ-агентов и позволяет сосредоточиться на функциональности без сложностей коммуникации.