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
51 changes: 35 additions & 16 deletions sdk/python/agentfield/router.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,34 +98,53 @@ def decorator(func: Callable) -> Callable:
return decorator

# ------------------------------------------------------------------
# Agent delegation helpers
async def ai(self, *args: Any, **kwargs: Any) -> Any:
agent = self._require_agent()
return await agent.ai(*args, **kwargs)
# Automatic delegation via __getattr__
def __getattr__(self, name: str) -> Any:
"""
Automatically delegate any unknown attribute/method to the attached agent.

This allows AgentRouter to transparently proxy all Agent methods (like ai(),
call(), memory, note(), discover(), etc.) without explicitly defining
delegation methods for each one.

Args:
name: The attribute/method name being accessed

Returns:
The attribute/method from the attached agent

Raises:
RuntimeError: If router is not attached to an agent
AttributeError: If the agent doesn't have the requested attribute
"""
# Avoid infinite recursion by accessing _agent through object.__getattribute__
try:
agent = object.__getattribute__(self, '_agent')
except AttributeError:
raise RuntimeError(
"Router not attached to an agent. Call Agent.include_router(router) first."
)

async def call(self, target: str, *args: Any, **kwargs: Any) -> Any:
agent = self._require_agent()
return await agent.call(target, *args, **kwargs)
if agent is None:
raise RuntimeError(
"Router not attached to an agent. Call Agent.include_router(router) first."
)

@property
def memory(self): # type: ignore[override]
agent = self._require_agent()
return agent.memory
# Delegate to the agent - will raise AttributeError if not found
return getattr(agent, name)

@property
def app(self) -> "Agent":
"""Access the underlying Agent instance."""
return self._require_agent()

# ------------------------------------------------------------------
# Internal helpers
def _require_agent(self) -> "Agent":
if not self._agent:
raise RuntimeError(
"Router not attached to an agent. Call Agent.include_router(router) first."
)
return self._agent

# ------------------------------------------------------------------
# Internal helpers

def _combine_path(
self,
default: Optional[str],
Expand Down
48 changes: 48 additions & 0 deletions sdk/python/tests/test_router.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,14 @@ async def call(self, target, *args, **kwargs):
self.calls.append((target, args, kwargs))
return "call-result"

def note(self, message: str, tags=None):
self.calls.append(("note", (message,), {"tags": tags}))
return "note-logged"

def discover(self, **kwargs):
self.calls.append(("discover", (), kwargs))
return "discovery-result"

@property
def memory(self):
return "memory-client"
Expand Down Expand Up @@ -92,3 +100,43 @@ def inline_skill():
def test_combine_path(prefix, default, custom, expected):
router = AgentRouter(prefix=prefix)
assert router._combine_path(default, custom) == expected


def test_router_automatic_delegation():
"""Test that AgentRouter automatically delegates all Agent methods via __getattr__."""
router = AgentRouter()
agent = DummyAgent()
router._attach_agent(agent)

# Test note() delegation (the original issue)
note_result = router.note("Test message", tags=["debug"])
assert note_result == "note-logged"
assert agent.calls[-1] == ("note", ("Test message",), {"tags": ["debug"]})

# Test discover() delegation (future-proofing)
discover_result = router.discover(agent="test_agent", tags=["api"])
assert discover_result == "discovery-result"
assert agent.calls[-1] == ("discover", (), {"agent": "test_agent", "tags": ["api"]})

# Test property access (memory)
assert router.memory == "memory-client"

# Test app property
assert router.app is agent


def test_router_delegation_without_agent_raises_error():
"""Test that accessing delegated methods without an attached agent raises RuntimeError."""
router = AgentRouter()

# Test that note() raises RuntimeError when no agent is attached
with pytest.raises(RuntimeError, match="Router not attached to an agent"):
router.note("Test message")

# Test that discover() raises RuntimeError when no agent is attached
with pytest.raises(RuntimeError, match="Router not attached to an agent"):
router.discover()

# Test that memory raises RuntimeError when no agent is attached
with pytest.raises(RuntimeError, match="Router not attached to an agent"):
_ = router.memory
Loading