diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 8492e05..4ead8b7 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -10,7 +10,8 @@ permissions:
   contents: read
 
 jobs:
-  build:
+  lint-and-unit-tests:
+    name: Lint & Unit Tests
     runs-on: ubuntu-latest
     timeout-minutes: 15
 
@@ -58,10 +59,44 @@ jobs:
         working-directory: ./plugins/examples/nemocheck
         run: uv run pytest tests
 
-      # Server tests
+      # Server unit tests (no proto generation needed — envoy modules are mocked)
       - name: Install server test dependencies
         run: uv sync --group dev
       - name: Run server unit tests
         run: |
           echo "Running server unit tests..."
-          uv run pytest tests
+          uv run pytest tests/ --ignore=tests/integration
+
+  integration-tests:
+    name: Integration Tests
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    needs: lint-and-unit-tests
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6
+
+      - name: Set up Python 3.11
+        uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405  # v6
+        with:
+          python-version: "3.11"
+
+      - name: Install uv
+        run: pip install uv
+
+      # Build generated protos (gitignored, needed for real envoy imports)
+      - name: Build protobuf files
+        run: |
+          uv sync --group proto
+          USE_HTTPS=true ./proto-build.sh
+
+      - name: Install test dependencies
+        run: uv sync --group dev
+
+      - name: Run integration tests
+        env:
+          PYTHONPATH: src
+        run: |
+          echo "Running integration tests..."
+          uv run pytest tests/integration/ -v
diff --git a/pyproject.toml b/pyproject.toml
index 9107a30..f8abe6f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -39,6 +39,10 @@ exclude = [
 [tool.ruff.lint]
 select = ["E", "F", "I", "W"]
 
+[tool.ruff.lint.isort]
+known-first-party = ["src", "tests"]
+known-third-party = ["cpex", "envoy", "grpc", "google"]
+
 [tool.pytest.ini_options]
 log_cli = false
 log_cli_level = "INFO"
@@ -49,6 +53,9 @@ log_format = "%(asctime)s [%(module)s] [%(levelname)s] %(message)s"
 log_date_format = "%Y-%m-%d %H:%M:%S"
 testpaths = ["tests"]
 pythonpath = [".", "src"]
+markers = [
+    "integration: integration tests (start real gRPC server)",
+]
 filterwarnings = [
     "ignore::DeprecationWarning",
 ]
diff --git a/tests/pytest.ini b/pytest.ini
similarity index 53%
rename from tests/pytest.ini
rename to pytest.ini
index 10a6de9..ee67c66 100644
--- a/tests/pytest.ini
+++ b/pytest.ini
@@ -6,6 +6,12 @@ log_cli_date_format = %Y-%m-%d %H:%M:%S
 log_level = INFO
 log_format = %(asctime)s [%(module)s] [%(levelname)s] %(message)s
 log_date_format = %Y-%m-%d %H:%M:%S
-pythonpath = . src
+# Paths relative to rootdir (repo root).
+#   tests  = for `from conftest import ...` in unit tests
+#   src    = for generated envoy/xds protos used by integration tests
+pythonpath = tests src
+testpaths = tests
+markers =
+    integration: integration tests (start real gRPC server)
 filterwarnings =
     ignore::DeprecationWarning
diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/integration/config.yaml b/tests/integration/config.yaml
new file mode 100644
index 0000000..ac6f5d8
--- /dev/null
+++ b/tests/integration/config.yaml
@@ -0,0 +1,17 @@
+plugins:
+  - name: "PassthroughPlugin"
+    kind: "tests.integration.passthrough_plugin.plugin.PassthroughPlugin"
+    description: "Passthrough plugin for integration testing"
+    version: "0.1.0"
+    hooks: ["tool_pre_invoke", "tool_post_invoke"]
+    mode: "sequential"
+    config: {}
+
+plugin_dirs:
+  - "tests/integration/passthrough_plugin"
+
+plugin_settings:
+  parallel_execution_within_band: false
+  plugin_timeout: 10
+  fail_on_plugin_error: true
+  enable_plugin_api: false
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
new file mode 100644
index 0000000..8d0bed4
--- /dev/null
+++ b/tests/integration/conftest.py
@@ -0,0 +1,41 @@
+"""Fixtures for integration tests — starts a real gRPC ext-proc server.
+
+Uses module-scoped state so the server starts once per test module on the
+first test's event loop, then reuses for subsequent tests.
+"""
+
+import os
+import pathlib
+
+import grpc
+import pytest_asyncio
+from cpex.framework import PluginManager
+from envoy.service.ext_proc.v3 import external_processor_pb2_grpc as ep_grpc
+
+INTEGRATION_DIR = pathlib.Path(__file__).parent
+CONFIG_PATH = str(INTEGRATION_DIR / "config.yaml")
+
+
+@pytest_asyncio.fixture
+async def grpc_stub():
+    """Start a gRPC server and yield a connected stub, then tear down."""
+    import src.server as server_module
+
+    os.environ["PLUGIN_MANAGER_CONFIG"] = CONFIG_PATH
+    manager = PluginManager(CONFIG_PATH)
+    await manager.initialize()
+    server_module.manager = manager
+
+    server = grpc.aio.server()
+    ep_grpc.add_ExternalProcessorServicer_to_server(server_module.ExtProcServicer(), server)
+    port = server.add_insecure_port("127.0.0.1:0")
+    await server.start()
+
+    channel = grpc.aio.insecure_channel(f"127.0.0.1:{port}")
+    stub = ep_grpc.ExternalProcessorStub(channel)
+
+    yield stub
+
+    await channel.close()
+    await server.stop(grace=1)
+    await manager.shutdown()
diff --git a/tests/integration/passthrough_plugin/__init__.py b/tests/integration/passthrough_plugin/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/integration/passthrough_plugin/plugin.py b/tests/integration/passthrough_plugin/plugin.py
new file mode 100644
index 0000000..25bdd42
--- /dev/null
+++ b/tests/integration/passthrough_plugin/plugin.py
@@ -0,0 +1,59 @@
+"""Passthrough test plugin for integration testing.
+
+A minimal cpex Plugin that either passes through or blocks requests
+based on a class-level toggle, allowing tests to control behavior.
+"""
+
+import logging
+
+from cpex.framework import (
+    Plugin,
+    PluginConfig,
+    PluginContext,
+    PluginViolation,
+    ToolPostInvokePayload,
+    ToolPostInvokeResult,
+    ToolPreInvokePayload,
+    ToolPreInvokeResult,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class PassthroughPlugin(Plugin):
+    """Test plugin that can be toggled between passthrough and blocking mode."""
+
+    # Class-level toggles so tests can control behavior
+    block_pre_invoke = False
+    block_post_invoke = False
+
+    def __init__(self, config: PluginConfig):
+        super().__init__(config)
+
+    @classmethod
+    def reset(cls):
+        """Reset toggles to default passthrough mode."""
+        cls.block_pre_invoke = False
+        cls.block_post_invoke = False
+
+    async def tool_pre_invoke(self, payload: ToolPreInvokePayload, context: PluginContext) -> ToolPreInvokeResult:
+        if self.block_pre_invoke:
+            violation = PluginViolation(
+                reason="Blocked by test",
+                description="Pre-invoke blocked for testing",
+                code="TEST_BLOCKED",
+                mcp_error_code=-32602,
+            )
+            return ToolPreInvokeResult(continue_processing=False, violation=violation)
+        return ToolPreInvokeResult(continue_processing=True)
+
+    async def tool_post_invoke(self, payload: ToolPostInvokePayload, context: PluginContext) -> ToolPostInvokeResult:
+        if self.block_post_invoke:
+            violation = PluginViolation(
+                reason="Blocked by test",
+                description="Post-invoke blocked for testing",
+                code="TEST_BLOCKED",
+                mcp_error_code=-32603,
+            )
+            return ToolPostInvokeResult(continue_processing=False, violation=violation)
+        return ToolPostInvokeResult(continue_processing=True)
diff --git a/tests/integration/test_ext_proc_e2e.py b/tests/integration/test_ext_proc_e2e.py
new file mode 100644
index 0000000..93c1703
--- /dev/null
+++ b/tests/integration/test_ext_proc_e2e.py
@@ -0,0 +1,217 @@
+"""End-to-end integration tests for the ext-proc gRPC server.
+
+These tests start a real gRPC server with a passthrough test plugin
+and exercise the full request/response flow.
+"""
+
+import json
+
+import pytest
+from envoy.config.core.v3 import base_pb2 as core
+from envoy.service.ext_proc.v3 import external_processor_pb2 as ep
+
+from tests.integration.passthrough_plugin.plugin import PassthroughPlugin
+
+pytestmark = pytest.mark.integration
+
+
+# ---------------------------------------------------------------------------
+# Helper: send a single request over bidi stream and read the first response
+# ---------------------------------------------------------------------------
+
+
+async def send_one(stub, request):
+    """Open a bidi stream, write one request, signal done, read one response."""
+    call = stub.Process()
+    await call.write(request)
+    await call.done_writing()
+    response = await call.read()
+    return response
+
+
+# ---------------------------------------------------------------------------
+# Request Headers
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_request_headers_adds_custom_header(grpc_stub):
+    """Sending request_headers should return a header mutation with x-ext-proc-header."""
+    request = ep.ProcessingRequest(
+        request_headers=ep.HttpHeaders(
+            headers=core.HeaderMap(headers=[]),
+        )
+    )
+    response = await send_one(grpc_stub, request)
+
+    assert response.HasField("request_headers")
+    mutations = response.request_headers.response.header_mutation.set_headers
+    header_keys = [h.header.key for h in mutations]
+    assert "x-ext-proc-header" in header_keys
+
+
+# ---------------------------------------------------------------------------
+# Response Headers
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_response_headers_adds_custom_header(grpc_stub):
+    """Sending response_headers should return x-ext-proc-response-header."""
+    request = ep.ProcessingRequest(
+        response_headers=ep.HttpHeaders(
+            headers=core.HeaderMap(headers=[]),
+        )
+    )
+    response = await send_one(grpc_stub, request)
+
+    assert response.HasField("response_headers")
+    mutations = response.response_headers.response.header_mutation.set_headers
+    header_keys = [h.header.key for h in mutations]
+    assert "x-ext-proc-response-header" in header_keys
+
+
+# ---------------------------------------------------------------------------
+# Request Body — tools/call passthrough
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_tools_call_passthrough(grpc_stub):
+    """A tools/call request with passthrough plugin should return request_body (not immediate_response)."""
+    PassthroughPlugin.reset()
+
+    body = {
+        "jsonrpc": "2.0",
+        "id": "int-test-1",
+        "method": "tools/call",
+        "params": {"name": "echo", "arguments": {"msg": "hello"}},
+    }
+    request = ep.ProcessingRequest(
+        request_body=ep.HttpBody(
+            body=json.dumps(body).encode("utf-8"),
+            end_of_stream=True,
+        )
+    )
+    response = await send_one(grpc_stub, request)
+
+    assert response.HasField("request_body"), f"Expected request_body, got: {response}"
+    assert not response.HasField("immediate_response")
+
+
+# ---------------------------------------------------------------------------
+# Request Body — tools/call blocked
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_tools_call_blocked(grpc_stub):
+    """When the plugin blocks, the server returns an immediate_response with an MCP error."""
+    PassthroughPlugin.block_pre_invoke = True
+    try:
+        body = {
+            "jsonrpc": "2.0",
+            "id": "int-test-2",
+            "method": "tools/call",
+            "params": {"name": "dangerous_tool", "arguments": {"x": 1}},
+        }
+        request = ep.ProcessingRequest(
+            request_body=ep.HttpBody(
+                body=json.dumps(body).encode("utf-8"),
+                end_of_stream=True,
+            )
+        )
+        response = await send_one(grpc_stub, request)
+
+        assert response.HasField("immediate_response"), f"Expected immediate_response, got: {response}"
+        error_body = json.loads(response.immediate_response.body)
+        assert "error" in error_body
+        assert error_body["error"]["code"] == -32602
+        assert "Blocked by test" in error_body["error"]["message"]
+    finally:
+        PassthroughPlugin.reset()
+
+
+# ---------------------------------------------------------------------------
+# Request Body — non-tool method passthrough
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_non_tool_request_body_passthrough(grpc_stub):
+    """A non-tools/call request body should pass through without plugin invocation."""
+    body = {
+        "jsonrpc": "2.0",
+        "id": "int-test-3",
+        "method": "resources/list",
+        "params": {},
+    }
+    request = ep.ProcessingRequest(
+        request_body=ep.HttpBody(
+            body=json.dumps(body).encode("utf-8"),
+            end_of_stream=True,
+        )
+    )
+    response = await send_one(grpc_stub, request)
+
+    assert response.HasField("request_body")
+    assert not response.HasField("immediate_response")
+
+
+# ---------------------------------------------------------------------------
+# Response Body — tool result passthrough
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_response_body_tool_result_passthrough(grpc_stub):
+    """A tool result in the response body should pass through the post-invoke hook."""
+    PassthroughPlugin.reset()
+
+    tool_result = {
+        "jsonrpc": "2.0",
+        "id": "int-test-4",
+        "result": {"content": [{"type": "text", "text": "Tool output data"}]},
+    }
+    request = ep.ProcessingRequest(
+        response_body=ep.HttpBody(
+            body=json.dumps(tool_result).encode("utf-8"),
+            end_of_stream=True,
+        )
+    )
+    response = await send_one(grpc_stub, request)
+
+    assert response.HasField("response_body")
+    assert not response.HasField("immediate_response")
+
+
+# ---------------------------------------------------------------------------
+# Response Body — tool result blocked
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_response_body_tool_result_blocked(grpc_stub):
+    """When post-invoke blocks, the server returns an immediate_response error."""
+    PassthroughPlugin.block_post_invoke = True
+    try:
+        tool_result = {
+            "jsonrpc": "2.0",
+            "id": "int-test-5",
+            "result": {"content": [{"type": "text", "text": "Sensitive output"}]},
+        }
+        request = ep.ProcessingRequest(
+            response_body=ep.HttpBody(
+                body=json.dumps(tool_result).encode("utf-8"),
+                end_of_stream=True,
+            )
+        )
+        response = await send_one(grpc_stub, request)
+
+        assert response.HasField("immediate_response"), f"Expected immediate_response, got: {response}"
+        error_body = json.loads(response.immediate_response.body)
+        assert "error" in error_body
+        assert error_body["error"]["code"] == -32603
+        assert "Blocked by test" in error_body["error"]["message"]
+    finally:
+        PassthroughPlugin.reset()