Perform a protocol transformation. Args: type: Transformation type (t2e, t2a, e2t, e2a, a2t, a2e) env_names: List of environment names (for e2t, e2a, a2e) tool_names: List of tool names (for t2e, t2a, a2t) agent_names: List of agent na
(self,
type: str,
env_names: Optional[List[str]] = None,
tool_names: Optional[List[str]] = None,
agent_names: Optional[List[str]] = None,
)
| 45 | logger.info("| 🔄 Transformation Server initialized") |
| 46 | |
| 47 | async def transform(self, |
| 48 | type: str, |
| 49 | env_names: Optional[List[str]] = None, |
| 50 | tool_names: Optional[List[str]] = None, |
| 51 | agent_names: Optional[List[str]] = None, |
| 52 | ) -> Any: |
| 53 | """Perform a protocol transformation. |
| 54 | |
| 55 | Args: |
| 56 | type: Transformation type (t2e, t2a, e2t, e2a, a2t, a2e) |
| 57 | env_names: List of environment names (for e2t, e2a, a2e) |
| 58 | tool_names: List of tool names (for t2e, t2a, a2t) |
| 59 | agent_names: List of agent names (for t2a, e2a, a2t, a2e) |
| 60 | |
| 61 | Returns: |
| 62 | Transformation response |
| 63 | """ |
| 64 | try: |
| 65 | logger.info(f"| 🔄 Starting transformation: {type}") |
| 66 | |
| 67 | # Route to appropriate transformation method |
| 68 | if type == TransformationType.E2T.value: |
| 69 | request = E2TRequest( |
| 70 | type=type, |
| 71 | env_names=env_names or [] |
| 72 | ) |
| 73 | result = await self.e2t_transformer.transform(request) |
| 74 | elif type == TransformationType.T2E.value: |
| 75 | request = T2ERequest( |
| 76 | type=type, |
| 77 | tool_names=tool_names or [] |
| 78 | ) |
| 79 | result = await self.t2e_transformer.transform(request) |
| 80 | elif type == TransformationType.T2A.value: |
| 81 | request = T2ARequest( |
| 82 | type=type, |
| 83 | tool_names=tool_names or [] |
| 84 | ) |
| 85 | result = await self.t2a_transformer.transform(request) |
| 86 | elif type == TransformationType.E2A.value: |
| 87 | request = E2ARequest( |
| 88 | type=type, |
| 89 | env_names=env_names or [] |
| 90 | ) |
| 91 | result = await self.e2a_transformer.transform(request) |
| 92 | elif type == TransformationType.A2T.value: |
| 93 | request = A2TRequest( |
| 94 | type=type, |
| 95 | agent_names=agent_names or [] |
| 96 | ) |
| 97 | result = await self.a2t_transformer.transform(request) |
| 98 | elif type == TransformationType.A2E.value: |
| 99 | request = A2ERequest( |
| 100 | type=type, |
| 101 | agent_names=agent_names or [] |
| 102 | ) |
| 103 | result = await self.a2e_transformer.transform(request) |
| 104 | else: |
no test coverage detected