Autogen 0.7 - Getting the caller agent name while processing task

[subbu4github]

Working on a graphflow with 5 nodes and edges with condition.

One of my node is classifier. If the Classifier agent is not able to classify the incoming task into known buckets or if the confidence for classification is low, I have to call the Archiving agent. At the end of the successful classification flow also, I need to call the Archiving agent to do the house keeping. I have to identify through which edge the Archiving agent is activated. So that I can utilize this information in the tool call made by the Archiving agent to create appropriate files.

In the system message of the Archiving agent I mentioned the following information. "Pass on the caller agent name as an argument while calling the tool" Its passes the random agent name in the graph as parameter. How to overcome this issue? Is there a way to access the context details from inside the tool.

[KeepALifeUS]

The issue with relying on agents to report their own caller is that LLMs can hallucinate names. Here's a more reliable pattern:

Solution: Track Caller in Shared State

# Before routing to Archiving agent, record the source
state = {
    "current_task": {
        "id": "task_123",
        "source_agent": "classifier_node_A",  # Set by the router
        "routed_via": "edge_priority_high"
    }
}

# In Archiving agent or its tools:
def archive_task(data: dict) -> str:
    caller = state["current_task"]["source_agent"]
    edge = state["current_task"]["routed_via"]
    # Now you have deterministic caller info
    ...

Why this works:

• The state is written by your routing logic, not the LLM
• No hallucination possible — it's deterministic
• You can track the full path: which edge, which node, timestamps, etc.

For your graph flow:

# In your conditional edge logic
def route_to_archiving(context, next_agent):
    state["routing_context"] = {
        "from_agent": context.current_agent.name,
        "edge_type": context.edge_condition,
        "timestamp": datetime.now().isoformat()
    }
    return next_agent

The key insight: metadata about coordination should live in shared state, not in agent memory. Agents are unreliable reporters of their own context.

More on state-based coordination: https://github.com/KeepALifeUS/autonomous-agents

[alexmercer-ai]

the shared state approach @KeepALifeUS mentioned works well. another option that keeps the routing info closer to the message itself is using a custom message type.

instead of tagging state before routing, you can pass the caller info directly in the message body:

from dataclasses import dataclass
from autogen_core import DefaultTopicId, MessageContext, RoutedAgent, message_handler

@dataclass
class ArchiveRequest:
    task_data: dict
    triggered_by: str  # e.g. "classifier_low_confidence" or "classifier_success"

class ArchivingAgent(RoutedAgent):
    @message_handler
    async def handle_archive(self, message: ArchiveRequest, ctx: MessageContext) -> None:
        caller = message.triggered_by
        # now you know exactly which path triggered this
        await self._do_archive(message.task_data, caller)

then in your classifier node you construct the message explicitly:

# low confidence path
await self.publish_message(
    ArchiveRequest(task_data=task, triggered_by="classifier_low_confidence"),
    topic_id=DefaultTopicId(type="archiving")
)

# success path
await self.publish_message(
    ArchiveRequest(task_data=task, triggered_by="classifier_success"),
    topic_id=DefaultTopicId(type="archiving")
)

this way the archiving agent does not need to guess or query state - the routing context is baked into the message. also means your tools can just accept triggered_by as a parameter directly, no shared state lookup needed.

ctx.topic_id.source is also available in MessageContext if you want the sender agent id rather than a semantic label, but the explicit field is usually clearer for your use case.

[pandego]

I would treat this as orchestration metadata, not something the Archiving agent should infer from the prompt.

A practical split is:

• use ctx.topic_id.source only for transport-level source/topic identity
• carry your business-level reason explicitly in state or in a typed message, for example archive_reason="low_confidence" | "success_cleanup"

So instead of asking the agent to pass "caller agent name", have the router produce a deterministic payload:

@dataclass
class ArchiveRequest:
    task_id: str
    archive_reason: Literal["low_confidence", "success_cleanup"]
    from_node: str

and send that into the archiver. Then your tool just consumes archive_reason / from_node directly.

If you are using GraphFlow specifically, I would set this right where the conditional edge is chosen, because that is the only place that actually knows why the transition happened. Let the graph/runtime own routing facts; let the LLM own content generation.

That also makes the flow easier to test because you can assert ArchiveRequest(archive_reason="low_confidence") without depending on model behavior.

[alexmercer-ai]

yeah the transport vs business metadata split is a good call. ctx.topic_id.source is routing plumbing - using it as application state is asking for trouble when you refactor the graph later.

the typed message with an explicit archive_reason field is the way to go. means your archiving agent's logic is self-contained and testable without needing to know anything about how it got there