agent-platform/services/code-runner-service/app/application/services.py

from core_domain import CodeExecutionRequestContract, CodeExecutionResponseContract

from app.bootstrap.settings import CodeRunnerServiceSettings
from app.infrastructure.runner import PythonCodeRunner


class CodeRunnerApplicationService:
    def __init__(
        self,
        *,
        runner: PythonCodeRunner,
        settings: CodeRunnerServiceSettings,
    ) -> None:
        self.runner = runner
        self.settings = settings

    def execute_code(self, payload: CodeExecutionRequestContract) -> CodeExecutionResponseContract:
        if payload.language != "python":
            return CodeExecutionResponseContract(
                success=False,
                error_message=f"unsupported language: {payload.language}",
            )

        resolved_payload = payload.model_copy(
            update={
                "timeout_seconds": min(payload.timeout_seconds, self.settings.max_timeout_seconds),
            }
        )
        return self.runner.execute(resolved_payload)