Coroutines look like generators and use yield to receive values. Note that coroutines are not generators, they are two separate concepts. Generators produce data, coroutines consume data. They just use the same keyword and type.

from typing import Generator

def grep(pattern: str) -> Generator[None, str, None]:
    while True:
        received = yield
        if pattern in received.lower():
            print(received)

my_grep = grep('coroutine')

# coroutines don't start automatically
# you need to start them with next()
next(my_grep)

my_grep.send('A: I like you.')
my_grep.send('J: I like coroutines.')  # => prints
my_grep.send('A: What did you say?')
my_grep.send('J: Coroutines are the best!')  # => prints

# you close a coroutine to stop it
# garbage collection also calls .close() if the variable goes out of scope
my_grep.close()

# now any extra calls to .send will raise StopIteration
my_grep.send('A: COME AGAIN?.')

It is easy to forget the next() initialization. You can write decorator for that.

from functools import wraps
from typing import Callable, Generator

def coroutine(func: Callable):
    @wraps(func)
    def primer(*args, **kwargs) -> Generator:
        cr = func(*args, **kwargs)
        next(cr)
        return cr

    return primer

@coroutine
def grep(pattern: str) -> Generator[None, str, None]:
    while True:
        received = yield
        if pattern in received.lower():
            print(received)

my_grep = grep('coroutine')
my_grep.send('A: I like you.')
my_grep.send('J: I like coroutines.')  # => prints

Don't mix generators and coroutines. You make it really hard to reason with the control flow if you mix these two concepts together.

from typing import Generator

def mirror() -> Generator[str, str, None]:
    while True:
        text = yield
        yield text

m = mirror()

next(m)  # moves pointer to coroutine receiver
assert m.send('hello') == 'hello'

next(m)  # continue after yield, ready to receive value
assert m.send('how are you') == 'how are you'

next(m)  # continue after yield, ready to receive value
assert m.send('bye') == 'bye'

A coroutine can be in one of four states.

• GEN_CREATED - waiting for execution
• GEN_STARTED - being executed by the interpreter
• GEN_SUSPENDED - suspended at a yield expression
• GEN_CLOSED - execution has completed
• You can only .send() if coroutine is GEN_SUSPENDED.

from inspect import getgeneratorstate
from typing import Generator

def echo() -> Generator[str, str, None]:
    previous = ''
    while True:
        text = yield previous.lower()  # this sends and receives
        previous = text

e = echo()
assert getgeneratorstate(e) == 'GEN_CREATED'
next(e)
assert getgeneratorstate(e) == 'GEN_SUSPENDED'
e.close()
assert getgeneratorstate(e) == 'GEN_CLOSED'

You can also raise errors on coroutines. If coroutine catches the error, it will continue until the next yield.

from typing import Generator

def echo() -> Generator[str, str, None]:
    previous = ''
    while True:
        try:
            text = yield previous.lower()  # this sends and receives
            previous = text
        except ValueError:
            pass

e = echo()
next(e)
assert e.send('HELLO?') == 'hello?'
assert e.throw(ValueError) == 'hello?'
assert e.throw(ValueError) == 'hello?'
assert e.throw(ValueError) == 'hello?'
e.throw(IndexError)
# => IndexError is raised

Coroutine return value is part of the StopIteration error. This is how yield from gets the value from subgenerator.

from typing import Generator, Optional, Tuple

def averager() -> Generator[None, Optional[int], Tuple[int, float]]:
    total: int = 0
    count: int = 0
    average: float = None
    while True:
        term = yield
        if term is None:
            break
        total += term
        count += 1
        average = total / count
    return count, average

avg = averager()
next(avg)
avg.send(10)
avg.send(30)
try:
    avg.send(None)
except StopIteration as exc:
    assert exc.value[0] == 2
    assert exc.value[1] == 20.0

Source

• Fluent Python, Luciano Ramalho
• Python Tips - Coroutines