Repository navigation
Expand file tree
/
Copy path_server_config.py
More file actions
1144 lines (969 loc) · 45.1 KB
/
Copy path_server_config.py
File metadata and controls
1144 lines (969 loc) · 45.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Typed configuration for the app-to-server subprocess communication channel.
The app spawns a `langgraph dev` subprocess and passes configuration via
environment variables prefixed with `DEEPAGENTS_CODE_SERVER_`. This module
provides a single
`ServerConfig` dataclass that both sides share so that the set of variables,
their serialization format, and their default values are defined in one place.
The app writes config with `to_env()` and the server graph reads it back
with `from_env()`.
"""
from __future__ import annotations
import json
import logging
import os
from dataclasses import dataclass, replace
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from deepagents_code._constants import DEFAULT_AGENT_NAME as DEFAULT_ASSISTANT_ID
from deepagents_code._env_vars import SERVER_ENV_PREFIX
if TYPE_CHECKING:
from collections.abc import Mapping
from deepagents import FsToolName
from deepagents_code.project_utils import ProjectContext
logger = logging.getLogger(__name__)
SESSION_WORKSPACE_FIELDS = frozenset(
{
"allow_fs_tools",
"assistant_id",
"auto_approve",
"enable_ask_user",
"enable_interpreter",
"enable_memory",
"enable_shell",
"enable_skills",
"interactive",
"interpreter_ptc",
"interpreter_ptc_acknowledge_unsafe",
"interrupt_shell_only",
"no_mcp",
"recursion_limit",
"sandbox_id",
"sandbox_snapshot_name",
"sandbox_type",
"shell_allow_list",
}
)
"""Policy a managed client may claim for its own command invocation.
These come from the client's own CLI flags, so the client already knows them
and claiming them proves only that both sides agree. Together with
`PROJECT_WORKSPACE_FIELDS` this must partition `to_workspace_payload()`
exactly: a payload field in neither set is never verified against a client
claim and never checked for project drift.
`test_workspace_claim_partitions_every_policy_field` pins that.
"""
PROJECT_WORKSPACE_FIELDS = frozenset(
{
"extension_paths",
"mcp_config_path",
"sandbox_setup",
"trust_project_extensions",
"trust_project_mcp",
}
)
"""Policy the server must resolve per project directory, never accept.
Each of these grants code execution scoped to a checkout -- MCP servers,
sandbox setup commands, Python extensions. A client that could claim them could
execute one directory's configuration against another directory's trust
decision.
"""
MODEL_COMPATIBLE_FIELDS = frozenset(
{
"auto_classifier_model",
"cli_max_retries",
"model",
"rubric_model",
"summarization_model",
}
)
"""Cosmetic model settings that do not affect durable access-policy compat.
Changing which model serves a bound thread — or the retry budget or auxiliary
grader/classifier models — does not change what the thread is *allowed* to do
(trust, tool, sandbox, and approval policy are unchanged). These fields are
therefore excluded from the policy fingerprint, so a model switch does not
invalidate a durable binding. They remain part of the runtime fingerprint so
the runtime rebuilds. `None` and an empty/unset value are normalized as
equivalent here.
"""
RUNTIME_ONLY_FIELDS = frozenset(
{
"assistant_id",
"model_params",
"profile_overrides",
"recursion_limit",
"rubric_max_iterations",
"system_prompt",
}
)
"""Runtime-identity fields that are not durable access policy.
These shape *how* the runtime executes (model kwargs, prompt, step budget,
graph id) but are not the trust/tool/sandbox/approval policy a binding guards.
They are fingerprinted into the runtime identity (so changing them rebuilds the
runtime) yet never persisted or reported — `model_params`, `profile_overrides`,
and `system_prompt` can carry secrets and prompt material.
"""
WORKSPACE_IDENTITY_FIELDS = frozenset({"cwd", "project_root"})
"""Fields that name *which* workspace a binding is for.
Workspace identity is part of durable access-policy compatibility: a binding
is meaningless if it can be replayed against a different directory. These are
compared exactly (canonical paths) and are never normalized away.
"""
def _classified_fields() -> frozenset[str]:
"""Return every field the policy/runtime split accounts for.
Returns:
The union of all classification sets plus workspace identity.
"""
return MODEL_COMPATIBLE_FIELDS | RUNTIME_ONLY_FIELDS | WORKSPACE_IDENTITY_FIELDS
def _same_workspace_project(first: str | None, second: str) -> bool:
"""Whether two paths name the same project directory.
Fails closed: an unset launch root, a missing path, or an undecidable
comparison counts as *different*, so the caller drops project policy rather
than carrying it across an unverified boundary. `_same_directory` compares
by device and inode, so a symlinked or differently cased spelling of one
directory still compares equal.
Returns:
`True` only when both paths name the same directory.
"""
if first is None:
return False
from deepagents_code._paths import DeepAgentsHomeError, _same_directory
try:
return _same_directory(Path(first), Path(second))
except DeepAgentsHomeError:
logger.warning(
"Could not compare project directories %s and %s; treating as "
"separate projects, so project-scoped policy will not apply",
first,
second,
exc_info=True,
)
return False
def _read_env_bool(suffix: str, *, default: bool = False) -> bool:
"""Read a `DEEPAGENTS_CODE_SERVER_*` boolean from the environment.
Boolean env vars use the `'true'` / `'false'` convention (case insensitive).
Missing variables fall back to *default*.
Args:
suffix: Variable name suffix after the `DEEPAGENTS_CODE_SERVER_` prefix.
default: Value when the variable is absent.
Returns:
Parsed boolean.
"""
raw = os.environ.get(f"{SERVER_ENV_PREFIX}{suffix}")
if raw is None:
return default
return raw.lower() == "true"
def _read_env_json(suffix: str) -> Any: # noqa: ANN401
"""Read a JSON-encoded `DEEPAGENTS_CODE_SERVER_*` variable.
Args:
suffix: Variable name suffix after the `DEEPAGENTS_CODE_SERVER_` prefix.
Returns:
Parsed JSON value, or `None` if the variable is absent.
Raises:
ValueError: If the variable is present but not valid JSON.
"""
raw = os.environ.get(f"{SERVER_ENV_PREFIX}{suffix}")
if raw is None:
return None
try:
return json.loads(raw)
except json.JSONDecodeError as exc:
msg = (
f"Failed to parse {SERVER_ENV_PREFIX}{suffix} as JSON: {exc}. "
f"Value was: {raw[:200]!r}"
)
raise ValueError(msg) from exc
def _read_env_str_list(suffix: str) -> tuple[str, ...]:
raw = _read_env_json(suffix)
if raw is None:
return ()
if isinstance(raw, list) and all(isinstance(item, str) for item in raw):
return tuple(raw)
msg = f"Invalid {SERVER_ENV_PREFIX}{suffix}: expected a JSON string list"
raise ValueError(msg)
def _read_env_allow_fs_tools() -> list[FsToolName] | None:
"""Read and shape-validate the `ALLOW_FS_TOOLS` filesystem allowlist.
The parent writes only an absent variable (unrestricted — `None`, which is
also what `--allow-fs-tools all` collapses to) or a non-empty JSON list of
tool names (`main._parse_allow_fs_tools_flag`). This runs in the server
subprocess, where the variable could be tampered with, so — because the
value is a security control — any unrecognized shape must fail closed
(raise) rather than fall through to an unrestricted filesystem.
(`_read_env_json` already fails closed on malformed JSON.)
`[]` and unknown tool names are rejected here, not deferred downstream, so
the returned list genuinely satisfies `list[FsToolName]` and the `cast`
asserts membership that was actually checked. Importing `deepagents` here is
fine: the subprocess already imports the SDK to build the agent (this is not
the arg-parsing hot path guarded in `main`). The `"read_file"` requirement
is not checked here: `ServerConfig.__post_init__` enforces it when the
returned value is placed on the config (with `FilesystemMiddleware` as a
final backstop), so a tampered list without `read_file` still fails closed
at construction.
Returns:
`None` when the variable is absent, or a non-empty list of filesystem
tool-name strings, each a valid `FsToolName`.
Raises:
ValueError: If the present variable parses to anything other than a
non-empty list of strings, or if any list element is not a
recognized filesystem tool name.
"""
env_name = f"{SERVER_ENV_PREFIX}ALLOW_FS_TOOLS"
if env_name not in os.environ:
return None
raw = _read_env_json("ALLOW_FS_TOOLS")
if isinstance(raw, list) and raw and all(isinstance(name, str) for name in raw):
from typing import get_args
from deepagents import FsToolName
valid_names = frozenset(get_args(FsToolName))
unknown = [name for name in raw if name not in valid_names]
if unknown:
msg = (
f"Invalid {SERVER_ENV_PREFIX}ALLOW_FS_TOOLS value: unknown "
f"filesystem tool name(s) {unknown!r}; valid names are "
f"{sorted(valid_names)}."
)
raise ValueError(msg)
return cast("list[FsToolName]", raw)
msg = (
f"Invalid {SERVER_ENV_PREFIX}ALLOW_FS_TOOLS value: {raw!r}; expected "
"a non-empty list of filesystem tool names."
)
raise ValueError(msg)
def _read_env_str(suffix: str) -> str | None:
"""Read an optional `DEEPAGENTS_CODE_SERVER_*` string variable.
Args:
suffix: Variable name suffix after the `DEEPAGENTS_CODE_SERVER_` prefix.
Returns:
The string value, or `None` if absent.
"""
return os.environ.get(f"{SERVER_ENV_PREFIX}{suffix}")
def _read_env_int(suffix: str, *, default: int | None) -> int | None:
"""Read a `DEEPAGENTS_CODE_SERVER_*` integer from the environment.
Args:
suffix: Variable name suffix after the `DEEPAGENTS_CODE_SERVER_` prefix.
default: Value when the variable is absent or malformed.
Returns:
Parsed integer, or the default when absent or parsing fails.
"""
raw = os.environ.get(f"{SERVER_ENV_PREFIX}{suffix}")
if raw is None:
return default
try:
return int(raw)
except ValueError:
return default
def _read_env_optional_bool(suffix: str) -> bool | None:
"""Read a tri-state `DEEPAGENTS_CODE_SERVER_*` boolean (`True` / `False` / `None`).
Used for settings where `None` carries a distinct meaning (e.g. "not
specified, use default logic").
Args:
suffix: Variable name suffix after the `DEEPAGENTS_CODE_SERVER_` prefix.
Returns:
`True`, `False`, or `None` when the variable is absent.
"""
raw = os.environ.get(f"{SERVER_ENV_PREFIX}{suffix}")
if raw is None:
return None
return raw.lower() == "true"
def _resolve_enable_interpreter(
enable_interpreter: bool | None, sandbox_type: str | None
) -> bool:
"""Resolve the interpreter's tri-state caller option to a concrete boolean.
Args:
enable_interpreter: Explicit caller preference, or `None` to use the
sandbox-aware default.
sandbox_type: Sandbox backend identifier. Any falsy value (`None`, `""`)
or `"none"` is treated as local execution.
Returns:
The explicit `enable_interpreter` value when not `None`; `False` for
remote-sandbox defaults; otherwise the configured local default
from `interpreter.enable_interpreter`.
Raises:
RuntimeError: If the interpreter option is absent from the manifest.
"""
if enable_interpreter is not None:
return enable_interpreter
if sandbox_type and sandbox_type != "none":
return False
from deepagents_code.config_manifest import _emit_ranked_diagnostics, get_option
from deepagents_code.configuration.resolver import get_config_resolver
option = get_option("interpreter.enable_interpreter")
if option is None:
msg = "interpreter.enable_interpreter is missing from the config manifest"
raise RuntimeError(msg)
resolved = get_config_resolver().get(option)
_emit_ranked_diagnostics(option, resolved)
return bool(resolved.value)
def _interpreter_suppressed_by_sandbox(
*, enable_interpreter: bool | None, sandbox_type: str | None, local_default: bool
) -> bool:
"""Whether a remote sandbox suppressed the otherwise-default interpreter.
Used to decide whether to surface an advisory: returns `True` only when the
user made no explicit choice, a remote sandbox is active, and the local
default would have enabled it — i.e. the sandbox (not an explicit
`--no-interpreter` opt-out, nor a disabled `[interpreter]` config) is why
`js_eval` is unavailable.
Takes the *raw* tri-state caller intent rather than the resolved boolean: a
sandbox-suppressed default and an explicit `--no-interpreter` both resolve to
`False`, so the resolved value cannot distinguish them. Any explicit choice
(`not None`) is the user's own decision and is left unannounced.
Args:
enable_interpreter: The raw tri-state caller intent (`--interpreter` →
`True`, `--no-interpreter` → `False`, unset → `None`).
sandbox_type: Sandbox backend identifier. Any falsy value (`None`, `""`)
or `"none"` is treated as local execution.
local_default: The resolver-backed local-mode default;
gating on it keeps the advisory quiet for users who disabled the
interpreter in config.
Returns:
`True` when the advisory should be shown, otherwise `False`.
"""
if enable_interpreter is not None:
return False
if not (sandbox_type and sandbox_type != "none"):
return False
return local_default
@dataclass(frozen=True)
class ServerConfig:
"""Full configuration payload passed from the app to the server subprocess.
Serialized to/from `DEEPAGENTS_CODE_SERVER_*` environment variables so
that the server graph (which runs in a separate Python interpreter)
can reconstruct the app's intent without sharing memory.
"""
model: str | None = None
"""Model spec string (e.g. `'anthropic:claude-opus-4-7'`); `None` lets the
server pick its default."""
summarization_model: str | None = None
"""Model spec used only for context-compaction summaries.
`None` reuses the main agent model.
"""
model_params: dict[str, Any] | None = None
"""Extra kwargs forwarded to the chat model constructor (temperature,
max_tokens, etc.)."""
cli_max_retries: int | None = None
"""Explicit `--max-retries` value, separate from provider model kwargs."""
profile_overrides: dict[str, Any] | None = None
"""Model profile metadata overrides resolved by the client."""
assistant_id: str = DEFAULT_ASSISTANT_ID
"""Identifier of the agent graph to invoke on the server."""
system_prompt: str | None = None
"""Override for the agent's system prompt; `None` uses the agent's default."""
auto_approve: bool = False
"""Auto-approve every tool call without human-in-the-loop interrupts."""
interrupt_shell_only: bool = False
"""Route only shell tool calls through HITL; validate others via middleware."""
shell_allow_list: list[str] | None = None
"""Restrictive allow-list of shell commands; `None` disables the allow-list.
Must be non-empty when set.
"""
interactive: bool = True
"""Whether the agent runs in an interactive session (vs.
one-shot/non-interactive)."""
enable_shell: bool = True
"""Enable the shell execution tool on the server."""
enable_ask_user: bool = False
"""Enable the `ask_user` tool that lets the agent prompt the user mid-run."""
enable_memory: bool = True
"""Enable the long-term memory subsystem."""
enable_skills: bool = True
"""Enable the skills subsystem (SKILL.md loading and skill tools)."""
enable_interpreter: bool = False
"""Enable `CodeInterpreterMiddleware` (`js_eval` REPL) on the main agent.
Always the resolved concrete value: `from_cli_args` collapses the tri-state
caller option via `_resolve_enable_interpreter` before constructing the
config, so the `bool | None` "defer to default" sentinel never reaches this
field. The `False` default here is only the bare-constructor/`from_env`
fallback; the user-facing default (on in local mode) is resolver-backed.
Local-mode only; the server graph raises if a sandbox is configured and
this flag is `True`.
"""
interpreter_ptc: str | list[str] | None = None
"""Invocation-scoped override for `interpreter.ptc`.
`None` means "fall through to whatever `interpreter.ptc` resolves
to from `~/.deepagents/config.toml`". A string is one of `"safe"`/`"all"`;
a list is an explicit allowlist of tool names that may also include the
`"safe"` preset (expanded at agent-build time); `"all"` is rejected inside
a list.
"""
interpreter_ptc_acknowledge_unsafe: bool = False
"""Override for `interpreter.ptc_acknowledge_unsafe` — required when
`interpreter_ptc="all"` is paired with non-`auto_approve` mode.
"""
allow_fs_tools: list[FsToolName] | None = None
"""Allowlist for `FilesystemMiddleware`'s `tools` param, from
`--allow-fs-tools`.
`None` means "all filesystem tools" and is also what `--allow-fs-tools all`
parses to: it leaves the SDK's own default `FilesystemMiddleware` in place
(no replacement). A list is an explicit allowlist of filesystem tool names,
must include `"read_file"`, and installs a restricted replacement (see
`create_cli_agent`).
"""
rubric_model: str | None = None
"""Grader model spec for `RubricMiddleware` (e.g. `'anthropic:...'`).
`None` reuses the main agent model.
"""
rubric_max_iterations: int | None = None
"""Explicit grader iterations per rubric attempt; `None` uses the SDK default."""
auto_classifier_model: str | None = None
"""Classifier model spec for Auto mode (e.g. `'anthropic:claude-haiku-4-5'`).
`None` falls through to `DEEPAGENTS_CODE_AUTO_CLASSIFIER_MODEL`, then
`[models].auto_classifier`, and then to the main agent model. An empty value
round-trips to `None`, so it means "inherit", never "empty spec".
"""
recursion_limit: int | None = None
"""Explicit main-agent LangGraph `recursion_limit` (graph step budget).
`None` resolves from runtime configuration. An explicit value from
`--recursion-limit` wins over the env var and `config.toml`, but managed
config outranks the flag. Must be a positive integer when set.
"""
sandbox_type: str | None = None
"""Sandbox backend identifier (e.g. `'daytona'`); `None` runs tools on the
host. `'none'` is normalized to `None` in `__post_init__`."""
sandbox_id: str | None = None
"""Existing sandbox ID to attach to; `None` creates a fresh sandbox."""
sandbox_snapshot_name: str | None = None
"""Sandbox snapshot (langsmith) or blueprint (runloop) name; must be `None`
when `sandbox_id` is set."""
sandbox_setup: str | None = None
"""Absolute path to a setup script executed inside the sandbox on first attach."""
cwd: str | None = None
"""User's original working directory, serialized as an absolute path."""
project_root: str | None = None
"""Detected project root (e.g. nearest git/uv/npm boundary), or `None` when
outside a project."""
mcp_config_path: str | None = None
"""Absolute path to the MCP server config file; `None` disables
MCP-from-config."""
no_mcp: bool = False
"""Disable all MCP server connections regardless of other config."""
trust_project_mcp: bool | None = None
"""Tri-state trust flag for project-scoped MCP servers: `True`/`False`/`None`
(prompt user)."""
trust_project_extensions: bool = False
"""Whether the project's Python extensions may execute for this run."""
extension_paths: tuple[str, ...] = ()
"""Absolute one-run extension files or directories from repeatable CLI flags."""
def to_workspace_payload(self) -> dict[str, Any]:
"""Return non-secret resource policy for a durable workspace binding."""
return {
"assistant_id": self.assistant_id,
"auto_approve": self.auto_approve,
"interrupt_shell_only": self.interrupt_shell_only,
"shell_allow_list": self.shell_allow_list,
"interactive": self.interactive,
"enable_shell": self.enable_shell,
"enable_ask_user": self.enable_ask_user,
"enable_memory": self.enable_memory,
"enable_skills": self.enable_skills,
"enable_interpreter": self.enable_interpreter,
"interpreter_ptc": self.interpreter_ptc,
"interpreter_ptc_acknowledge_unsafe": (
self.interpreter_ptc_acknowledge_unsafe
),
"allow_fs_tools": self.allow_fs_tools,
"recursion_limit": self.recursion_limit,
"sandbox_type": self.sandbox_type,
"sandbox_id": self.sandbox_id,
"sandbox_snapshot_name": self.sandbox_snapshot_name,
"sandbox_setup": self.sandbox_setup,
"mcp_config_path": self.mcp_config_path,
"no_mcp": self.no_mcp,
"trust_project_mcp": self.trust_project_mcp,
"trust_project_extensions": self.trust_project_extensions,
"extension_paths": list(self.extension_paths),
}
def to_session_workspace_claim(self) -> dict[str, Any]:
"""Return the command-scoped policy a managed client may claim.
Returns:
The session-scoped subset of the workspace policy.
"""
return {
key: value
for key, value in self.to_workspace_payload().items()
if key in SESSION_WORKSPACE_FIELDS
}
def to_project_workspace_policy(self) -> dict[str, Any]:
"""Return policy that must be resolved for each project directory.
Returns:
The project-scoped subset of the workspace policy.
"""
return {
key: value
for key, value in self.to_workspace_payload().items()
if key in PROJECT_WORKSPACE_FIELDS
}
def session_workspace_fingerprint(self) -> str:
"""Fingerprint the exact client-claimable session policy.
Returns:
The canonical SHA-256 fingerprint.
"""
from deepagents_code.workspace import canonical_fingerprint
return canonical_fingerprint(self.to_session_workspace_claim())
def resolve_workspace(
self,
cwd: str,
project_root: str | None,
) -> ServerConfig:
"""Resolve directory-bound policy for one server workspace.
Project-scoped policy (`PROJECT_WORKSPACE_FIELDS`) is valid only for
the directory it was resolved against: it came from the launch-time CLI
and that project's trust decisions. Reusing it for another directory
would apply one project's MCP servers, sandbox setup, and extensions to
a different, possibly untrusted, checkout.
So the launch project keeps its policy verbatim, and any other project
starts from nothing: MCP and sandbox setup are *dropped* rather than
rediscovered, and extension trust is re-read from the trust store for
that project. `_same_workspace_project` fails closed, so an
unresolvable path also takes the drop branch.
Args:
cwd: Absolute, canonical working directory for the workspace.
project_root: Canonical project root, or `None` when the workspace
has none. The launch cwd uses the server's explicit root when
configured. Otherwise, extension trust is keyed on `cwd` when
no root exists.
Returns:
A config whose session policy is unchanged and whose project policy
is either the launch project's or empty.
"""
if self.project_root is not None and _same_workspace_project(self.cwd, cwd):
project_root = str(Path(self.project_root).expanduser().resolve())
launch_root = self.project_root or self.cwd
target_root = project_root or cwd
if _same_workspace_project(launch_root, target_root):
return replace(self, cwd=cwd, project_root=project_root)
from deepagents_code.extensions.trust import is_project_extensions_trusted
return replace(
self,
cwd=cwd,
project_root=project_root,
sandbox_setup=None,
mcp_config_path=None,
trust_project_mcp=None,
trust_project_extensions=is_project_extensions_trusted(target_root),
extension_paths=(),
)
def preserve_bound_extension_trust(
self, bound_policy: Mapping[str, object]
) -> ServerConfig:
"""Keep an existing thread's extension trust when a new grant appears.
Args:
bound_policy: Server policy persisted when the thread was bound.
Returns:
A config that defers new grants to new threads. Revocations remain
visible so binding and runtime validation can reject them.
"""
if bound_policy.get("trust_project_extensions") is False and (
self.trust_project_extensions is True
):
return replace(self, trust_project_extensions=False)
return self
def policy_fingerprint(self) -> str:
"""Fingerprint durable access policy plus workspace identity.
Covers the trust/tool/sandbox/approval payload (the
`to_workspace_payload()` keys) and the workspace's `cwd`/`project_root`.
Cosmetic model settings (`MODEL_COMPATIBLE_FIELDS`) are excluded so a
model switch does not invalidate a durable binding; runtime-only fields
(`RUNTIME_ONLY_FIELDS`) are excluded because they are not persisted.
Returns:
The canonical SHA-256 fingerprint.
"""
from deepagents_code.workspace import canonical_fingerprint
return canonical_fingerprint(
{
"cwd": self.cwd,
"policy": self.to_workspace_payload(),
"project_root": self.project_root,
}
)
def runtime_fingerprint(self) -> str:
"""Fingerprint the full runtime identity, including model settings.
Any change to model, model params, prompt, policy, or runtime fields
yields a different value, so the runtime cache can rebuild on any of
them while the binding only guards the policy fingerprint.
Returns:
The canonical SHA-256 fingerprint.
"""
values = self.to_env()
values.pop("CWD")
values.pop("PROJECT_ROOT")
from deepagents_code.workspace import canonical_fingerprint
return canonical_fingerprint(values)
def workspace_fingerprint(self) -> str:
"""Fingerprint the resolved runtime config except workspace identity.
Retained for compatibility; equivalent to `runtime_fingerprint()`.
Returns:
The canonical SHA-256 fingerprint.
"""
return self.runtime_fingerprint()
def __post_init__(self) -> None:
"""Normalize fields and validate invariants.
Raises:
TypeError: If `rubric_max_iterations` or `recursion_limit` is a
boolean.
ValueError: If `shell_allow_list` is an empty list,
`allow_fs_tools` is an empty list or omits `"read_file"`, or
`rubric_max_iterations` / `recursion_limit` is non-positive.
"""
if self.sandbox_type == "none":
object.__setattr__(self, "sandbox_type", None)
if self.shell_allow_list is not None and len(self.shell_allow_list) == 0:
msg = "shell_allow_list must be None or non-empty"
raise ValueError(msg)
# `allow_fs_tools` is a security control: `None` means unrestricted, but
# an explicit list must be a usable allowlist. Own the non-empty +
# `read_file`-required invariant here (the single authoritative point
# for both the env round-trip via `from_env` and direct construction)
# rather than deferring to `FilesystemMiddleware`, which would only
# surface the violation a process boundary away. `_parse_allow_fs_tools_flag`
# still enforces the same rule at the CLI for a friendlier error.
if self.allow_fs_tools is not None:
if len(self.allow_fs_tools) == 0:
msg = "allow_fs_tools must be None or a non-empty list"
raise ValueError(msg)
if "read_file" not in self.allow_fs_tools:
msg = "allow_fs_tools must include 'read_file'"
raise ValueError(msg)
if isinstance(self.rubric_max_iterations, bool):
msg = "rubric_max_iterations must be None or a positive integer"
raise TypeError(msg)
if self.rubric_max_iterations is not None and self.rubric_max_iterations <= 0:
msg = "rubric_max_iterations must be None or a positive integer"
raise ValueError(msg)
if isinstance(self.cli_max_retries, bool):
msg = "cli_max_retries must be None or a non-negative integer"
raise TypeError(msg)
if self.cli_max_retries is not None and self.cli_max_retries < 0:
msg = "cli_max_retries must be None or a non-negative integer"
raise ValueError(msg)
if isinstance(self.recursion_limit, bool):
msg = "recursion_limit must be None or a positive integer"
raise TypeError(msg)
if self.recursion_limit is not None and self.recursion_limit <= 0:
msg = "recursion_limit must be None or a positive integer"
raise ValueError(msg)
# ------------------------------------------------------------------
# Serialization
# ------------------------------------------------------------------
def to_env(self) -> dict[str, str | None]:
"""Serialize this config to a `DEEPAGENTS_CODE_SERVER_*` env-var mapping.
`None` values signal that the variable should be *cleared* from the
environment (rather than set to an empty string), so callers can
iterate and set or clear each variable in `os.environ`.
Returns:
Dict mapping env-var suffixes (without the prefix) to their
string values or `None`.
"""
return {
"MODEL": self.model,
"SUMMARIZATION_MODEL": self.summarization_model,
"MODEL_PARAMS": (
json.dumps(self.model_params) if self.model_params is not None else None
),
"MAX_RETRIES": (
str(self.cli_max_retries) if self.cli_max_retries is not None else None
),
"PROFILE_OVERRIDES": (
json.dumps(self.profile_overrides)
if self.profile_overrides is not None
else None
),
"ASSISTANT_ID": self.assistant_id,
"SYSTEM_PROMPT": self.system_prompt,
"AUTO_APPROVE": str(self.auto_approve).lower(),
"INTERRUPT_SHELL_ONLY": str(self.interrupt_shell_only).lower(),
"SHELL_ALLOW_LIST": (
",".join(self.shell_allow_list)
if self.shell_allow_list is not None
else None
),
"INTERACTIVE": str(self.interactive).lower(),
"ENABLE_SHELL": str(self.enable_shell).lower(),
"ENABLE_ASK_USER": str(self.enable_ask_user).lower(),
"ENABLE_MEMORY": str(self.enable_memory).lower(),
"ENABLE_SKILLS": str(self.enable_skills).lower(),
"ENABLE_INTERPRETER": str(self.enable_interpreter).lower(),
"INTERPRETER_PTC": (
json.dumps(self.interpreter_ptc)
if self.interpreter_ptc is not None
else None
),
"INTERPRETER_PTC_ACKNOWLEDGE_UNSAFE": str(
self.interpreter_ptc_acknowledge_unsafe
).lower(),
"ALLOW_FS_TOOLS": (
json.dumps(self.allow_fs_tools)
if self.allow_fs_tools is not None
else None
),
"RUBRIC_MODEL": self.rubric_model,
"AUTO_CLASSIFIER_MODEL": self.auto_classifier_model,
"RUBRIC_MAX_ITERATIONS": (
str(self.rubric_max_iterations)
if self.rubric_max_iterations is not None
else None
),
"RECURSION_LIMIT": (
str(self.recursion_limit) if self.recursion_limit is not None else None
),
"SANDBOX_TYPE": self.sandbox_type,
"SANDBOX_ID": self.sandbox_id,
"SANDBOX_SNAPSHOT_NAME": self.sandbox_snapshot_name,
"SANDBOX_SETUP": self.sandbox_setup,
"CWD": self.cwd,
"PROJECT_ROOT": self.project_root,
"MCP_CONFIG_PATH": self.mcp_config_path,
"NO_MCP": str(self.no_mcp).lower(),
"TRUST_PROJECT_MCP": (
str(self.trust_project_mcp).lower()
if self.trust_project_mcp is not None
else None
),
"TRUST_PROJECT_EXTENSIONS": str(self.trust_project_extensions).lower(),
"EXTENSION_PATHS": (
json.dumps(self.extension_paths) if self.extension_paths else None
),
}
@classmethod
def from_env(cls) -> ServerConfig:
"""Reconstruct a `ServerConfig` from `DEEPAGENTS_CODE_SERVER_*` env vars.
This is the inverse of `to_env()` and is called inside the server
subprocess to recover the app's configuration.
Returns:
A `ServerConfig` populated from the environment.
"""
return cls(
model=_read_env_str("MODEL"),
summarization_model=_read_env_str("SUMMARIZATION_MODEL") or None,
model_params=_read_env_json("MODEL_PARAMS"),
cli_max_retries=_read_env_int("MAX_RETRIES", default=None),
profile_overrides=_read_env_json("PROFILE_OVERRIDES"),
assistant_id=_read_env_str("ASSISTANT_ID") or DEFAULT_ASSISTANT_ID,
system_prompt=_read_env_str("SYSTEM_PROMPT"),
auto_approve=_read_env_bool("AUTO_APPROVE"),
interrupt_shell_only=_read_env_bool("INTERRUPT_SHELL_ONLY"),
shell_allow_list=(
[cmd.strip() for cmd in raw.split(",") if cmd.strip()]
if (raw := _read_env_str("SHELL_ALLOW_LIST"))
else None
)
or None,
interactive=_read_env_bool("INTERACTIVE", default=True),
enable_shell=_read_env_bool("ENABLE_SHELL", default=True),
enable_ask_user=_read_env_bool("ENABLE_ASK_USER"),
enable_memory=_read_env_bool("ENABLE_MEMORY", default=True),
enable_skills=_read_env_bool("ENABLE_SKILLS", default=True),
enable_interpreter=_read_env_bool("ENABLE_INTERPRETER"),
interpreter_ptc=_read_env_json("INTERPRETER_PTC"),
interpreter_ptc_acknowledge_unsafe=_read_env_bool(
"INTERPRETER_PTC_ACKNOWLEDGE_UNSAFE"
),
allow_fs_tools=_read_env_allow_fs_tools(),
rubric_model=_read_env_str("RUBRIC_MODEL") or None,
auto_classifier_model=_read_env_str("AUTO_CLASSIFIER_MODEL") or None,
rubric_max_iterations=_read_env_int("RUBRIC_MAX_ITERATIONS", default=None),
recursion_limit=_read_env_int("RECURSION_LIMIT", default=None),
sandbox_type=_read_env_str("SANDBOX_TYPE"),
sandbox_id=_read_env_str("SANDBOX_ID"),
sandbox_snapshot_name=_read_env_str("SANDBOX_SNAPSHOT_NAME") or None,
sandbox_setup=_read_env_str("SANDBOX_SETUP"),
cwd=_read_env_str("CWD"),
project_root=_read_env_str("PROJECT_ROOT"),
mcp_config_path=_read_env_str("MCP_CONFIG_PATH"),
no_mcp=_read_env_bool("NO_MCP"),
trust_project_mcp=_read_env_optional_bool("TRUST_PROJECT_MCP"),
trust_project_extensions=_read_env_bool("TRUST_PROJECT_EXTENSIONS"),
extension_paths=_read_env_str_list("EXTENSION_PATHS"),
)
# ------------------------------------------------------------------
# Factory
# ------------------------------------------------------------------
@classmethod
def from_cli_args(
cls,
*,
project_context: ProjectContext | None,
model_name: str | None,
summarization_model: str | None = None,
model_params: dict[str, Any] | None,
cli_max_retries: int | None = None,
profile_overrides: dict[str, Any] | None = None,
assistant_id: str,
auto_approve: bool,
interrupt_shell_only: bool = False,
shell_allow_list: list[str] | None = None,
sandbox_type: str = "none",
sandbox_id: str | None,
sandbox_snapshot_name: str | None,
sandbox_setup: str | None,
enable_shell: bool,
enable_ask_user: bool,
enable_interpreter: bool | None = None,
interpreter_ptc: str | list[str] | None = None,
interpreter_ptc_acknowledge_unsafe: bool = False,
allow_fs_tools: list[FsToolName] | None = None,
rubric_model: str | None = None,
rubric_max_iterations: int | None = None,
auto_classifier_model: str | None = None,
recursion_limit: int | None = None,
mcp_config_path: str | None,
no_mcp: bool,
trust_project_mcp: bool | None,
interactive: bool,
trust_project_extensions: bool = False,
extension_paths: tuple[str, ...] = (),
) -> ServerConfig:
"""Build a `ServerConfig` from parsed CLI arguments.
Handles path normalization (e.g. resolving relative MCP config paths
against the user's working directory) so that the raw serialized values
are always absolute and unambiguous.
Args:
project_context: Explicit user/project path context.
model_name: Model spec string.
summarization_model: Model spec used only for context-compaction
summaries; `None` reuses the main model.
model_params: Extra model kwargs.
cli_max_retries: Explicit `--max-retries` value.
profile_overrides: Model profile metadata overrides.
assistant_id: Agent identifier.
auto_approve: Auto-approve all tools.
interrupt_shell_only: Validate shell commands via middleware instead
of HITL.
shell_allow_list: Restrictive shell allow-list to forward to the
server subprocess for `ShellAllowListMiddleware`.
sandbox_type: Sandbox type.
sandbox_id: Existing sandbox ID to reuse.
sandbox_snapshot_name: Snapshot (langsmith) or blueprint (runloop)
name to use or create.
sandbox_setup: Path to setup script for the sandbox.
enable_shell: Enable shell execution tools.
enable_ask_user: Enable ask_user tool.
enable_interpreter: Enable `CodeInterpreterMiddleware` on the main
agent. `None` uses the sandbox-aware default.
interpreter_ptc: Invocation-scoped PTC allowlist override.