execution_registry.py 1.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051
  1. from __future__ import annotations
  2. from dataclasses import dataclass
  3. from .commands import PORTED_COMMANDS, execute_command
  4. from .tools import PORTED_TOOLS, execute_tool
  5. @dataclass(frozen=True)
  6. class MirroredCommand:
  7. name: str
  8. source_hint: str
  9. def execute(self, prompt: str) -> str:
  10. return execute_command(self.name, prompt).message
  11. @dataclass(frozen=True)
  12. class MirroredTool:
  13. name: str
  14. source_hint: str
  15. def execute(self, payload: str) -> str:
  16. return execute_tool(self.name, payload).message
  17. @dataclass(frozen=True)
  18. class ExecutionRegistry:
  19. commands: tuple[MirroredCommand, ...]
  20. tools: tuple[MirroredTool, ...]
  21. def command(self, name: str) -> MirroredCommand | None:
  22. lowered = name.lower()
  23. for command in self.commands:
  24. if command.name.lower() == lowered:
  25. return command
  26. return None
  27. def tool(self, name: str) -> MirroredTool | None:
  28. lowered = name.lower()
  29. for tool in self.tools:
  30. if tool.name.lower() == lowered:
  31. return tool
  32. return None
  33. def build_execution_registry() -> ExecutionRegistry:
  34. return ExecutionRegistry(
  35. commands=tuple(MirroredCommand(module.name, module.source_hint) for module in PORTED_COMMANDS),
  36. tools=tuple(MirroredTool(module.name, module.source_hint) for module in PORTED_TOOLS),
  37. )