# First Flow

# Write Flow

Below is an example flow you may use for this tutorial.

import prefect
from prefect import task, Flow

@task
def hello_task():
    logger = prefect.context.get("logger")
    logger.info("Hello, Cloud!")

flow = Flow("hello-flow", tasks=[hello_task])

Note that this Flow can be run locally by calling flow.run(), however, it currently does not yet use any of the features provided by the Prefect API.

# Register Flow with the Prefect API

In order to take advantage of the Prefect API for your flow, that flow must first be registered. Registration of a flow sends a flow's metadata to the Prefect API in order to support its orchestration.

Projects

Registering a flow with a backend requires users to organize flows into projects. In this case we are using the Hello, World! Project created in the "creating a project" tutorial.

Add the following line to the bottom of the example flow to register the flow with the Prefect API:

flow.register(project_name="Hello, World!")

Flow Code

Registration only sends data about the existence and format of your flow; no actual code from the flow is sent to the Prefect API. Your code remains safe, secure, and private in your own infrastructure!

# Run Flow with the Prefect API

Now that your flow is registered with the Prefect API, we will use an Agent to watch for flow runs that are scheduled by the Prefect API and execute them accordingly.

The simplest way to start a Local agent is right from within your script. Add the following line to the bottom of your flow:

flow.run_agent()

Runner Token Cloud

This Local Agent will use the RUNNER token stored in your environment but if you want to manually pass it a token you may do so with run_agent(token=<YOUR_RUNNER_TOKEN>).

Lastly, we need to indicate to the API to schedule a flow run; there are a few options at your disposal to do this:

prefect run flow --name hello-flow --project 'Hello, World!'

Navigate to the UI and click Run on your flow's page

mutation {
  create_flow_run(input: { flow_id: "<flow id>" }) {
    id
  }
}

See flow runs for more details.

from prefect import Client

c = Client()
c.create_flow_run(flow_id="<flow id>")

Notice the result of your flow run in the agent output in your terminal.

Remember to stop the agent with Ctrl-C.