Update Cloud Client for new USER-scoped API tokens

CHANGELOG.md

@@ -26,6 +26,7 @@ These changes are available in the [master branch](https://github.com/PrefectHQ/
 - Add `task_slug`, `flow_id`, and `flow_run_id` to context - [#1405](https://github.com/PrefectHQ/prefect/pull/1405)
 - Support persistent `scheduled_start_time` for scheduled flow runs when run locally with `flow.run()` - [#1418](https://github.com/PrefectHQ/prefect/pull/1418)
 - Add `task_args` to `Task.map` - [#1390](https://github.com/PrefectHQ/prefect/issues/1390)
+- Add auth flows for `USER`-scoped Cloud API tokens - [#1423](https://github.com/PrefectHQ/prefect/pull/1423)
 
 ### Task Library
 

docs/cloud/agent/kubernetes.md

@@ -45,8 +45,8 @@ The Kubernetes Agent can be started either through the Prefect CLI or by importi
 There are a few ways in which you can specify a `RUNNER` API token:
 
 - command argument `prefect agent start kubernetes -t MY_TOKEN`
-- environment variable `export PREFECT__CLOUD__AGENT__API_TOKEN=MY_TOKEN`
-- token will be used from `prefect.config.cloud.api_token` if not provided from one of the two previous methods
+- environment variable `export PREFECT__CLOUD__AGENT__AUTH_TOKEN=MY_TOKEN`
+- token will be used from `prefect.config.cloud.auth_token` if not provided from one of the two previous methods
 
 :::
 

docs/cloud/agent/local.md

@@ -32,8 +32,8 @@ The Local Agent can be started either through the Prefect CLI or by importing th
 There are a few ways in which you can specify a `RUNNER` API token:
 
 - command argument `prefect agent start -t MY_TOKEN`
-- environment variable `export PREFECT__CLOUD__AGENT__API_TOKEN=MY_TOKEN`
-- token will be used from `prefect.config.cloud.api_token` if not provided from one of the two previous methods
+- environment variable `export PREFECT__CLOUD__AGENT__AUTH_TOKEN=MY_TOKEN`
+- token will be used from `prefect.config.cloud.auth_token` if not provided from one of the two previous methods
 
 :::
 

docs/cloud/cloud_concepts/api.md

@@ -0,0 +1,110 @@
+# Prefect Cloud API
+
+Prefect Cloud exposes a powerful GraphQL API for interacting with the platform. There are a variety of ways you can access the API.
+
+# Authentication
+
+In order to interact with Cloud from your local machine, you'll need to generate an API token.
+
+To generate an API token, use the Cloud UI or the following GraphQL call (from an already authenticated client!):
+
+```graphql
+mutation {
+  createAPIToken(input: { name: "My API token", role: USER }) {
+    token
+  }
+}
+```
+
+## API Token Scopes
+
+Prefect Cloud can generate API tokens with three different scopes.
+
+### `USER`
+
+`USER`-scoped API tokens function as personal access tokens. These tokens have very few permissions on their own, but can be used to authenticate with the Cloud API. Once authenticated, `USER` tokens can be used to generate short-lived JWT auth tokens for any tenant the user belongs to. These auth tokens inherit any permissions the user has in that tenant, allowing full API access. The Client manages the process of provisioning and refreshing these tokens.
+
+### `TENANT`
+
+`TENANT`-scoped API tokens are used for long-lived programmatic access to a specific Cloud tenant. Unlike `USER` tokens, which can adopt any tenant membership the user has, `TENANT` tokens are fixed to a specific membership in a specific tenant. They adopt whatever permissions the user has in the tenant.
+
+### `RUNNER`
+
+`RUNNER`-scoped API tokens are used for processes like the Prefect Agent, which require the ability to execute flows on behalf of a tenant. Unlike the other token types, `RUNNER` tokens are not scoped to a particular user. Consequently, they can only be generated by tenant admins.
+
+# Python Client
+
+## About the Client
+
+Prefect Core includes a Python client for Prefect Cloud. The Python client was designed for both interactive and programmatic use, and includes convenience methods for transparently managing authentication when used with `USER`-scoped tokens.
+
+## Getting started
+
+For interactive use, the most common way to use the Cloud Client is to generate a `USER`-scoped token and provide it to the client. After doing so, users can save the token so it persists across all Python sessions:
+
+```python
+import prefect
+client = prefect.Client(api_token="YOUR_USER_TOKEN")
+client.save_api_token()
+```
+
+Now, starting a client in another session will automatically reload the token:
+
+```python
+client = prefect.Client()
+assert client._api_token == "YOUR_USER_TOKEN"  # True
+```
+
+Note that a token can be provided by environment variable (`PREFECT__CLOUD__AUTH_TOKEN`) or in your Prefect config (under `cloud.auth_token`).
+
+:::tip Using `USER` tokens
+The steps shown here were designed to be used with `USER`-scoped tokens. They will work with `TENANT` scoped tokens as well, but unexpected errors could result.
+:::
+
+Once provisioned with a `USER` token, the Cloud Client can query for available tenants and login to those tenants. In order to query for tenants, call:
+
+```python
+client.get_available_tenants()
+```
+
+This will print the id, name, and slug of all the tenants the user can login to.
+
+```python
+client.login_to_tenant(tenant_slug='a-tenant-slug')
+# OR
+client.login_to_tenant(tenant_id='A_TENANT_ID')
+```
+
+Both of these calls persist the active tenant in local storage, so you won't have to login again until you're ready to switch tenants.
+
+Once logged in, you can make any GraphQL query against the Cloud API:
+
+```python
+client.graphql(
+    {
+        'query': {
+            'flow': {
+                'id'
+            }
+        }
+    }
+)
+```
+
+(Note that this illustrates how Prefect can parse Python structures to construct GraphQL query strings!)
+
+Finally, you can logout:
+
+```python
+client.logout_from_tenant()
+```
+
+# GraphQL
+
+The Cloud API also supports direct GraphQL access. While you can still use `USER`-scoped tokens to access and log in to tenants, you will need to manage the short-lived auth and refresh tokens yourself. Therefore, we recommend using the Python client for `USER`-scoped access.
+
+For `TENANT`-scoped tokens, simply include the token as the authorization header of your GraphQL requests:
+
+```json
+{ "authorization": "Bearer YOUR_TOKEN" }
+```

docs/cloud/cloud_concepts/debug.md

@@ -11,7 +11,7 @@ The most likely culprit when a flow is stuck in scheduled is agent misconfigurat
 2. Check that the API token given to the agent is scoped to the same tenant as your flow
 
 ```
-$ export PREFECT__CLOUD__API_TOKEN=YOUR_TOKEN
+$ export PREFECT__CLOUD__AUTH_TOKEN=YOUR_TOKEN
 $ prefect get flows
 # if you do not see your flow then there is a tenant mismatch
 ```

src/prefect/agent/agent.py

@@ -36,7 +36,7 @@ class Agent:
     """
 
     def __init__(self) -> None:
-        self.client = Client(token=config.cloud.agent.get("auth_token"))
+        self.client = Client(api_token=config.cloud.agent.get("auth_token"))
 
         logger = logging.getLogger("agent")
         logger.setLevel(logging.DEBUG)

src/prefect/agent/kubernetes/resource_manager.py

@@ -25,7 +25,7 @@ def __init__(self) -> None:
         self.loop_interval = prefect_config.cloud.agent.resource_manager.get(
             "loop_interval"
         )
-        self.client = Client(token=prefect_config.cloud.agent.get("auth_token"))
+        self.client = Client(api_token=prefect_config.cloud.agent.get("auth_token"))
         self.namespace = os.getenv("NAMESPACE", "default")
 
         logger = logging.getLogger("resource-manager")

src/prefect/cli/auth.py

@@ -44,8 +44,7 @@ def login(token):
             abort=True,
         )
 
-    client = Client()
-    client.login(api_token=token)
+    client = Client(api_token=token)
 
     # Verify login obtained a valid api token
     try:
@@ -59,4 +58,7 @@ def login(token):
         click.secho("Error attempting to communicate with Prefect Cloud", fg="red")
         return
 
+    # save token
+    client.save_api_token()
+
     click.secho("Login successful", fg="green")