diff --git a/src/carnot/planner/logical_planner_test.cc b/src/carnot/planner/logical_planner_test.cc
index 3280df6c5bc..7203d3d4456 100644
--- a/src/carnot/planner/logical_planner_test.cc
+++ b/src/carnot/planner/logical_planner_test.cc
@@ -373,6 +373,85 @@ TEST_F(LogicalPlannerTest, PlanWithExecFuncs) {
   EXPECT_OK(plan->ToProto());
 }
 
+constexpr char kBPFTraceProgramMaxKernel[] = R"bpftrace(
+kprobe:tcp_drop
+{
+  ...
+}
+)bpftrace";
+
+constexpr char kBPFTraceProgramMinKernel[] = R"bpftrace(
+tracepoint:skb:kfree_skb
+{
+  ...
+}
+)bpftrace";
+
+constexpr char kTwoTraceProgramsPxl[] = R"pxl(
+import pxtrace
+import px
+
+before_518_trace_program = pxtrace.TraceProgram(
+  program="""$0""",
+  max_kernel='5.18',
+)
+
+after_519_trace_program = pxtrace.TraceProgram(
+  program="""$1""",
+  min_kernel='5.19',
+)
+
+table_name = 'tcp_drop_table'
+pxtrace.UpsertTracepoint('tcp_drop_tracer',
+                          table_name,
+                          [before_518_trace_program, after_519_trace_program],
+                          pxtrace.kprobe(),
+                          '10m')
+)pxl";
+
+constexpr char kBPFTwoTraceProgramsPb[] = R"proto(
+name: "tcp_drop_tracer"
+ttl {
+  seconds: 600
+}
+programs {
+  table_name: "tcp_drop_table"
+  bpftrace {
+    program: "\nkprobe:tcp_drop\n{\n  ...\n}\n"
+  }
+  selectors {
+    selector_type: MAX_KERNEL
+    value: "5.18"
+  }
+}
+programs {
+  table_name: "tcp_drop_table"
+  bpftrace {
+    program: "\ntracepoint:skb:kfree_skb\n{\n  ...\n}\n"
+  }
+  selectors {
+    selector_type: MIN_KERNEL
+    value: "5.19"
+  }
+}
+)proto";
+
+TEST_F(LogicalPlannerTest, CompileTwoTracePrograms) {
+  auto planner = LogicalPlanner::Create(info_).ConsumeValueOrDie();
+  plannerpb::CompileMutationsRequest req;
+  req.set_query_str(
+      absl::Substitute(kTwoTraceProgramsPxl, kBPFTraceProgramMaxKernel, kBPFTraceProgramMinKernel));
+  *req.mutable_logical_planner_state() =
+      testutils::CreateTwoPEMsOneKelvinPlannerState(testutils::kHttpEventsSchema);
+  auto trace_ir_or_s = planner->CompileTrace(req);
+  ASSERT_OK(trace_ir_or_s);
+  auto trace_ir = trace_ir_or_s.ConsumeValueOrDie();
+  plannerpb::CompileMutationsResponse resp;
+  ASSERT_OK(trace_ir->ToProto(&resp));
+  ASSERT_EQ(resp.mutations_size(), 1);
+  EXPECT_THAT(resp.mutations()[0].trace(), EqualsProto(kBPFTwoTraceProgramsPb));
+}
+
 constexpr char kSingleProbePxl[] = R"pxl(
 import pxtrace
 import px
@@ -391,7 +470,7 @@ pxtrace.UpsertTracepoint('http_return',
                          "5m")
 )pxl";
 
-constexpr char kSingleProbeProgramPb[] = R"pxl(
+constexpr char kSingleProbeProgramPb[] = R"proto(
 name: "http_return"
 ttl {
   seconds: 300
@@ -435,7 +514,7 @@ programs {
     }
   }
 }
-)pxl";
+)proto";
 
 TEST_F(LogicalPlannerTest, CompileTrace) {
   auto planner = LogicalPlanner::Create(info_).ConsumeValueOrDie();
diff --git a/src/carnot/planner/objects/qlobject.h b/src/carnot/planner/objects/qlobject.h
index ec1d900763c..4231fb78b0e 100644
--- a/src/carnot/planner/objects/qlobject.h
+++ b/src/carnot/planner/objects/qlobject.h
@@ -55,6 +55,7 @@ enum class QLObjectType {
   // General module type.
   kModule,
   kTraceModule,
+  kTraceProgram,
   kDict,
   kTracingVariable,
   kProbe,
diff --git a/src/carnot/planner/probes/probes.cc b/src/carnot/planner/probes/probes.cc
index 1991b2d89a8..fbe21a674d5 100644
--- a/src/carnot/planner/probes/probes.cc
+++ b/src/carnot/planner/probes/probes.cc
@@ -197,12 +197,17 @@ StatusOr<TracepointDeployment*> MutationsIR::CreateKProbeTracepointDeployment(
   return raw;
 }
 
-Status TracepointDeployment::AddBPFTrace(const std::string& bpftrace,
-                                         const std::string& output_name) {
+Status TracepointDeployment::AddBPFTrace(const std::string& bpftrace_str,
+                                         const std::string& output_name,
+                                         const std::vector<TracepointSelector>& selectors) {
   carnot::planner::dynamic_tracing::ir::logical::TracepointDeployment::TracepointProgram
       tracepoint_pb;
-  tracepoint_pb.mutable_bpftrace()->set_program(bpftrace);
+  tracepoint_pb.mutable_bpftrace()->set_program(bpftrace_str);
+  // set the output table to write program results to
   tracepoint_pb.set_table_name(output_name);
+  for (const auto& selector : selectors) {
+    *tracepoint_pb.add_selectors() = selector;
+  }
   tracepoints_.push_back(tracepoint_pb);
   return Status::OK();
 }
diff --git a/src/carnot/planner/probes/probes.h b/src/carnot/planner/probes/probes.h
index 34a3f0f42f3..3578cdb33d6 100644
--- a/src/carnot/planner/probes/probes.h
+++ b/src/carnot/planner/probes/probes.h
@@ -36,6 +36,8 @@ namespace carnot {
 namespace planner {
 namespace compiler {
 
+using TracepointSelector = carnot::planner::dynamic_tracing::ir::logical::TracepointSelector;
+
 class ProbeOutput {
  public:
   ProbeOutput() = delete;
@@ -192,9 +194,11 @@ class TracepointDeployment {
    *
    * @param bpftrace_program the program in string format.
    * @param output_name the output table to write program results.
+   * @param selectors the selectors to use for the program.
    * @return Status
    */
-  Status AddBPFTrace(const std::string& bpftrace_program, const std::string& output_name);
+  Status AddBPFTrace(const std::string& bpftrace_str, const std::string& output_name,
+                     const std::vector<TracepointSelector>& selectors);
 
   std::string name() const { return name_; }
 
diff --git a/src/carnot/planner/probes/probes_test.cc b/src/carnot/planner/probes/probes_test.cc
index 77ead0a6792..02914616b11 100644
--- a/src/carnot/planner/probes/probes_test.cc
+++ b/src/carnot/planner/probes/probes_test.cc
@@ -648,6 +648,219 @@ TEST_F(ProbeCompilerTest, parse_bpftrace) {
               testing::proto::EqualsProto(absl::Substitute(kBPFTraceProgramPb, literal_bpf_trace)));
 }
 
+constexpr char kBPFTraceProgramMaxKernel[] = R"bpftrace(
+kprobe:tcp_drop
+{
+  ...
+}
+)bpftrace";
+
+constexpr char kBPFTraceProgramMinKernel[] = R"bpftrace(
+tracepoint:skb:kfree_skb
+{
+  ...
+}
+)bpftrace";
+
+// Test that we can compile/parse a single TraceProgram object with a valid selector
+constexpr char kBPFSingleTraceProgramObjectPxl[] = R"pxl(
+import pxtrace
+import px
+
+after_519_trace_program = pxtrace.TraceProgram(
+  program="""$0""",
+  min_kernel='5.19',
+)
+
+table_name = 'tcp_drop_table'
+pxtrace.UpsertTracepoint('tcp_drop_tracer',
+                          table_name,
+                          after_519_trace_program,
+                          pxtrace.kprobe(),
+                          '10m')
+)pxl";
+
+constexpr char kBPFSingleTraceProgramObjectPb[] = R"proto(
+name: "tcp_drop_tracer"
+ttl {
+  seconds: 600
+}
+programs {
+  table_name: "tcp_drop_table"
+  bpftrace {
+    program: "$0"
+  }
+  selectors {
+    selector_type: MIN_KERNEL
+    value: "5.19"
+  }
+}
+)proto";
+
+TEST_F(ProbeCompilerTest, parse_single_bpftrace_program_object) {
+  ASSERT_OK_AND_ASSIGN(auto probe_ir,
+                       CompileProbeScript(absl::Substitute(kBPFSingleTraceProgramObjectPxl,
+                                                           kBPFTraceProgramMinKernel)));
+  plannerpb::CompileMutationsResponse pb;
+  EXPECT_OK(probe_ir->ToProto(&pb));
+  ASSERT_EQ(pb.mutations_size(), 1);
+
+  std::string literal_bpf_trace_min = kBPFTraceProgramMinKernel;
+  literal_bpf_trace_min = std::regex_replace(literal_bpf_trace_min, std::regex("\n"), "\\n");
+
+  EXPECT_THAT(pb.mutations()[0].trace(),
+              testing::proto::EqualsProto(
+                  absl::Substitute(kBPFSingleTraceProgramObjectPb, literal_bpf_trace_min)));
+}
+
+// Test that we can compile a list of TraceProgram objects with valid selectors
+constexpr char kBPFTraceProgramObjectsPxl[] = R"pxl(
+import pxtrace
+import px
+
+before_518_trace_program = pxtrace.TraceProgram(
+  program="""$0""",
+  max_kernel='5.18',
+)
+
+after_519_trace_program = pxtrace.TraceProgram(
+  program="""$1""",
+  min_kernel='5.19',
+)
+
+table_name = 'tcp_drop_table'
+pxtrace.UpsertTracepoint('tcp_drop_tracer',
+                          table_name,
+                          [before_518_trace_program, after_519_trace_program],
+                          pxtrace.kprobe(),
+                          '10m')
+)pxl";
+
+constexpr char kBPFTraceProgramObjectsPb[] = R"proto(
+name: "tcp_drop_tracer"
+ttl {
+  seconds: 600
+}
+programs {
+  table_name: "tcp_drop_table"
+  bpftrace {
+    program: "$0"
+  }
+  selectors {
+    selector_type: MAX_KERNEL
+    value: "5.18"
+  }
+}
+programs {
+  table_name: "tcp_drop_table"
+  bpftrace {
+    program: "$1"
+  }
+  selectors {
+    selector_type: MIN_KERNEL
+    value: "5.19"
+  }
+}
+)proto";
+
+TEST_F(ProbeCompilerTest, parse_multiple_bpftrace_program_objects) {
+  ASSERT_OK_AND_ASSIGN(auto probe_ir, CompileProbeScript(absl::Substitute(
+                                          kBPFTraceProgramObjectsPxl, kBPFTraceProgramMinKernel,
+                                          kBPFTraceProgramMaxKernel)));
+  plannerpb::CompileMutationsResponse pb;
+  EXPECT_OK(probe_ir->ToProto(&pb));
+  ASSERT_EQ(pb.mutations_size(), 1);
+
+  std::string literal_bpf_trace_min = kBPFTraceProgramMinKernel;
+  literal_bpf_trace_min = std::regex_replace(literal_bpf_trace_min, std::regex("\n"), "\\n");
+
+  std::string literal_bpf_trace_max = kBPFTraceProgramMaxKernel;
+  literal_bpf_trace_max = std::regex_replace(literal_bpf_trace_max, std::regex("\n"), "\\n");
+
+  EXPECT_THAT(pb.mutations()[0].trace(),
+              testing::proto::EqualsProto(absl::Substitute(
+                  kBPFTraceProgramObjectsPb, literal_bpf_trace_min, literal_bpf_trace_max)));
+}
+
+// Test that passing an unsupported selector type to TraceProgram throws a compiler error
+constexpr char kBPFUnsupportedTraceProgramObjectSelectorPxl[] = R"pxl(
+import pxtrace
+import px
+
+after_519_trace_program = pxtrace.TraceProgram(
+  program="""$0""",
+  min_kernel='5.19',
+  my_unsupported_selector='12345',
+)
+
+table_name = 'tcp_drop_table'
+pxtrace.UpsertTracepoint('tcp_drop_tracer',
+                          table_name,
+                          after_519_trace_program,
+                          pxtrace.kprobe(),
+                          '10m')
+)pxl";
+
+TEST_F(ProbeCompilerTest, parse_unsupported_selector_in_trace_program_object) {
+  auto probe_ir_or_s = CompileProbeScript(kBPFUnsupportedTraceProgramObjectSelectorPxl);
+  ASSERT_NOT_OK(probe_ir_or_s);
+  EXPECT_THAT(
+      probe_ir_or_s.status(),
+      HasCompilerError("Unsupported selector argument provided \'my_unsupported_selector\'"));
+}
+
+// Test that an invalid selector value throws a compiler error (currently needs to be a string)
+constexpr char kBPFInvalidTraceProgramObjectSelectorPxl[] = R"pxl(
+import pxtrace
+import px
+
+after_519_trace_program = pxtrace.TraceProgram(
+  program="""$0""",
+  min_kernel='5.19',
+  max_kernel=None,
+)
+
+table_name = 'tcp_drop_table'
+pxtrace.UpsertTracepoint('tcp_drop_tracer',
+                          table_name,
+                          after_519_trace_program,
+                          pxtrace.kprobe(),
+                          '10m')
+)pxl";
+
+TEST_F(ProbeCompilerTest, parse_invalid_trace_program_object) {
+  auto probe_ir_or_s = CompileProbeScript(kBPFInvalidTraceProgramObjectSelectorPxl);
+  ASSERT_NOT_OK(probe_ir_or_s);
+  EXPECT_THAT(probe_ir_or_s.status(),
+              HasCompilerError("Expected \'String\' in arg \'max_kernel\', got \'none\'"));
+}
+
+// Test that an empty selector value throws a compiler error
+constexpr char kBPFEmptyTraceProgramObjectSelectorPxl[] = R"pxl(
+import pxtrace
+import px
+
+after_519_trace_program = pxtrace.TraceProgram(
+  program="""$0""",
+  min_kernel='5.19',
+  max_kernel='',
+)
+
+table_name = 'tcp_drop_table'
+pxtrace.UpsertTracepoint('tcp_drop_tracer',
+                          table_name,
+                          after_519_trace_program,
+                          pxtrace.kprobe(),
+                          '10m')
+)pxl";
+
+TEST_F(ProbeCompilerTest, parse_empty_trace_program_object) {
+  auto probe_ir_or_s = CompileProbeScript(kBPFEmptyTraceProgramObjectSelectorPxl);
+  ASSERT_NOT_OK(probe_ir_or_s);
+  EXPECT_THAT(probe_ir_or_s.status(),
+              HasCompilerError("Empty selector value provided for \'max_kernel\'"));
+}
+
 constexpr char kConfigChangePxl[] = R"pxl(
 import pxconfig
 import px
diff --git a/src/carnot/planner/probes/tracing_module.cc b/src/carnot/planner/probes/tracing_module.cc
index c1bd0aa8d8d..0ee12c66cd0 100644
--- a/src/carnot/planner/probes/tracing_module.cc
+++ b/src/carnot/planner/probes/tracing_module.cc
@@ -49,6 +49,12 @@ class UpsertHandler {
                                     const ParsedArgs& args, ASTVisitor* visitor);
 };
 
+class TraceProgramHandler {
+ public:
+  static StatusOr<QLObjectPtr> Eval(const pypa::AstPtr& ast, const ParsedArgs& args,
+                                    ASTVisitor* visitor);
+};
+
 class SharedObjectHandler {
  public:
   static StatusOr<QLObjectPtr> Eval(const pypa::AstPtr& ast, const ParsedArgs& args,
@@ -153,6 +159,17 @@ Status TraceModule::Init() {
   PX_RETURN_IF_ERROR(upsert_fn->SetDocString(kUpsertTracepointDocstring));
   AddMethod(kUpsertTraceID, upsert_fn);
 
+  // Add pxtrace.TraceProgram object (FuncObject)
+  PX_ASSIGN_OR_RETURN(std::shared_ptr<FuncObject> program_fn,
+                      FuncObject::Create(kTraceProgramID, {"program"}, {},
+                                         /* has_variable_len_args */ false,
+                                         /* has_variable_len_kwargs */ true,
+                                         std::bind(TraceProgramHandler::Eval, std::placeholders::_1,
+                                                   std::placeholders::_2, std::placeholders::_3),
+                                         ast_visitor()));
+  // add method to the pxtrace module, pxtrace.TraceProgram
+  AddMethod(kTraceProgramID, program_fn);
+
   PX_ASSIGN_OR_RETURN(std::shared_ptr<FuncObject> shared_object_fn,
                       FuncObject::Create(kSharedObjectID, {"name", "upid"}, {},
                                          /* has_variable_len_args */ false,
@@ -341,6 +358,45 @@ StatusOr<QLObjectPtr> ReturnHandler::Eval(MutationsIR* mutations_ir, const pypa:
       std::make_shared<TracingVariableObject>(ast, visitor, id));
 }
 
+// Construct a TraceProgram object
+StatusOr<QLObjectPtr> TraceProgramHandler::Eval(const pypa::AstPtr& ast, const ParsedArgs& args,
+                                                ASTVisitor* visitor) {
+  std::vector<TracepointSelector> selectors;
+  // Check if supported selectors are passed in kwargs
+  const std::vector<NameToNode>& kwargs = args.kwargs();
+  const google::protobuf::EnumDescriptor* selector_type_descriptor =
+      carnot::planner::dynamic_tracing::ir::logical::SelectorType_descriptor();
+  for (const auto& [name, node] : kwargs) {
+    if (name == "program") {
+      continue;
+    }
+    // Enums are stored in uppercase, so we convert the argument key
+    const google::protobuf::EnumValueDescriptor* selector_value =
+        selector_type_descriptor->FindValueByName(absl::AsciiStrToUpper(name));
+    if (selector_value) {
+      // Selector type found
+      carnot::planner::dynamic_tracing::ir::logical::TracepointSelector tracepoint_selector;
+      tracepoint_selector.set_selector_type(
+          static_cast<carnot::planner::dynamic_tracing::ir::logical::SelectorType>(
+              selector_value->number()));
+      // Set user provided restriction, taken from the argument value
+      PX_ASSIGN_OR_RETURN(auto selector_value_ir, GetArgAs<StringIR>(node, name));
+      // Selector value is empty
+      if (selector_value_ir->str().empty()) {
+        return CreateAstError(ast, "Empty selector value provided for '$0'", name);
+      }
+      tracepoint_selector.set_value(selector_value_ir->str());
+      selectors.push_back(tracepoint_selector);
+    } else {
+      return CreateAstError(ast, "Unsupported selector argument provided '$0'", name);
+    }
+  }
+  // extract BPFTrace program string
+  PX_ASSIGN_OR_RETURN(auto program_ir, GetArgAs<StringIR>(ast, args, "program"));
+  return std::static_pointer_cast<QLObject>(
+      std::make_shared<TraceProgramObject>(ast, visitor, program_ir->str(), selectors));
+}
+
 StatusOr<QLObjectPtr> UpsertHandler::Eval(MutationsIR* mutations_ir, const pypa::AstPtr& ast,
                                           const ParsedArgs& args, ASTVisitor* visitor) {
   DCHECK(mutations_ir);
@@ -360,41 +416,41 @@ StatusOr<QLObjectPtr> UpsertHandler::Eval(MutationsIR* mutations_ir, const pypa:
   // const auto& pod_name = pod_name_ir->str();
   // const auto& container_name = pod_name_ir->str();
   // const auto& binary_name = binary_name_ir->str();
-  TracepointDeployment* trace_program;
+  TracepointDeployment* trace_deployment;
   auto target = args.GetArg("target");
   if (SharedObjectTarget::IsSharedObject(target)) {
     auto shared_object = std::static_pointer_cast<SharedObjectTarget>(target);
-    auto trace_program_or_s = mutations_ir->CreateTracepointDeployment(
+    auto trace_deployment_or_s = mutations_ir->CreateTracepointDeployment(
         tp_deployment_name, shared_object->shared_object(), ttl_ns);
-    PX_RETURN_IF_ERROR(WrapAstError(ast, trace_program_or_s.status()));
-    trace_program = trace_program_or_s.ConsumeValueOrDie();
+    PX_RETURN_IF_ERROR(WrapAstError(ast, trace_deployment_or_s.status()));
+    trace_deployment = trace_deployment_or_s.ConsumeValueOrDie();
   } else if (KProbeTarget::IsKProbeTarget(target)) {
-    auto trace_program_or_s =
+    auto trace_deployment_or_s =
         mutations_ir->CreateKProbeTracepointDeployment(tp_deployment_name, ttl_ns);
-    PX_RETURN_IF_ERROR(WrapAstError(ast, trace_program_or_s.status()));
-    trace_program = trace_program_or_s.ConsumeValueOrDie();
+    PX_RETURN_IF_ERROR(WrapAstError(ast, trace_deployment_or_s.status()));
+    trace_deployment = trace_deployment_or_s.ConsumeValueOrDie();
   } else if (ProcessTarget::IsProcessTarget(target)) {
     auto process_target = std::static_pointer_cast<ProcessTarget>(target);
-    auto trace_program_or_s = mutations_ir->CreateTracepointDeploymentOnProcessSpec(
+    auto trace_deployment_or_s = mutations_ir->CreateTracepointDeploymentOnProcessSpec(
         tp_deployment_name, process_target->target(), ttl_ns);
-    PX_RETURN_IF_ERROR(WrapAstError(ast, trace_program_or_s.status()));
-    trace_program = trace_program_or_s.ConsumeValueOrDie();
+    PX_RETURN_IF_ERROR(WrapAstError(ast, trace_deployment_or_s.status()));
+    trace_deployment = trace_deployment_or_s.ConsumeValueOrDie();
   } else if (LabelSelectorTarget::IsLabelSelectorTarget(target)) {
     auto label_selector_target = std::static_pointer_cast<LabelSelectorTarget>(target);
-    auto trace_program_or_s = mutations_ir->CreateTracepointDeploymentOnLabelSelectorSpec(
+    auto trace_deployment_or_s = mutations_ir->CreateTracepointDeploymentOnLabelSelectorSpec(
         tp_deployment_name, label_selector_target->target(), ttl_ns);
-    PX_RETURN_IF_ERROR(WrapAstError(ast, trace_program_or_s.status()));
-    trace_program = trace_program_or_s.ConsumeValueOrDie();
+    PX_RETURN_IF_ERROR(WrapAstError(ast, trace_deployment_or_s.status()));
+    trace_deployment = trace_deployment_or_s.ConsumeValueOrDie();
   } else if (ExprObject::IsExprObject(target)) {
     auto expr_object = std::static_pointer_cast<ExprObject>(target);
     if (Match(expr_object->expr(), UInt128Value())) {
       PX_ASSIGN_OR_RETURN(UInt128IR * upid_ir, GetArgAs<UInt128IR>(ast, args, "target"));
       md::UPID upid(upid_ir->val());
 
-      auto trace_program_or_s =
+      auto trace_deployment_or_s =
           mutations_ir->CreateTracepointDeployment(tp_deployment_name, upid, ttl_ns);
-      PX_RETURN_IF_ERROR(WrapAstError(ast, trace_program_or_s.status()));
-      trace_program = trace_program_or_s.ConsumeValueOrDie();
+      PX_RETURN_IF_ERROR(WrapAstError(ast, trace_deployment_or_s.status()));
+      trace_deployment = trace_deployment_or_s.ConsumeValueOrDie();
     } else {
       return CreateAstError(ast, "Unexpected type '$0' for arg '$1'",
                             expr_object->expr()->type_string(), "target");
@@ -404,18 +460,43 @@ StatusOr<QLObjectPtr> UpsertHandler::Eval(MutationsIR* mutations_ir, const pypa:
                           QLObjectTypeString(target->type()), "target");
   }
 
+  // looking at probe_fn arg of the UpsertTracepoint function and check what kind of object it is
+  // (Checking for FuncObject is a legacy thing, usually it's a bpftrace program as a string)
   if (FuncObject::IsFuncObject(args.GetArg("probe_fn"))) {
     PX_ASSIGN_OR_RETURN(auto probe_fn, GetCallMethod(ast, args.GetArg("probe_fn")));
     PX_ASSIGN_OR_RETURN(auto probe, probe_fn->Call({}, ast));
     CHECK(ProbeObject::IsProbe(probe));
     auto probe_ir = std::static_pointer_cast<ProbeObject>(probe)->probe();
     PX_RETURN_IF_ERROR(WrapAstError(
-        ast, trace_program->AddTracepoint(probe_ir.get(), tp_deployment_name, output_name)));
+        ast, trace_deployment->AddTracepoint(probe_ir.get(), tp_deployment_name, output_name)));
+    // If passing UpsertTracepoint a TraceProgram object or a list of TraceProgram objects,
+    // then we add the bpftrace script string and we create selectors from
+    // the arguments on the TraceProgram object, populating TracepointDeployment.selectors
+  } else if (CollectionObject::IsCollection(args.GetArg("probe_fn"))) {
+    // The probe_fn (QL object) is a list of TraceProgram objects.
+    // for each of the TraceProgram objects in the list, add the bpftrace script and selectors
+    for (const auto& item :
+         static_cast<CollectionObject*>(args.GetArg("probe_fn").get())->items()) {
+      if (!TraceProgramObject::IsTraceProgram(item)) {
+        return item->CreateError("Expected TraceProgram, got $0", item->name());
+      }
+      auto trace_program = static_cast<TraceProgramObject*>(item.get());
+      PX_RETURN_IF_ERROR(
+          WrapAstError(ast, trace_deployment->AddBPFTrace(trace_program->program(), output_name,
+                                                          trace_program->selectors())));
+    }
+  } else if (TraceProgramObject::IsTraceProgram(args.GetArg("probe_fn"))) {
+    // The probe_fn (QL object) is a single TraceProgram object.
+    auto trace_program = static_cast<TraceProgramObject*>(args.GetArg("probe_fn").get());
+    PX_RETURN_IF_ERROR(
+        WrapAstError(ast, trace_deployment->AddBPFTrace(trace_program->program(), output_name,
+                                                        trace_program->selectors())));
   } else {
     // The probe_fn is a string.
     PX_ASSIGN_OR_RETURN(auto program_str_ir, GetArgAs<StringIR>(ast, args, "probe_fn"));
-    PX_RETURN_IF_ERROR(
-        WrapAstError(ast, trace_program->AddBPFTrace(program_str_ir->str(), output_name)));
+    std::vector<TracepointSelector> empty_selectors;
+    PX_RETURN_IF_ERROR(WrapAstError(
+        ast, trace_deployment->AddBPFTrace(program_str_ir->str(), output_name, empty_selectors)));
   }
 
   return std::static_pointer_cast<QLObject>(std::make_shared<NoneObject>(ast, visitor));
diff --git a/src/carnot/planner/probes/tracing_module.h b/src/carnot/planner/probes/tracing_module.h
index 729169eb731..6b9bcc19df0 100644
--- a/src/carnot/planner/probes/tracing_module.h
+++ b/src/carnot/planner/probes/tracing_module.h
@@ -81,6 +81,30 @@ class ProbeObject : public QLObject {
   std::shared_ptr<TracepointIR> probe_;
 };
 
+class TraceProgramObject : public QLObject {
+ public:
+  static constexpr TypeDescriptor TracePointProgramType = {
+      /* name */ "TraceProgram",
+      /* type */ QLObjectType::kTraceProgram,
+  };
+
+  static bool IsTraceProgram(const QLObjectPtr& ptr) {
+    return ptr->type() == TracePointProgramType.type();
+  }
+  const std::string& program() const { return program_; }
+  const std::vector<TracepointSelector>& selectors() const { return selectors_; }
+
+  TraceProgramObject(const pypa::AstPtr& ast, ASTVisitor* visitor, const std::string& program,
+                     const std::vector<TracepointSelector>& selectors)
+      : QLObject(TracePointProgramType, ast, visitor),
+        program_(std::move(program)),
+        selectors_(selectors) {}
+
+ private:
+  std::string program_;
+  std::vector<TracepointSelector> selectors_;
+};
+
 class TraceModule : public QLObject {
  public:
   static constexpr TypeDescriptor TraceModuleType = {
@@ -173,6 +197,24 @@ class TraceModule : public QLObject {
       which it will be removed.
   )doc";
 
+  inline static constexpr char kTraceProgramID[] = "TraceProgram";
+  inline static constexpr char kTraceProgramDocstring[] = R"doc(
+  Creates a trace program.
+
+  :topic: pixie_state_management
+
+  Args:
+    program (str): The BPFtrace program string.
+    min_kernel (str, optional): The minimum kernel version that the tracepoint is supported on. Format is <version>.<major>.<minor>.
+    max_kernel (str, optional): The maximum kernel version that the tracepoint is supported on. Format is <version>.<major>.<minor>.
+    host_name (str, optional): Restrict the tracepoint to a specific host.
+    (Additional selectors may be added in the future.)
+
+  Returns:
+    TraceProgram: A pointer to the TraceProgram that can be passed as a probe_fn
+    to UpsertTracepoint.
+  )doc";
+
   inline static constexpr char kDeleteTracepointID[] = "DeleteTracepoint";
   inline static constexpr char kDeleteTracepointDocstring[] = R"doc(
   Deletes a tracepoint.