Attach a source to an agent Args: agent_id (str): ID of the agent source_id (str): ID of the source source_name (str): Name of the source
(self, source_id: str, agent_id: str)
| 1462 | return Source(**response.json()) |
| 1463 | |
| 1464 | def attach_source(self, source_id: str, agent_id: str) -> AgentState: |
| 1465 | """ |
| 1466 | Attach a source to an agent |
| 1467 | |
| 1468 | Args: |
| 1469 | agent_id (str): ID of the agent |
| 1470 | source_id (str): ID of the source |
| 1471 | source_name (str): Name of the source |
| 1472 | """ |
| 1473 | params = {"agent_id": agent_id} |
| 1474 | response = requests.patch( |
| 1475 | f"{self.base_url}/{self.api_prefix}/agents/{agent_id}/sources/attach/{source_id}", params=params, headers=self.headers |
| 1476 | ) |
| 1477 | assert response.status_code == 200, f"Failed to attach source to agent: {response.text}" |
| 1478 | return AgentState(**response.json()) |
| 1479 | |
| 1480 | def detach_source(self, source_id: str, agent_id: str) -> AgentState: |
| 1481 | """Detach a source from an agent""" |
no test coverage detected