Begin reworking local agent

src/prefect/agent/__init__.py

@@ -2,6 +2,7 @@
 # others must be explicitly imported so they can raise helpful errors if appropriate
 
 from prefect.agent.agent import Agent
+import prefect.agent.docker
 import prefect.agent.fargate
 import prefect.agent.local
 import prefect.agent.kubernetes

src/prefect/agent/agent.py

@@ -50,7 +50,6 @@ def __init__(self, name: str = None, labels: Iterable[str] = None) -> None:
         token = config.cloud.agent.get("auth_token")
 
         self.client = Client(api_token=token)
-        self._verify_token(token)
 
         logger = logging.getLogger(self.name)
         logger.setLevel(config.cloud.agent.get("level"))
@@ -63,27 +62,6 @@ def __init__(self, name: str = None, labels: Iterable[str] = None) -> None:
 
         self.logger = logger
 
-    def _verify_token(self, token: str) -> None:
-        """
-        Checks whether a token with a `RUNNER` scope was provided
-
-        Args:
-            - token (str): The provided agent token to verify
-
-        Raises:
-            - AuthorizationError: if token is empty or does not have a RUNNER role
-        """
-        if not token:
-            raise AuthorizationError("No agent API token provided.")
-
-        # Check if RUNNER role
-        result = self.client.graphql(query="query { authInfo { apiTokenScope } }")
-        if (
-            not result.data  # type: ignore
-            or result.data.authInfo.apiTokenScope != "RUNNER"  # type: ignore
-        ):
-            raise AuthorizationError("Provided token does not have a RUNNER scope.")
-
     def start(self) -> None:
         """
         The main entrypoint to the agent. This function loops and constantly polls for

src/prefect/agent/local/agent.py

@@ -12,58 +12,19 @@
 
 class LocalAgent(Agent):
     """
-    Agent which deploys flow runs locally as Docker containers. Information on using the
+    Agent which deploys flow runs locally. Information on using the
     Local Agent can be found at https://docs.prefect.io/cloud/agent/local.html
 
     Args:
         - name (str, optional): An optional name to give this agent. Can also be set through
             the environment variable `PREFECT__CLOUD__AGENT__NAME`. Defaults to "agent"
         - labels (List[str], optional): a list of labels, which are arbitrary string identifiers used by Prefect
             Agents when polling for work
-        - base_url (str, optional): URL for a Docker daemon server. Defaults to
-            `unix:///var/run/docker.sock` however other hosts such as
-            `tcp://0.0.0.0:2375` can be provided
-        - no_pull (bool, optional): Flag on whether or not to pull flow images.
-            Defaults to `False` if not provided here or in context.
     """
 
-    def __init__(
-        self,
-        name: str = None,
-        labels: Iterable[str] = None,
-        base_url: str = None,
-        no_pull: bool = None,
-    ) -> None:
+    def __init__(self, name: str = None, labels: Iterable[str] = None,) -> None:
         super().__init__(name=name, labels=labels)
 
-        if platform == "win32":
-            default_url = "npipe:////./pipe/docker_engine"
-        else:
-            default_url = "unix://var/run/docker.sock"
-        self.logger.debug(
-            "Platform {} and default docker daemon {}".format(platform, default_url)
-        )
-
-        # Determine Daemon URL
-        self.base_url = base_url or context.get("base_url", default_url)
-        self.logger.debug("Base docker daemon url {}".format(self.base_url))
-
-        # Determine pull specification
-        self.no_pull = no_pull or context.get("no_pull", False)
-        self.logger.debug("no_pull set to {}".format(self.no_pull))
-
-        self.docker_client = docker.APIClient(base_url=self.base_url, version="auto")
-
-        # Ping Docker daemon for connection issues
-        try:
-            self.logger.debug("Pinging docker daemon")
-            self.docker_client.ping()
-        except Exception as exc:
-            self.logger.exception(
-                "Issue connecting to the Docker daemon. Make sure it is running."
-            )
-            raise exc
-
     def deploy_flows(self, flow_runs: list) -> None:
         """
         Deploy flow runs on your local machine as Docker containers
@@ -111,28 +72,6 @@ def deploy_flows(self, flow_runs: list) -> None:
             )
             self.docker_client.start(container=container.get("Id"))
 
-    def populate_env_vars(self, flow_run: GraphQLResult) -> dict:
-        """
-        Populate metadata and variables in the environment variables for a flow run
-
-        Args:
-            - flow_run (GraphQLResult): A flow run object
-
-        Returns:
-            - dict: a dictionary representing the populated environment variables
-        """
-        return {
-            "PREFECT__CLOUD__API": config.cloud.api,
-            "PREFECT__CLOUD__AUTH_TOKEN": config.cloud.agent.auth_token,
-            "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
-            "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
-            "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
-            "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
-            "PREFECT__LOGGING__LEVEL": "DEBUG",
-            "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
-            "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
-        }
-
 
 if __name__ == "__main__":
     LocalAgent().start()

src/prefect/core/flow.py

@@ -1233,7 +1233,10 @@ def run_agent(self, executor=None, labels: list = None):
                 "local" along with a safe version of the current Flow's name
         """
         labels = labels or ["local", slugify(self.name)]
-        agent = prefect.agent.agent.Agent(labels=labels)
+        with set_temporary_config(
+            {"cloud.agent.auth_token": prefect.config.cloud.auth_token}
+        ):
+            agent = prefect.agent.agent.Agent(labels=labels)
 
         while True:
             flow_runs = agent.query_flow_runs(str(uuid.uuid4()))
@@ -1245,7 +1248,6 @@ def run_agent(self, executor=None, labels: list = None):
                         {
                             "cloud.use_local_secrets": False,
                             "logging.log_to_cloud": True,
-                            "cloud.auth_token": agent.client._api_token,
                         }
                     ):
                         runner_cls = prefect.engine.cloud.flow_runner.CloudFlowRunner