Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 77 additions & 0 deletions tests/test_identity_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,83 @@ def test_deregister_uses_authenticated_identity_not_path_text(self):


class WrapperLaunchTests(unittest.TestCase):
def test_named_claude_identity_derives_injection_from_command(self):
with tempfile.TemporaryDirectory() as tmp:
args, _env, inject_env, settings_path = wrapper._build_provider_launch(
agent="claude2",
agent_cfg={"command": "claude"},
instance_name="claude2",
data_dir=Path(tmp),
proxy_url=None,
extra_args=[],
env={},
token="named_claude_token",
mcp_cfg={"http_port": 8200},
)

self.assertEqual(args[:1], ["--mcp-config"])
self.assertEqual(inject_env, {})
payload = json.loads(Path(settings_path).read_text("utf-8"))
self.assertEqual(
payload["mcpServers"]["agentchattr"]["headers"]["Authorization"],
"Bearer named_claude_token",
)

def test_named_codex_identity_derives_injection_from_command_path(self):
args, _env, inject_env, settings_path = wrapper._build_provider_launch(
agent="codex2",
agent_cfg={"command": r"C:\Tools\codex.exe"},
instance_name="codex2",
data_dir=Path(tempfile.gettempdir()),
proxy_url="http://127.0.0.1:7777/mcp",
extra_args=[],
env={},
)

self.assertEqual(args[0], "-c")
self.assertIn('mcp_servers.agentchattr.url="http://127.0.0.1:7777/mcp"', args[1])
self.assertEqual(inject_env, {})
self.assertIsNone(settings_path)

def test_command_exe_suffix_resolves_provider_defaults(self):
resolved = wrapper._resolve_mcp_inject("claude2", {"command": "claude.exe"})

self.assertEqual(resolved["mcp_inject"], "flag")
self.assertEqual(resolved["mcp_flag"], "--mcp-config")

def test_explicit_mcp_override_precedes_command_provider_defaults(self):
resolved = wrapper._resolve_mcp_inject(
"claude2",
{
"command": "claude",
"mcp_inject": "proxy_flag",
"mcp_proxy_flag_template": "--server={url}",
},
)

self.assertEqual(resolved["mcp_inject"], "proxy_flag")
self.assertEqual(resolved["mcp_proxy_flag_template"], "--server={url}")
self.assertNotIn("mcp_flag", resolved)

def test_named_provider_merges_partial_mcp_overrides(self):
resolved = wrapper._resolve_mcp_inject(
"claude2",
{"command": "claude", "mcp_transport": "sse", "mcp_flag": "--custom"},
)

self.assertEqual(resolved["mcp_inject"], "flag")
self.assertEqual(resolved["mcp_transport"], "sse")
self.assertEqual(resolved["mcp_flag"], "--custom")

def test_unknown_custom_command_does_not_inherit_provider_defaults(self):
self.assertEqual(
wrapper._resolve_mcp_inject(
"custom-agent",
{"command": "claude-compatible"},
),
{},
)

def test_build_provider_launch_for_claude_uses_direct_server_auth(self):
"""Claude bypasses proxy — connects directly to MCP server with bearer token."""
with tempfile.TemporaryDirectory() as tmp:
Expand Down
29 changes: 24 additions & 5 deletions wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -163,13 +163,32 @@ def _write_claude_mcp_config(
_VALID_INJECT_MODES = {"settings_file", "env", "flag", "proxy_flag", "env_content"}


def _provider_from_command(command: object) -> str:
"""Return a built-in provider for an exact configured executable name.

Named agent identities (for example ``claude2``) are independent of
the CLI they launch. Resolve only exact known executable basenames; do
not guess from prefixes or arbitrary command text.
"""
if not isinstance(command, str):
return ""
executable = command.strip().replace("\\", "/").rsplit("/", 1)[-1].lower()
for suffix in (".exe", ".cmd", ".bat"):
if executable.endswith(suffix):
executable = executable[:-len(suffix)]
break
return executable if executable in _BUILTIN_DEFAULTS else ""


def _resolve_mcp_inject(agent: str, agent_cfg: dict) -> dict:
"""Resolve MCP injection config: explicit agent_cfg > built-in defaults > None."""
inject_mode = agent_cfg.get("mcp_inject")
if inject_mode:
"""Resolve MCP injection config: explicit config > provider defaults > none."""
if agent_cfg.get("mcp_inject"):
return dict(agent_cfg)
if agent in _BUILTIN_DEFAULTS:
merged = dict(_BUILTIN_DEFAULTS[agent])
provider = agent
if provider not in _BUILTIN_DEFAULTS:
provider = _provider_from_command(agent_cfg.get("command", ""))
if provider:
merged = dict(_BUILTIN_DEFAULTS[provider])
merged.update({k: v for k, v in agent_cfg.items() if k.startswith("mcp_")})
return merged
return {}
Expand Down