diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@carryover.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@carryover.graphql-2.snap
index 83cf15a887..be4d749610 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@carryover.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@carryover.graphql-2.snap
@@ -166,6 +166,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ca
         spec: V0_1,
         request_variables: {},
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "one_Query_t_0": Connector {
         id: ConnectId {
@@ -380,6 +382,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ca
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "one_T_r_0": Connector {
         id: ConnectId {
@@ -520,5 +524,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ca
             $this,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@interface-object.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@interface-object.graphql-2.snap
index edd2b57f5f..fad26e4cdd 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@interface-object.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@interface-object.graphql-2.snap
@@ -147,6 +147,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/in
             $this,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_Query_itfs_0": Connector {
         id: ConnectId {
@@ -240,6 +242,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/in
         spec: V0_1,
         request_variables: {},
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_Query_itf_0": Connector {
         id: ConnectId {
@@ -396,5 +400,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/in
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@keys.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@keys.graphql-2.snap
index 04f4c88382..3c37e970cb 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@keys.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@keys.graphql-2.snap
@@ -155,6 +155,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ke
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "one_Query_t2_0": Connector {
         id: ConnectId {
@@ -367,6 +369,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ke
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "one_Query_unselected_0": Connector {
         id: ConnectId {
@@ -519,6 +523,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ke
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "one_T_r1_0": Connector {
         id: ConnectId {
@@ -659,6 +665,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ke
             $this,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "one_T_r2_0": Connector {
         id: ConnectId {
@@ -859,6 +867,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ke
             $this,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "one_T_r3_0": Connector {
         id: ConnectId {
@@ -1043,6 +1053,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ke
         response_variables: {
             $this,
         },
+        request_headers: {},
+        response_headers: {},
     },
     "one_T_r4_0": Connector {
         id: ConnectId {
@@ -1200,6 +1212,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ke
             $this,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "one_T_r5_0": Connector {
         id: ConnectId {
@@ -1401,5 +1415,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ke
         response_variables: {
             $this,
         },
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@nested_inputs.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@nested_inputs.graphql-2.snap
index bd83daf098..19ad3df6d0 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@nested_inputs.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@nested_inputs.graphql-2.snap
@@ -262,5 +262,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ne
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@normalize_names.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@normalize_names.graphql-2.snap
index 6318b65e64..991eac5ed8 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@normalize_names.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@normalize_names.graphql-2.snap
@@ -83,6 +83,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/no
         spec: V0_1,
         request_variables: {},
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors-subgraph_Query_user_0": Connector {
         id: ConnectId {
@@ -227,5 +229,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/no
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@realistic.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@realistic.graphql-2.snap
index 91d300e6d0..1cea7f9292 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@realistic.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@realistic.graphql-2.snap
@@ -301,6 +301,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/re
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_Query_filterUsersByEmailDomain_0": Connector {
         id: ConnectId {
@@ -468,6 +470,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/re
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_Query_usersByCompany_0": Connector {
         id: ConnectId {
@@ -679,6 +683,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/re
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_Query_user_0": Connector {
         id: ConnectId {
@@ -1039,5 +1045,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/re
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@sibling_fields.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@sibling_fields.graphql-2.snap
index 8fc72357a1..3ec4b276cc 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@sibling_fields.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@sibling_fields.graphql-2.snap
@@ -100,6 +100,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/si
         spec: V0_1,
         request_variables: {},
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_T_b_0": Connector {
         id: ConnectId {
@@ -243,5 +245,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/si
             $this,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@simple.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@simple.graphql-2.snap
index 0323ff7192..45755b3616 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@simple.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@simple.graphql-2.snap
@@ -83,6 +83,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/si
         spec: V0_1,
         request_variables: {},
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_Query_user_0": Connector {
         id: ConnectId {
@@ -227,6 +229,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/si
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_User_d_1": Connector {
         id: ConnectId {
@@ -423,5 +427,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/si
             $this,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@steelthread.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@steelthread.graphql-2.snap
index be3b10be8b..cb8875ac51 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@steelthread.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@steelthread.graphql-2.snap
@@ -94,6 +94,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/st
         spec: V0_1,
         request_variables: {},
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_Query_user_0": Connector {
         id: ConnectId {
@@ -248,6 +250,8 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/st
             $args,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
     "connectors_User_d_1": Connector {
         id: ConnectId {
@@ -395,5 +399,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/st
             $this,
         },
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@types_used_twice.graphql-2.snap b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@types_used_twice.graphql-2.snap
index 7514920800..22f6303e0b 100644
--- a/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@types_used_twice.graphql-2.snap
+++ b/apollo-federation/src/sources/connect/expand/tests/snapshots/it_expand_supergraph@types_used_twice.graphql-2.snap
@@ -143,5 +143,7 @@ input_file: apollo-federation/src/sources/connect/expand/tests/schemas/expand/ty
         spec: V0_1,
         request_variables: {},
         response_variables: {},
+        request_headers: {},
+        response_headers: {},
     },
 }
diff --git a/apollo-federation/src/sources/connect/json_selection/parser.rs b/apollo-federation/src/sources/connect/json_selection/parser.rs
index 9527ed8441..fd91641e59 100644
--- a/apollo-federation/src/sources/connect/json_selection/parser.rs
+++ b/apollo-federation/src/sources/connect/json_selection/parser.rs
@@ -32,6 +32,7 @@ use super::location::WithRange;
 use crate::sources::connect::variable::VariableNamespace;
 use crate::sources::connect::variable::VariablePathPart;
 use crate::sources::connect::variable::VariableReference;
+use crate::sources::connect::Namespace;
 
 // ParseResult is the internal type returned by most ::parse methods, as it is
 // convenient to use with nom's combinators. The top-level JSONSelection::parse
@@ -239,6 +240,32 @@ impl JSONSelection {
             .flat_map(|var_path| var_path.variable_reference())
             .map(|var_ref| var_ref.namespace.namespace)
     }
+
+    /// Get any headers referenced in the variable references by looking at both Request and Response namespaces.
+    pub fn header_references(&self) -> impl Iterator<Item = String> + '_ {
+        self.external_var_paths()
+            .into_iter()
+            .flat_map(|var_path| var_path.variable_reference())
+            .filter_map(|var_ref: VariableReference<'_, Namespace>| {
+                if var_ref.namespace.namespace != Namespace::Request
+                    && var_ref.namespace.namespace != Namespace::Response
+                {
+                    return None;
+                }
+
+                // We only care if the path references starts with "headers"
+                if !var_ref
+                    .path
+                    .first()
+                    .map_or(false, |path| path.part == "headers")
+                {
+                    return None;
+                }
+
+                // Grab the name of the header from the path
+                var_ref.path.get(1).map(|path| path.part.to_string())
+            })
+    }
 }
 
 impl ExternalVarPaths for JSONSelection {
diff --git a/apollo-federation/src/sources/connect/models.rs b/apollo-federation/src/sources/connect/models.rs
index a6161ecd6e..d5af0e56cc 100644
--- a/apollo-federation/src/sources/connect/models.rs
+++ b/apollo-federation/src/sources/connect/models.rs
@@ -64,6 +64,11 @@ pub struct Connector {
 
     pub request_variables: HashSet<Namespace>,
     pub response_variables: HashSet<Namespace>,
+
+    /// The request headers referenced in the connectors request mapping
+    pub request_headers: HashSet<String>,
+    /// The request or response headers referenced in the connectors response mapping
+    pub response_headers: HashSet<String>,
 }
 
 pub type CustomConfiguration = Arc<HashMap<String, Value>>;
@@ -159,7 +164,9 @@ impl Connector {
         };
 
         let request_variables = transport.variables().collect();
+        let request_headers = transport.header_references().collect();
         let response_variables = connect.selection.external_variables().collect();
+        let response_headers = connect.selection.header_references().collect();
 
         let connector = Connector {
             id: id.clone(),
@@ -171,6 +178,8 @@ impl Connector {
             spec,
             request_variables,
             response_variables,
+            request_headers,
+            response_headers,
         };
 
         Ok((id, connector))
@@ -320,6 +329,29 @@ impl HttpJsonTransport {
                     .flat_map(PathSelection::variable_reference)
             })
     }
+
+    /// Get any headers referenced in the variable references by looking at both Request and Response namespaces.
+    fn header_references(&self) -> impl Iterator<Item = String> + '_ {
+        self.variable_references().filter_map(|var_ref| {
+            if var_ref.namespace.namespace != Namespace::Request
+                && var_ref.namespace.namespace != Namespace::Response
+            {
+                return None;
+            }
+
+            // We only care if the path references starts with "headers"
+            if !var_ref
+                .path
+                .first()
+                .map_or(false, |path| path.part == "headers")
+            {
+                return None;
+            }
+
+            // Grab the name of the header from the path
+            var_ref.path.get(1).map(|path| path.part.to_string())
+        })
+    }
 }
 
 /// The HTTP arguments needed for a connect request
@@ -583,7 +615,7 @@ mod tests {
         let connectors =
             Connector::from_schema(subgraph.schema.schema(), "connectors", ConnectSpec::V0_1)
                 .unwrap();
-        assert_debug_snapshot!(&connectors, @r###"
+        assert_debug_snapshot!(&connectors, @r#"
         {
             ConnectId {
                 label: "connectors.json http: GET /users",
@@ -704,6 +736,8 @@ mod tests {
                 spec: V0_1,
                 request_variables: {},
                 response_variables: {},
+                request_headers: {},
+                response_headers: {},
             },
             ConnectId {
                 label: "connectors.json http: GET /posts",
@@ -836,8 +870,10 @@ mod tests {
                 spec: V0_1,
                 request_variables: {},
                 response_variables: {},
+                request_headers: {},
+                response_headers: {},
             },
         }
-        "###);
+        "#);
     }
 }
diff --git a/apollo-federation/src/sources/connect/validation/expression.rs b/apollo-federation/src/sources/connect/validation/expression.rs
index 8b4c399476..da3f36f1b4 100644
--- a/apollo-federation/src/sources/connect/validation/expression.rs
+++ b/apollo-federation/src/sources/connect/validation/expression.rs
@@ -53,6 +53,7 @@ impl<'schema> Context<'schema> {
             ),
             (Namespace::Config, Shape::unknown()),
             (Namespace::Context, Shape::unknown()),
+            (Namespace::Request, Shape::unknown()),
         ]
         .into_iter()
         .collect();
@@ -299,6 +300,46 @@ mod tests {
         assert!(validate(&expression(selection), &context).is_err());
     }
 
+    #[rstest]
+    #[case("$args.int")]
+    #[case("$config.abc")]
+    #[case("$context.def")]
+    #[case("$request.headers.'apollo-client-name'")]
+    fn valid_variables(#[case] selection: &str) {
+        let schema = Schema::parse(SCHEMA, "schema").unwrap();
+        let connect = name!("connect");
+        let source = name!("source");
+        let schema_info = SchemaInfo::new(&schema, "", &connect, &source);
+        let object = schema.get_object("Query").unwrap();
+        let field = object.fields.get("aField").unwrap();
+        let directive = field.directives.get("connect").unwrap();
+        let coordinate = ConnectDirectiveCoordinate {
+            field_coordinate: FieldCoordinate { field, object },
+            directive,
+        };
+        let context = Context::for_connect_request(&schema_info, coordinate);
+        validate(&expression(selection), &context).unwrap();
+    }
+
+    #[rstest]
+    #[case("$status")]
+    #[case("$response.headers.etag")]
+    fn invalid_variables(#[case] selection: &str) {
+        let schema = Schema::parse(SCHEMA, "schema").unwrap();
+        let connect = name!("connect");
+        let source = name!("source");
+        let schema_info = SchemaInfo::new(&schema, "", &connect, &source);
+        let object = schema.get_object("Query").unwrap();
+        let field = object.fields.get("aField").unwrap();
+        let directive = field.directives.get("connect").unwrap();
+        let coordinate = ConnectDirectiveCoordinate {
+            field_coordinate: FieldCoordinate { field, object },
+            directive,
+        };
+        let context = Context::for_connect_request(&schema_info, coordinate);
+        assert!(validate(&expression(selection), &context).is_err());
+    }
+
     #[rstest]
     #[case("$args.int")]
     #[case("$args.string")]
diff --git a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@headers__invalid_namespace_in_header_variables.graphql.snap b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@headers__invalid_namespace_in_header_variables.graphql.snap
index 8e0e20d6b6..60102c5e3e 100644
--- a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@headers__invalid_namespace_in_header_variables.graphql.snap
+++ b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@headers__invalid_namespace_in_header_variables.graphql.snap
@@ -27,21 +27,21 @@ input_file: apollo-federation/src/sources/connect/validation/test_data/headers/i
     },
     Message {
         code: InvalidHeader,
-        message: "In `@connect(http.headers:)` on `Query.scalar`: unknown variable `$foo`, must be one of $args, $config, $context",
+        message: "In `@connect(http.headers:)` on `Query.scalar`: unknown variable `$foo`, must be one of $args, $config, $context, $request",
         locations: [
             24:49..24:57,
         ],
     },
     Message {
         code: InvalidHeader,
-        message: "In `@connect(http.headers:)` on `Query.scalar`: $status is not valid here, must be one of $args, $config, $context",
+        message: "In `@connect(http.headers:)` on `Query.scalar`: $status is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             25:62..25:69,
         ],
     },
     Message {
         code: InvalidHeader,
-        message: "In `@connect(http.headers:)` on `Query.scalar`: $this is not valid here, must be one of $args, $config, $context",
+        message: "In `@connect(http.headers:)` on `Query.scalar`: $this is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             26:47..26:52,
         ],
diff --git a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@invalid_namespace_in_body_selection.graphql.snap b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@invalid_namespace_in_body_selection.graphql.snap
index cf30789ed7..4d05f7e196 100644
--- a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@invalid_namespace_in_body_selection.graphql.snap
+++ b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@invalid_namespace_in_body_selection.graphql.snap
@@ -1,12 +1,12 @@
 ---
 source: apollo-federation/src/sources/connect/validation/mod.rs
-expression: "format!(\"{:#?}\", errors)"
+expression: "format!(\"{:#?}\", result.errors)"
 input_file: apollo-federation/src/sources/connect/validation/test_data/invalid_namespace_in_body_selection.graphql
 ---
 [
     Message {
         code: InvalidJsonSelection,
-        message: "In `@connect(http: {body:})` on `Mutation.createUser`: variable `$status` is not valid at this location, must be one of $args, $config, $context, $this",
+        message: "In `@connect(http: {body:})` on `Mutation.createUser`: variable `$status` is not valid at this location, must be one of $args, $config, $context, $request, $this",
         locations: [
             21:17..21:24,
         ],
diff --git a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__invalid-path-parameter.graphql.snap b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__invalid-path-parameter.graphql.snap
index b3ad291d4c..4c5e1da162 100644
--- a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__invalid-path-parameter.graphql.snap
+++ b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__invalid-path-parameter.graphql.snap
@@ -6,7 +6,7 @@ input_file: apollo-federation/src/sources/connect/validation/test_data/uri_templ
 [
     Message {
         code: InvalidUrl,
-        message: "In `GET` in `@connect(http:)` on `Query.resources`: unknown variable `$blah`, must be one of $args, $config, $context",
+        message: "In `GET` in `@connect(http:)` on `Query.resources`: unknown variable `$blah`, must be one of $args, $config, $context, $request",
         locations: [
             12:23..12:28,
         ],
diff --git a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__invalid_namespace_in_url_template_variables.graphql.snap b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__invalid_namespace_in_url_template_variables.graphql.snap
index 092dc1a392..e790a59c24 100644
--- a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__invalid_namespace_in_url_template_variables.graphql.snap
+++ b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__invalid_namespace_in_url_template_variables.graphql.snap
@@ -6,14 +6,14 @@ input_file: apollo-federation/src/sources/connect/validation/test_data/uri_templ
 [
     Message {
         code: InvalidUrl,
-        message: "In `GET` in `@connect(http:)` on `Query.unknown`: unknown variable `$foo`, must be one of $args, $config, $context",
+        message: "In `GET` in `@connect(http:)` on `Query.unknown`: unknown variable `$foo`, must be one of $args, $config, $context, $request",
         locations: [
             11:31..11:39,
         ],
     },
     Message {
         code: InvalidUrl,
-        message: "In `GET` in `@connect(http:)` on `Query.invalid`: $status is not valid here, must be one of $args, $config, $context",
+        message: "In `GET` in `@connect(http:)` on `Query.invalid`: $status is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             18:31..18:42,
         ],
diff --git a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__this_on_root_types.graphql.snap b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__this_on_root_types.graphql.snap
index 27bfc422dd..1775e27221 100644
--- a/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__this_on_root_types.graphql.snap
+++ b/apollo-federation/src/sources/connect/validation/snapshots/validation_tests@uri_templates__this_on_root_types.graphql.snap
@@ -1,19 +1,19 @@
 ---
 source: apollo-federation/src/sources/connect/validation/mod.rs
-expression: "format!(\"{:#?}\", errors)"
+expression: "format!(\"{:#?}\", result.errors)"
 input_file: apollo-federation/src/sources/connect/validation/test_data/uri_templates/this_on_root_types.graphql
 ---
 [
     Message {
         code: InvalidUrl,
-        message: "In `GET` in `@connect(http:)` on `Query.requiresThis`: $this is not valid here, must be one of $args, $config, $context",
+        message: "In `GET` in `@connect(http:)` on `Query.requiresThis`: $this is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             11:39..11:51,
         ],
     },
     Message {
         code: InvalidUrl,
-        message: "In `GET` in `@connect(http:)` on `Mutation.requiresThis`: $this is not valid here, must be one of $args, $config, $context",
+        message: "In `GET` in `@connect(http:)` on `Mutation.requiresThis`: $this is not valid here, must be one of $args, $config, $context, $request",
         locations: [
             20:37..20:49,
         ],
diff --git a/apollo-federation/src/sources/connect/variable.rs b/apollo-federation/src/sources/connect/variable.rs
index 930c047ab8..b604018b88 100644
--- a/apollo-federation/src/sources/connect/variable.rs
+++ b/apollo-federation/src/sources/connect/variable.rs
@@ -51,6 +51,8 @@ impl<'schema> VariableContext<'schema> {
                     Namespace::Context,
                     Namespace::Status,
                     Namespace::This,
+                    Namespace::Request,
+                    Namespace::Response,
                 ]
             }
             Phase::Request => {
@@ -59,6 +61,7 @@ impl<'schema> VariableContext<'schema> {
                     Namespace::Context,
                     Namespace::This,
                     Namespace::Args,
+                    Namespace::Request,
                 ]
             }
         }
@@ -115,6 +118,8 @@ pub enum Namespace {
     Context,
     Status,
     This,
+    Request,
+    Response,
 }
 
 impl Namespace {
@@ -125,6 +130,8 @@ impl Namespace {
             Self::Context => "$context",
             Self::Status => "$status",
             Self::This => "$this",
+            Self::Request => "$request",
+            Self::Response => "$response",
         }
     }
 }
@@ -139,6 +146,8 @@ impl FromStr for Namespace {
             "$context" => Ok(Self::Context),
             "$status" => Ok(Self::Status),
             "$this" => Ok(Self::This),
+            "$request" => Ok(Self::Request),
+            "$response" => Ok(Self::Response),
             _ => Err(()),
         }
     }
diff --git a/apollo-router/src/plugins/connectors/handle_responses.rs b/apollo-router/src/plugins/connectors/handle_responses.rs
index caaa817353..c68b0b81f1 100644
--- a/apollo-router/src/plugins/connectors/handle_responses.rs
+++ b/apollo-router/src/plugins/connectors/handle_responses.rs
@@ -76,6 +76,7 @@ impl RawResponse {
         connector: Arc<Connector>,
         context: &Context,
         debug_context: &Option<Arc<Mutex<ConnectorContext>>>,
+        supergraph_request: Arc<http::Request<crate::graphql::Request>>,
     ) -> connector::request_service::Response {
         let mapped_response = match self {
             RawResponse::Error { error, key } => MappedResponse::Error { error, key },
@@ -87,9 +88,12 @@ impl RawResponse {
             } => {
                 let inputs = key.inputs().merge(
                     &connector.response_variables,
+                    &connector.response_headers,
                     connector.config.as_ref(),
                     context,
                     Some(parts.status.as_u16()),
+                    supergraph_request,
+                    Some(&parts),
                 );
 
                 let (res, apply_to_errors) = key.selection().apply_with_vars(&data, &inputs);
@@ -297,6 +301,7 @@ pub(crate) async fn process_response<T: HttpBody>(
     context: &Context,
     debug_request: Option<ConnectorDebugHttpRequest>,
     debug_context: &Option<Arc<Mutex<ConnectorContext>>>,
+    supergraph_request: Arc<http::Request<crate::graphql::Request>>,
 ) -> connector::request_service::Response {
     match result {
         // This occurs when we short-circuit the request when over the limit
@@ -346,7 +351,13 @@ pub(crate) async fn process_response<T: HttpBody>(
             };
             if is_success {
                 Span::current().record(OTEL_STATUS_CODE, OTEL_STATUS_CODE_OK);
-                raw.map_response(result, connector, context, debug_context)
+                raw.map_response(
+                    result,
+                    connector,
+                    context,
+                    debug_context,
+                    supergraph_request,
+                )
             } else {
                 Span::current().record(OTEL_STATUS_CODE, OTEL_STATUS_CODE_ERROR);
                 raw.map_error(result, connector, context, debug_context)
@@ -549,6 +560,7 @@ mod tests {
     use insta::assert_debug_snapshot;
     use url::Url;
 
+    use crate::graphql;
     use crate::plugins::connectors::handle_responses::process_response;
     use crate::plugins::connectors::make_requests::ResponseKey;
     use crate::services::router;
@@ -580,6 +592,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         });
 
         let response1: http::Response<RouterBody> = http::Response::builder()
@@ -600,6 +614,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response1),
@@ -608,6 +628,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -618,6 +639,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,
@@ -680,6 +702,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         });
 
         let response1: http::Response<RouterBody> = http::Response::builder()
@@ -700,6 +724,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response1),
@@ -708,6 +738,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -718,6 +749,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,
@@ -786,6 +818,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         });
 
         let response1: http::Response<RouterBody> = http::Response::builder()
@@ -810,6 +844,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response1),
@@ -818,6 +858,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -828,6 +869,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,
@@ -902,6 +944,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         });
 
         let response_plaintext: http::Response<RouterBody> = http::Response::builder()
@@ -942,6 +986,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response_plaintext),
@@ -950,6 +1000,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -960,6 +1011,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -970,6 +1022,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request.clone(),
             )
             .await
             .mapped_response,
@@ -980,6 +1033,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,
@@ -1156,6 +1210,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: selection.external_variables().collect(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         });
 
         let response1: http::Response<RouterBody> = http::Response::builder()
@@ -1168,6 +1224,12 @@ mod tests {
             selection: Arc::new(JSONSelection::parse("$status").unwrap()),
         };
 
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         let res = super::aggregate_responses(vec![
             process_response(
                 Ok(response1),
@@ -1176,6 +1238,7 @@ mod tests {
                 &Context::default(),
                 None,
                 &None,
+                supergraph_request,
             )
             .await
             .mapped_response,
diff --git a/apollo-router/src/plugins/connectors/make_requests.rs b/apollo-router/src/plugins/connectors/make_requests.rs
index f8e204f594..39590469ce 100644
--- a/apollo-router/src/plugins/connectors/make_requests.rs
+++ b/apollo-router/src/plugins/connectors/make_requests.rs
@@ -9,6 +9,7 @@ use apollo_federation::sources::connect::CustomConfiguration;
 use apollo_federation::sources::connect::EntityResolver;
 use apollo_federation::sources::connect::JSONSelection;
 use apollo_federation::sources::connect::Namespace;
+use http::response::Parts;
 use parking_lot::Mutex;
 use serde_json_bytes::json;
 use serde_json_bytes::ByteString;
@@ -38,12 +39,16 @@ impl RequestInputs {
     /// Creates a map for use in JSONSelection::apply_with_vars. It only clones
     /// values into the map if the variable namespaces (`$args`, `$this`, etc.)
     /// are actually referenced in the expressions for URLs, headers, body, or selection.
+    #[allow(clippy::too_many_arguments)]
     pub(crate) fn merge(
         &self,
         variables_used: &HashSet<Namespace>,
+        headers_used: &HashSet<String>,
         config: Option<&CustomConfiguration>,
         context: &Context,
         status: Option<u16>,
+        supergraph_request: Arc<http::Request<crate::graphql::Request>>,
+        response_parts: Option<&Parts>,
     ) -> IndexMap<String, Value> {
         let mut map = IndexMap::with_capacity_and_hasher(variables_used.len(), Default::default());
 
@@ -92,6 +97,54 @@ impl RequestInputs {
             }
         }
 
+        // Add headers from the original router request.
+        // Only include headers that are actually referenced to save on passing around unused headers in memory.
+        if variables_used.contains(&Namespace::Request) {
+            let headers: Map<ByteString, Value> = supergraph_request
+                .headers()
+                .iter()
+                .filter_map(|(key, value)| {
+                    if headers_used.contains(key.as_str()) {
+                        return Some((
+                            key.as_str().into(),
+                            value.to_str().unwrap_or_default().into(),
+                        ));
+                    }
+
+                    None
+                })
+                .collect();
+            let request_object = json!({
+                "headers": Value::Object(headers)
+            });
+            map.insert(Namespace::Request.as_str().into(), request_object);
+        }
+
+        // Add headers from the connectors response
+        // Only include headers that are actually referenced to save on passing around unused headers in memory.
+        if variables_used.contains(&Namespace::Response) {
+            if let Some(response_parts) = response_parts {
+                let headers: Map<ByteString, Value> = response_parts
+                    .headers
+                    .iter()
+                    .filter_map(|(key, value)| {
+                        if headers_used.contains(key.as_str()) {
+                            return Some((
+                                key.as_str().into(),
+                                value.to_str().unwrap_or_default().into(),
+                            ));
+                        }
+
+                        None
+                    })
+                    .collect();
+                let response_object = json!({
+                    "headers": Value::Object(headers)
+                });
+                map.insert(Namespace::Response.as_str().into(), response_object);
+            }
+        }
+
         map
     }
 }
@@ -187,7 +240,7 @@ pub(crate) fn make_requests(
         connector,
         service_name,
         request_params,
-        &request,
+        request,
         debug,
     )
 }
@@ -197,7 +250,7 @@ fn request_params_to_requests(
     connector: Arc<Connector>,
     service_name: &str,
     request_params: Vec<ResponseKey>,
-    original_request: &connect::Request,
+    original_request: connect::Request,
     debug: &Option<Arc<Mutex<ConnectorContext>>>,
 ) -> Result<Vec<Request>, MakeRequestError> {
     let mut results = vec![];
@@ -207,11 +260,14 @@ fn request_params_to_requests(
             &connector.transport,
             response_key.inputs().merge(
                 &connector.request_variables,
+                &connector.request_headers,
                 connector.config.as_ref(),
                 &original_request.context,
                 None,
+                original_request.supergraph_request.clone(),
+                None,
             ),
-            original_request,
+            &original_request,
             debug,
         )?;
 
@@ -222,6 +278,7 @@ fn request_params_to_requests(
             transport_request,
             key: response_key,
             mapping_problems,
+            supergraph_request: original_request.supergraph_request.clone(),
         });
     }
 
@@ -636,6 +693,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         assert_debug_snapshot!(super::root_fields(Arc::new(connector), &req), @r###"
@@ -767,6 +826,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         assert_debug_snapshot!(super::root_fields(Arc::new(connector), &req), @r###"
@@ -926,6 +987,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         assert_debug_snapshot!(super::root_fields(Arc::new(connector), &req), @r###"
@@ -1155,6 +1218,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         assert_debug_snapshot!(super::entities_from_request(Arc::new(connector), &req).unwrap(), @r###"
@@ -1473,6 +1538,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         assert_debug_snapshot!(super::entities_from_request(Arc::new(connector), &req).unwrap(), @r###"
@@ -1772,6 +1839,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         assert_debug_snapshot!(super::entities_from_request(Arc::new(connector), &req).unwrap(), @r###"
@@ -1991,6 +2060,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         assert_debug_snapshot!(super::entities_with_fields_from_request(Arc::new(connector), &req).unwrap(), @r###"
@@ -2269,6 +2340,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         assert_debug_snapshot!(super::entities_with_fields_from_request(Arc::new(connector), &req).unwrap(), @r###"
@@ -2544,6 +2617,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         assert_debug_snapshot!(super::entities_with_fields_from_request(Arc::new(connector), &req).unwrap(), @r###"
@@ -2684,6 +2759,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         let requests: Vec<_> = super::make_requests(
diff --git a/apollo-router/src/plugins/connectors/testdata/variables-subgraph.graphql b/apollo-router/src/plugins/connectors/testdata/variables-subgraph.graphql
index f0745e80bb..9cd3003a38 100644
--- a/apollo-router/src/plugins/connectors/testdata/variables-subgraph.graphql
+++ b/apollo-router/src/plugins/connectors/testdata/variables-subgraph.graphql
@@ -1,9 +1,6 @@
 extend schema
   @link(url: "https://specs.apollo.dev/federation/v2.10")
-  @link(
-    url: "https://specs.apollo.dev/connect/v0.1"
-    import: ["@connect", "@source"]
-  )
+  @link(url: "https://specs.apollo.dev/connect/v0.1", import: ["@connect", "@source"])
   @source(
     name: "v1"
     http: {
@@ -20,7 +17,7 @@ type Query {
     @connect(
       source: "v1"
       http: {
-        POST: "/f?arg={$args.arg->slice(1)}&context={$context.value}&config={$config.value}"
+        POST: "/f?arg={$args.arg->slice(1)}&context={$context.value}&config={$config.value}&header={$request.headers.value}"
         headers: [
           { name: "x-connect-context", value: "{$context.value}" }
           { name: "x-connect-config", value: "{$config.value}" }
@@ -30,6 +27,7 @@ type Query {
         arg: $args.arg
         context: $context.value
         config: $config.value
+        request: $request.headers.value
         """
       }
       selection: """
@@ -39,6 +37,8 @@ type Query {
       status: $status
       sibling: $("D")
       extra: $->echo({ arg: $args.arg, context: $context.value, config: $config.value, status: $status })
+      request: $request.headers.value
+      response: $response.headers.value
       """
     )
 }
@@ -50,6 +50,8 @@ type T {
   status: Int
   sibling: String
   extra: JSON
+  request: String
+  response: String
   f(arg: String): U
     @connect(
       source: "v1"
diff --git a/apollo-router/src/plugins/connectors/testdata/variables.graphql b/apollo-router/src/plugins/connectors/testdata/variables.graphql
index eba9f46791..80d7782bc2 100644
--- a/apollo-router/src/plugins/connectors/testdata/variables.graphql
+++ b/apollo-router/src/plugins/connectors/testdata/variables.graphql
@@ -61,7 +61,7 @@ enum link__Purpose {
 type Query
   @join__type(graph: CONNECTORS)
 {
-  f(arg: String!): T @join__directive(graphs: [CONNECTORS], name: "connect", args: {source: "v1", http: {POST: "/f?arg={$args.arg->slice(1)}&context={$context.value}&config={$config.value}", headers: [{name: "x-connect-context", value: "{$context.value}"}, {name: "x-connect-config", value: "{$config.value}"}, {name: "x-connect-arg", value: "{$args.arg->last}"}], body: "arg: $args.arg\ncontext: $context.value\nconfig: $config.value"}, selection: "arg: $args.arg\ncontext: $context.value\nconfig: $config.value\nstatus: $status\nsibling: $(\"D\")\nextra: $->echo({ arg: $args.arg, context: $context.value, config: $config.value, status: $status })"})
+  f(arg: String!): T @join__directive(graphs: [CONNECTORS], name: "connect", args: {source: "v1", http: {POST: "/f?arg={$args.arg->slice(1)}&context={$context.value}&config={$config.value}&header={$request.headers.value}", headers: [{name: "x-connect-context", value: "{$context.value}"}, {name: "x-connect-config", value: "{$config.value}"}, {name: "x-connect-arg", value: "{$args.arg->last}"}], body: "arg: $args.arg\ncontext: $context.value\nconfig: $config.value\nrequest: $request.headers.value"}, selection: "arg: $args.arg\ncontext: $context.value\nconfig: $config.value\nstatus: $status\nsibling: $(\"D\")\nextra: $->echo({ arg: $args.arg, context: $context.value, config: $config.value, status: $status })\nrequest: $request.headers.value\nresponse: $response.headers.value"})
 }
 
 type T
@@ -73,6 +73,8 @@ type T
   status: Int
   sibling: String
   extra: JSON
+  request: String
+  response: String
   f(arg: String): U @join__directive(graphs: [CONNECTORS], name: "connect", args: {source: "v1", http: {POST: "/f?arg={$args.arg->slice(2)}&context={$context.value}&config={$config.value}&sibling={$this.sibling}", headers: [{name: "x-connect-context", value: "{$context.value}"}, {name: "x-connect-config", value: "{$config.value}"}, {name: "x-connect-arg", value: "{$args.arg->first}"}, {name: "x-connect-sibling", value: "{$this.sibling}"}], body: "arg: $args.arg\ncontext: $context.value\nconfig: $config.value\nsibling: $this.sibling"}, selection: "arg: $args.arg\ncontext: $context.value\nconfig: $config.value\nsibling: $this.sibling\nstatus: $status"})
 }
 
diff --git a/apollo-router/src/plugins/connectors/tests/mod.rs b/apollo-router/src/plugins/connectors/tests/mod.rs
index a6fa81ed8c..fa45373a9d 100644
--- a/apollo-router/src/plugins/connectors/tests/mod.rs
+++ b/apollo-router/src/plugins/connectors/tests/mod.rs
@@ -1461,7 +1461,11 @@ async fn test_variables() {
         .await;
     Mock::given(method("POST"))
         .and(path("/f"))
-        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({})))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .set_body_json(serde_json::json!({}))
+                .insert_header("value", "myothercoolheader"),
+        )
         .mount(&mock_server)
         .await;
     let uri = mock_server.uri();
@@ -1469,7 +1473,7 @@ async fn test_variables() {
     let response = execute(
         &VARIABLES_SCHEMA.replace("http://localhost:4001/", &mock_server.uri()),
         &uri,
-        "{ f(arg: \"arg\") { arg context config sibling status extra f(arg: \"arg\") { arg context config sibling status } } }",
+        "{ f(arg: \"arg\") { arg context config sibling status extra request response f(arg: \"arg\") { arg context config sibling status } } }",
         Default::default(),
         Some(json!({
           "preview_connectors": {
@@ -1490,7 +1494,10 @@ async fn test_variables() {
             }
           }
         })),
-        |_| {},
+        |request| {
+          let headers = request.router_request.headers_mut();
+          headers.insert("value", "coolheader".parse().unwrap());
+        },
     )
     .await;
 
@@ -1509,6 +1516,8 @@ async fn test_variables() {
             "config": "C",
             "status": 200
           },
+          "request": "coolheader",
+          "response": "myothercoolheader",
           "f": {
             "arg": "arg",
             "context": "B",
@@ -1528,13 +1537,13 @@ async fn test_variables() {
             Matcher::new()
                 .method("POST")
                 .path("/f")
-                .query("arg=rg&context=B&config=C")
+                .query("arg=rg&context=B&config=C&header=coolheader")
                 .header("x-source-context".into(), "B".try_into().unwrap())
                 .header("x-source-config".into(), "C".try_into().unwrap())
                 .header("x-connect-arg".into(), "g".try_into().unwrap())
                 .header("x-connect-context".into(), "B".try_into().unwrap())
                 .header("x-connect-config".into(), "C".try_into().unwrap())
-                .body(serde_json::json!({ "arg": "arg", "context": "B", "config": "C" }))
+                .body(serde_json::json!({ "arg": "arg", "context": "B", "config": "C", "request": "coolheader" }))
                 ,
             Matcher::new()
                 .method("POST")
diff --git a/apollo-router/src/plugins/connectors/tracing.rs b/apollo-router/src/plugins/connectors/tracing.rs
index 3615c11cf4..13c73fba3c 100644
--- a/apollo-router/src/plugins/connectors/tracing.rs
+++ b/apollo-router/src/plugins/connectors/tracing.rs
@@ -87,6 +87,8 @@ mod tests {
             max_requests: None,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         };
 
         let connectors = Connectors {
diff --git a/apollo-router/src/plugins/telemetry/config_new/connector/selectors.rs b/apollo-router/src/plugins/telemetry/config_new/connector/selectors.rs
index 8116a66b3c..d4341e12d5 100644
--- a/apollo-router/src/plugins/telemetry/config_new/connector/selectors.rs
+++ b/apollo-router/src/plugins/telemetry/config_new/connector/selectors.rs
@@ -313,6 +313,7 @@ mod tests {
     use super::ConnectorSelector;
     use super::ConnectorSource;
     use super::MappingProblems;
+    use crate::graphql;
     use crate::plugins::connectors::handle_responses::MappedResponse;
     use crate::plugins::connectors::make_requests::ResponseKey;
     use crate::plugins::connectors::mapping::Problem;
@@ -362,6 +363,8 @@ mod tests {
             spec: ConnectSpec::V0_1,
             request_variables: Default::default(),
             response_variables: Default::default(),
+            request_headers: Default::default(),
+            response_headers: Default::default(),
         }
     }
 
@@ -394,6 +397,12 @@ mod tests {
         http_request: http::Request<RouterBody>,
         mapping_problems: Vec<Problem>,
     ) -> Request {
+        let supergraph_request = Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        );
+
         Request {
             context: context(),
             connector: Arc::new(connector()),
@@ -404,6 +413,7 @@ mod tests {
             }),
             key: response_key(),
             mapping_problems,
+            supergraph_request,
         }
     }
 
diff --git a/apollo-router/src/plugins/telemetry/config_new/events.rs b/apollo-router/src/plugins/telemetry/config_new/events.rs
index 7ea219bf2a..8a957cc7fa 100644
--- a/apollo-router/src/plugins/telemetry/config_new/events.rs
+++ b/apollo-router/src/plugins/telemetry/config_new/events.rs
@@ -1232,12 +1232,19 @@ mod tests {
                 spec: ConnectSpec::V0_1,
                 request_variables: Default::default(),
                 response_variables: Default::default(),
+                request_headers: Default::default(),
+                response_headers: Default::default(),
             };
             let response_key = ResponseKey::RootField {
                 name: "hello".to_string(),
                 inputs: Default::default(),
                 selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
             };
+            let supergraph_request = Arc::new(
+                http::Request::builder()
+                    .body(graphql::Request::builder().build())
+                    .unwrap(),
+            );
             let connector_request = Request {
                 context: context.clone(),
                 connector: Arc::new(connector.clone()),
@@ -1245,6 +1252,7 @@ mod tests {
                 transport_request,
                 key: response_key.clone(),
                 mapping_problems: vec![],
+                supergraph_request,
             };
             test_harness
                 .call_connector_request_service(connector_request, |request| Response {
@@ -1314,12 +1322,19 @@ mod tests {
                 spec: ConnectSpec::V0_1,
                 request_variables: Default::default(),
                 response_variables: Default::default(),
+                request_headers: Default::default(),
+                response_headers: Default::default(),
             };
             let response_key = ResponseKey::RootField {
                 name: "hello".to_string(),
                 inputs: Default::default(),
                 selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
             };
+            let supergraph_request = Arc::new(
+                http::Request::builder()
+                    .body(graphql::Request::builder().build())
+                    .unwrap(),
+            );
             let connector_request = Request {
                 context: context.clone(),
                 connector: Arc::new(connector.clone()),
@@ -1327,6 +1342,7 @@ mod tests {
                 transport_request,
                 key: response_key.clone(),
                 mapping_problems: vec![],
+                supergraph_request,
             };
             test_harness
                 .call_connector_request_service(connector_request, |request| Response {
diff --git a/apollo-router/src/plugins/telemetry/config_new/instruments.rs b/apollo-router/src/plugins/telemetry/config_new/instruments.rs
index c47d11ac87..1fadf91bd9 100644
--- a/apollo-router/src/plugins/telemetry/config_new/instruments.rs
+++ b/apollo-router/src/plugins/telemetry/config_new/instruments.rs
@@ -3276,6 +3276,8 @@ mod tests {
                                         spec: ConnectSpec::V0_1,
                                         request_variables: Default::default(),
                                         response_variables: Default::default(),
+                                        request_headers: Default::default(),
+                                        response_headers: Default::default(),
                                     };
                                     let response_key = ResponseKey::RootField {
                                         name: "hello".to_string(),
@@ -3284,6 +3286,11 @@ mod tests {
                                             JSONSelection::parse("$.data").unwrap(),
                                         ),
                                     };
+                                    let supergraph_request = Arc::new(
+                                        http::Request::builder()
+                                            .body(graphql::Request::builder().build())
+                                            .unwrap(),
+                                    );
                                     let request = Request {
                                         context: Context::default(),
                                         connector: Arc::new(connector),
@@ -3291,6 +3298,7 @@ mod tests {
                                         transport_request,
                                         key: response_key.clone(),
                                         mapping_problems,
+                                        supergraph_request,
                                     };
                                     connector_instruments = Some({
                                         let connector_instruments = config
@@ -3338,6 +3346,8 @@ mod tests {
                                         spec: ConnectSpec::V0_1,
                                         request_variables: Default::default(),
                                         response_variables: Default::default(),
+                                        request_headers: Default::default(),
+                                        response_headers: Default::default(),
                                     };
                                     let response_key = ResponseKey::RootField {
                                         name: "hello".to_string(),
diff --git a/apollo-router/src/plugins/telemetry/fmt_layer.rs b/apollo-router/src/plugins/telemetry/fmt_layer.rs
index 1a091d2e4e..bf22506b37 100644
--- a/apollo-router/src/plugins/telemetry/fmt_layer.rs
+++ b/apollo-router/src/plugins/telemetry/fmt_layer.rs
@@ -878,12 +878,19 @@ connector:
                     spec: ConnectSpec::V0_1,
                     request_variables: Default::default(),
                     response_variables: Default::default(),
+                    request_headers: Default::default(),
+                    response_headers: Default::default(),
                 });
                 let response_key = ResponseKey::RootField {
                     name: "hello".to_string(),
                     inputs: Default::default(),
                     selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
                 };
+                let supergraph_request = Arc::new(
+                    http::Request::builder()
+                        .body(graphql::Request::builder().build())
+                        .unwrap(),
+                );
                 let connector_request = Request {
                     context: context.clone(),
                     connector: connector.clone(),
@@ -907,6 +914,7 @@ connector:
                             path: "@.id".to_string(),
                         },
                     ],
+                    supergraph_request,
                 };
                 let connector_events = event_config.new_connector_events();
                 connector_events.on_request(&connector_request);
@@ -1126,12 +1134,19 @@ connector:
                     spec: ConnectSpec::V0_1,
                     request_variables: Default::default(),
                     response_variables: Default::default(),
+                    request_headers: Default::default(),
+                    response_headers: Default::default(),
                 });
                 let response_key = ResponseKey::RootField {
                     name: "hello".to_string(),
                     inputs: Default::default(),
                     selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
                 };
+                let supergraph_request = Arc::new(
+                    http::Request::builder()
+                        .body(graphql::Request::builder().build())
+                        .unwrap(),
+                );
                 let connector_request = Request {
                     context: context.clone(),
                     connector: connector.clone(),
@@ -1155,6 +1170,7 @@ connector:
                             path: "@.id".to_string(),
                         },
                     ],
+                    supergraph_request,
                 };
                 let connector_events = event_config.new_connector_events();
                 connector_events.on_request(&connector_request);
diff --git a/apollo-router/src/services/connect.rs b/apollo-router/src/services/connect.rs
index 40b1f990d1..4053d5122a 100644
--- a/apollo-router/src/services/connect.rs
+++ b/apollo-router/src/services/connect.rs
@@ -1,5 +1,6 @@
 //! Connect service request and response types.
 
+use std::fmt::Debug;
 use std::sync::Arc;
 
 use apollo_compiler::validation::Valid;
@@ -23,6 +24,18 @@ pub(crate) struct Request {
     pub(crate) variables: Variables,
 }
 
+impl Debug for Request {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Request")
+            .field("service_name", &self.service_name)
+            .field("context", &self.context)
+            .field("operation", &self.operation)
+            .field("supergraph_request", &self.supergraph_request)
+            .field("variables", &self.variables.variables)
+            .finish()
+    }
+}
+
 assert_impl_all!(Response: Send);
 #[derive(Debug)]
 #[non_exhaustive]
diff --git a/apollo-router/src/services/connector/request_service.rs b/apollo-router/src/services/connector/request_service.rs
index 65643b91f0..7beec5e57f 100644
--- a/apollo-router/src/services/connector/request_service.rs
+++ b/apollo-router/src/services/connector/request_service.rs
@@ -76,6 +76,9 @@ pub(crate) struct Request {
 
     /// Mapping problems encountered when creating the transport request
     pub(crate) mapping_problems: Vec<Problem>,
+
+    /// The original supergraph request from the supergraph service
+    pub(crate) supergraph_request: Arc<http::Request<crate::graphql::Request>>,
 }
 
 /// Response type for a connector
@@ -318,6 +321,7 @@ impl tower::Service<Request> for ConnectorRequestService {
                 &request.context,
                 debug_request,
                 &debug,
+                request.supergraph_request,
             )
             .await)
         })