prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Add optional pull to local agent
  If the image is present locally we don't want to attempt a pull from a registry
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `src/prefect/cli/agent.py`
  Content:
  ```
  1 import click
  2 
  3 from prefect import config
  4 from prefect.utilities.configuration import set_temporary_config
  5 from prefect.utilities.serialization import from_qualified_name
  6 
  7 _agents = {
  8     "local": "prefect.agent.local.LocalAgent",
  9     "kubernetes": "prefect.agent.kubernetes.KubernetesAgent",
  10     "nomad": "prefect.agent.nomad.NomadAgent",
  11 }
  12 
  13 
  14 @click.group(hidden=True)
  15 def agent():
  16     """
  17     Manage Prefect agents.
  18 
  19     \b
  20     Usage:
  21         $ prefect agent [COMMAND]
  22 
  23     \b
  24     Arguments:
  25         start       Start a Prefect agent
  26 
  27     \b
  28     Examples:
  29         $ prefect agent start
  30 
  31     \b
  32         $ prefect agent start kubernetes --token MY_TOKEN
  33     """
  34     pass
  35 
  36 
  37 @agent.command(hidden=True)
  38 @click.argument("name", default="local")
  39 @click.option(
  40     "--token", "-t", required=False, help="A Prefect Cloud API token.", hidden=True
  41 )
  42 def start(name, token):
  43     """
  44     Start an agent.
  45 
  46     \b
  47     Arguments:
  48         name            TEXT    The name of an agent to start (e.g. `local`, `kubernetes`, `nomad`)
  49                                 Defaults to `local`
  50 
  51     \b
  52     Options:
  53         --token, -t     TEXT    A Prefect Cloud api token
  54     """
  55     with set_temporary_config(
  56         {"cloud.agent.auth_token": token or config.cloud.agent.auth_token}
  57     ):
  58         retrieved_agent = _agents.get(name, None)
  59 
  60         if not retrieved_agent:
  61             click.secho("{} is not a valid agent".format(name), fg="red")
  62             return
  63 
  64         from_qualified_name(retrieved_agent)().start()
  65
  ```
  
  Path: `src/prefect/agent/local/agent.py`
  Content:
  ```
  1 import docker
  2 
  3 from prefect import config
  4 from prefect.agent import Agent
  5 from prefect.environments.storage import Docker
  6 from prefect.serialization.storage import StorageSchema
  7 from prefect.utilities.graphql import GraphQLResult
  8 
  9 
  10 class LocalAgent(Agent):
  11     """
  12     Agent which deploys flow runs locally as Docker containers.
  13 
  14     Args:
  15         - base_url (str, optional): URL for a Docker daemon server. Defaults to
  16             `unix:///var/run/docker.sock` however other hosts such as
  17             `tcp://0.0.0.0:2375` can be provided
  18     """
  19 
  20     def __init__(self, base_url: str = None) -> None:
  21         super().__init__()
  22 
  23         base_url = base_url or "unix://var/run/docker.sock"
  24         self.docker_client = docker.APIClient(base_url=base_url, version="auto")
  25 
  26         # Ping Docker daemon for connection issues
  27         try:
  28             self.docker_client.ping()
  29         except Exception as exc:
  30             self.logger.error(
  31                 "Issue connecting to the Docker daemon. Make sure it is running."
  32             )
  33             raise exc
  34 
  35     def deploy_flows(self, flow_runs: list) -> None:
  36         """
  37         Deploy flow runs on your local machine as Docker containers
  38 
  39         Args:
  40             - flow_runs (list): A list of GraphQLResult flow run objects
  41         """
  42         for flow_run in flow_runs:
  43 
  44             storage = StorageSchema().load(flow_run.flow.storage)
  45             if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):
  46                 self.logger.error(
  47                     "Storage for flow run {} is not of type Docker.".format(flow_run.id)
  48                 )
  49                 continue
  50 
  51             env_vars = self.populate_env_vars(flow_run=flow_run)
  52 
  53             # Pull image if it doesn't exist locally
  54             self.docker_client.pull(storage.name)
  55 
  56             # Create a container
  57             container = self.docker_client.create_container(
  58                 storage.name, command="prefect execute cloud-flow", environment=env_vars
  59             )
  60 
  61             # Start the container
  62             self.docker_client.start(container=container.get("Id"))
  63 
  64     def populate_env_vars(self, flow_run: GraphQLResult) -> dict:
  65         """
  66         Populate metadata and variables in the environment variables for a flow run
  67 
  68         Args:
  69             - flow_run (GraphQLResult): A flow run object
  70 
  71         Returns:
  72             - dict: a dictionary representing the populated environment variables
  73         """
  74         return {
  75             "PREFECT__CLOUD__API": config.cloud.api,
  76             "PREFECT__CLOUD__AUTH_TOKEN": config.cloud.agent.auth_token,
  77             "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
  78             "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
  79             "PREFECT__LOGGING__LOG_TO_CLOUD": "true",
  80             "PREFECT__LOGGING__LEVEL": "DEBUG",
  81             "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
  82             "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
  83         }
  84 
  85 
  86 if __name__ == "__main__":
  87     LocalAgent().start()
  88
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py
  --- a/src/prefect/agent/local/agent.py
  +++ b/src/prefect/agent/local/agent.py
  @@ -1,6 +1,6 @@
   import docker
   
  -from prefect import config
  +from prefect import config, context
   from prefect.agent import Agent
   from prefect.environments.storage import Docker
   from prefect.serialization.storage import StorageSchema
  @@ -15,9 +15,11 @@
           - 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, base_url: str = None) -> None:
  +    def __init__(self, base_url: str = None, no_pull: bool = None) -> None:
           super().__init__()
   
           base_url = base_url or "unix://var/run/docker.sock"
  @@ -32,6 +34,10 @@
               )
               raise exc
   
  +        self.no_pull = no_pull or context.get("no_pull")
  +        if self.no_pull is None:
  +            self.no_pull = False
  +
       def deploy_flows(self, flow_runs: list) -> None:
           """
           Deploy flow runs on your local machine as Docker containers
  @@ -50,8 +56,8 @@
   
               env_vars = self.populate_env_vars(flow_run=flow_run)
   
  -            # Pull image if it doesn't exist locally
  -            self.docker_client.pull(storage.name)
  +            if not self.no_pull:
  +                self.docker_client.pull(storage.name)
   
               # Create a container
               container = self.docker_client.create_container(
  diff --git a/src/prefect/cli/agent.py b/src/prefect/cli/agent.py
  --- a/src/prefect/cli/agent.py
  +++ b/src/prefect/cli/agent.py
  @@ -1,6 +1,6 @@
   import click
   
  -from prefect import config
  +from prefect import config, context
   from prefect.utilities.configuration import set_temporary_config
   from prefect.utilities.serialization import from_qualified_name
   
  @@ -39,7 +39,8 @@
   @click.option(
       "--token", "-t", required=False, help="A Prefect Cloud API token.", hidden=True
   )
  -def start(name, token):
  +@click.option("--no-pull", is_flag=True, help="Pull images flag.", hidden=True)
  +def start(name, token, no_pull):
       """
       Start an agent.
   
  @@ -51,6 +52,8 @@
       \b
       Options:
           --token, -t     TEXT    A Prefect Cloud api token
  +        --no-pull               Pull images for a LocalAgent
  +                                Defaults to pulling if not provided
       """
       with set_temporary_config(
           {"cloud.agent.auth_token": token or config.cloud.agent.auth_token}
  @@ -61,4 +64,5 @@
               click.secho("{} is not a valid agent".format(name), fg="red")
               return
   
  -        from_qualified_name(retrieved_agent)().start()
  +        with context(no_pull=no_pull):
  +            from_qualified_name(retrieved_agent)().start()
  
verification_info: {"golden_diff": "diff --git a/src/prefect/agent/local/agent.py b/src/prefect/agent/local/agent.py\n--- a/src/prefect/agent/local/agent.py\n+++ b/src/prefect/agent/local/agent.py\n@@ -1,6 +1,6 @@\n import docker\n \n-from prefect import config\n+from prefect import config, context\n from prefect.agent import Agent\n from prefect.environments.storage import Docker\n from prefect.serialization.storage import StorageSchema\n@@ -15,9 +15,11 @@\n         - base_url (str, optional): URL for a Docker daemon server. Defaults to\n             `unix:///var/run/docker.sock` however other hosts such as\n             `tcp://0.0.0.0:2375` can be provided\n+        - no_pull (bool, optional): Flag on whether or not to pull flow images.\n+            Defaults to `False` if not provided here or in context.\n     \"\"\"\n \n-    def __init__(self, base_url: str = None) -> None:\n+    def __init__(self, base_url: str = None, no_pull: bool = None) -> None:\n         super().__init__()\n \n         base_url = base_url or \"unix://var/run/docker.sock\"\n@@ -32,6 +34,10 @@\n             )\n             raise exc\n \n+        self.no_pull = no_pull or context.get(\"no_pull\")\n+        if self.no_pull is None:\n+            self.no_pull = False\n+\n     def deploy_flows(self, flow_runs: list) -> None:\n         \"\"\"\n         Deploy flow runs on your local machine as Docker containers\n@@ -50,8 +56,8 @@\n \n             env_vars = self.populate_env_vars(flow_run=flow_run)\n \n-            # Pull image if it doesn't exist locally\n-            self.docker_client.pull(storage.name)\n+            if not self.no_pull:\n+                self.docker_client.pull(storage.name)\n \n             # Create a container\n             container = self.docker_client.create_container(\ndiff --git a/src/prefect/cli/agent.py b/src/prefect/cli/agent.py\n--- a/src/prefect/cli/agent.py\n+++ b/src/prefect/cli/agent.py\n@@ -1,6 +1,6 @@\n import click\n \n-from prefect import config\n+from prefect import config, context\n from prefect.utilities.configuration import set_temporary_config\n from prefect.utilities.serialization import from_qualified_name\n \n@@ -39,7 +39,8 @@\n @click.option(\n     \"--token\", \"-t\", required=False, help=\"A Prefect Cloud API token.\", hidden=True\n )\n-def start(name, token):\n+@click.option(\"--no-pull\", is_flag=True, help=\"Pull images flag.\", hidden=True)\n+def start(name, token, no_pull):\n     \"\"\"\n     Start an agent.\n \n@@ -51,6 +52,8 @@\n     \\b\n     Options:\n         --token, -t     TEXT    A Prefect Cloud api token\n+        --no-pull               Pull images for a LocalAgent\n+                                Defaults to pulling if not provided\n     \"\"\"\n     with set_temporary_config(\n         {\"cloud.agent.auth_token\": token or config.cloud.agent.auth_token}\n@@ -61,4 +64,5 @@\n             click.secho(\"{} is not a valid agent\".format(name), fg=\"red\")\n             return\n \n-        from_qualified_name(retrieved_agent)().start()\n+        with context(no_pull=no_pull):\n+            from_qualified_name(retrieved_agent)().start()\n", "issue": "Add optional pull to local agent\nIf the image is present locally we don't want to attempt a pull from a registry\n", "before_files": [{"content": "import click\n\nfrom prefect import config\nfrom prefect.utilities.configuration import set_temporary_config\nfrom prefect.utilities.serialization import from_qualified_name\n\n_agents = {\n    \"local\": \"prefect.agent.local.LocalAgent\",\n    \"kubernetes\": \"prefect.agent.kubernetes.KubernetesAgent\",\n    \"nomad\": \"prefect.agent.nomad.NomadAgent\",\n}\n\n\n@click.group(hidden=True)\ndef agent():\n    \"\"\"\n    Manage Prefect agents.\n\n    \\b\n    Usage:\n        $ prefect agent [COMMAND]\n\n    \\b\n    Arguments:\n        start       Start a Prefect agent\n\n    \\b\n    Examples:\n        $ prefect agent start\n\n    \\b\n        $ prefect agent start kubernetes --token MY_TOKEN\n    \"\"\"\n    pass\n\n\n@agent.command(hidden=True)\n@click.argument(\"name\", default=\"local\")\n@click.option(\n    \"--token\", \"-t\", required=False, help=\"A Prefect Cloud API token.\", hidden=True\n)\ndef start(name, token):\n    \"\"\"\n    Start an agent.\n\n    \\b\n    Arguments:\n        name            TEXT    The name of an agent to start (e.g. `local`, `kubernetes`, `nomad`)\n                                Defaults to `local`\n\n    \\b\n    Options:\n        --token, -t     TEXT    A Prefect Cloud api token\n    \"\"\"\n    with set_temporary_config(\n        {\"cloud.agent.auth_token\": token or config.cloud.agent.auth_token}\n    ):\n        retrieved_agent = _agents.get(name, None)\n\n        if not retrieved_agent:\n            click.secho(\"{} is not a valid agent\".format(name), fg=\"red\")\n            return\n\n        from_qualified_name(retrieved_agent)().start()\n", "path": "src/prefect/cli/agent.py"}, {"content": "import docker\n\nfrom prefect import config\nfrom prefect.agent import Agent\nfrom prefect.environments.storage import Docker\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass LocalAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs locally as Docker containers.\n\n    Args:\n        - base_url (str, optional): URL for a Docker daemon server. Defaults to\n            `unix:///var/run/docker.sock` however other hosts such as\n            `tcp://0.0.0.0:2375` can be provided\n    \"\"\"\n\n    def __init__(self, base_url: str = None) -> None:\n        super().__init__()\n\n        base_url = base_url or \"unix://var/run/docker.sock\"\n        self.docker_client = docker.APIClient(base_url=base_url, version=\"auto\")\n\n        # Ping Docker daemon for connection issues\n        try:\n            self.docker_client.ping()\n        except Exception as exc:\n            self.logger.error(\n                \"Issue connecting to the Docker daemon. Make sure it is running.\"\n            )\n            raise exc\n\n    def deploy_flows(self, flow_runs: list) -> None:\n        \"\"\"\n        Deploy flow runs on your local machine as Docker containers\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        for flow_run in flow_runs:\n\n            storage = StorageSchema().load(flow_run.flow.storage)\n            if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n                self.logger.error(\n                    \"Storage for flow run {} is not of type Docker.\".format(flow_run.id)\n                )\n                continue\n\n            env_vars = self.populate_env_vars(flow_run=flow_run)\n\n            # Pull image if it doesn't exist locally\n            self.docker_client.pull(storage.name)\n\n            # Create a container\n            container = self.docker_client.create_container(\n                storage.name, command=\"prefect execute cloud-flow\", environment=env_vars\n            )\n\n            # Start the container\n            self.docker_client.start(container=container.get(\"Id\"))\n\n    def populate_env_vars(self, flow_run: GraphQLResult) -> dict:\n        \"\"\"\n        Populate metadata and variables in the environment variables for a flow run\n\n        Args:\n            - flow_run (GraphQLResult): A flow run object\n\n        Returns:\n            - dict: a dictionary representing the populated environment variables\n        \"\"\"\n        return {\n            \"PREFECT__CLOUD__API\": config.cloud.api,\n            \"PREFECT__CLOUD__AUTH_TOKEN\": config.cloud.agent.auth_token,\n            \"PREFECT__CONTEXT__FLOW_RUN_ID\": flow_run.id,  # type: ignore\n            \"PREFECT__CLOUD__USE_LOCAL_SECRETS\": \"false\",\n            \"PREFECT__LOGGING__LOG_TO_CLOUD\": \"true\",\n            \"PREFECT__LOGGING__LEVEL\": \"DEBUG\",\n            \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudFlowRunner\",\n            \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudTaskRunner\",\n        }\n\n\nif __name__ == \"__main__\":\n    LocalAgent().start()\n", "path": "src/prefect/agent/local/agent.py"}], "after_files": [{"content": "import click\n\nfrom prefect import config, context\nfrom prefect.utilities.configuration import set_temporary_config\nfrom prefect.utilities.serialization import from_qualified_name\n\n_agents = {\n    \"local\": \"prefect.agent.local.LocalAgent\",\n    \"kubernetes\": \"prefect.agent.kubernetes.KubernetesAgent\",\n    \"nomad\": \"prefect.agent.nomad.NomadAgent\",\n}\n\n\n@click.group(hidden=True)\ndef agent():\n    \"\"\"\n    Manage Prefect agents.\n\n    \\b\n    Usage:\n        $ prefect agent [COMMAND]\n\n    \\b\n    Arguments:\n        start       Start a Prefect agent\n\n    \\b\n    Examples:\n        $ prefect agent start\n\n    \\b\n        $ prefect agent start kubernetes --token MY_TOKEN\n    \"\"\"\n    pass\n\n\n@agent.command(hidden=True)\n@click.argument(\"name\", default=\"local\")\n@click.option(\n    \"--token\", \"-t\", required=False, help=\"A Prefect Cloud API token.\", hidden=True\n)\n@click.option(\"--no-pull\", is_flag=True, help=\"Pull images flag.\", hidden=True)\ndef start(name, token, no_pull):\n    \"\"\"\n    Start an agent.\n\n    \\b\n    Arguments:\n        name            TEXT    The name of an agent to start (e.g. `local`, `kubernetes`, `nomad`)\n                                Defaults to `local`\n\n    \\b\n    Options:\n        --token, -t     TEXT    A Prefect Cloud api token\n        --no-pull               Pull images for a LocalAgent\n                                Defaults to pulling if not provided\n    \"\"\"\n    with set_temporary_config(\n        {\"cloud.agent.auth_token\": token or config.cloud.agent.auth_token}\n    ):\n        retrieved_agent = _agents.get(name, None)\n\n        if not retrieved_agent:\n            click.secho(\"{} is not a valid agent\".format(name), fg=\"red\")\n            return\n\n        with context(no_pull=no_pull):\n            from_qualified_name(retrieved_agent)().start()\n", "path": "src/prefect/cli/agent.py"}, {"content": "import docker\n\nfrom prefect import config, context\nfrom prefect.agent import Agent\nfrom prefect.environments.storage import Docker\nfrom prefect.serialization.storage import StorageSchema\nfrom prefect.utilities.graphql import GraphQLResult\n\n\nclass LocalAgent(Agent):\n    \"\"\"\n    Agent which deploys flow runs locally as Docker containers.\n\n    Args:\n        - base_url (str, optional): URL for a Docker daemon server. Defaults to\n            `unix:///var/run/docker.sock` however other hosts such as\n            `tcp://0.0.0.0:2375` can be provided\n        - no_pull (bool, optional): Flag on whether or not to pull flow images.\n            Defaults to `False` if not provided here or in context.\n    \"\"\"\n\n    def __init__(self, base_url: str = None, no_pull: bool = None) -> None:\n        super().__init__()\n\n        base_url = base_url or \"unix://var/run/docker.sock\"\n        self.docker_client = docker.APIClient(base_url=base_url, version=\"auto\")\n\n        # Ping Docker daemon for connection issues\n        try:\n            self.docker_client.ping()\n        except Exception as exc:\n            self.logger.error(\n                \"Issue connecting to the Docker daemon. Make sure it is running.\"\n            )\n            raise exc\n\n        self.no_pull = no_pull or context.get(\"no_pull\")\n        if self.no_pull is None:\n            self.no_pull = False\n\n    def deploy_flows(self, flow_runs: list) -> None:\n        \"\"\"\n        Deploy flow runs on your local machine as Docker containers\n\n        Args:\n            - flow_runs (list): A list of GraphQLResult flow run objects\n        \"\"\"\n        for flow_run in flow_runs:\n\n            storage = StorageSchema().load(flow_run.flow.storage)\n            if not isinstance(StorageSchema().load(flow_run.flow.storage), Docker):\n                self.logger.error(\n                    \"Storage for flow run {} is not of type Docker.\".format(flow_run.id)\n                )\n                continue\n\n            env_vars = self.populate_env_vars(flow_run=flow_run)\n\n            if not self.no_pull:\n                self.docker_client.pull(storage.name)\n\n            # Create a container\n            container = self.docker_client.create_container(\n                storage.name, command=\"prefect execute cloud-flow\", environment=env_vars\n            )\n\n            # Start the container\n            self.docker_client.start(container=container.get(\"Id\"))\n\n    def populate_env_vars(self, flow_run: GraphQLResult) -> dict:\n        \"\"\"\n        Populate metadata and variables in the environment variables for a flow run\n\n        Args:\n            - flow_run (GraphQLResult): A flow run object\n\n        Returns:\n            - dict: a dictionary representing the populated environment variables\n        \"\"\"\n        return {\n            \"PREFECT__CLOUD__API\": config.cloud.api,\n            \"PREFECT__CLOUD__AUTH_TOKEN\": config.cloud.agent.auth_token,\n            \"PREFECT__CONTEXT__FLOW_RUN_ID\": flow_run.id,  # type: ignore\n            \"PREFECT__CLOUD__USE_LOCAL_SECRETS\": \"false\",\n            \"PREFECT__LOGGING__LOG_TO_CLOUD\": \"true\",\n            \"PREFECT__LOGGING__LEVEL\": \"DEBUG\",\n            \"PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudFlowRunner\",\n            \"PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS\": \"prefect.engine.cloud.CloudTaskRunner\",\n        }\n\n\nif __name__ == \"__main__\":\n    LocalAgent().start()\n", "path": "src/prefect/agent/local/agent.py"}]}