@@ -49,11 +49,13 @@ def __init__(self, client) -> None:
     def create(
         self,
         *,
-        agent: str,
+        agent: Any,
         environment_id: Optional[str] = None,
         title: Optional[str] = None,
         resources: Optional[Sequence[Mapping[str, Any]]] = None,
         vault_ids: Optional[Sequence[str]] = None,
+        environment_variables: Optional[Mapping[str, str]] = None,
+        mcp_configs: Optional[Sequence[Mapping[str, Any]]] = None,
         metadata: Optional[Mapping[str, Any]] = None,
     ) -> Session:
         body = SessionCreateParams(
@@ -62,6 +64,8 @@ def create(
             title=title,
             resources=resources,
             vault_ids=vault_ids,
+            environment_variables=environment_variables,
+            mcp_configs=mcp_configs,
             metadata=metadata,
         ).to_dict()
         resp = self._client.transport.request(
@@ -90,10 +94,18 @@ def update(
         *,
         title: Optional[str] = None,
         metadata: Optional[Mapping[str, Any]] = None,
+        vault_ids: Optional[Sequence[str]] = None,
+        environment_variables: Optional[Mapping[str, str]] = None,
+        agent: Any = None,
+        mcp_configs: Optional[Sequence[Mapping[str, Any]]] = None,
     ) -> Session:
         body = SessionUpdateParams(
             title=title,
             metadata=metadata,
+            vault_ids=vault_ids,
+            environment_variables=environment_variables,
+            agent=agent,
+            mcp_configs=mcp_configs,
         ).to_dict()
         resp = self._client.transport.request(
             "POST",
@@ -176,11 +188,13 @@ def __init__(self, client) -> None:
     async def create(
         self,
         *,
-        agent: str,
+        agent: Any,
         environment_id: Optional[str] = None,
         title: Optional[str] = None,
         resources: Optional[Sequence[Mapping[str, Any]]] = None,
         vault_ids: Optional[Sequence[str]] = None,
+        environment_variables: Optional[Mapping[str, str]] = None,
+        mcp_configs: Optional[Sequence[Mapping[str, Any]]] = None,
         metadata: Optional[Mapping[str, Any]] = None,
     ) -> Session:
         body = SessionCreateParams(
@@ -189,6 +203,8 @@ async def create(
             title=title,
             resources=resources,
             vault_ids=vault_ids,
+            environment_variables=environment_variables,
+            mcp_configs=mcp_configs,
             metadata=metadata,
         ).to_dict()
         resp = await self._client.transport.request(
@@ -217,10 +233,18 @@ async def update(
         *,
         title: Optional[str] = None,
         metadata: Optional[Mapping[str, Any]] = None,
+        vault_ids: Optional[Sequence[str]] = None,
+        environment_variables: Optional[Mapping[str, str]] = None,
+        agent: Any = None,
+        mcp_configs: Optional[Sequence[Mapping[str, Any]]] = None,
     ) -> Session:
         body = SessionUpdateParams(
             title=title,
             metadata=metadata,
+            vault_ids=vault_ids,
+            environment_variables=environment_variables,
+            agent=agent,
+            mcp_configs=mcp_configs,
         ).to_dict()
         resp = await self._client.transport.request(
             "POST",
 
@@ -558,6 +558,8 @@ class Session(BaseModel):
         "stats",
         "usage",
         "vault_ids",
+        "environment_variables",
+        "mcp_configs",
         "archived_at",
         "created_at",
         "updated_at",
 
@@ -259,12 +259,14 @@ def __init__(  # pylint: disable=useless-parent-delegation
 class SessionCreateParams(BaseModel):
     """Request body for ``POST /sessions``.
 
-    ``agent`` is the agent ID string (not the full agent object).
-    ``resources`` is an optional list of file mounts; each item is a
-    mapping with ``type``, ``file_id`` and ``mount_path`` keys.
-    ``vault_ids`` is create-only — attach vaults (``vlt_*``) whose
-    credentials are substituted at egress; the session update path does
-    not accept it.
+    ``agent`` accepts the Agent ID string (backward compatible) **or** an
+    Agent reference/override object (a mapping with
+    ``type="agent_with_overrides"``) that establishes Session-level
+    overrides for ``system``/``tools``/``mcp_servers``/``skills``.
+
+    ``environment_variables`` and ``mcp_configs`` are Session-level
+    settings independent of the agent. ``resources`` is create-only — it
+    cannot be modified via ``updateSession``.
     """
 
     _fields = (
@@ -273,45 +275,92 @@ class SessionCreateParams(BaseModel):
         "title",
         "resources",
         "vault_ids",
+        "environment_variables",
+        "mcp_configs",
         "metadata",
     )
 
     def __init__(  # pylint: disable=useless-parent-delegation
         self,
         *,
-        agent: str,
+        agent: Any,
         environment_id: Optional[str] = None,
         title: Optional[str] = None,
         resources: Optional[Sequence[Mapping[str, Any]]] = None,
         vault_ids: Optional[Sequence[str]] = None,
+        environment_variables: Optional[Mapping[str, str]] = None,
+        mcp_configs: Optional[Sequence[Mapping[str, Any]]] = None,
         metadata: Optional[Mapping[str, Any]] = None,
     ) -> None:
         super().__init__(
-            agent=agent,
+            agent=_to_mapping(agent),
             environment_id=environment_id,
             title=title,
             resources=(
                 [dict(r) for r in resources] if resources is not None else None
             ),
             vault_ids=(list(vault_ids) if vault_ids is not None else None),
+            environment_variables=(
+                dict(environment_variables)
+                if environment_variables is not None
+                else None
+            ),
+            mcp_configs=(
+                [_to_mapping(m) for m in mcp_configs]
+                if mcp_configs is not None
+                else None
+            ),
             metadata=(dict(metadata) if metadata is not None else None),
         )
 
 
 class SessionUpdateParams(BaseModel):
-    """Request body for ``POST /sessions/{id}``."""
+    """Request body for ``POST /sessions/{id}``.
+
+    Omitted (``None``) fields are not sent — the existing value is kept;
+    ``[]`` / ``{}`` explicitly clear the value. ``agent`` is a Session-level
+    Agent config patch (a plain mapping carrying only ``system``/``tools``/
+    ``mcp_servers``/``skills``); the mapping is passed through verbatim so
+    an explicit ``None`` sub-field serializes to JSON ``null`` (delete the
+    override and re-inherit), an omitted sub-field stays absent (keep the
+    override), and ``[]`` / ``""`` override to empty. The server enforces
+    the override principle; the SDK does not read or merge the agent source.
+    """
 
-    _fields = ("title", "metadata")
+    _fields = (
+        "title",
+        "metadata",
+        "vault_ids",
+        "environment_variables",
+        "agent",
+        "mcp_configs",
+    )
 
     def __init__(  # pylint: disable=useless-parent-delegation
         self,
         *,
         title: Optional[str] = None,
         metadata: Optional[Mapping[str, Any]] = None,
+        vault_ids: Optional[Sequence[str]] = None,
+        environment_variables: Optional[Mapping[str, str]] = None,
+        agent: Any = None,
+        mcp_configs: Optional[Sequence[Mapping[str, Any]]] = None,
     ) -> None:
         super().__init__(
             title=title,
             metadata=(dict(metadata) if metadata is not None else None),
+            vault_ids=(list(vault_ids) if vault_ids is not None else None),
+            environment_variables=(
+                dict(environment_variables)
+                if environment_variables is not None
+                else None
+            ),
+            agent=(_to_mapping(agent) if agent is not None else None),
+            mcp_configs=(
+                [_to_mapping(m) for m in mcp_configs]
+                if mcp_configs is not None
+                else None
+            ),
         )
 
 
 
@@ -715,6 +715,148 @@ def test_session_thread_model_fields():
     assert "title" not in d  # legacy field removed
 
 
+def test_session_create_params_agent_string_and_object():
+    """createSession ``agent`` accepts the Agent ID string (forward-compat)
+    or an override object; ``environment_variables`` / ``mcp_configs`` are
+    new top-level fields. Omitted fields are not sent (no null leak)."""
+    from dashscope.agentstudio.types.params import SessionCreateParams
+
+    # String agent — backward compatible.
+    assert SessionCreateParams(agent="agent_01", title="x").to_dict() == {
+        "agent": "agent_01",
+        "title": "x",
+    }
+    # Override object + new top-level fields (doc §2.3 example).
+    body = SessionCreateParams(
+        agent={
+            "type": "agent_with_overrides",
+            "id": "agent_xxx",
+            "version": 3,
+            "system": "你是此 Session 专用的助手。",
+            "tools": [],
+            "mcp_servers": [{"type": "official", "name": "weather"}],
+            "skills": [
+                {
+                    "type": "customer",
+                    "skill_id": "skill_xxx",
+                    "version": "1.0",
+                }
+            ],
+        },
+        environment_variables={"LANG": "zh_CN.UTF-8"},
+        mcp_configs=[
+            {"mcp_server_name": "weather", "headers": {"key1": "value1"}}
+        ],
+    ).to_dict()
+    assert body["agent"]["type"] == "agent_with_overrides"
+    assert body["agent"]["id"] == "agent_xxx"
+    assert body["environment_variables"] == {"LANG": "zh_CN.UTF-8"}
+    assert body["mcp_configs"] == [
+        {"mcp_server_name": "weather", "headers": {"key1": "value1"}}
+    ]
+    # Omitted optional fields are absent — never sent as null.
+    for absent in (
+        "environment_id",
+        "resources",
+        "vault_ids",
+        "metadata",
+    ):
+        assert absent not in body
+
+
+def test_session_update_params_patch_and_clear_semantics():
+    """updateSession: omitted top-level fields are not sent (keep existing);
+    ``[]`` / ``{}`` clear; the ``agent`` patch is passed through verbatim so
+    an explicit ``None`` sub-field serializes to ``null`` (revoke override),
+    an omitted sub-field stays absent (keep), ``[]`` / ``""`` override empty.
+    """
+    from dashscope.agentstudio.types.params import SessionUpdateParams
+
+    # Only title — everything else omitted, no null uploaded.
+    assert SessionUpdateParams(title="x").to_dict() == {"title": "x"}
+    # [] / {} explicitly clear (not null).
+    assert SessionUpdateParams(
+        vault_ids=[], environment_variables={}, mcp_configs=[], metadata={}
+    ).to_dict() == {
+        "vault_ids": [],
+        "environment_variables": {},
+        "mcp_configs": [],
+        "metadata": {},
+    }
+    # Full doc §3.3 example body.
+    body = SessionUpdateParams(
+        title="更新后的项目分析",
+        metadata={"project": "demo", "phase": "review"},
+        vault_ids=["vault_project", "vault_shared"],
+        environment_variables={"LANG": "zh_CN.UTF-8", "REPORT_MODE": "strict"},
+        agent={
+            "system": "你是本 Session 专用的项目分析助手。",
+            "tools": [],
+            "mcp_servers": [],
+            "skills": [
+                {
+                    "type": "customer",
+                    "skill_id": "skill_report_writer",
+                    "version": "1.0",
+                }
+            ],
+        },
+        mcp_configs=[
+            {
+                "mcp_server_name": "weather",
+                "headers": {"X-Tenant-Id": "demo"},
+            }
+        ],
+    ).to_dict()
+    assert body["agent"]["system"] == "你是本 Session 专用的项目分析助手。"
+    assert body["agent"]["skills"] == [
+        {
+            "type": "customer",
+            "skill_id": "skill_report_writer",
+            "version": "1.0",
+        }
+    ]
+    assert body["environment_variables"] == {
+        "LANG": "zh_CN.UTF-8",
+        "REPORT_MODE": "strict",
+    }
+    # Agent patch: omitted sub-field absent (keep), None → null (revoke),
+    # [] / "" override to empty. Mapping passed through verbatim.
+    patch = SessionUpdateParams(
+        agent={"system": "", "tools": None, "skills": []}
+    ).to_dict()["agent"]
+    assert patch == {"system": "", "tools": None, "skills": []}
+    assert "mcp_servers" not in patch
+
+
+def test_session_model_hydrates_override_agent_and_new_fields():
+    """Session response parses the agent override object and the new
+    ``environment_variables`` / ``mcp_configs`` fields."""
+    from dashscope.agentstudio.types import Agent, Session
+
+    s = Session(
+        id="sesn_1",
+        agent={
+            "id": "agent_01",
+            "type": "agent_with_overrides",
+            "version": 3,
+            "system": "hi",
+        },
+        environment_variables={"LANG": "zh_CN.UTF-8"},
+        mcp_configs=[
+            {"mcp_server_name": "weather", "headers": {"k": "v"}}
+        ],
+    )
+    assert isinstance(s.agent, Agent)
+    assert s.agent.id == "agent_01"
+    assert s.agent.type == "agent_with_overrides"
+    assert s.agent_id == "agent_01"  # works for string and object forms
+    assert s.environment_variables == {"LANG": "zh_CN.UTF-8"}
+    assert s.mcp_configs == [
+        {"mcp_server_name": "weather", "headers": {"k": "v"}}
+    ]
+
+
 def test_message_error_and_pending_signal():
     from dashscope.agentstudio.types import parse_message