MCPcopy Create free account
hub / github.com/HKUDS/OpenOPC / _ScriptAdapter

Class _ScriptAdapter

tests/test_external_agent_monitoring.py:156–181  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

154 rationale="tool ok" if self.allow_tool else "tool denied",
155 confidence=1.0,
156 policy_source="test_tool",
157 metadata={**(metadata or {}), "human_reply": self.human_reply},
158 )
159
160
161class _ScriptAdapter(ExternalAgentAdapter):
162 agent_type = "script_agent"
163 default_command = sys.executable
164
165 def __init__(self, script: str, idle_timeout_seconds: int = 900) -> None:
166 super().__init__()
167 self.script = script
168 self.config.command = sys.executable
169 self.config.idle_timeout_seconds = idle_timeout_seconds
170 self.config.status_heartbeat_seconds = 1
171 self._process = None
172
173 async def is_available(self) -> bool:
174 return True
175
176 async def execute(self, task: Task, workspace_path: str) -> TaskResult:
177 raise NotImplementedError("Broker monitoring path should be used")
178
179 def build_invocation(self, task: Task, workspace_path: str | None = None):
180 cmd = [self.configured_command(), "-c", self.script]
181 return cmd, self.build_invocation_metadata(cmd)
182
183 async def get_status(self) -> AgentStatus:
184 if self._process and self._process.returncode is None:

Calls

no outgoing calls