title: Airflow task argument return from previous task
body: <p>How can I set the function argument to a task that is the return from a previous task / function that was run.  Note that these tasks are defined programmatically, therefore I cannot simply use <code>xcom_pull(task_id="some_task")</code> because the tasks are defined in a loop (as shown below):</p>
  
  <pre><code>def scrape(site):
      return requests.get(site).content
  
  def echo(**kwargs):
      ti = kwargs['ti']
      # how can I use the input from `scrape()` here?
  
  for idx, site in enumerate(sites):
      myop = PythonOperator(task_id='scape_%s' % str(idx), python_callable=scrape, op_args=[site], dag=dag)
      echo_op = PythonOperator(task_id='echo_%s' % str(idx), dag=dag, provide_context=True, python_callable=echo)
      myop.set_downstream(echo_op)
  </code></pre>