Build a Task Tracker gRPC Service

Define a gRPC service in Python with unary and server-streaming RPCs.

Introduction

30 Second Summary

Every time you use an app, services behind the scenes are constantly sending data to each other. They need a fast, reliable way to communicate.

In this project, you will build a task tracker service where a Python client talks to a Python server over gRPC. You will define the service, generate Python code from it, and make both single-request and streaming calls.

What You'll Build

Picture running a Python script that creates tasks on a server, fetches one back by ID, and then streams every task straight to your terminal in a single call.

By the end of this project, you'll have:

  • A running gRPC server that accepts requests to create and fetch tasks, printing each action to the terminal as it happens.
  • A gRPC client that talks to your server from a separate terminal, creating tasks, fetching them individually, and streaming all tasks back at once.
  • Hands-on experience with the full gRPC workflow: writing a .proto file, running the code generator, and seeing the Python stubs it produces.
  • Secret Mission: Add error handling with gRPC status codes so the server rejects bad requests and the client handles failures gracefully.

Are there any prerequisites?

You just need basic Python knowledge (functions, classes, loops). No prior API or networking experience is required.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

You're going to build a task tracker service where a Python client talks to a Python server over gRPC. By the end, you'll be able to create tasks, fetch them by ID, and stream all tasks back at once.

Set Up Your gRPC Python Environment

You've committed to building a task tracker service with gRPC. Before you can define your service or write any server code, you need the gRPC libraries and code generation tools installed on your machine.

In this step, you'll create a project directory, set up a Python virtual environment, and install the two core packages that power gRPC in Python.

In this step, get ready to:
  • Create a task-tracker project directory with a Python virtual environment.
  • Install grpcio and grpcio-tools.
  • Verify the installation by printing the grpc version.
Create your project directory and virtual environment

Every project needs a home. You'll create a dedicated folder on your Desktop and set up a virtual environment inside it to keep your gRPC packages separate from the rest of your system.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop and create your project folder by running these commands:
cd ~/Desktop
mkdir task-tracker
cd task-tracker
  • Create a Python virtual environment inside the project folder by running:
python3 -m venv venv
  • Confirm the virtual environment was created by running:
ls

You should see a venv folder listed in the output.

What is a virtual environment?

A virtual environment is an isolated Python workspace. Packages you install inside it stay contained to this project, so they never conflict with other Python projects on your machine.

The venv folder holds a private copy of the Python interpreter and a place for installed packages to live.

  • Activate the virtual environment by running:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

Linux

source venv/bin/activate

Your terminal prompt now starts with (venv). This means the virtual environment is active and any packages you install will go into this project only.

Don't see (venv) in your prompt?

  • Make sure you ran the activation command for your operating system. macOS and Linux use source, while Windows uses a backslash path with no source prefix.
  • Check that you're still inside the task-tracker directory. Run pwd to confirm your location.
  • If you're on Windows using PowerShell, you may need to run Set-ExecutionPolicy -Scope Process -ExecutionPolicy Bypass first to allow script execution.

Still stuck? Help me activate my Python virtual environment

Install the gRPC packages

gRPC in Python relies on two packages. grpcio is the core runtime that handles connections and message serialization. grpcio-tools is the code generator that turns your service definitions into Python classes.

  • Install both packages by running:
python3 -m pip install grpcio grpcio-tools

What does this command do?

  • python3 -m pip runs pip through your virtual environment's Python, ensuring the packages land in the right place.
  • grpcio provides the gRPC runtime for sending and receiving messages over the network.
  • grpcio-tools includes the Protocol Buffers compiler that generates Python code from .proto files.

You should see pip download and install several packages. The output ends with a line confirming all packages were successfully installed.

Seeing an error during installation?

  • Check that your virtual environment is still active. Your terminal prompt should start with (venv). If it is not showing, re-run the activation command from the previous substep.
  • If you see a permissions error, make sure you are using python3 -m pip (not a system-level pip). The virtual environment should handle permissions automatically.

Need more help? Help me troubleshoot my gRPC package installation

Verify the installation

The packages are installed, but a quick sanity check makes sure everything is wired up correctly before you start building.

Before you run the next command, what do you think will happen when Python tries to import the grpc module and print its version?

  • Verify the gRPC installation by running:
python3 -c "import grpc; print(grpc.__version__)"

You should see a version number printed, such as 1.83.0. The exact number may differ, but any version number means gRPC is ready to go.

Seeing ModuleNotFoundError?

  • Make sure your virtual environment is active. Look for (venv) at the start of your prompt. If it is missing, activate the environment again and re-run the install command.
  • Confirm that the install step completed without errors. Scroll up in your terminal to check the install output for any failure messages.

Still seeing errors? Help me fix my gRPC import error

Your gRPC environment is ready. Next up, you'll write your first .proto file to define the task tracker service and generate Python code from it.

Define Your Service in a Proto File and Generate Python Code

Your environment is ready with grpcio and grpcio-tools installed. Now you need to tell gRPC what your service actually does.

Every gRPC service starts as a .proto file written in Protocol Buffers. This file defines what methods the server offers and what data travels between client and server. From that single file, gRPC generates all the Python networking code for you.

In this step, get ready to:
  • Write a .proto file that defines your task tracker service and its data shapes.
  • Generate Python code from the proto file using the grpcio-tools compiler.
  • Explore the generated code to understand what gRPC built for you.
Create the task.proto file

A .proto file is a blueprint for your service. It declares the methods clients can call (RPCs) and the structure of data each method sends and receives (messages).

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the task-tracker folder, then click Open.

You should see the task-tracker folder name in the VS Code title bar, with the venv folder visible in the Explorer sidebar on the left.

  • In the Explorer sidebar, hover over the TASK-TRACKER heading and click the New File icon (a page with a plus sign).
  • Type task.proto as the file name and press Enter.
  • Copy the following proto definition into task.proto:
syntax = "proto3";

package tasktracker;

service TaskService {
  rpc CreateTask (CreateTaskRequest) returns (TaskResponse);
  rpc GetTask (GetTaskRequest) returns (TaskResponse);
}

message CreateTaskRequest {
  string title = 1;
  string description = 2;
}

message GetTaskRequest {
  int32 id = 1;
}

message TaskResponse {
  int32 id = 1;
  string title = 2;
  string description = 3;
  bool completed = 4;
}

What does this proto file define?

This file has three sections that work together to describe your entire service contract.

  • service TaskService declares two RPC methods. CreateTask takes a title and description and returns a task. GetTask takes an ID and returns the matching task. Both are unary RPCs: one request in, one response out.
  • The message blocks define the shape of data that travels over the wire. Each field has a type, a name, and a unique number that Protocol Buffers uses to encode the data efficiently.
  • syntax = "proto3" tells the compiler to use version 3 of the Protocol Buffers language. package tasktracker groups everything under a single namespace to avoid naming collisions.
  • Save task.proto by pressing Cmd+S (macOS) or Ctrl+S (Windows).

✔️ Awesome, I've got everything!

Great. Make sure you have saved task.proto before moving on.

ⓧ I'd like to double check the full code

Here is the complete task.proto file. Compare it with yours line by line.

syntax = "proto3";

package tasktracker;

service TaskService {
  rpc CreateTask (CreateTaskRequest) returns (TaskResponse);
  rpc GetTask (GetTaskRequest) returns (TaskResponse);
}

message CreateTaskRequest {
  string title = 1;
  string description = 2;
}

message GetTaskRequest {
  int32 id = 1;
}

message TaskResponse {
  int32 id = 1;
  string title = 2;
  string description = 3;
  bool completed = 4;
}
Generate Python code from the proto file

The proto file is a blueprint, but Python cannot read it directly. The grpcio-tools package includes a compiler that reads your .proto file and generates ready-to-use Python classes and functions.

Before you run the next command, what files do you think it will create?

  • Generate the Python code by running this command in your terminal (make sure you are inside the task-tracker directory with the venv still active):
python3 -m grpc_tools.protoc -I. --python_out=. --pyi_out=. --grpc_python_out=. task.proto

What does this command do?

  • python3 -m grpc_tools.protoc runs the Protocol Buffers compiler bundled with grpcio-tools.
  • -I. tells the compiler to look in the current directory for .proto files.
  • --python_out=. generates message classes. --pyi_out=. generates type stubs for IDE autocompletion. --grpc_python_out=. generates the service stub and servicer classes.

The command produces no output on success. It silently creates the files in your current directory.

  • Confirm the files were generated by running:
ls

You should see three new files alongside your task.proto: task_pb2.py, task_pb2.pyi, and task_pb2_grpc.py.

Not seeing the generated files?

  • Make sure your terminal is inside the task-tracker folder, not a parent or child directory. Run pwd to check.
  • Confirm your virtual environment is active. You should see (venv) at the start of your terminal prompt. If not, run source venv/bin/activate to reactivate it.
  • If you see an error mentioning that the file does not reside within any path, double-check that task.proto is saved in the same directory where you ran the command.
  • help me troubleshoot the protoc code generation command
Explore the generated code

gRPC generated three files from your one proto definition. The most important one is task_pb2_grpc.py because it contains the building blocks you will use to create both the server and the client.

  • In VS Code's Explorer sidebar, click on task_pb2_grpc.py to open it.
  • Scroll through the file and find the three key pieces that gRPC created for you: TaskServiceStub, TaskServiceServicer, and add_TaskServiceServicer_to_server.

What are these three generated pieces?

  • TaskServiceStub is for clients. It provides methods like CreateTask() and GetTask() that your client calls to talk to the server.
  • TaskServiceServicer is the base class your server will extend. It defines empty versions of each RPC method. Your job is to fill them with real logic.
  • add_TaskServiceServicer_to_server is a registration function. It connects your custom server logic to the gRPC server so incoming requests get routed to your code.

You never edit these generated files. They are rebuilt every time you run the protoc command.

What about the other two files?

  • task_pb2.py contains the Python classes for your messages: CreateTaskRequest, GetTaskRequest, and TaskResponse. You will use these to construct request and response objects.
  • task_pb2.pyi is a type stub file that helps VS Code offer autocompletion and type hints as you write your server and client code.

Your service contract is defined and the generated code is ready. Next up, you will use that generated TaskServiceServicer base class to build a real server that creates and stores tasks.

Implement the gRPC Server

Your proto file is defined and the generated Python code is ready to use. The generated task_pb2_grpc.py gave you a TaskServiceServicer base class with placeholder methods that raise "not implemented" errors.

Now it's time to bring that skeleton to life. You'll create a server that stores tasks in memory and responds to client requests over gRPC.

In this step, get ready to:
  • Create server.py with the TaskServiceServicer class.
  • Implement the CreateTask and GetTask RPC methods.
  • Add the server startup function and run it.
Create the server file and class skeleton

The generated TaskServiceServicer base class has methods that do nothing by default. Your server subclasses it and fills in the actual task logic.

  • In VS Code, create a new file called server.py in the task-tracker directory and add the following code:
from concurrent import futures

import grpc

import task_pb2
import task_pb2_grpc


class TaskServiceServicer(task_pb2_grpc.TaskServiceServicer):
    def __init__(self):
        # In-memory storage for tasks, keyed by ID
        self.tasks = {}
        # Auto-incrementing counter for new task IDs
        self.next_id = 1

What does this code do?

  • from concurrent import futures imports the thread pool you'll use to handle multiple client connections at once.
  • import task_pb2 and import task_pb2_grpc pull in the generated message classes and service base class from your proto file.
  • TaskServiceServicer subclasses the generated base class. This is where you'll implement the actual RPC methods.
  • self.tasks is a dictionary that stores tasks in memory, keyed by their ID.
  • self.next_id tracks the next available task ID, starting at 1.
  • Save server.py.
  • Confirm the file exists by running:
ls server.py
  • You should see server.py printed in the output.

Don't see server.py?

  • Make sure you created the file inside the task-tracker directory, not a parent folder.
  • Check that your terminal is in the right directory by running pwd. You should see a path ending in task-tracker.
  • Help me find my server.py file
Implement the CreateTask and GetTask methods

Each RPC method you defined in task.proto becomes a Python method on your servicer class. Each method receives a request object (the incoming message) and a context object (gRPC metadata), and returns a response message.

  • In server.py, add the CreateTask method inside the TaskServiceServicer class, below the __init__ method:
    def CreateTask(self, request, context):
        # Assign an auto-incremented ID to the new task
        task_id = self.next_id
        self.next_id += 1
        # Build a TaskResponse from the request data
        task = task_pb2.TaskResponse(
            id=task_id,
            title=request.title,
            description=request.description,
            completed=False,
        )
        # Store the task in memory and log it
        self.tasks[task_id] = task
        print(f"Created task {task_id}: {request.title}")
        return task

What does CreateTask do?

  • request.title and request.description are the fields from the CreateTaskRequest message you defined in task.proto.
  • The method builds a TaskResponse with an auto-incremented ID and stores it in the self.tasks dictionary.
  • return task sends the response back to the client as a single message. This is a unary RPC: one request in, one response out.
  • Save server.py. The CreateTask method is ready. Next, you'll add GetTask to complete the servicer.
  • Still in server.py, add the GetTask method below CreateTask, inside the same class:
    def GetTask(self, request, context):
        # Look up the task by the ID the client sent
        task = self.tasks.get(request.id)
        if task:
            return task
        # Return an empty response if the task doesn't exist
        return task_pb2.TaskResponse()

What does GetTask do?

  • self.tasks.get(request.id) looks up a task by the ID the client sent in the GetTaskRequest message.
  • If the task exists, it returns it directly to the client.
  • If not, it returns an empty TaskResponse with default values (empty strings, zero ID). The Secret Mission later improves this with proper error codes.
  • Save server.py. Both RPC methods are now implemented.

Seeing an IndentationError?

  • Both CreateTask and GetTask must be indented one level inside the TaskServiceServicer class, at the same level as __init__.
  • Python uses indentation to define class structure. Make sure you're using consistent spaces (4 spaces per level) and not mixing tabs with spaces.
  • Help me fix my IndentationError in server.py
Add the server startup function

The servicer class has the logic, but it needs a gRPC server to receive and route incoming connections. The serve() function wires everything together.

  • In server.py, add the serve() function and the __main__ block below the TaskServiceServicer class (at the top level, not inside the class):
def serve():
    # Create a gRPC server with a thread pool for handling requests
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    # Register the servicer so gRPC knows which class handles RPCs
    task_pb2_grpc.add_TaskServiceServicer_to_server(
        TaskServiceServicer(), server
    )
    # Listen on all network interfaces, port 50051
    server.add_insecure_port("[::]:50051")
    server.start()
    print("Task service running on port 50051...")
    # Block the main thread so the server keeps running
    server.wait_for_termination()


if __name__ == "__main__":
    serve()

What does the serve function do?

  • grpc.server(futures.ThreadPoolExecutor(max_workers=10)) creates a gRPC server that uses 10 threads to handle incoming requests concurrently.
  • add_TaskServiceServicer_to_server registers your servicer class with the server. This tells gRPC which Python methods to call when a client sends an RPC.
  • add_insecure_port("[::]:50051") tells the server to listen on port 50051 on all network interfaces.
  • wait_for_termination() blocks the main thread so the server stays alive until you stop it with Ctrl+C.
  • Save server.py.

✔️ Awesome, I've got everything!

Great. Make sure server.py is saved before running it.

ⓧ I'd like to double check the full code

Your server.py file should have these sections in order. Scroll up and confirm each one is present:

  • Imports: from concurrent import futures, import grpc, import task_pb2, import task_pb2_grpc.
  • TaskServiceServicer class: __init__ (with self.tasks and self.next_id), CreateTask, and GetTask methods.
  • Server startup: serve() function and if __name__ == "__main__": block at the bottom.

Before you run the server, take a guess. What do you think you'll see printed in the terminal?

  • Start the server by running:
python3 server.py

You should see Task service running on port 50051... printed in the terminal. The server is now running and waiting for client connections.

Why doesn't the terminal give me my prompt back?

The wait_for_termination() call blocks the terminal on purpose. The server stays alive, listening for incoming gRPC calls. Press Ctrl+C to stop it when you need to. Leave it running for now.

Server won't start?

  • If you see ModuleNotFoundError: No module named 'grpc', your virtual environment is not active. Look for (venv) at the start of your terminal prompt. If it's missing, run source venv/bin/activate and try again.
  • If you see ModuleNotFoundError: No module named 'task_pb2', the generated files are not in the same directory as server.py. Run ls and confirm you see task_pb2.py, task_pb2_grpc.py, and server.py all listed.
  • If you see a SyntaxError, double check the serve() function is outside the class (no indentation), and that the if __name__ block is also at the top level.
  • Help me get my gRPC server running

Your server is up and listening for gRPC connections. Next up, you'll build a client to talk to it and make your first RPC calls.

Build the Client and Make Your First RPC Calls

Your gRPC server is up and running, waiting for connections. Now you need something on the other end to actually talk to it.

In this step, you'll build a Python client that connects to the server, creates tasks, and retrieves them. By the end, you'll spot a limitation in how the client fetches tasks that sets up the next step perfectly.

In this step, get ready to:
  • Create a gRPC client that connects to the server and creates tasks.
  • Fetch individual tasks using unary RPC calls.
  • Discover why fetching tasks one by one doesn't scale.
Create the client and send your first calls

The server is occupying your current terminal, so you need a second terminal to run the client.

  • In the VS Code terminal panel, click the + button to open a new terminal.
  • Activate the virtual environment in the new terminal by running this command:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

What if my terminal opened in a different folder?

If your terminal prompt doesn't show task-tracker, make sure you opened the task-tracker folder in VS Code. You can also navigate there manually with cd ~/Desktop/task-tracker.

  • Create a new file called client.py in the task-tracker folder.
  • Add the following code to client.py:
import grpc

import task_pb2
import task_pb2_grpc

def run():
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = task_pb2_grpc.TaskServiceStub(channel)

        # Create some tasks
        for title in ["Buy groceries", "Write report", "Fix bug #42"]:
            response = stub.CreateTask(
                task_pb2.CreateTaskRequest(
                    title=title,
                    description=f"Description for: {title}",
                )
            )
            print(f"Created task {response.id}: {response.title}")

        print("\n--- Fetching task 1 ---")
        task = stub.GetTask(task_pb2.GetTaskRequest(id=1))
        print(f"Got task: {task.title} (completed: {task.completed})")

if __name__ == "__main__":
    run()

What does this code do?

  • grpc.insecure_channel("localhost:50051") opens a connection to the server on port 50051. The with block ensures the channel closes cleanly when you're done.
  • TaskServiceStub is the client-side class that gRPC generated from your proto file. It has methods matching each RPC you defined: CreateTask and GetTask.
  • The for loop creates three tasks by calling stub.CreateTask() with a CreateTaskRequest message. Each call is a unary RPC: one request in, one response back.
  • stub.GetTask() fetches a single task by its ID. This is also a unary RPC.
  • Save client.py.
  • Run the client in terminal 2 (the new terminal you just opened):
python3 client.py

You should see three created tasks printed, followed by the details of task 1.

What should I see?

Your client terminal should show output like this:

Created task 1: Buy groceries
Created task 2: Write report
Created task 3: Fix bug #42

--- Fetching task 1 ---
Got task: Buy groceries (completed: False)

Switch to terminal 1 (where the server is running). You should also see three Created task lines printed by the server, confirming both sides of the connection are working.

Not seeing any output?

  • Make sure the server is still running in terminal 1. You should see Task service running on port 50051... in that terminal.
  • Check that you activated the virtual environment in terminal 2. Your prompt should show (venv) at the beginning.
  • Verify that client.py is saved in the same task-tracker folder as the generated task_pb2.py and task_pb2_grpc.py files.

Why isn't my gRPC client connecting to the server?

Fetch all tasks with hardcoded IDs

You fetched a single task by its ID. But what if you want all the tasks at once?

  • In client.py, find this line:
        print(f"Got task: {task.title} (completed: {task.completed})")
  • Add the following code directly below it, at the same indentation level (inside the with block):
        # Fetch all tasks by hardcoded IDs (demonstrates limitation)
        for task_id in [1, 2, 3]:
            task = stub.GetTask(task_pb2.GetTaskRequest(id=task_id))
            print(f"Task {task.id}: {task.title}")

What does this code do?

This loop calls GetTask three separate times, once for each hardcoded ID in [1, 2, 3]. Each call is its own unary RPC to the server.

  • Save client.py.

The server still has tasks 1 through 3 in memory from your first client run. Stop and restart it for a clean slate so the output is easy to read.

  • Switch to terminal 1 (where the server is running).
  • Press Ctrl+C to stop the server.
  • Restart the server by running:
python3 server.py

You should see Task service running on port 50051... printed again, confirming the server is ready.

Before you run the client, think about this: the client makes a total of 7 RPC calls (3 creates, 1 fetch, then 3 more fetches). What would happen if there were 100 tasks instead of 3?

  • Switch to terminal 2 and run the client:
python3 client.py

You should see all 3 created tasks, the single fetched task, and then all 3 tasks fetched individually by their hardcoded IDs.

What should I see?

Your full client output should look like this:

Created task 1: Buy groceries
Created task 2: Write report
Created task 3: Fix bug #42

--- Fetching task 1 ---
Got task: Buy groceries (completed: False)
Task 1: Buy groceries
Task 2: Write report
Task 3: Fix bug #42

Why doesn't this approach scale?

The loop works because you hardcoded the exact IDs [1, 2, 3]. In a real application, you wouldn't know how many tasks exist or what their IDs are.

Making a separate RPC call for each task also means more network round trips. With 100 tasks, that's 100 individual requests to the server. There has to be a better way.

✔️ Awesome, I've got everything!

Great. Double check that you've saved client.py and that both terminals are still active.

ⓧ I'd like to double check the full code

Here is the complete client.py so far:

import grpc

import task_pb2
import task_pb2_grpc

def run():
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = task_pb2_grpc.TaskServiceStub(channel)

        # Create some tasks
        for title in ["Buy groceries", "Write report", "Fix bug #42"]:
            response = stub.CreateTask(
                task_pb2.CreateTaskRequest(
                    title=title,
                    description=f"Description for: {title}",
                )
            )
            print(f"Created task {response.id}: {response.title}")

        print("\n--- Fetching task 1 ---")
        task = stub.GetTask(task_pb2.GetTaskRequest(id=1))
        print(f"Got task: {task.title} (completed: {task.completed})")

        # Fetch all tasks by hardcoded IDs (demonstrates limitation)
        for task_id in [1, 2, 3]:
            task = stub.GetTask(task_pb2.GetTaskRequest(id=task_id))
            print(f"Task {task.id}: {task.title}")

if __name__ == "__main__":
    run()

Your gRPC client is talking to the server, creating tasks and fetching them back. But that hardcoded ID loop is a dead end. Next up, you'll add server-side streaming so the client can ask for all tasks in a single call and receive them one by one as they arrive.

Add Server-Side Streaming to List All Tasks

In the previous step, your client created tasks and fetched them back. But fetching required hardcoding every ID and making a separate RPC call for each task.

Server-side streaming solves this problem. Instead of multiple individual calls, the client sends one request and the server streams back all matching responses one at a time. In this step, you'll add a ListTasks streaming RPC to your gRPC service.

In this step, get ready to:
  • Update the proto file with a server-streaming RPC definition.
  • Implement the streaming method on the server using Python's yield.
  • Replace the hardcoded fetch loop in the client with a streaming call.
Add a streaming RPC to your proto file

Every new gRPC method starts in the .proto file. You need a new RPC called ListTasks that returns a stream of tasks instead of a single response.

  • Switch back to task.proto in VS Code.
  • Replace the entire file contents with this updated definition:
syntax = "proto3";

package tasktracker;

service TaskService {
  rpc CreateTask (CreateTaskRequest) returns (TaskResponse);
  rpc GetTask (GetTaskRequest) returns (TaskResponse);
  rpc ListTasks (ListTasksRequest) returns (stream TaskResponse);
}

message CreateTaskRequest {
  string title = 1;
  string description = 2;
}

message GetTaskRequest {
  int32 id = 1;
}

message ListTasksRequest {}

message TaskResponse {
  int32 id = 1;
  string title = 2;
  string description = 3;
  bool completed = 4;
}

What's new here?

Two additions make this a streaming RPC.

  • rpc ListTasks (ListTasksRequest) returns (stream TaskResponse) uses the stream keyword before TaskResponse. This tells gRPC the server will send back multiple TaskResponse messages from a single call.
  • ListTasksRequest {} is an empty message. The client doesn't need to send any parameters to ask for all tasks.
  • Save task.proto.

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before running the code generation command next.

ⓧ I'd like to double check the full code

syntax = "proto3";

package tasktracker;

service TaskService {
  rpc CreateTask (CreateTaskRequest) returns (TaskResponse);
  rpc GetTask (GetTaskRequest) returns (TaskResponse);
  rpc ListTasks (ListTasksRequest) returns (stream TaskResponse);
}

message CreateTaskRequest {
  string title = 1;
  string description = 2;
}

message GetTaskRequest {
  int32 id = 1;
}

message ListTasksRequest {}

message TaskResponse {
  int32 id = 1;
  string title = 2;
  string description = 3;
  bool completed = 4;
}

The proto file defines the contract, but you still need to regenerate the Python code so gRPC knows about the new ListTasks method.

  • In your second terminal (not the one running the server), regenerate the Python code by running:
python3 -m grpc_tools.protoc -I. --python_out=. --pyi_out=. --grpc_python_out=. task.proto
  • Confirm the files were regenerated by running:
ls *.py *.pyi

You should see task_pb2.py, task_pb2.pyi, and task_pb2_grpc.py listed alongside your server.py and client.py.

Seeing a file not found error?

  • Make sure your terminal is inside the task-tracker directory where task.proto lives.
  • Make sure you saved task.proto before running the command.

Help me debug my protoc code generation command.

Implement the streaming method on the server

The regenerated code added a ListTasks skeleton to the base servicer class. Now you need to override it with your actual logic.

  • In server.py, find the GetTask method. Add the ListTasks method directly below it, inside the TaskServiceServicer class:
    def ListTasks(self, request, context):
        # Stream each stored task back to the client
        for task in self.tasks.values():
            print(f"Streaming task {task.id}: {task.title}")
            yield task

What does yield do here?

In a unary RPC like CreateTask, you return a single response. In a streaming RPC, yield sends each response to the client one at a time.

Python's yield keyword turns this method into a generator. gRPC reads from the generator and streams each yielded value to the client as it becomes available.

  • Save server.py.

✔️ Awesome, I've got everything!

Great. Make sure the new method is indented inside the class, at the same level as your other methods.

ⓧ I'd like to double check the full code

Your server.py is longer than 30 lines, so scroll up and confirm these three methods are present inside your TaskServiceServicer class, in this order:

  • __init__ (initializes self.tasks and self.next_id).
  • CreateTask (stores a new task and returns it).
  • GetTask (looks up a task by ID).
  • ListTasks (the new streaming method using yield).

The serve() function and if __name__ == "__main__": block at the bottom should be unchanged from the previous step.

The running server doesn't pick up code changes automatically. You need to restart it.

  • Switch to the terminal where your server is running.
  • Press Ctrl+C to stop the server.
  • Restart the server by running:
python3 server.py

You should see Task service running on port 50051... printed again.

Server crashes on startup?

  • Check that your ListTasks method is indented at the same level as CreateTask and GetTask. It must be inside the TaskServiceServicer class.
  • Make sure you re-ran the protoc command after saving the updated task.proto. The server imports the generated code, so it must match.

Help me debug my gRPC server startup error.

Stream tasks from the client

The server is ready to stream. Now update the client to use the new ListTasks call instead of fetching tasks one by one with hardcoded IDs.

  • In client.py, scroll to the bottom of the run() function.
  • Find this block of code:
        # Fetch all tasks by hardcoded IDs (demonstrates limitation)
        for task_id in [1, 2, 3]:
            task = stub.GetTask(task_pb2.GetTaskRequest(id=task_id))
            print(f"Task {task.id}: {task.title}")
  • Delete those four lines and replace them with this streaming call:
        print("\n--- Streaming all tasks ---")
        # Iterate over streamed responses from the server
        for task in stub.ListTasks(task_pb2.ListTasksRequest()):
            print(f"  Task {task.id}: {task.title}")

What changed?

The old code needed three separate GetTask calls with hardcoded IDs [1, 2, 3]. The new code makes one ListTasks call and iterates over the stream with a simple for loop.

The client doesn't need to know how many tasks exist or what their IDs are. The server streams them all back automatically.

  • Save client.py.

✔️ Awesome, I've got everything!

Great. Make sure the hardcoded loop is fully removed and replaced with the streaming code.

ⓧ I'd like to double check the full code

import grpc

import task_pb2
import task_pb2_grpc


def run():
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = task_pb2_grpc.TaskServiceStub(channel)

        # Create some tasks
        for title in ["Buy groceries", "Write report", "Fix bug #42"]:
            response = stub.CreateTask(
                task_pb2.CreateTaskRequest(
                    title=title,
                    description=f"Description for: {title}",
                )
            )
            print(f"Created task {response.id}: {response.title}")

        print("\n--- Fetching task 1 ---")
        task = stub.GetTask(task_pb2.GetTaskRequest(id=1))
        print(f"Got task: {task.title} (completed: {task.completed})")

        print("\n--- Streaming all tasks ---")
        for task in stub.ListTasks(task_pb2.ListTasksRequest()):
            print(f"  Task {task.id}: {task.title}")


if __name__ == "__main__":
    run()

Everything is in place. Before you run the client, think about what you expect to see. Will the output look different from the previous step's hardcoded approach?

  • In your second terminal (not the server terminal), run the client by running:
python3 client.py

You should see output showing 3 created tasks, the fetched task 1, and then a "Streaming all tasks" section listing all 3 tasks received from a single ListTasks call.

What should I see?

In the client terminal, you should see the three created tasks, the fetched task, and the streaming section printing each task with an indented format. No hardcoded IDs needed.

In the server terminal, look for the Streaming task log lines. These confirm the server is yielding each task one at a time through the stream.

Seeing an error about ListTasks?

  • Make sure you re-ran the protoc command after updating task.proto. The client imports task_pb2_grpc which must include the new ListTasks stub.
  • Make sure you restarted the server after updating server.py. The old server process doesn't know about the new method.
  • Check that ListTasksRequest() in the client uses parentheses, not curly braces.

Help me debug my gRPC streaming client error.

Your gRPC service now supports both unary and server-side streaming RPCs. You can create individual tasks, fetch them by ID, and stream all tasks back in a single call. Next up, take on the Secret Mission to add error handling with gRPC status codes.

Secret mission

Add Error Handling with gRPC Status Codes

Your gRPC service works, but what happens when a client asks for a task that doesn't exist? Or tries to create one without a title? Right now, the server silently returns an empty response with no indication anything went wrong. In this secret mission, you'll add proper error handling using gRPC status codes so the server communicates exactly what failed.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.

Resources you used:

  • The task-tracker project directory (contains your proto file, server, client, generated code, and the virtual environment with grpcio and grpcio-tools).
  • The gRPC server process (if still running in your terminal).

Keep everything running

No action needed. Choose this if you are still actively building or want to experiment more with your gRPC service.

  • Your project files stay in the task-tracker directory on your Desktop.
  • The virtual environment holds your installed packages (grpcio and grpcio-tools). These only exist inside the venv folder and do not affect your system Python.
  • If the server is still running, it will continue listening on port 50051 until you stop it with Ctrl+C.

Pause - I'll come back to this later

Shut down running processes to free up your terminal. Your project files will stay on disk so you can pick up where you left off.

  • If the gRPC server is still running, press Ctrl+C in its terminal to stop it.
  • Deactivate your virtual environment by running:
deactivate
  • When you come back, navigate to your task-tracker directory and reactivate the virtual environment with source venv/bin/activate.

Delete - I don't want to use this again

Remove all project files and start fresh. This deletes everything you built in this project.

  • If the gRPC server is still running, press Ctrl+C in its terminal to stop it.
  • Deactivate your virtual environment and move out of the project directory by running:
deactivate
cd ~/Desktop
  • Delete the task-tracker directory by running:

macOS

rm -rf task-tracker

Windows

rmdir /s /q task-tracker
  • Confirm the directory is gone by running:
ls
  • You should no longer see task-tracker in the output.

Nice Work!

Nice Work!

Nice work! You built a fully functional task tracker service that communicates over gRPC in Python.

You've learned how to:

  • Wrote a .proto file defining a gRPC service and generated Python server and client code using grpc_tools.protoc.
  • Built a gRPC server that handles unary RPCs (CreateTask, GetTask) and a gRPC client that connects, creates tasks, and fetches them.
  • Added server-side streaming (ListTasks) to send multiple responses from a single request using Python's yield.
  • Secret Mission: Added gRPC error handling with status codes (NOT_FOUND, INVALID_ARGUMENT) and client-side exception handling.

Ready to quiz yourself?