how can i take the inputs for data inside the data...
# ask-aia
ambitious-refrigerator-86200
09/17/2025, 3:45 AMhow can i take the inputs for data inside the data class with the help of an agent or task or function tool in livekit when i'm using nova sonic realtime model
t
tall-airline-16756
09/17/2025, 3:45 AMThanks for reaching out, Ok. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
tall-airline-16756
09/17/2025, 3:46 AMI'm glad you're interested in taking inputs for data inside a data class using an agent, task, or function tool in LiveKit with the Nova Sonic real-time model. Let's walk through the steps together:
1. Define a Data Class: First, you'll want to create a data class to hold session-specific user data. Here's a simple example:
Copy code
@dataclass
class UserData:
customer_name: Optional[str] = None
customer_phone: Optional[str] = NoneRunContext Copy code
@function_tool()
async def update_name(name: str, context: RunContext[UserData]) -> str:
context.userdata.customer_name = name
return f"The name is updated to {name}"userdata Copy code
session = AgentSession(
llm=aws.realtime.RealtimeModel(),
userdata=UserData(),
)aws.realtime.RealtimeModel Copy code
class CollectConsent(AgentTask[bool]):
...a
ambitious-refrigerator-86200
09/17/2025, 3:58 AMi tried the this but i get below error
aws_sdk_bedrock_runtime.models.ValidationException: RequestId=90c-9b571c92f1fd : Tool Response parsing error
t
tall-airline-16756
09/17/2025, 3:59 AMIt looks like you're encountering the error
aws_sdk_bedrock_runtime.models.ValidationException: Tool Response parsing error Copy code
@function_tool()
async def update_name(name: str, context: RunContext[UserData]) -> dict:
context.userdata.customer_name = name
return {"status": "success", "name": name}json.dumps Copy code
import json
return json.dumps({"result": "your string"})LK_BEDROCK_DEBUG=1a
ambitious-refrigerator-86200
09/17/2025, 4:04 AMthis is the error i see now:
ERROR livekit.agents - AI function
get_user_profilet
tall-airline-16756
09/17/2025, 4:05 AMIt looks like the error is happening because the tool function
get_user_profilePolicyAgent Copy code
@function_tool()
async def transfer_to_policy_agent(context: RunContext):
return PolicyAgent(), "Transferring to the policy specialist to assist with your inquiry"a
ambitious-refrigerator-86200
09/17/2025, 4:08 AMthis is my complete code, I am not returning any raw object
@dataclass
class UserProfile:
name: Optional[str]=None
class PolicyAgent(Agent):
def _init_(self, chat_ctx: Optional[ChatContext] = None):
super()._init_(
instructions="You are a policy help agent, You have to collect the user profile data with the help of get_user_profile tool at the start of the session, without that don't answer any questions.",
chat_ctx=chat_ctx,
)
@function_tool
async def get_user_profile(name: str, context: RunContext[UserProfile]) -> dict:
context.userdata.name=name
return {"status":"Profile Updated", "name": name}
async def entrypoint(ctx):
await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
session = AgentSession(llm=RealtimeModel(), userdata=UserProfile())
await session.start(agent=PolicyAgent(), room=ctx.room)
if name == "_main_":
agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))
t
tall-airline-16756
09/17/2025, 4:09 AMIt looks like the issue is happening because the function tool
get_user_profilePolicyAgent Copy code
@function_tool()
async def get_user_profile(name: str, context: RunContext[UserProfile]) -> dict:
context.userdata.name = name
return {"status": "Profile Updated", "name": name}<http://logger.info|logger.info>(f"Returning from get_user_profile: {type(output)} = {output}")PolicyAgenta
ambitious-refrigerator-86200
09/17/2025, 4:25 AMcan you give me the complete code with agent task and my current logic to collect name with it, also integrating it with nova sonic agent
t
tall-airline-16756
09/17/2025, 4:26 AMHere’s the complete code integrating your logic to collect the name using an agent task and Nova Sonic:
Copy code
from dataclasses import dataclass
from typing import Optional
from livekit.agents import Agent, AgentSession, AgentTask, function_tool, RunContext
from livekit.aws.realtime import RealtimeModel
from livekit.core import AutoSubscribe
# Define the UserProfile data class
@dataclass
class UserProfile:
name: Optional[str] = None
# Define the PolicyAgent class
class PolicyAgent(Agent):
def __init__(self, chat_ctx=None):
super().__init__(
instructions=(
"You are a policy help agent. You must collect the user profile data with the help of the 'get_user_profile' tool at the start of the session. Without that, don't answer any questions."
),
chat_ctx=chat_ctx,
)
@function_tool()
async def get_user_profile(self, name: str, context: RunContext[UserProfile]) -> dict:
"""Collects the user's name and updates the profile."""
context.userdata.name = name
return {"status": "Profile Updated", "name": name}
# Define the AgentTask to collect the name
class CollectNameTask(AgentTask[bool]):
async def run(self, context: RunContext[UserProfile]) -> bool:
"""Runs the task to collect the user's name."""
if not context.userdata.name:
await context.agent.get_user_profile(name="John Doe", context=context) # Example name
return True
# Define the entrypoint function
async def entrypoint(ctx):
await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
# Initialize the agent session with Nova Sonic and the user profile
session = AgentSession(
llm=RealtimeModel(),
userdata=UserProfile(),
)
# Start the session with the PolicyAgent and run the CollectNameTask
await session.start(agent=PolicyAgent(), room=ctx.room)
await session.run_task(CollectNameTask())
# Main entrypoint
if __name__ == "__main__":
from livekit.agents import cli
cli.run_app(cli.WorkerOptions(entrypoint_fnc=entrypoint))UserProfilePolicyAgentget_user_profileAgentTaskCollectNameTaskRealtimeModela
ambitious-refrigerator-86200
09/17/2025, 4:50 AMgetting this error
await session.run_task(CollectNameTask())
^^^^^^^^^^^^^^^^
AttributeError: 'AgentSession' object has no attribute 'run_task'
t
tall-airline-16756
09/17/2025, 4:51 AMIt looks like the error happens because the
AgentSessionrun_taskAgentTask Copy code
# Inside the PolicyAgent class
class PolicyAgent(Agent):
...
async def on_start(self, context: RunContext[UserProfile]):
# Run the task logic here
if not context.userdata.name:
await self.get_user_profile(name="John Doe", context=context)2 Views