diff --git a/demo/use_case/graphql_server.py b/demo/use_case/graphql_server.py
index efc712d..28a3323 100644
--- a/demo/use_case/graphql_server.py
+++ b/demo/use_case/graphql_server.py
@@ -8,7 +8,7 @@
 Routing inside the POST handler:
 - ``is_introspection_query(query)`` → ``compose_introspect(schema, query)``
   (services ``__schema`` / ``__type`` / ``__typename`` so GraphiQL can boot)
-- otherwise → ``execute_compose_query(app, schema, query, context)``
+- otherwise → ``execute_compose_query(app, schema, query, context, variables)``
   (real data fetch; ``__schema`` etc. are rejected here per spec FR-008,
   redirecting clients to the schema-discovery MCP layers when relevant)
 
@@ -65,24 +65,6 @@ async def graphql_endpoint(request: Request) -> JSONResponse:
     query: str = body.get("query", "")
     variables: dict[str, Any] | None = body.get("variables")
     operation_name: str | None = body.get("operationName")
-
-    # Variables aren't currently supported by execute_compose_query — inline
-    # them if you need parametrized queries. Fail loudly so callers notice.
-    if variables:
-        return JSONResponse(
-            status_code=400,
-            content={
-                "data": None,
-                "errors": [
-                    {
-                        "message": (
-                            "Variables are not yet supported by the compose "
-                            "executor; inline arguments in the query string."
-                        )
-                    }
-                ],
-            },
-        )
     if operation_name:
         # operationName selection is silently accepted (single-op queries
         # work without it); multi-op queries aren't currently supported.
@@ -96,6 +78,7 @@ async def graphql_endpoint(request: Request) -> JSONResponse:
         schema=SCHEMA,
         query=query,
         context={},  # no FromContext params in this demo
+        variables=variables,
     )
     # ``execute_compose_query`` returns Pydantic subset-model instances inside
     # the ``data`` tree; JSONResponse can't serialize them directly. Run them
diff --git a/docs/changelog.md b/docs/changelog.md
index 52ce649..cb3de31 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -10,6 +10,28 @@ description: "Release-by-release changelog for nexusx, following semver — majo
 
 > Pre-3.0 history is not included here. See `git log` and the historical tags for changes before 3.0.0.
 
+## 6.3
+
+### 6.3.0 (2026-9-9)
+
+- feat:
+  - **`compose_query` accepts GraphQL variables**: `execute_compose_query`
+    and the MCP Layer 3 `compose_query` tool now take a `variables` dict;
+    `$var` references in arguments resolve to their values through the
+    existing `QueryParser` path (the same one `GraphQLHandler` uses).
+    Pass string arguments this way instead of inlining GraphQL literals —
+    inline strings containing quotes, backslashes or newlines are the #1
+    source of agent-authored parse errors, and variables sidestep escaping
+    entirely. A query that declares variables fails fast with a clear
+    message naming the missing ones, instead of dying later in argument
+    coercion with a cryptic error. Variable default values
+    (`$t: String = "x"`) are not applied — they never were on any
+    execution path; an omitted defaulted variable now also fails fast,
+    with the error naming the limitation (the note appears only when the
+    query actually declares a default), instead of silently becoming
+    `Undefined`. Backward compatible: `variables` is optional and
+    inline-literal queries are unaffected.
+
 ## 6.2
 
 ### 6.2.1 (2026-9-3)
diff --git a/src/nexusx/use_case/compose_executor.py b/src/nexusx/use_case/compose_executor.py
index bcffd09..e3b7f5f 100644
--- a/src/nexusx/use_case/compose_executor.py
+++ b/src/nexusx/use_case/compose_executor.py
@@ -50,6 +50,16 @@
     "to discover the schema."
 )
 
+# Appended to variables-contract errors when the query declares a variable
+# default ($t: String = "x"). GraphQL spec would apply the default; nexusx's
+# parser resolves variables purely from the provided dict (value_from_ast_
+# untyped never reads default_value), so defaulted variables are required too
+# — the error must say so instead of contradicting the query.
+_VARIABLE_DEFAULTS_NOTE = (
+    " Variable default values are not supported; every declared variable "
+    "must be passed explicitly."
+)
+
 
 # ---------------------------------------------------------------------------
 # Public API
@@ -74,6 +84,7 @@ async def execute_compose_query(
     schema: ComposeSchema,
     query: str,
     context: dict[str, Any] | None = None,
+    variables: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Execute a UseCase compose query, returning graphql-standard ``{data, errors}``.
 
@@ -86,6 +97,12 @@ async def execute_compose_query(
         schema: The ``ComposeSchema`` derived from ``app``.
         query: Standard GraphQL query string.
         context: ``FromContext`` parameter values, keyed by parameter name.
+        variables: Values for ``$variables`` declared by the query. Pass string
+            arguments this way instead of inlining them as GraphQL literals —
+            inline strings containing quotes, backslashes or newlines are the
+            #1 source of agent-authored parse errors. Every declared variable
+            must be provided: variable default values (``$t: String = "x"``)
+            are not applied.
 
     Returns:
         ``{"data": <nested service→method→result>, "errors": []}`` on success;
@@ -98,6 +115,39 @@ async def execute_compose_query(
     except Exception as exc:  # noqa: BLE001 — graphql parse errors vary in shape
         return _error_response(f"Failed to parse query: {exc}")
 
+    # 1.5 Variables contract check — friendly failure before any execution.
+    #     Without it, a missing variable silently resolves to graphql's
+    #     Undefined and dies later inside argument coercion with a cryptic
+    #     message. Variables belong to the (single) operation's definitions.
+    #     Declared defaults ($t: String = "x") count as declared variables:
+    #     they are never auto-applied (see _VARIABLE_DEFAULTS_NOTE), so the
+    #     error names the limitation only when the query uses one — a plain
+    #     $t: String! omission keeps a to-the-point message.
+    defined_vars: list[str] = []
+    defaulted_vars: set[str] = set()
+    for definition in document.definitions:
+        if isinstance(definition, OperationDefinitionNode):
+            for vd in definition.variable_definitions or []:
+                defined_vars.append(vd.variable.name.value)
+                if vd.default_value is not None:
+                    defaulted_vars.add(vd.variable.name.value)
+            break
+    if defined_vars and variables is None:
+        message = (
+            f"Query declares variables {defined_vars} but none were provided — "
+            "pass them via the 'variables' argument (recommended for any "
+            "string containing quotes, backslashes or newlines)."
+        )
+        if defaulted_vars:
+            message += _VARIABLE_DEFAULTS_NOTE
+        return _error_response(message)
+    missing_vars = [name for name in defined_vars if name not in (variables or {})]
+    if missing_vars:
+        message = f"Missing variables: {missing_vars}."
+        if set(missing_vars) & defaulted_vars:
+            message += _VARIABLE_DEFAULTS_NOTE
+        return _error_response(message)
+
     # 2. Reject introspection (FR-008) before any service call.
     if _document_uses_introspection(document):
         return _error_response(_INTROSPECTION_REJECTION_HINT)
@@ -109,9 +159,11 @@ async def execute_compose_query(
     #    contaminated same-name groups across operations). compose_query
     #    takes a bare query string with no operationName channel, so the
     #    document must contain exactly one operation.
+    #    Variables resolve to their values during argument extraction
+    #    (QueryParser.parse_operations forwards them).
     parser = QueryParser()
     try:
-        operations = parser.parse_operations(document)
+        operations = parser.parse_operations(document, variables)
     except ValueError as exc:
         return _error_response(str(exc), code="ALIAS_CONFLICT")
     if not operations:
diff --git a/src/nexusx/use_case/compose_mcp_server.py b/src/nexusx/use_case/compose_mcp_server.py
index 1486690..8dc19d4 100644
--- a/src/nexusx/use_case/compose_mcp_server.py
+++ b/src/nexusx/use_case/compose_mcp_server.py
@@ -289,6 +289,7 @@ def _register_compose_query(
     async def compose_query(
         app_name: str,
         query: str,
+        variables: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         """Execute a GraphQL query against an app's UseCase compose schema.
 
@@ -297,6 +298,19 @@ async def compose_query(
         Introspection queries (``__schema``, ``__type``, ``__typename``) are
         rejected — use ``describe_compose_schema`` and
         ``describe_compose_method`` for schema discovery.
+
+        Pass string arguments via ``variables`` — never inline them as GraphQL
+        string literals. Inline strings containing quotes (``"``), backslashes
+        or newlines produce unparseable queries; ``variables`` sidesteps all
+        escaping::
+
+            query: 'mutation($t: String!) { TaskService { create_task(title: $t) { id } } }'
+            variables: {"t": 'He said "hi" \\ done'}
+
+        A query that declares ``$variables`` fails fast with a clear error if
+        the values are missing. Variable default values
+        (``$t: String = "x"``) are not applied — pass every declared variable
+        explicitly.
         """
         entry = _get_app(registry, app_name)
         if entry is None:
@@ -334,4 +348,5 @@ async def compose_query(
             schema=entry.schema,
             query=query,
             context=context,
+            variables=variables,
         )
diff --git a/tests/mcp/test_application.py b/tests/mcp/test_application.py
index 9571859..972891c 100644
--- a/tests/mcp/test_application.py
+++ b/tests/mcp/test_application.py
@@ -12,8 +12,6 @@
 
 from __future__ import annotations
 
-import warnings
-
 import pytest
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlmodel import Field, SQLModel
diff --git a/tests/test_compose_mcp_server.py b/tests/test_compose_mcp_server.py
index 6cc6f80..30163c0 100644
--- a/tests/test_compose_mcp_server.py
+++ b/tests/test_compose_mcp_server.py
@@ -297,6 +297,88 @@ async def test_wrapper_field_is_rejected(self, mcp_server) -> None:
         assert data["data"] is None
         assert "Service 'Op' not found" in data["errors"][0]["message"]
 
+    async def test_variables_pass_string_with_quotes(self, mcp_server) -> None:
+        """Regression: strings with quotes/backslashes/newlines via variables.
+
+        Inline GraphQL literals with such characters are the #1 source of
+        agent-authored parse errors; variables sidestep all escaping.
+        """
+        nasty = 'He said "hi" \\ done\n(tabs\ttoo)'
+        data = await _call(
+            mcp_server,
+            "compose_query",
+            {
+                "app_name": "project",
+                "query": (
+                    "mutation($t: String!) { TaskService { create_task(title: $t) { id title } } }"
+                ),
+                "variables": {"t": nasty},
+            },
+        )
+        assert data["errors"] == []
+        assert data["data"]["TaskService"]["create_task"]["title"] == nasty
+
+    async def test_query_with_variables_but_none_provided(self, mcp_server) -> None:
+        data = await _call(
+            mcp_server,
+            "compose_query",
+            {
+                "app_name": "project",
+                "query": "mutation($t: String!) { TaskService { create_task(title: $t) { id } } }",
+            },
+        )
+        assert data["data"] is None
+        msg = data["errors"][0]["message"]
+        assert "declares variables ['t']" in msg
+        assert "'variables'" in msg  # 指路：用 variables 参数传值
+
+    async def test_partial_variables_report_missing_names(self, mcp_server) -> None:
+        data = await _call(
+            mcp_server,
+            "compose_query",
+            {
+                "app_name": "project",
+                "query": (
+                    "mutation($a: String!, $b: String!) "
+                    "{ TaskService { create_task(title: $a) { id } } }"
+                ),
+                "variables": {"a": "x"},
+            },
+        )
+        assert data["data"] is None
+        assert "Missing variables: ['b']" in data["errors"][0]["message"]
+
+    async def test_variable_default_value_names_the_limitation(self, mcp_server) -> None:
+        """Declared defaults ($t: String = "x") are required too — but the
+        error must say defaults aren't supported, not contradict the query
+        (GraphQL spec would silently apply the default; the parser never did).
+        """
+        data = await _call(
+            mcp_server,
+            "compose_query",
+            {
+                "app_name": "project",
+                "query": (
+                    'mutation($t: String = "fallback") '
+                    "{ TaskService { create_task(title: $t) { id } } }"
+                ),
+            },
+        )
+        assert data["data"] is None
+        msg = data["errors"][0]["message"]
+        assert "declares variables ['t']" in msg
+        assert "default values are not supported" in msg
+        # 反例：纯 $t: String! 漏传时不附加默认值说明（信息保持切题）
+        plain = await _call(
+            mcp_server,
+            "compose_query",
+            {
+                "app_name": "project",
+                "query": "mutation($t: String!) { TaskService { create_task(title: $t) { id } } }",
+            },
+        )
+        assert "default values" not in plain["errors"][0]["message"]
+
 
 # ──────────────────────────────────────────────────────────────────────
 # FromContext plumbing through Layer 3
diff --git a/tests/test_composed_federation.py b/tests/test_composed_federation.py
index 6961dcf..c8afcf5 100644
--- a/tests/test_composed_federation.py
+++ b/tests/test_composed_federation.py
@@ -13,7 +13,6 @@
 """
 
 import httpx
-import pytest
 import pytest_asyncio
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlalchemy.pool import StaticPool
@@ -22,14 +21,12 @@
 from starlette.applications import Starlette
 from starlette.routing import Mount
 
-from nexusx import AutoQueryConfig, ComposedErManager, DefineSubset, ErManager
-from nexusx import GraphQLHandler
+from nexusx import AutoQueryConfig, ComposedErManager, DefineSubset, ErManager, GraphQLHandler
 from nexusx import Relationship as NxRelationship
 from nexusx.federation import RemoteRelationship, RemoteService
 from nexusx.federation.http import GraphQLTransport
 from nexusx.federation.introspect import build_federable_app
 
-
 # ── RemoteService 声明（cfreviews = composed-federation reviews）──
 cfreviews = RemoteService("cfreviews", url="http://test/cfreviews")
 
diff --git a/tests/test_composed_handler.py b/tests/test_composed_handler.py
index 8269717..6c32f9a 100644
--- a/tests/test_composed_handler.py
+++ b/tests/test_composed_handler.py
@@ -20,7 +20,6 @@
 from nexusx.loader import LoaderRegistry
 from nexusx.mcp import Application
 
-
 # ── 实体（Ch 前缀，带静态 @query，不查 db）──
 
 class ChUser(SQLModel, table=True):
diff --git a/tests/test_composed_voyager.py b/tests/test_composed_voyager.py
index 4ffef6b..e54d408 100644
--- a/tests/test_composed_voyager.py
+++ b/tests/test_composed_voyager.py
@@ -351,10 +351,10 @@ def test_remote_type_and_members_keep_separate_clusters(self):
 
 # ── US3 — UseCase page clusters registered DTOs by member (FR-005) ─────
 
+from nexusx import query  # noqa: E402
 from nexusx.use_case.business import UseCaseService  # noqa: E402
 from nexusx.voyager.use_case_voyager import UseCaseVoyager  # noqa: E402
 from nexusx.voyager.voyager_context import VoyagerContext  # noqa: E402
-from nexusx import query  # noqa: E402
 
 
 class CvSummary(BaseModel):
diff --git a/tests/test_federation_remote_loader.py b/tests/test_federation_remote_loader.py
index 8a49103..511a4a4 100644
--- a/tests/test_federation_remote_loader.py
+++ b/tests/test_federation_remote_loader.py
@@ -301,11 +301,12 @@ def test_fetch_passes_params_key_for_isolation(self):
         limit=5 / limit=10 loads (defect 1)."""
         import asyncio
 
+        from pydantic import BaseModel
+
         from nexusx.federation.remote_loader import fetch_remote_subtree
         from nexusx.loader.pagination import Paged
         from nexusx.loader.registry import ErManager
         from nexusx.query_parser import FieldSelection
-        from pydantic import BaseModel
 
         class Target(BaseModel):
             id: int
@@ -378,9 +379,10 @@ class L:
     def test_paged_selection_alone_yields_none_type_key(self):
         """The root cause, locked: the {items, pagination} wrapper is not a
         target-entity field set, so the selection fingerprint is None."""
+        from pydantic import BaseModel
+
         from nexusx.loader.query_meta import generate_type_key_from_selection
         from nexusx.query_parser import FieldSelection
-        from pydantic import BaseModel
 
         class Target(BaseModel):
             id: int
@@ -419,6 +421,8 @@ def test_clamp_caps_limit(self):
     async def test_fetch_clamps_to_rel_max_page_size(self):
         """A β fetch with limit=100000 against a relationship whose
         max_page_size is the default 100 must send limit: 100 on the wire."""
+        from pydantic import BaseModel
+
         from nexusx.federation.remote_loader import (
             create_paginated_remote_loader,
             paged_from_selection,
@@ -426,7 +430,6 @@ async def test_fetch_clamps_to_rel_max_page_size(self):
             set_remote_selection,
         )
         from nexusx.query_parser import FieldSelection
-        from pydantic import BaseModel
 
         class Target(BaseModel):
             id: int
diff --git a/tests/test_fetch_primitive_symmetry.py b/tests/test_fetch_primitive_symmetry.py
index 8b17ad1..d7336df 100644
--- a/tests/test_fetch_primitive_symmetry.py
+++ b/tests/test_fetch_primitive_symmetry.py
@@ -19,8 +19,8 @@
 from pathlib import Path
 
 from nexusx.federation.remote_loader import (
-    prepare_dto_loader,
     fetch_remote_subtree,
+    prepare_dto_loader,
 )
 
 # fetch_remote_subtree.__module__ is the string "nexusx.federation.remote_loader"
diff --git a/tests/test_mcp_pagination_guard.py b/tests/test_mcp_pagination_guard.py
index 2616884..6018447 100644
--- a/tests/test_mcp_pagination_guard.py
+++ b/tests/test_mcp_pagination_guard.py
@@ -21,7 +21,6 @@
 
 from nexusx import AutoQueryConfig, GraphQLHandler, query
 from nexusx.mcp import Application, create_multi_app_mcp_server, create_single_app_mcp_server
-from nexusx.mcp.application import _coerce_to_application
 from nexusx.mcp.managers.single_app_manager import SingleAppManager
 
 try:
diff --git a/tests/test_pagination_items_regression.py b/tests/test_pagination_items_regression.py
index c03a8ca..79deb65 100644
--- a/tests/test_pagination_items_regression.py
+++ b/tests/test_pagination_items_regression.py
@@ -6,7 +6,6 @@
 读 RelationshipInfo，运行时 isinstance(PaginatedPackage)——关系命名自由。
 """
 
-import pytest
 import pytest_asyncio
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlalchemy.pool import StaticPool
diff --git a/tests/test_query_parser.py b/tests/test_query_parser.py
index c545d51..f1d7ef1 100644
--- a/tests/test_query_parser.py
+++ b/tests/test_query_parser.py
@@ -265,7 +265,7 @@ class TestParseOperations:
 
     def test_same_name_group_in_different_operations_coexist(self):
         from graphql import parse
-        from nexusx.query_parser import ResponseKeyConflictError
+
 
         ops = QueryParser().parse_operations(
             parse("mutation M { S { f { id } } } query Q { S { g { id } } }")
@@ -282,6 +282,7 @@ def test_same_name_group_in_different_operations_coexist(self):
 
     def test_duplicate_within_one_operation_still_conflicts(self):
         from graphql import parse
+
         from nexusx.query_parser import ResponseKeyConflictError
 
         with pytest.raises(ResponseKeyConflictError, match="conflict"):