Spaces:
Running
Running
File size: 2,514 Bytes
dba8b8e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | import xml.etree.ElementTree as ET
from typing import Any, Dict, Optional
import httpx
from app.tool.base import BaseTool
from app.utils.spatial_grid import SpatialGridManager
from app.logger import logger
class RSSMatrixSync(BaseTool):
"""
Synchronizes RSS feed items to the Agent Matrix via spatial indexing.
Mimics MCP functionality by populating grid nodes with live data.
"""
name: str = "rss_matrix_sync"
description: str = "Fetch RSS feed and sync items to matrix nodes using URI-based spatial indexing."
parameters: dict = {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The URL of the RSS feed to synchronize."
}
},
"required": ["url"]
}
async def execute(self, url: str) -> str:
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url)
response.raise_for_status()
root = ET.fromstring(response.content)
items = root.findall(".//item")
grid_mgr = SpatialGridManager()
sync_count = 0
assignments = []
for item in items:
title = item.find("title").text if item.find("title") is not None else "No Title"
link = item.find("link").text if item.find("link") is not None else ""
pub_date = item.find("pubDate").text if item.find("pubDate") is not None else ""
description = item.find("description").text if item.find("description") is not None else ""
if not link:
continue
content = {
"title": title,
"published": pub_date,
"summary": description[:500] if description else "" # Truncate for context efficiency
}
result = grid_mgr.sync_data_to_node(link, content)
if result:
agent_key, coords = result
assignments.append(f"Item '{title}' -> Node {coords} ({agent_key})")
sync_count += 1
return f"Successfully synchronized {sync_count} items to the spatial grid.\n" + "\n".join(assignments[:10])
except Exception as e:
logger.error(f"RSS Spatial Sync failed: {e}")
return f"Error during spatial synchronization: {str(e)}"
|