user_id and session_id on tools

[DDani]

Hi! I have a question on tools: how I can access to the the user_id and session_id from a tool?

In my use case, I have to call to an API where I need this information to get some data about the user.

sorry if the question is to naive, but I’m relatively new to Python and this awesome framework.

Thanks!

[monali7-d]

Hi @DDani
thanks for reaching out and supporting Agno. I’ve shared this with the team, we’re working through all requests one by one and will get back to you soon.
If it’s urgent, please let us know. We appreciate your patience!

[dirkbrnd]

Hi @DDani
I assume you are writing a custom tool? And you set the session_id and user_id on the Agent itself?

[DDani]

Hi, yes, I 'm writing custom tools like this one:

@tool(
show_result=False,
stop_after_tool_call=False,
cache_results=False,
)
async def set_user_not_new(user_id: str, session_id: str) -> str:
"""
A tool to set the user as not new.
"""
await self._user_service.set_user_not_new(user_id)
return f"User {user_id} set as not new"

I've added the user_id and session_id just to see if those params arrive (like Agent, if is set on the tool as param) but no success. the LLM is just passing random ids.

Regarding how I'm calling agents and teams, I'm doing it as follows:

Agents:

 run_response = await agent.arun(
            user_id=user_id,
            session_id=session_id,
            debug_mode=debug_mode,
            stream=True,
            message="Introduce yourself and explain how you work",
            **kwargs
        )
        async for response in run_response:
            yield response

Teams:

run_response = await self.team.arun(
            message=message,
            stream=True,
           session_id=session_id,
            debug_mode=debug_mode,
            user_id=user_id
        );

        async for chunk in run_response:
            # chunk.content only contains the text response from the Agent.
            # For advanced use cases, we should yield the entire chunk
            # that contains the tool calls and intermediate steps.
            yield f"data: {json.dumps({'content': chunk.content})}\n\n"

Both user_id and session_id have proper values.

What I was expecting is that the agent / team forward the user_id and the session_id to the tools, so I can there, identify the user, for example to set the user as not new, call an external API to retrieve data about the user, etc.

Thanks in advance @dirkbrnd !

[Mustafa-Esoofally]

Hey @DDani ! I think you'll need to modify the method a little bit. Could you try this?

@tool(
      show_result=False,
      stop_after_tool_call=False,
      cache_results=False,
  )
  async def set_user_not_new(self, agent: Agent) -> str: # Changed signature
      """
      A tool to set the user associated with the current agent run as not new.
      Retrieves user_id from the agent context.
      """
      # Access user_id and session_id from the injected agent instance
      user_id = agent.user_id
      session_id = agent.session_id # You can also access session_id if needed

      if not user_id:
          return "Error: Could not determine user_id from agent context."

      print(f"Tool 'set_user_not_new' called for user_id: {user_id} in session: {session_id}")

      # Call the user service with the correct user_id
      await self._user_service.set_user_not_new(user_id)

      return f"User {user_id} set as not new"

[DDani]

Hi @Mustafa-Esoofally ,
thanks!, I have give it a try, and it doesn't works:

This is the tool definition:

 @tool(
    show_result=False,
    stop_after_tool_call=False,
    cache_results=False,
)   
def user_verification_tool2(self, agent: Agent) -> str:
  
    """
    A tool to set the user associated with the current agent run as not new.
    Retrieves user_id from the agent context.
    """
    
    # Access user_id and session_id from the injected agent instance
    user_id = agent.user_id
    session_id = agent.session_id # You can also access session_id if needed

    if not user_id:
        return "Error: Could not determine user_id from agent context."

    print(f"Tool 'set_user_not_new' called for user_id: {user_id} in session: {session_id}")

This is the call to the agent call (please, notice that I'm calling to arun; can it be this related?)) :

  # Use the agent's run method directly for streaming
           run_response = await agent.arun(
               user_id=user_id,
               session_id=session_id,
               debug_mode=debug_mode,
               stream=True,
               message="HF_CHAT_COMMAND_INSTRUCTIONS: Introduce yourself and explain how you work. End with, let's begin with your first update.",
               **kwargs
           )
           async for response in run_response:
               yield response

And this is the trace:

WARNING  Could not run function user_verification_tool2()                                                                                                 
ERROR    1 validation error for user_verification_tool2                                                                                                   
         self                                                                                                                                             
           Missing required argument [type=missing_argument, input_value=ArgsKwargs((), {'agent': ...False, telemetry=True)}), input_type=ArgsKwargs]     
             For further information visit https://errors.pydantic.dev/2.11/v/missing_argument                                                            
         Traceback (most recent call last):                                                                                                               
           File "/Users/dani/Developer/Happyforce/happyforce-ai/.venv/lib/python3.13/site-packages/agno/tools/function.py", line 473, in execute          
             result = self.function.entrypoint(**entrypoint_args)                          

Also I've tried to define the function as def user_verification_tool2(agent: Agent) -> str without success: the agent.user_id and agent.sessionId are empty.

In my case, this is a blocking issue; I'm writing a chatbot and I need to be able to call to some tools by using the user_id :-(.

Any workaround? (maybe injecting on each agent call prompt the user_id and the session_id)?

[Mustafa-Esoofally]

Hi @DDani I believe you need correct instructions. Try this:

@ tool(show_result=False, stop_after_tool_call=False, cache_results=False)
async def user_verification_tool(agent: Agent) -> str:
    # A tool to retrieve the current user_id and session_id from the agent context.
    
    user_id = agent.user_id or 'undefined'
    session_id = agent.session_id or "undefined'
    return f'Tool accessed user_id: {user_id}, session_id: {session_id}'
async def main():
    # Initialize the model (replace with your actual configuration)
    user_id = 'example_user'
    session_id = 'example_session'

    # Create an Agent and register the tool
    agent = Agent(
        model=OpenAIChat(),
        tools=[user_verification_tool],
        user_id=user_id,
        session_id=session_id,
        debug_mode=True,
        stream=True,
    )

    # Call the agent and stream responses (await the async call first)
    response_stream = await agent.arun(
        user_id=user_id,
        session_id=session_id,
        stream=True,
        message='Please invoke the user_verification_tool to retrieve context.'
    )
    async for run_response in response_stream:
        if run_response.content:
            print(run_response.content)

if __name__ == '__main__':
    asyncio.run(main())

[DDani]

Hi @Mustafa-Esoofally thanks for your quick reply!

I have tested your code, and it works, but if you remove the user_id and the session_id from the Agent initialisation, the tool is called without those params, even if I pass them on the agent.arun

Here is the modified version:

import asyncio
from agno.agent import Agent
from agno.tools import tool
from agno.models.openai import OpenAIChat

@tool(show_result=False, stop_after_tool_call=False, cache_results=False)
async def user_verification_tool(agent: Agent) -> str:
    # A tool to retrieve the current user_id and session_id from the agent context.
    
    user_id = agent.user_id or 'undefined'
    session_id = agent.session_id or "undefined"

    print(f'Tool accessed user_id: {user_id}, session_id: {session_id}')

    return f'Tool accessed user_id: {user_id}, session_id: {session_id}'

async def main():
    # Initialize the model (replace with your actual configuration)
    user_id = 'example_user'
    session_id = 'example_session'

    # Create an Agent and register the tool
    agent = Agent(
        model= OpenAIChat(),
        tools=[user_verification_tool],
        # Uncomment this to pass user_id and session_id to the tool to the agent
        #  user_id=user_id,
        #  session_id=session_id,
        debug_mode=True,
        stream=True,
    )

    # Call the agent and stream responses (await the async call first)
    response_stream = await agent.arun(
        user_id=user_id,
        session_id=session_id,
        stream=True,
        message='Please invoke the user_verification_tool to retrieve context.'
    )
    async for run_response in response_stream:
        if run_response.content:
            print(run_response.content)

if __name__ == '__main__':
    asyncio.run(main())`

PD: I've tried with a "sync" version (removing all the async code, Stream=true, and calling the agent.run method and same behaviour: user_id and session_id is only passed if is set during the Agent construction).

[DDani]

Checking the internals of function.py I don't see an easy way to achieve it but modifying the agent.py, and base.py to pass session_id and user_id to propagate this to the function.execute, and do some work on _build_entrypoint_args.

I'm completely new to Python (and contribution to OS), but I would like to help. What do you think @dirkbrnd and @Mustafa-Esoofally ? Something too complex for a newbie 😃?

[DDani]

Hi @dirkbrnd and @Mustafa-Esoofally,

I've created a pull request with the changes I proposed. I am not sure if is the best approach, so I'm open to modify it with your suggestions.

Thanks :-)

[DDani]

Hi, sorry for insisting, but this issue is currently blocking our development. If you could please take a few minutes to review the pull request and validate whether the approach is correct—whether it ends up being merged or the problem is resolved in a different way—we would be able to continue our development with Agno.

Thanks a lot!

[Mustafa-Esoofally]

Hi @DDani ! Thanks for bringing this up. We are planning to support this with the help of your PR.
We will also add a detailed example here.

[dirkbrnd]

Hi @DDani
This was released with 1.4.6! See here

[DDani]

Thanks @dirkbrnd!! 🔝