Spaces:
Running
Running
File size: 1,821 Bytes
78e921e 34c6191 78e921e | 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 | import re
import sys
from typing import Dict, Any
# Support both standalone (core.*) and OpenManus (app.utils.*) deployments
try:
from app.utils.signal_registry import signal_registry # OpenManus full stack
from app.logger import logger
except (ImportError, AttributeError):
from core.signal_registry import signal_registry # standalone API
try:
from app.logger import logger
except (ImportError, AttributeError):
import logging
logger = logging.getLogger("signalmesh")
class AgentifiedToolParser:
"""
Parses natural language output from agents to detect and execute
tool-less actions (e.g., broadcasting to the mesh) passively.
"""
ACTION_PATTERN = re.compile(r'<execute\s+type="([^"]+)"(?:\s+topic="([^"]+)")?>\s*(.*?)\s*</execute>', re.DOTALL)
@classmethod
def parse_and_execute(cls, text: str) -> bool:
"""
Scans the text for <execute> tags and performs the corresponding action.
Returns True if an action was executed.
"""
executed = False
matches = cls.ACTION_PATTERN.finditer(text)
for match in matches:
action_type = match.group(1)
topic = match.group(2)
content = match.group(3).strip()
if action_type == "broadcast":
if not topic:
topic = "general"
logger.info(f"Agentified Tool: Passively executing broadcast to '{topic}'")
signal_registry.broadcast(
name=topic,
source_type="passive_agent_action",
data=content
)
executed = True
else:
logger.warning(f"Agentified Tool: Unknown action type '{action_type}'")
return executed
|