# This file was auto-generated by Fern from our API Definition. import typing from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.request_options import RequestOptions from ..types.json_node import JsonNode from ..types.local_runner import LocalRunner from ..types.local_runner_connect_response import LocalRunnerConnectResponse from ..types.local_runner_heartbeat_response import LocalRunnerHeartbeatResponse from ..types.local_runner_job import LocalRunnerJob from ..types.local_runner_job_metadata import LocalRunnerJobMetadata from ..types.local_runner_job_page import LocalRunnerJobPage from ..types.local_runner_log_entry import LocalRunnerLogEntry from ..types.local_runner_page import LocalRunnerPage from ..types.local_runner_pair_response import LocalRunnerPairResponse from .raw_client import AsyncRawRunnersClient, RawRunnersClient from .types.local_runner_job_result_request_status import LocalRunnerJobResultRequestStatus # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RunnersClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawRunnersClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawRunnersClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawRunnersClient """ return self._raw_client def get_job_logs( self, job_id: str, *, offset: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.List[LocalRunnerLogEntry]: """ Get log entries for a local runner job Parameters ---------- job_id : str offset : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[LocalRunnerLogEntry] Log entries Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.get_job_logs(job_id='jobId', ) """ _response = self._raw_client.get_job_logs(job_id, offset=offset, request_options=request_options) return _response.data def append_job_logs( self, job_id: str, *, request: typing.Sequence[LocalRunnerLogEntry], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Append log entries for a running local runner job Parameters ---------- job_id : str request : typing.Sequence[LocalRunnerLogEntry] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi from Opik import LocalRunnerLogEntry client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.append_job_logs(job_id='jobId', request=[LocalRunnerLogEntry(stream='stream', text='text', )], ) """ _response = self._raw_client.append_job_logs(job_id, request=request, request_options=request_options) return _response.data def cancel_job(self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Cancel a pending or running local runner job Parameters ---------- job_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.cancel_job(job_id='jobId', ) """ _response = self._raw_client.cancel_job(job_id, request_options=request_options) return _response.data def connect_runner( self, *, pairing_code: str, runner_name: str, request_options: typing.Optional[RequestOptions] = None ) -> LocalRunnerConnectResponse: """ Exchange a pairing code or API key for local runner credentials Parameters ---------- pairing_code : str runner_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerConnectResponse Runner connected Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.connect_runner(pairing_code='pairing_code', runner_name='runner_name', ) """ _response = self._raw_client.connect_runner( pairing_code=pairing_code, runner_name=runner_name, request_options=request_options ) return _response.data def create_job( self, *, agent_name: str, project_id: str, inputs: typing.Optional[JsonNode] = OMIT, mask_id: typing.Optional[str] = OMIT, metadata: typing.Optional[LocalRunnerJobMetadata] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Create a local runner job and enqueue it for execution Parameters ---------- agent_name : str project_id : str inputs : typing.Optional[JsonNode] mask_id : typing.Optional[str] metadata : typing.Optional[LocalRunnerJobMetadata] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.create_job(agent_name='agent_name', project_id='project_id', ) """ _response = self._raw_client.create_job( agent_name=agent_name, project_id=project_id, inputs=inputs, mask_id=mask_id, metadata=metadata, request_options=request_options, ) return _response.data def generate_pairing_code( self, *, project_id: str, request_options: typing.Optional[RequestOptions] = None ) -> LocalRunnerPairResponse: """ Generate a pairing code for a local runner in the current workspace Parameters ---------- project_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerPairResponse Pairing code generated Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.generate_pairing_code(project_id='project_id', ) """ _response = self._raw_client.generate_pairing_code(project_id=project_id, request_options=request_options) return _response.data def get_job(self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> LocalRunnerJob: """ Get a single local runner job's status and results Parameters ---------- job_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerJob Job details Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.get_job(job_id='jobId', ) """ _response = self._raw_client.get_job(job_id, request_options=request_options) return _response.data def get_runner(self, runner_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> LocalRunner: """ Get a single local runner with its registered agents Parameters ---------- runner_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunner Runner details Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.get_runner(runner_id='runnerId', ) """ _response = self._raw_client.get_runner(runner_id, request_options=request_options) return _response.data def heartbeat( self, runner_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> LocalRunnerHeartbeatResponse: """ Refresh local runner heartbeat Parameters ---------- runner_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerHeartbeatResponse Heartbeat response Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.heartbeat(runner_id='runnerId', ) """ _response = self._raw_client.heartbeat(runner_id, request_options=request_options) return _response.data def list_jobs( self, runner_id: str, *, project_id: typing.Optional[str] = None, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> LocalRunnerJobPage: """ List jobs for a local runner Parameters ---------- runner_id : str project_id : typing.Optional[str] page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerJobPage Jobs list Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.list_jobs(runner_id='runnerId', ) """ _response = self._raw_client.list_jobs( runner_id, project_id=project_id, page=page, size=size, request_options=request_options ) return _response.data def list_runners( self, *, project_id: str, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> LocalRunnerPage: """ List local runners owned by the current user in the workspace Parameters ---------- project_id : str page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerPage Runners list Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.list_runners(project_id='project_id', ) """ _response = self._raw_client.list_runners( project_id=project_id, page=page, size=size, request_options=request_options ) return _response.data def next_job(self, runner_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> LocalRunnerJob: """ Long-poll for the next pending local runner job Parameters ---------- runner_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerJob Job available Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.next_job(runner_id='runnerId', ) """ _response = self._raw_client.next_job(runner_id, request_options=request_options) return _response.data def register_agents( self, runner_id: str, *, request: typing.Dict[str, typing.Optional[typing.Any]], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Register or update the local runner's agent list Parameters ---------- runner_id : str request : typing.Dict[str, typing.Optional[typing.Any]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.register_agents(runner_id='runnerId', request={'key': 'value' }, ) """ _response = self._raw_client.register_agents(runner_id, request=request, request_options=request_options) return _response.data def report_job_result( self, job_id: str, *, status: LocalRunnerJobResultRequestStatus, result: typing.Optional[JsonNode] = OMIT, error: typing.Optional[str] = OMIT, trace_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Report local runner job completion or failure Parameters ---------- job_id : str status : LocalRunnerJobResultRequestStatus result : typing.Optional[JsonNode] error : typing.Optional[str] trace_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import OpikApi client = OpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) client.runners.report_job_result(job_id='jobId', status="pending", ) """ _response = self._raw_client.report_job_result( job_id, status=status, result=result, error=error, trace_id=trace_id, request_options=request_options ) return _response.data class AsyncRunnersClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawRunnersClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawRunnersClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawRunnersClient """ return self._raw_client async def get_job_logs( self, job_id: str, *, offset: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.List[LocalRunnerLogEntry]: """ Get log entries for a local runner job Parameters ---------- job_id : str offset : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.List[LocalRunnerLogEntry] Log entries Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.get_job_logs(job_id='jobId', ) asyncio.run(main()) """ _response = await self._raw_client.get_job_logs(job_id, offset=offset, request_options=request_options) return _response.data async def append_job_logs( self, job_id: str, *, request: typing.Sequence[LocalRunnerLogEntry], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Append log entries for a running local runner job Parameters ---------- job_id : str request : typing.Sequence[LocalRunnerLogEntry] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi from Opik import LocalRunnerLogEntry import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.append_job_logs(job_id='jobId', request=[LocalRunnerLogEntry(stream='stream', text='text', )], ) asyncio.run(main()) """ _response = await self._raw_client.append_job_logs(job_id, request=request, request_options=request_options) return _response.data async def cancel_job(self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None: """ Cancel a pending or running local runner job Parameters ---------- job_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.cancel_job(job_id='jobId', ) asyncio.run(main()) """ _response = await self._raw_client.cancel_job(job_id, request_options=request_options) return _response.data async def connect_runner( self, *, pairing_code: str, runner_name: str, request_options: typing.Optional[RequestOptions] = None ) -> LocalRunnerConnectResponse: """ Exchange a pairing code or API key for local runner credentials Parameters ---------- pairing_code : str runner_name : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerConnectResponse Runner connected Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.connect_runner(pairing_code='pairing_code', runner_name='runner_name', ) asyncio.run(main()) """ _response = await self._raw_client.connect_runner( pairing_code=pairing_code, runner_name=runner_name, request_options=request_options ) return _response.data async def create_job( self, *, agent_name: str, project_id: str, inputs: typing.Optional[JsonNode] = OMIT, mask_id: typing.Optional[str] = OMIT, metadata: typing.Optional[LocalRunnerJobMetadata] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Create a local runner job and enqueue it for execution Parameters ---------- agent_name : str project_id : str inputs : typing.Optional[JsonNode] mask_id : typing.Optional[str] metadata : typing.Optional[LocalRunnerJobMetadata] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.create_job(agent_name='agent_name', project_id='project_id', ) asyncio.run(main()) """ _response = await self._raw_client.create_job( agent_name=agent_name, project_id=project_id, inputs=inputs, mask_id=mask_id, metadata=metadata, request_options=request_options, ) return _response.data async def generate_pairing_code( self, *, project_id: str, request_options: typing.Optional[RequestOptions] = None ) -> LocalRunnerPairResponse: """ Generate a pairing code for a local runner in the current workspace Parameters ---------- project_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerPairResponse Pairing code generated Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.generate_pairing_code(project_id='project_id', ) asyncio.run(main()) """ _response = await self._raw_client.generate_pairing_code(project_id=project_id, request_options=request_options) return _response.data async def get_job(self, job_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> LocalRunnerJob: """ Get a single local runner job's status and results Parameters ---------- job_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerJob Job details Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.get_job(job_id='jobId', ) asyncio.run(main()) """ _response = await self._raw_client.get_job(job_id, request_options=request_options) return _response.data async def get_runner( self, runner_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> LocalRunner: """ Get a single local runner with its registered agents Parameters ---------- runner_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunner Runner details Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.get_runner(runner_id='runnerId', ) asyncio.run(main()) """ _response = await self._raw_client.get_runner(runner_id, request_options=request_options) return _response.data async def heartbeat( self, runner_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> LocalRunnerHeartbeatResponse: """ Refresh local runner heartbeat Parameters ---------- runner_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerHeartbeatResponse Heartbeat response Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.heartbeat(runner_id='runnerId', ) asyncio.run(main()) """ _response = await self._raw_client.heartbeat(runner_id, request_options=request_options) return _response.data async def list_jobs( self, runner_id: str, *, project_id: typing.Optional[str] = None, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> LocalRunnerJobPage: """ List jobs for a local runner Parameters ---------- runner_id : str project_id : typing.Optional[str] page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerJobPage Jobs list Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.list_jobs(runner_id='runnerId', ) asyncio.run(main()) """ _response = await self._raw_client.list_jobs( runner_id, project_id=project_id, page=page, size=size, request_options=request_options ) return _response.data async def list_runners( self, *, project_id: str, page: typing.Optional[int] = None, size: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> LocalRunnerPage: """ List local runners owned by the current user in the workspace Parameters ---------- project_id : str page : typing.Optional[int] size : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerPage Runners list Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.list_runners(project_id='project_id', ) asyncio.run(main()) """ _response = await self._raw_client.list_runners( project_id=project_id, page=page, size=size, request_options=request_options ) return _response.data async def next_job( self, runner_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> LocalRunnerJob: """ Long-poll for the next pending local runner job Parameters ---------- runner_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- LocalRunnerJob Job available Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.next_job(runner_id='runnerId', ) asyncio.run(main()) """ _response = await self._raw_client.next_job(runner_id, request_options=request_options) return _response.data async def register_agents( self, runner_id: str, *, request: typing.Dict[str, typing.Optional[typing.Any]], request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Register or update the local runner's agent list Parameters ---------- runner_id : str request : typing.Dict[str, typing.Optional[typing.Any]] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.register_agents(runner_id='runnerId', request={'key': 'value' }, ) asyncio.run(main()) """ _response = await self._raw_client.register_agents(runner_id, request=request, request_options=request_options) return _response.data async def report_job_result( self, job_id: str, *, status: LocalRunnerJobResultRequestStatus, result: typing.Optional[JsonNode] = OMIT, error: typing.Optional[str] = OMIT, trace_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> None: """ Report local runner job completion or failure Parameters ---------- job_id : str status : LocalRunnerJobResultRequestStatus result : typing.Optional[JsonNode] error : typing.Optional[str] trace_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- None Examples -------- from Opik import AsyncOpikApi import asyncio client = AsyncOpikApi(api_key="YOUR_API_KEY", workspace_name="YOUR_WORKSPACE_NAME", ) async def main() -> None: await client.runners.report_job_result(job_id='jobId', status="pending", ) asyncio.run(main()) """ _response = await self._raw_client.report_job_result( job_id, status=status, result=result, error=error, trace_id=trace_id, request_options=request_options ) return _response.data