diff --git a/.gitignore b/.gitignore
index 3179ebc..780f20e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,4 @@ dist/
 ffmpeg/tests/sample_data/out*.mp4
 ffmpeg_python.egg-info/
 venv*
+build/
diff --git a/ffmpeg/_ffmpeg.py b/ffmpeg/_ffmpeg.py
index 7ea3be2..7928930 100644
--- a/ffmpeg/_ffmpeg.py
+++ b/ffmpeg/_ffmpeg.py
@@ -16,6 +16,11 @@ from .nodes import (
 def input(filename, **kwargs):
     """Input file URL (ffmpeg ``-i`` option)
 
+    Any supplied kwargs are passed to ffmpeg verbatim (e.g. ``t=20``,
+    ``f='mp4'``, ``acodec='pcm'``, etc.).
+
+    To tell ffmpeg to read from stdin, use ``pipe:`` as the filename.
+
     Official documentation: `Main options <https://ffmpeg.org/ffmpeg.html#Main-options>`__
     """
     kwargs['filename'] = filename
@@ -57,7 +62,13 @@ def output(*streams_and_filename, **kwargs):
     Syntax:
         `ffmpeg.output(stream1[, stream2, stream3...], filename, **ffmpeg_args)`
 
-        If multiple streams are provided, they are mapped to the same output.
+    If multiple streams are provided, they are mapped to the same
+    output.
+
+    Any supplied kwargs are passed to ffmpeg verbatim (e.g. ``t=20``,
+    ``f='mp4'``, ``acodec='pcm'``, etc.).
+
+    To tell ffmpeg to write to stdout, use ``pipe:`` as the filename.
 
     Official documentation: `Synopsis <https://ffmpeg.org/ffmpeg.html#Synopsis>`__
     """
diff --git a/ffmpeg/_probe.py b/ffmpeg/_probe.py
old mode 100755
new mode 100644
index ea3a52c..d3658fd
--- a/ffmpeg/_probe.py
+++ b/ffmpeg/_probe.py
@@ -1,29 +1,25 @@
 import json
 import subprocess
-
-
-class ProbeException(Exception):
-    def __init__(self, stderr_output):
-        super(ProbeException, self).__init__('ffprobe error')
-        self.stderr_output = stderr_output
+from ._run import Error
 
 
 def probe(filename):
     """Run ffprobe on the specified file and return a JSON representation of the output.
 
     Raises:
-        ProbeException: if ffprobe returns a non-zero exit code, a ``ProbeException`` is returned with a generic error
-            message.  The stderr output can be retrieved by accessing the ``stderr_output`` property of the exception.
+        :class:`ffmpeg.Error`: if ffprobe returns a non-zero exit code,
+            an :class:`Error` is returned with a generic error message.
+            The stderr output can be retrieved by accessing the
+            ``stderr`` property of the exception.
     """
     args = ['ffprobe', '-show_format', '-show_streams', '-of', 'json', filename]
     p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
     out, err = p.communicate()
     if p.returncode != 0:
-        raise ProbeException(err)
+        raise Error('ffprobe', out, err)
     return json.loads(out.decode('utf-8'))
 
 
 __all__ = [
     'probe',
-    'ProbeException',
 ]
diff --git a/ffmpeg/_run.py b/ffmpeg/_run.py
index cec8bfb..c5a2b52 100644
--- a/ffmpeg/_run.py
+++ b/ffmpeg/_run.py
@@ -1,13 +1,13 @@
 from __future__ import unicode_literals
 
-from builtins import str
-from past.builtins import basestring
 from .dag import get_outgoing_edges, topo_sort
-from functools import reduce
 from ._utils import basestring
+from builtins import str
+from functools import reduce
+from past.builtins import basestring
 import copy
 import operator
-import subprocess as _subprocess
+import subprocess
 
 from ._ffmpeg import (
     input,
@@ -23,6 +23,13 @@ from .nodes import (
 )
 
 
+class Error(Exception):
+    def __init__(self, cmd, stdout, stderr):
+        super(Error, self).__init__('{} error (see stderr output for detail)'.format(cmd))
+        self.stdout = stdout
+        self.stderr = stderr
+
+
 def _convert_kwargs_to_cmd_line_args(kwargs):
     args = []
     for k in sorted(kwargs.keys()):
@@ -80,8 +87,9 @@ def _allocate_filter_stream_names(filter_nodes, outgoing_edge_maps, stream_name_
         for upstream_label, downstreams in list(outgoing_edge_map.items()):
             if len(downstreams) > 1:
                 # TODO: automatically insert `splits` ahead of time via graph transformation.
-                raise ValueError('Encountered {} with multiple outgoing edges with same upstream label {!r}; a '
-                                 '`split` filter is probably required'.format(upstream_node, upstream_label))
+                raise ValueError(
+                    'Encountered {} with multiple outgoing edges with same upstream label {!r}; a '
+                    '`split` filter is probably required'.format(upstream_node, upstream_label))
             stream_name_map[upstream_node, upstream_label] = 's{}'.format(stream_count)
             stream_count += 1
 
@@ -122,7 +130,7 @@ def _get_output_args(node, stream_name_map):
 
 @output_operator()
 def get_args(stream_spec, overwrite_output=False):
-    """Get command-line arguments for ffmpeg."""
+    """Build command-line arguments to be passed to ffmpeg."""
     nodes = get_stream_spec_nodes(stream_spec)
     args = []
     # TODO: group nodes together, e.g. `-i somefile -r somerate`.
@@ -144,27 +152,57 @@ def get_args(stream_spec, overwrite_output=False):
 
 
 @output_operator()
-def compile(stream_spec, cmd='ffmpeg', **kwargs):
-    """Build command-line for ffmpeg."""
+def compile(stream_spec, cmd='ffmpeg', overwrite_output=False):
+    """Build command-line for invoking ffmpeg.
+
+    The :meth:`run` function uses this to build the commnad line
+    arguments and should work in most cases, but calling this function
+    directly is useful for debugging or if you need to invoke ffmpeg
+    manually for whatever reason.
+
+    This is the same as calling :meth:`get_args` except that it also
+    includes the ``ffmpeg`` command as the first argument.
+    """
     if isinstance(cmd, basestring):
         cmd = [cmd]
     elif type(cmd) != list:
         cmd = list(cmd)
-    return cmd + get_args(stream_spec, **kwargs)
+    return cmd + get_args(stream_spec, overwrite_output=overwrite_output)
 
 
 @output_operator()
-def run(stream_spec, cmd='ffmpeg', **kwargs):
-    """Run ffmpeg on node graph.
+def run(
+        stream_spec, cmd='ffmpeg', capture_stdout=False, capture_stderr=False, input=None,
+        quiet=False, overwrite_output=False):
+    """Ivoke ffmpeg for the supplied node graph.
 
     Args:
-        **kwargs: keyword-arguments passed to ``get_args()`` (e.g. ``overwrite_output=True``).
+        capture_stdout: if True, capture stdout (to be used with
+            ``pipe:`` ffmpeg outputs).
+        capture_stderr: if True, capture stderr.
+        quiet: shorthand for setting ``capture_stdout`` and ``capture_stderr``.
+        input: text to be sent to stdin (to be used with ``pipe:``
+            ffmpeg inputs)
+        **kwargs: keyword-arguments passed to ``get_args()`` (e.g.
+            ``overwrite_output=True``).
+
+    Returns: (out, err) tuple containing captured stdout and stderr data.
     """
-    _subprocess.check_call(compile(stream_spec, cmd, **kwargs))
+    args = compile(stream_spec, cmd, overwrite_output=overwrite_output)
+    stdin_stream = subprocess.PIPE if input else None
+    stdout_stream = subprocess.PIPE if capture_stdout or quiet else None
+    stderr_stream = subprocess.PIPE if capture_stderr or quiet else None
+    p = subprocess.Popen(args, stdin=stdin_stream, stdout=stdout_stream, stderr=stderr_stream)
+    out, err = p.communicate(input)
+    retcode = p.poll()
+    if retcode:
+        raise Error('ffmpeg', out, err)
+    return out, err
 
 
 __all__ = [
     'compile',
+    'Error',
     'get_args',
     'run',
 ]
diff --git a/ffmpeg/tests/test_ffmpeg.py b/ffmpeg/tests/test_ffmpeg.py
index c7805fa..25aedd1 100644
--- a/ffmpeg/tests/test_ffmpeg.py
+++ b/ffmpeg/tests/test_ffmpeg.py
@@ -101,7 +101,7 @@ def test_stream_repr():
     assert repr(dummy_out) == 'dummy()[{!r}] <{}>'.format(dummy_out.label, dummy_out.node.short_hash)
 
 
-def test_get_args_simple():
+def test__get_args__simple():
     out_file = ffmpeg.input('dummy.mp4').output('dummy2.mp4')
     assert out_file.get_args() == ['-i', 'dummy.mp4', 'dummy2.mp4']
 
@@ -111,6 +111,10 @@ def test_global_args():
     assert out_file.get_args() == ['-i', 'dummy.mp4', 'dummy2.mp4', '-progress', 'someurl']
 
 
+def _get_simple_example():
+    return ffmpeg.input(TEST_INPUT_FILE1).output(TEST_OUTPUT_FILE1)
+
+
 def _get_complex_filter_example():
     split = (ffmpeg
         .input(TEST_INPUT_FILE1)
@@ -134,7 +138,7 @@ def _get_complex_filter_example():
     )
 
 
-def test_get_args_complex_filter():
+def test__get_args__complex_filter():
     out = _get_complex_filter_example()
     args = ffmpeg.get_args(out)
     assert args == ['-i', TEST_INPUT_FILE1,
@@ -305,41 +309,81 @@ def test_filter_text_arg_str_escape():
 #    subprocess.check_call(['ffmpeg', '-version'])
 
 
-def test_compile():
+def test__compile():
     out_file = ffmpeg.input('dummy.mp4').output('dummy2.mp4')
     assert out_file.compile() == ['ffmpeg', '-i', 'dummy.mp4', 'dummy2.mp4']
     assert out_file.compile(cmd='ffmpeg.old') == ['ffmpeg.old', '-i', 'dummy.mp4', 'dummy2.mp4']
 
 
-def test_run():
+def test__run():
     stream = _get_complex_filter_example()
-    ffmpeg.run(stream)
+    out, err = ffmpeg.run(stream)
+    assert out is None
+    assert err is None
 
 
-def test_run_multi_output():
+@pytest.mark.parametrize('capture_stdout', [True, False])
+@pytest.mark.parametrize('capture_stderr', [True, False])
+def test__run__capture_out(mocker, capture_stdout, capture_stderr):
+    mocker.patch.object(ffmpeg._run, 'compile', return_value=['echo', 'test'])
+    stream = _get_simple_example()
+    out, err = ffmpeg.run(stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr)
+    if capture_stdout:
+        assert out == 'test\n'.encode()
+    else:
+        assert out is None
+    if capture_stderr:
+        assert err == ''.encode()
+    else:
+        assert err is None
+
+
+def test__run__input_output(mocker):
+    mocker.patch.object(ffmpeg._run, 'compile', return_value=['cat'])
+    stream = _get_simple_example()
+    out, err = ffmpeg.run(stream, input='test'.encode(), capture_stdout=True)
+    assert out == 'test'.encode()
+    assert err is None
+
+
+@pytest.mark.parametrize('capture_stdout', [True, False])
+@pytest.mark.parametrize('capture_stderr', [True, False])
+def test__run__error(mocker, capture_stdout, capture_stderr):
+    mocker.patch.object(ffmpeg._run, 'compile', return_value=['ffmpeg'])
+    stream = _get_complex_filter_example()
+    with pytest.raises(ffmpeg.Error) as excinfo:
+        out, err = ffmpeg.run(stream, capture_stdout=capture_stdout, capture_stderr=capture_stderr)
+    assert str(excinfo.value) == 'ffmpeg error (see stderr output for detail)'
+    out = excinfo.value.stdout
+    err = excinfo.value.stderr
+    if capture_stdout:
+        assert out == ''.encode()
+    else:
+        assert out is None
+    if capture_stderr:
+        assert err.decode().startswith('ffmpeg version')
+    else:
+        assert err is None
+
+
+def test__run__multi_output():
     in_ = ffmpeg.input(TEST_INPUT_FILE1)
     out1 = in_.output(TEST_OUTPUT_FILE1)
     out2 = in_.output(TEST_OUTPUT_FILE2)
     ffmpeg.run([out1, out2], overwrite_output=True)
 
 
-def test_run_dummy_cmd():
+def test__run__dummy_cmd():
     stream = _get_complex_filter_example()
     ffmpeg.run(stream, cmd='true')
 
 
-def test_run_dummy_cmd_list():
+def test__run__dummy_cmd_list():
     stream = _get_complex_filter_example()
     ffmpeg.run(stream, cmd=['true', 'ignored'])
 
 
-def test_run_failing_cmd():
-    stream = _get_complex_filter_example()
-    with pytest.raises(subprocess.CalledProcessError):
-        ffmpeg.run(stream, cmd='false')
-
-
-def test_custom_filter():
+def test__filter__custom():
     stream = ffmpeg.input('dummy.mp4')
     stream = ffmpeg.filter_(stream, 'custom_filter', 'a', 'b', kwarg1='c')
     stream = ffmpeg.output(stream, 'dummy2.mp4')
@@ -351,7 +395,7 @@ def test_custom_filter():
     ]
 
 
-def test_custom_filter_fluent():
+def test__filter__custom_fluent():
     stream = (ffmpeg
         .input('dummy.mp4')
         .filter_('custom_filter', 'a', 'b', kwarg1='c')
@@ -365,7 +409,7 @@ def test_custom_filter_fluent():
     ]
 
 
-def test_merge_outputs():
+def test__merge_outputs():
     in_ = ffmpeg.input('in.mp4')
     out1 = in_.output('out1.mp4')
     out2 = in_.output('out2.mp4')
@@ -441,14 +485,14 @@ def test_pipe():
     assert out_data == in_data[start_frame*frame_size:]
 
 
-def test_ffprobe():
+def test__probe():
     data = ffmpeg.probe(TEST_INPUT_FILE1)
     assert set(data.keys()) == {'format', 'streams'}
     assert data['format']['duration'] == '7.036000'
 
 
-def test_ffprobe_exception():
-    with pytest.raises(ffmpeg.ProbeException) as excinfo:
+def test__probe__exception():
+    with pytest.raises(ffmpeg.Error) as excinfo:
         ffmpeg.probe(BOGUS_INPUT_FILE)
-    assert str(excinfo.value) == 'ffprobe error'
-    assert b'No such file or directory' in excinfo.value.stderr_output
+    assert str(excinfo.value) == 'ffprobe error (see stderr output for detail)'
+    assert 'No such file or directory'.encode() in excinfo.value.stderr
diff --git a/requirements-base.txt b/requirements-base.txt
index b5e8ae3..6afd152 100644
--- a/requirements-base.txt
+++ b/requirements-base.txt
@@ -1,5 +1,6 @@
 future
 pytest
+pytest-mock
 pytest-runner
 sphinx
 tox
diff --git a/requirements.txt b/requirements.txt
index 84710e2..dfec4a1 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,18 +1,26 @@
 alabaster==0.7.10
+apipkg==1.4
 Babel==2.5.1
 certifi==2017.7.27.1
 chardet==3.0.4
 docutils==0.14
+execnet==1.5.0
+funcsigs==1.0.2
 future==0.16.0
 idna==2.6
 imagesize==0.7.1
 Jinja2==2.9.6
 MarkupSafe==1.0
+mock==2.0.0
+pbr==4.0.3
 pluggy==0.5.2
 py==1.4.34
 Pygments==2.2.0
 pytest==3.2.3
+pytest-forked==0.2
+pytest-mock==1.10.0
 pytest-runner==3.0
+pytest-xdist==1.22.2
 pytz==2017.3
 requests==2.18.4
 six==1.11.0
diff --git a/setup.py b/setup.py
index 6ee7bb0..3e4ee38 100644
--- a/setup.py
+++ b/setup.py
@@ -57,7 +57,7 @@ setup(
     name='ffmpeg-python',
     packages=['ffmpeg'],
     setup_requires=['pytest-runner'],
-    tests_require=['pytest'],
+    tests_require=['pytest', 'pytest-mock'],
     version=version,
     description='Python bindings for FFmpeg - with support for complex filtering',
     author='Karl Kroening',
diff --git a/tox.ini b/tox.ini
index c187ba3..f86ec4b 100644
--- a/tox.ini
+++ b/tox.ini
@@ -11,3 +11,4 @@ commands = py.test -vv
 deps =
     future
     pytest
+    pytest-mock