Skip to content

fix(sdk/python): allow skipping container IP detection #970

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mikemikimike wants to merge 1 commit into from
Closed

fix(sdk/python): allow skipping container IP detection #970

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
104 changes: 66 additions & 38 deletions sdk/python/agentfield/agent.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,6 +182,14 @@ def _detect_container_ip() -> Optional[str]:
Returns:
External IP address if detected, None otherwise
"""
if os.getenv("AGENTFIELD_SKIP_IP_DETECTION", "").lower() in {
"1",
"true",
"yes",
"on",
}:
return None

try:
# Try to get IP from container metadata (works in many hosted environments)
import requests
Expand Down Expand Up @@ -753,6 +761,7 @@ def __init__(
# one, and a re-import in the same process keeps the same one (since the
# same Agent instance is being used).
import uuid as _uuid

self.agent_instance_id = _uuid.uuid4().hex

# Memory-efficient handler registries (replaces old list-based storage)
Expand Down Expand Up @@ -835,6 +844,7 @@ def __init__(
# before any user-defined reasoners are registered so the path is
# always available for the control-plane callback.
from .cancel import install_cancel_route

install_cancel_route(self)

# Initialize async execution manager (will be lazily created when needed)
Expand Down Expand Up @@ -1070,7 +1080,11 @@ def _entry_to_metadata(
metadata["accepts_webhook"] = "true"
elif accepts_webhook is False:
metadata["accepts_webhook"] = "false"
elif isinstance(accepts_webhook, str) and accepts_webhook in ("true", "false", "warn"):
elif isinstance(accepts_webhook, str) and accepts_webhook in (
"true",
"false",
"warn",
):
metadata["accepts_webhook"] = accepts_webhook
else:
metadata["accepts_webhook"] = "warn"
Expand Down Expand Up @@ -1709,8 +1723,7 @@ def _build_agent_metadata(self) -> Optional[Dict[str, Any]]:
@property
def sessions(self) -> List[Dict[str, Any]]:
return [
entry["definition"].to_dict()
for entry in self._session_registry.values()
entry["definition"].to_dict() for entry in self._session_registry.values()
]

def session(
Expand All @@ -1725,7 +1738,10 @@ def session(
tools: Optional[List[str]] = None,
tags: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> Callable[[Callable[[RealtimeSession], Awaitable[Any]]], Callable[[RealtimeSession], Awaitable[Any]]]:
) -> Callable[
[Callable[[RealtimeSession], Awaitable[Any]]],
Callable[[RealtimeSession], Awaitable[Any]],
]:
"""Register a realtime/voice session endpoint.

Provider and transport are both explicit; AgentField does not infer or
Expand All @@ -1746,7 +1762,7 @@ def session(
)

def decorator(
func: Callable[[RealtimeSession], Awaitable[Any]]
func: Callable[[RealtimeSession], Awaitable[Any]],
) -> Callable[[RealtimeSession], Awaitable[Any]]:
self._session_registry[name] = {"definition": definition, "handler": func}
setattr(func, "_agentfield_session", definition)
Expand Down Expand Up @@ -2154,6 +2170,7 @@ async def run_reasoner() -> Any:
# execute_async_with_callback handles its own
# cancellation accounting on top of this.
from .cancel import register_execution_task

await register_execution_task(self, execution_id_header, task)
return JSONResponse(
status_code=202,
Expand All @@ -2180,6 +2197,7 @@ async def run_reasoner() -> Any:
register_execution_task,
deregister_execution,
)

sync_task = asyncio.create_task(run_reasoner())
await register_execution_task(
self, execution_id_header, sync_task
Expand Down Expand Up @@ -2258,7 +2276,7 @@ async def tracked_func(*args, **kwargs):
vc_setting = self._effective_component_vc_setting(
reasoner_id, self._reasoner_vc_overrides
)

self._reasoner_registry[reasoner_id] = ReasonerEntry(
id=reasoner_id,
func=func,
Expand Down Expand Up @@ -2294,7 +2312,9 @@ async def tracked_func(*args, **kwargs):

return decorator

def _detect_and_unwrap_trigger_envelope(self, payload_dict: Dict[str, Any]) -> tuple[Dict[str, Any], Optional[Any]]:
def _detect_and_unwrap_trigger_envelope(
self, payload_dict: Dict[str, Any]
) -> tuple[Dict[str, Any], Optional[Any]]:
"""
Detect dispatcher webhook envelope {event: ..., _meta: ...} and unwrap it.
Returns (unwrapped_input, trigger_context_or_none).
Expand All @@ -2303,23 +2323,25 @@ def _detect_and_unwrap_trigger_envelope(self, payload_dict: Dict[str, Any]) -> t
# Check if this looks like a dispatcher envelope
if not isinstance(payload_dict, dict):
return payload_dict, None

if "event" in payload_dict and "_meta" in payload_dict:
# This is a dispatcher envelope
event_data = payload_dict.get("event", {})
meta_data = payload_dict.get("_meta", {})

# Parse metadata into TriggerContext
try:
from datetime import datetime
from .triggers import TriggerContext

received_at_str = meta_data.get("received_at", "")
if received_at_str:
received_at = datetime.fromisoformat(received_at_str.replace('Z', '+00:00'))
received_at = datetime.fromisoformat(
received_at_str.replace("Z", "+00:00")
)
else:
received_at = datetime.utcnow()

trigger_ctx = TriggerContext(
trigger_id=meta_data.get("trigger_id", ""),
source=meta_data.get("source", ""),
Expand All @@ -2333,37 +2355,39 @@ def _detect_and_unwrap_trigger_envelope(self, payload_dict: Dict[str, Any]) -> t
except Exception:
# If parsing fails, return raw envelope for compatibility
return payload_dict, None

# Not an envelope
return payload_dict, None

def _apply_trigger_transform(self, trigger_ctx, bindings: list, input_data: dict) -> dict:
def _apply_trigger_transform(
self, trigger_ctx, bindings: list, input_data: dict
) -> dict:
"""
Match trigger context against reasoner bindings and apply transform if found.
Returns transformed input or original input if no match.

Matching logic:
1. Find bindings where binding.source == trigger_ctx.source
2. Check event_type: binding.types empty OR trigger_ctx.event_type matches (exact or prefix)
3. If multiple match, prefer most specific (non-empty types)
4. Apply transform if binding has one
"""
from .triggers import EventTrigger

if not bindings or not trigger_ctx:
return input_data

# Find best-matching binding
best_match = None
best_specificity = -1 # -1 = no match, 0 = broad (empty types), 1+ = specific

for binding in bindings:
if not isinstance(binding, EventTrigger):
continue

if binding.source != trigger_ctx.source:
continue

# Check event_type match
if binding.types:
# binding has specific types — check for match
Expand All @@ -2378,21 +2402,23 @@ def _apply_trigger_transform(self, trigger_ctx, bindings: list, input_data: dict
else:
# binding accepts all types
specificity = 0

# This binding matches; is it better than current best?
if specificity > best_specificity:
best_match = binding
best_specificity = specificity

# Apply transform if found
if best_match and best_match.transform:
try:
return best_match.transform(input_data)
except Exception as e:
if self.dev_mode:
log_warn(f"Transform failed for {trigger_ctx.source}/{trigger_ctx.event_type}: {e}; using raw input")
log_warn(
f"Transform failed for {trigger_ctx.source}/{trigger_ctx.event_type}: {e}; using raw input"
)
return input_data

return input_data

async def _execute_reasoner_endpoint(
Expand All @@ -2411,7 +2437,9 @@ async def _execute_reasoner_endpoint(
execution_context = ExecutionContext.from_request(request, self.node_id)
payload_dict = input_data # Already a dict from runtime validation
# Unwrap dispatcher envelope if present (Phase 5 webhook DX)
payload_dict, trigger_context = self._detect_and_unwrap_trigger_envelope(payload_dict)
payload_dict, trigger_context = self._detect_and_unwrap_trigger_envelope(
payload_dict
)
if trigger_context:
execution_context.trigger = trigger_context

Expand Down Expand Up @@ -2460,9 +2488,7 @@ async def _execute_reasoner_endpoint(
trigger_bindings = getattr(func, "_reasoner_triggers", [])
if execution_context.trigger and trigger_bindings:
payload_dict = self._apply_trigger_transform(
execution_context.trigger,
trigger_bindings,
payload_dict
execution_context.trigger, trigger_bindings, payload_dict
)

# When invoked via an inbound trigger, the (possibly transformed)
Expand Down Expand Up @@ -2787,6 +2813,7 @@ async def _watchdog() -> None:
# plane records status=failed WITHOUT discarding the rich result
# (it stores the result payload regardless of terminal status).
from .exceptions import ReasonerFailed

if isinstance(exc, ReasonerFailed) and exc.result is not None:
payload["result"] = jsonable_encoder(exc.result)
log_error(f"Execution {execution_id} failed asynchronously: {exc}")
Expand All @@ -2808,6 +2835,7 @@ async def _watchdog() -> None:
self._pause_clocks.pop(execution_id, None)
# Deregister the cancel hook regardless of outcome.
from .cancel import deregister_execution

await deregister_execution(self, execution_id)
# Attach usage on non-success terminal states too — a reasoner that
# failed or was cancelled may still have consumed tokens before ending.
Expand Down Expand Up @@ -2863,7 +2891,9 @@ def _build_execution_callback_url(self, execution_id: str) -> Optional[str]:
+ f"/api/v1/executions/{execution_id}/status"
)

def on_change(self, pattern: Union[str, List[str]]) -> Callable[[Callable[P, Awaitable[T]]], Callable[P, Awaitable[T]]]:
def on_change(
self, pattern: Union[str, List[str]]
) -> Callable[[Callable[P, Awaitable[T]]], Callable[P, Awaitable[T]]]:
"""
Decorator to mark a function as a memory event listener.

Expand Down Expand Up @@ -3822,9 +3852,10 @@ def _record_harness_usage(
cache_creation = getattr(result, "cache_creation_tokens", 0) or 0
cost = getattr(result, "cost_usd", None)

if not any(
(input_tokens, output_tokens, cache_read, cache_creation)
) and cost is None:
if (
not any((input_tokens, output_tokens, cache_read, cache_creation))
and cost is None
):
return

tracker = get_current_cost_tracker()
Expand All @@ -3845,9 +3876,7 @@ def _record_harness_usage(
or self._harness_model_name()
or (resolved_provider or "harness")
)
total = getattr(result, "total_tokens", 0) or (
input_tokens + output_tokens
)
total = getattr(result, "total_tokens", 0) or (input_tokens + output_tokens)
ctx = self._get_current_execution_context()
tracker.record(
model=str(model_name),
Expand Down Expand Up @@ -4589,6 +4618,7 @@ async def _push_self_running() -> None:
ExecutionFailedError,
ExecutionTimeoutError,
)

if isinstance(
async_error,
(
Expand Down Expand Up @@ -4807,9 +4837,7 @@ async def _send_note():
if self.dev_mode:
from agentfield.logger import log_debug

log_debug(
f"NOTE DEBUG: api_base: {self.client.api_base}"
)
log_debug(f"NOTE DEBUG: api_base: {self.client.api_base}")
log_debug(
f"NOTE DEBUG: Full URL: {self.client.api_base}/executions/note"
)
Expand Down
27 changes: 24 additions & 3 deletions sdk/python/tests/test_agent_networking.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,11 +29,22 @@ def json(self):
def fake_get(url, headers=None, timeout=None):
calls.append(url)
parsed = urlparse(url)
if parsed.netloc == "169.254.169.254" and parsed.path == "/latest/meta-data/public-ipv4":
if (
parsed.netloc == "169.254.169.254"
and parsed.path == "/latest/meta-data/public-ipv4"
):
return DummyResponse(200, "198.51.100.5")
if parsed.netloc == "metadata.google.internal" and parsed.path == "/computeMetadata/v1/instance/network-interfaces/0/access-configs/0/external-ip":
if (
parsed.netloc == "metadata.google.internal"
and parsed.path
== "/computeMetadata/v1/instance/network-interfaces/0/access-configs/0/external-ip"
):
return DummyResponse(200, "203.0.113.7")
if parsed.scheme == "https" and parsed.netloc == "api.ipify.org" and parsed.path in {"", "/"}:
if (
parsed.scheme == "https"
and parsed.netloc == "api.ipify.org"
and parsed.path in {"", "/"}
):
return DummyResponse(200, "192.0.2.9")
return DummyResponse(404, "")

Expand Down Expand Up @@ -67,6 +78,16 @@ def fake_get(url, headers=None, timeout=None):
assert agent_mod._detect_container_ip() == "203.0.113.9"


def test_detect_container_ip_can_be_disabled(monkeypatch):
def fail_get(*args, **kwargs):
raise AssertionError("IP detection should be skipped")

monkeypatch.setenv("AGENTFIELD_SKIP_IP_DETECTION", "true")
monkeypatch.setattr("requests.get", fail_get)

assert agent_mod._detect_container_ip() is None


def test_is_running_in_container_checks_dockerenv(monkeypatch, tmp_path):
monkeypatch.setattr(agent_mod.os.path, "exists", lambda path: path == "/.dockerenv")
monkeypatch.setattr(agent_mod.os, "environ", {})
Expand Down