diff --git a/src/cli.rs b/src/cli.rs
index 92dce34e7b..a59aef3216 100644
--- a/src/cli.rs
+++ b/src/cli.rs
@@ -80,8 +80,7 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
             Arc::new(Config),
             Arc::new(SkipWhile),
             command("sort-by", sort_by::sort_by),
-            command("inc", |x| plugin::plugin("inc".into(), x)),
-            command("sum", |x| plugin::plugin("sum".into(), x)),
+            command("inc", |x| plugin::filter_plugin("inc".into(), x)),
         ]);
 
         context.add_sinks(vec![
@@ -91,6 +90,7 @@ pub async fn cli() -> Result<(), Box<dyn Error>> {
             sink("table", table::table),
             sink("tree", tree::tree),
             sink("vtable", vtable::vtable),
+            sink("sum", |x| plugin::sink_plugin("sum".into(), x)),
         ]);
     }
 
diff --git a/src/commands/command.rs b/src/commands/command.rs
index 36e521de67..e9eac51b58 100644
--- a/src/commands/command.rs
+++ b/src/commands/command.rs
@@ -82,6 +82,10 @@ pub trait Command {
             optional_positional: vec![],
             rest_positional: true,
             named: indexmap::IndexMap::new(),
+            is_filter: true,
+            is_sink: false,
+            can_load: vec![],
+            can_save: vec![],
         }
     }
 }
@@ -97,6 +101,10 @@ pub trait Sink {
             optional_positional: vec![],
             rest_positional: true,
             named: indexmap::IndexMap::new(),
+            is_filter: false,
+            is_sink: true,
+            can_load: vec![],
+            can_save: vec![],
         }
     }
 }
diff --git a/src/commands/config.rs b/src/commands/config.rs
index 3e8ec1992b..d4c793e7ca 100644
--- a/src/commands/config.rs
+++ b/src/commands/config.rs
@@ -33,6 +33,10 @@ impl Command for Config {
             optional_positional: vec![],
             rest_positional: false,
             named,
+            is_sink: true,
+            is_filter: false,
+            can_load: vec![],
+            can_save: vec![],
         }
     }
 }
diff --git a/src/commands/open.rs b/src/commands/open.rs
index 87d1d959e7..a030863979 100644
--- a/src/commands/open.rs
+++ b/src/commands/open.rs
@@ -28,11 +28,19 @@ impl Command for Open {
             optional_positional: vec![],
             rest_positional: false,
             named,
+            is_filter: true,
+            is_sink: false,
+            can_load: vec![],
+            can_save: vec![],
         }
     }
 }
 
-pub fn fetch(cwd: &PathBuf, location: &str, span: Span) -> Result<(Option<String>, String), ShellError> {
+pub fn fetch(
+    cwd: &PathBuf,
+    location: &str,
+    span: Span,
+) -> Result<(Option<String>, String), ShellError> {
     let mut cwd = cwd.clone();
     if location.starts_with("http:") || location.starts_with("https:") {
         let response = reqwest::get(location);
@@ -154,9 +162,7 @@ pub fn parse_as_value(
                     name_span,
                 )
             }),
-        _ => {
-            Ok(Value::string(contents))
-        }
+        _ => Ok(Value::string(contents)),
     }
 }
 
diff --git a/src/commands/plugin.rs b/src/commands/plugin.rs
index 8d7b2c5d84..60b972ad38 100644
--- a/src/commands/plugin.rs
+++ b/src/commands/plugin.rs
@@ -1,3 +1,4 @@
+use crate::commands::command::SinkCommandArgs;
 use crate::errors::ShellError;
 use crate::prelude::*;
 use serde::{self, Deserialize, Serialize};
@@ -26,21 +27,23 @@ impl<T> JsonRpc<T> {
 #[serde(tag = "method")]
 #[allow(non_camel_case_types)]
 pub enum NuResult {
-    response { params: VecDeque<ReturnValue> },
+    response {
+        params: Result<VecDeque<ReturnValue>, ShellError>,
+    },
 }
 
-pub fn plugin(plugin_name: String, args: CommandArgs) -> Result<OutputStream, ShellError> {
-    let input = args.input;
-    let args = if let Some(ref positional) = args.args.positional {
-        positional.clone()
-    } else {
-        vec![]
-    };
-
+pub fn filter_plugin(plugin_name: String, args: CommandArgs) -> Result<OutputStream, ShellError> {
     let mut path = std::path::PathBuf::from(".");
     path.push("target");
     path.push("debug");
     path.push(format!("nu_plugin_{}", plugin_name));
+
+    path = if path.exists() {
+        path
+    } else {
+        std::path::PathBuf::from(format!("nu_plugin_{}", plugin_name))
+    };
+
     let mut child = std::process::Command::new(path)
         .stdin(std::process::Stdio::piped())
         .stdout(std::process::Stdio::piped())
@@ -53,7 +56,7 @@ pub fn plugin(plugin_name: String, args: CommandArgs) -> Result<OutputStream, Sh
 
         let _reader = BufReader::new(stdout);
 
-        let request = JsonRpc::new("init", args.clone());
+        let request = JsonRpc::new("begin_filter", args.args);
         let request_raw = serde_json::to_string(&request).unwrap();
         stdin.write(format!("{}\n", request_raw).as_bytes())?;
     }
@@ -61,7 +64,8 @@ pub fn plugin(plugin_name: String, args: CommandArgs) -> Result<OutputStream, Sh
     let mut eos = VecDeque::new();
     eos.push_back(Value::Primitive(Primitive::EndOfStream));
 
-    let stream = input
+    let stream = args
+        .input
         .chain(eos)
         .map(move |v| match v {
             Value::Primitive(Primitive::EndOfStream) => {
@@ -90,20 +94,30 @@ pub fn plugin(plugin_name: String, args: CommandArgs) -> Result<OutputStream, Sh
                     Ok(_) => {
                         let response = serde_json::from_str::<NuResult>(&input);
                         match response {
-                            Ok(NuResult::response { params }) => params,
-                            Err(_) => {
+                            Ok(NuResult::response { params }) => match params {
+                                Ok(params) => params,
+                                Err(e) => {
+                                    let mut result = VecDeque::new();
+                                    result.push_back(ReturnValue::Value(Value::Error(Box::new(e))));
+                                    result
+                                }
+                            },
+                            Err(e) => {
                                 let mut result = VecDeque::new();
                                 result.push_back(ReturnValue::Value(Value::Error(Box::new(
-                                    ShellError::string("Error while processing input"),
+                                    ShellError::string(format!(
+                                        "Error while processing input: {:?} {}",
+                                        e, input
+                                    )),
                                 ))));
                                 result
                             }
                         }
                     }
-                    Err(_) => {
+                    Err(e) => {
                         let mut result = VecDeque::new();
                         result.push_back(ReturnValue::Value(Value::Error(Box::new(
-                            ShellError::string("Error while processing input"),
+                            ShellError::string(format!("Error while processing input: {:?}", e)),
                         ))));
                         result
                     }
@@ -114,3 +128,31 @@ pub fn plugin(plugin_name: String, args: CommandArgs) -> Result<OutputStream, Sh
 
     Ok(stream.boxed())
 }
+
+pub fn sink_plugin(plugin_name: String, args: SinkCommandArgs) -> Result<(), ShellError> {
+    let mut path = std::path::PathBuf::from(".");
+    path.push("target");
+    path.push("debug");
+    path.push(format!("nu_plugin_{}", plugin_name));
+
+    path = if path.exists() {
+        path
+    } else {
+        std::path::PathBuf::from(format!("nu_plugin_{}", plugin_name))
+    };
+
+    let mut child = std::process::Command::new(path)
+        .stdin(std::process::Stdio::piped())
+        .spawn()
+        .expect("Failed to spawn child process");
+
+    let stdin = child.stdin.as_mut().expect("Failed to open stdin");
+
+    let request = JsonRpc::new("sink", (args.args, args.input));
+    let request_raw = serde_json::to_string(&request).unwrap();
+    stdin.write(format!("{}\n", request_raw).as_bytes())?;
+
+    let _ = child.wait();
+
+    Ok(())
+}
diff --git a/src/commands/skip_while.rs b/src/commands/skip_while.rs
index eb6a948823..466f8d9569 100644
--- a/src/commands/skip_while.rs
+++ b/src/commands/skip_while.rs
@@ -20,6 +20,10 @@ impl Command for SkipWhile {
             optional_positional: vec![],
             rest_positional: false,
             named: indexmap::IndexMap::new(),
+            is_filter: true,
+            is_sink: false,
+            can_load: vec![],
+            can_save: vec![],
         }
     }
 }
diff --git a/src/commands/where_.rs b/src/commands/where_.rs
index e5c2279ad3..afc0d901c4 100644
--- a/src/commands/where_.rs
+++ b/src/commands/where_.rs
@@ -19,6 +19,10 @@ impl Command for Where {
             optional_positional: vec![],
             rest_positional: false,
             named: indexmap::IndexMap::new(),
+            is_filter: true,
+            is_sink: false,
+            can_load: vec![],
+            can_save: vec![],
         }
     }
 }
diff --git a/src/lib.rs b/src/lib.rs
index 1f801a4355..c843906722 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -15,6 +15,7 @@ mod format;
 mod git;
 mod object;
 mod parser;
+mod plugin;
 mod prelude;
 mod shell;
 mod stream;
@@ -22,7 +23,9 @@ mod stream;
 pub use crate::commands::command::ReturnValue;
 pub use crate::parser::parse::span::SpannedItem;
 pub use crate::parser::Spanned;
+pub use crate::plugin::{serve_plugin, Plugin};
 pub use cli::cli;
 pub use errors::ShellError;
 pub use object::base::{Primitive, Value};
 pub use parser::parse::text::Text;
+pub use parser::registry::{Args, CommandConfig};
diff --git a/src/parser/registry.rs b/src/parser/registry.rs
index c3ee5ebf0f..52cc9f0497 100644
--- a/src/parser/registry.rs
+++ b/src/parser/registry.rs
@@ -5,17 +5,18 @@ use derive_new::new;
 use getset::Getters;
 use indexmap::IndexMap;
 use log::trace;
+use serde::{Deserialize, Serialize};
 use std::fmt;
 
 #[allow(unused)]
-#[derive(Debug)]
+#[derive(Debug, Serialize, Deserialize)]
 pub enum NamedType {
     Switch,
     Mandatory(NamedValue),
     Optional(NamedValue),
 }
 
-#[derive(Debug)]
+#[derive(Debug, Serialize, Deserialize)]
 pub enum NamedValue {
     Single,
 
@@ -33,7 +34,7 @@ impl NamedValue {
 }
 
 #[allow(unused)]
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum PositionalType {
     Value(String),
     Block(String),
@@ -55,17 +56,21 @@ impl PositionalType {
     }
 }
 
-#[derive(Debug, Getters)]
+#[derive(Debug, Getters, Serialize, Deserialize)]
 #[get = "crate"]
 pub struct CommandConfig {
-    crate name: String,
-    crate mandatory_positional: Vec<PositionalType>,
-    crate optional_positional: Vec<PositionalType>,
-    crate rest_positional: bool,
-    crate named: IndexMap<String, NamedType>,
+    pub name: String,
+    pub mandatory_positional: Vec<PositionalType>,
+    pub optional_positional: Vec<PositionalType>,
+    pub rest_positional: bool,
+    pub named: IndexMap<String, NamedType>,
+    pub is_filter: bool,
+    pub is_sink: bool,
+    pub can_load: Vec<String>,
+    pub can_save: Vec<String>,
 }
 
-#[derive(Debug, Default, new)]
+#[derive(Debug, Default, new, Serialize, Deserialize)]
 pub struct Args {
     pub positional: Option<Vec<Spanned<Value>>>,
     pub named: Option<IndexMap<String, Spanned<Value>>>,
diff --git a/src/plugin.rs b/src/plugin.rs
new file mode 100644
index 0000000000..412e5aa8a7
--- /dev/null
+++ b/src/plugin.rs
@@ -0,0 +1,101 @@
+use crate::{Args, CommandConfig, ReturnValue, ShellError, Value};
+use serde::{Deserialize, Serialize};
+use std::io;
+
+pub trait Plugin {
+    fn config(&mut self) -> Result<CommandConfig, ShellError> {
+        Err(ShellError::string("`config` not implemented in plugin"))
+    }
+    #[allow(unused)]
+    fn begin_filter(&mut self, args: Args) -> Result<(), ShellError> {
+        Err(ShellError::string(
+            "`begin_filter` not implemented in plugin",
+        ))
+    }
+    #[allow(unused)]
+    fn filter(&mut self, input: Value) -> Result<Vec<ReturnValue>, ShellError> {
+        Err(ShellError::string("`filter` not implemented in plugin"))
+    }
+    #[allow(unused)]
+    fn sink(&mut self, args: Args, input: Vec<Value>) {}
+
+    fn quit(&mut self) {
+        return;
+    }
+}
+
+pub fn serve_plugin(plugin: &mut dyn Plugin) {
+    loop {
+        let mut input = String::new();
+        match io::stdin().read_line(&mut input) {
+            Ok(_) => {
+                let command = serde_json::from_str::<NuCommand>(&input);
+                match command {
+                    Ok(NuCommand::config) => {
+                        send_response(plugin.config());
+                    }
+                    Ok(NuCommand::begin_filter { params }) => {
+                        let _ = plugin.begin_filter(params);
+                    }
+                    Ok(NuCommand::filter { params }) => {
+                        send_response(plugin.filter(params));
+                    }
+                    Ok(NuCommand::sink { params }) => {
+                        plugin.sink(params.0, params.1);
+                        break;
+                    }
+                    Ok(NuCommand::quit) => {
+                        plugin.quit();
+                        break;
+                    }
+                    e => {
+                        send_response(ShellError::string(format!(
+                            "Could not handle plugin message: {:?}",
+                            e,
+                        )));
+                        break;
+                    }
+                }
+            }
+            e => {
+                send_response(ShellError::string(format!(
+                    "Could not handle plugin message: {:?}",
+                    e,
+                )));
+                break;
+            }
+        }
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct JsonRpc<T> {
+    jsonrpc: String,
+    pub method: String,
+    pub params: T,
+}
+impl<T> JsonRpc<T> {
+    pub fn new<U: Into<String>>(method: U, params: T) -> Self {
+        JsonRpc {
+            jsonrpc: "2.0".into(),
+            method: method.into(),
+            params,
+        }
+    }
+}
+
+fn send_response<T: Serialize>(result: T) {
+    let response = JsonRpc::new("response", result);
+    let response_raw = serde_json::to_string(&response).unwrap();
+    println!("{}", response_raw);
+}
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(tag = "method")]
+#[allow(non_camel_case_types)]
+pub enum NuCommand {
+    config,
+    begin_filter { params: Args },
+    filter { params: Value },
+    sink { params: (Args, Vec<Value>) },
+    quit,
+}
diff --git a/src/plugins/inc.rs b/src/plugins/inc.rs
index d6ac3d84bf..afb0abeaf0 100644
--- a/src/plugins/inc.rs
+++ b/src/plugins/inc.rs
@@ -1,102 +1,49 @@
-use nu::{Primitive, ReturnValue, ShellError, Spanned, Value};
-use serde::{Deserialize, Serialize};
-use std::io;
+use nu::{serve_plugin, Args, Plugin, Primitive, ReturnValue, ShellError, Spanned, Value};
 
-/// A wrapper for proactive notifications to the IDE (eg. diagnostics). These must
-/// follow the JSON 2.0 RPC spec
-
-#[derive(Debug, Serialize, Deserialize)]
-pub struct JsonRpc<T> {
-    jsonrpc: String,
-    pub method: String,
-    pub params: Vec<T>,
+struct Inc {
+    inc_by: i64,
 }
-impl<T> JsonRpc<T> {
-    pub fn new<U: Into<String>>(method: U, params: Vec<T>) -> Self {
-        JsonRpc {
-            jsonrpc: "2.0".into(),
-            method: method.into(),
-            params,
-        }
+impl Inc {
+    fn new() -> Inc {
+        Inc { inc_by: 1 }
     }
 }
 
-fn send_response<T: Serialize>(result: Vec<T>) {
-    let response = JsonRpc::new("response", result);
-    let response_raw = serde_json::to_string(&response).unwrap();
-    println!("{}", response_raw);
-}
-#[derive(Debug, Serialize, Deserialize)]
-#[serde(tag = "method")]
-#[allow(non_camel_case_types)]
-pub enum NuCommand {
-    init { params: Vec<Spanned<Value>> },
-    filter { params: Value },
-    quit,
-}
-
-fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let mut inc_by = 1;
-
-    loop {
-        let mut input = String::new();
-        match io::stdin().read_line(&mut input) {
-            Ok(_) => {
-                let command = serde_json::from_str::<NuCommand>(&input);
-
-                match command {
-                    Ok(NuCommand::init { params }) => {
-                        for param in params {
-                            match param {
-                                Spanned {
-                                    item: Value::Primitive(Primitive::Int(i)),
-                                    ..
-                                } => {
-                                    inc_by = i;
-                                }
-                                _ => {
-                                    send_response(vec![ReturnValue::Value(Value::Error(
-                                        Box::new(ShellError::string("Unrecognized type in params")),
-                                    ))]);
-                                }
-                            }
-                        }
-                    }
-                    Ok(NuCommand::filter { params }) => match params {
-                        Value::Primitive(Primitive::Int(i)) => {
-                            send_response(vec![ReturnValue::Value(Value::int(i + inc_by))]);
-                        }
-                        Value::Primitive(Primitive::Bytes(b)) => {
-                            send_response(vec![ReturnValue::Value(Value::bytes(
-                                b + inc_by as u64,
-                            ))]);
-                        }
-                        x => {
-                            send_response(vec![ReturnValue::Value(Value::Error(Box::new(
-                                ShellError::string(format!("Unrecognized type in stream: {:?}", x)),
-                            )))]);
-                        }
-                    },
-                    Ok(NuCommand::quit) => {
-                        break;
-                    }
-                    Err(e) => {
-                        send_response(vec![ReturnValue::Value(Value::Error(Box::new(
-                            ShellError::string(format!(
-                                "Unrecognized type in stream: {} {:?}",
-                                input, e
-                            )),
-                        )))]);
+impl Plugin for Inc {
+    fn begin_filter(&mut self, args: Args) -> Result<(), ShellError> {
+        if let Some(args) = args.positional {
+            for arg in args {
+                match arg {
+                    Spanned {
+                        item: Value::Primitive(Primitive::Int(i)),
+                        ..
+                    } => {
+                        self.inc_by = i;
                     }
+                    _ => return Err(ShellError::string("Unrecognized type in params")),
                 }
             }
-            Err(_) => {
-                send_response(vec![ReturnValue::Value(Value::Error(Box::new(
-                    ShellError::string(format!("Unrecognized type in stream: {}", input)),
-                )))]);
-            }
         }
+
+        Ok(())
     }
 
-    Ok(())
+    fn filter(&mut self, input: Value) -> Result<Vec<ReturnValue>, ShellError> {
+        match input {
+            Value::Primitive(Primitive::Int(i)) => {
+                Ok(vec![ReturnValue::Value(Value::int(i + self.inc_by))])
+            }
+            Value::Primitive(Primitive::Bytes(b)) => Ok(vec![ReturnValue::Value(Value::bytes(
+                b + self.inc_by as u64,
+            ))]),
+            x => Err(ShellError::string(format!(
+                "Unrecognized type in stream: {:?}",
+                x
+            ))),
+        }
+    }
+}
+
+fn main() {
+    serve_plugin(&mut Inc::new());
 }
diff --git a/src/plugins/sum.rs b/src/plugins/sum.rs
index 0bbfc6a5ff..4576ad1d52 100644
--- a/src/plugins/sum.rs
+++ b/src/plugins/sum.rs
@@ -1,83 +1,33 @@
-use nu::{Primitive, ReturnValue, ShellError, Spanned, Value};
-use serde::{Deserialize, Serialize};
-use std::io;
+use nu::{serve_plugin, Args, Plugin, Primitive, Value};
 
-/// A wrapper for proactive notifications to the IDE (eg. diagnostics). These must
-/// follow the JSON 2.0 RPC spec
+struct Sum;
 
-#[derive(Debug, Serialize, Deserialize)]
-pub struct JsonRpc<T> {
-    jsonrpc: String,
-    pub method: String,
-    pub params: Vec<T>,
-}
-impl<T> JsonRpc<T> {
-    pub fn new<U: Into<String>>(method: U, params: Vec<T>) -> Self {
-        JsonRpc {
-            jsonrpc: "2.0".into(),
-            method: method.into(),
-            params,
-        }
+impl Sum {
+    fn new() -> Sum {
+        Sum
     }
 }
 
-fn send_response<T: Serialize>(result: Vec<T>) {
-    let response = JsonRpc::new("response", result);
-    let response_raw = serde_json::to_string(&response).unwrap();
-    println!("{}", response_raw);
-}
-#[derive(Debug, Serialize, Deserialize)]
-#[serde(tag = "method")]
-#[allow(non_camel_case_types)]
-pub enum NuCommand {
-    init { params: Vec<Spanned<Value>> },
-    filter { params: Value },
-    quit,
-}
+impl Plugin for Sum {
+    fn sink(&mut self, _args: Args, input: Vec<Value>) {
+        let mut total = 0i64;
 
-fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let mut total = 0i64;
-
-    loop {
-        let mut input = String::new();
-        match io::stdin().read_line(&mut input) {
-            Ok(_) => {
-                let command = serde_json::from_str::<NuCommand>(&input);
-
-                match command {
-                    Ok(NuCommand::init { .. }) => {}
-                    Ok(NuCommand::filter { params }) => match params {
-                        Value::Primitive(Primitive::Int(i)) => {
-                            total += i as i64;
-                            send_response(vec![ReturnValue::Value(Value::int(total))]);
-                        }
-                        Value::Primitive(Primitive::Bytes(b)) => {
-                            total += b as i64;
-                            send_response(vec![ReturnValue::Value(Value::bytes(total as u64))]);
-                        }
-                        _ => {
-                            send_response(vec![ReturnValue::Value(Value::Error(Box::new(
-                                ShellError::string("Unrecognized type in stream"),
-                            )))]);
-                        }
-                    },
-                    Ok(NuCommand::quit) => {
-                        break;
-                    }
-                    Err(_) => {
-                        send_response(vec![ReturnValue::Value(Value::Error(Box::new(
-                            ShellError::string("Unrecognized type in stream"),
-                        )))]);
-                    }
+        for v in input {
+            match v {
+                Value::Primitive(Primitive::Int(i)) => {
+                    total += i;
                 }
-            }
-            Err(_) => {
-                send_response(vec![ReturnValue::Value(Value::Error(Box::new(
-                    ShellError::string("Unrecognized type in stream"),
-                )))]);
+                Value::Primitive(Primitive::Bytes(i)) => {
+                    total += i as i64;
+                }
+                _ => {}
             }
         }
-    }
 
-    Ok(())
+        println!("Result: {}", total);
+    }
+}
+
+fn main() {
+    serve_plugin(&mut Sum::new());
 }