Hi could you help me create an agent according to json schema { "conversation_flow_id": "conversation_flow_621cc80f0db0", "version": 0, "nodes": [ { "instruction": { "type": "prompt", "text": "You are an weather Forcast assistant great the user and help them to provide weather updates" }, "model_choice": { "type": "cascading", "model": "gpt-4.1" }, "name": "Welcome Node", "edges": [ { "destination_node_id": "node-1757970372359", "id": "edge-1", "transition_condition": { "type": "prompt", "prompt": "Check current weather" } } ], "start_speaker": "agent", "id": "start-node-1757840296673", "type": "conversation", "display_position": { "x": 437.53846153846155, "y": 190.53846153846152 } }, { "tool_id": "tool-1757970320005", "name": "Weather", "edges": [ { "destination_node_id": "node-1757978077029", "id": "edge-1757971826346", "transition_condition": { "type": "prompt", "prompt": "If we have the temperature information" } } ], "id": "node-1757970372359", "type": "function", "tool_type": "local", "speak_during_execution": false, "display_position": { "x": 922.2728489336864, "y": 242.4811811808407 }, "wait_for_result": true }, { "name": "End Call", "id": "node-1757972035549", "type": "end", "display_position": { "x": 1772.4408983930457, "y": 297.19759979126945 }, "instruction": { "type": "prompt", "text": "Politely end the call" } }, { "name": "Conversation", "edges": [ { "destination_node_id": "node-1757972035549", "id": "edge-1757978077029", "transition_condition": { "type": "prompt", "prompt": "If user want to end the conversation" } } ], "id": "node-1757978077029", "type": "conversation", "display_position": { "x": 1393.6923076923076, "y": 230.53846153846155 }, "instruction": { "type": "prompt", "text": "explain user that the current temperature is {{temperature}}." } } ], "start_node_id": "start-node-1757840296673", "start_speaker": "agent", "tools": [ { "headers": {}, "parameter_type": "json", "tool_id": "tool-1757970320005", "method": "GET", "query_params": { "q": "New York,US", "units": "metric", "appid": "a4c3da3f5a7c97acba02069a3401d0f3" }, "args_at_root": false, "timeout_ms": 120000, "name": "Weather", "description": "API to get weather update", "response_variables": { "temperature": "list.main.temp" }, "type": "custom", "url": "https://api.openweathermap.org/data/2.5/forecast" } ], "model_choice": { "type": "cascading", "model": "gpt-4.1" }, "kb_config": { "top_k": 3, "filter_score": 0.6 }, "begin_tag_display_position": { "x": 99.69230769230771, "y": 404.53846153846155 }, "is_published": false }

Thanks for reaching out, Umair. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

please provide complete code

I'd love to walk you through a complete example of how you can create an agent using the provided JSON schema. Here it is:

import { llm, voice } from '@livekit/agents';
import { z } from 'zod';

// Define the tool using JSON schema
const weatherTool = llm.tool({
  description: 'Get the weather for a specific location.',
  parameters: z.object({
    location: z.string().describe('The city and state, e.g., New York, US'),
    units: z.enum(['metric', 'imperial']).optional().describe('Units for temperature (metric or imperial)'),
  }),
  execute: async (args) => {
    const { location, units } = args;
    const response = await fetch(`${process.env.WEATHER_API_URL}?q=${location}&units=${units || 'metric'}&appid=${process.env.WEATHER_API_KEY}`);
    const data = await response.json();
    return { temperature: data.list[0].main.temp, description: data.list[0].weather[0].description };
  },
});

// Create the agent
const agent = new voice.Agent({
  instructions: 'You are a weather forecast assistant. Greet the user and provide weather updates.',
  tools: { weatherTool },
});

// Deploy the agent (use LiveKit CLI for deployment)
// Ensure you have set up LIVEKIT_API_KEY and LIVEKIT_API_SECRET in your environment

When you're ready to deploy this agent, just save the code in your project directory and run the LiveKit CLI command:

lk agent create [working-dir]

Just a quick note: make sure to replace

process.env.WEATHER_API_URL

and

process.env.WEATHER_API_KEY

with your actual API URL and key in your environment variables. If you have any questions or need a hand with the deployment, please don’t hesitate to reach out. I’m here to help!

provide python sdk code

I'd love to guide you through creating an agent using the Python SDK in LiveKit. Here's a friendly example to get you started:

from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, cli
from livekit.plugins import deepgram, openai, silero

# Define the entrypoint function
async def entrypoint(ctx: JobContext):
    await ctx.connect()
    session = AgentSession(
        vad=silero.VAD.load(),
        stt=deepgram.STT(model="nova-3"),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=elevenlabs.TTS(),
    )
    await session.start(agent=Agent(instructions="You are a friendly voice assistant."), room=ctx.room)
    await session.generate_reply(instructions="Greet the user and ask about their day.")

# Run the agent
if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

Here are some simple steps to follow: 1. Install Dependencies: Just run

pip install "livekit-agents[openai,silero,deepgram,cartesia,turn-detector]~=1.0"

. This will get everything you need set up. 2. Set Environment Variables: Make sure to add your API keys like

DEEPGRAM_API_KEY

,

OPENAI_API_KEY

, and

ELEVEN_API_KEY

into a

.env

file or directly into your environment. This helps the agent connect smoothly. 3. Run the Agent: To see it in action, start the agent in development mode with

python3 agent.py dev

. If you have any questions or want me to help tweak this example, just let me know. I'm here to help! Sources: Accepting incoming calls | LiveKit Docs | Accepting incoming calls | LiveKit Docs | livekit-agents/README.md | Agents Overview | LiveKit Docs | Agents Overview | LiveKit Docs

this code is not reflecting what we have in json lide nodes, edges and tools etc { "conversation_flow_id": "conversation_flow_621cc80f0db0", "version": 0, "nodes": [ { "instruction": { "type": "prompt", "text": "You are an weather Forcast assistant great the user and help them to provide weather updates" }, "model_choice": { "type": "cascading", "model": "gpt-4.1" }, "name": "Welcome Node", "edges": [ { "destination_node_id": "node-1757970372359", "id": "edge-1", "transition_condition": { "type": "prompt", "prompt": "Check current weather" } } ], "start_speaker": "agent", "id": "start-node-1757840296673", "type": "conversation", "display_position": { "x": 437.53846153846155, "y": 190.53846153846152 } }, { "tool_id": "tool-1757970320005", "name": "Weather", "edges": [ { "destination_node_id": "node-1757978077029", "id": "edge-1757971826346", "transition_condition": { "type": "prompt", "prompt": "If we have the temperature information" } } ], "id": "node-1757970372359", "type": "function", "tool_type": "local", "speak_during_execution": false, "display_position": { "x": 922.2728489336864, "y": 242.4811811808407 }, "wait_for_result": true }, { "name": "End Call", "id": "node-1757972035549", "type": "end", "display_position": { "x": 1772.4408983930457, "y": 297.19759979126945 }, "instruction": { "type": "prompt", "text": "Politely end the call" } }, { "name": "Conversation", "edges": [ { "destination_node_id": "node-1757972035549", "id": "edge-1757978077029", "transition_condition": { "type": "prompt", "prompt": "If user want to end the conversation" } } ], "id": "node-1757978077029", "type": "conversation", "display_position": { "x": 1393.6923076923076, "y": 230.53846153846155 }, "instruction": { "type": "prompt", "text": "explain user that the current temperature is {{temperature}}." } } ], "start_node_id": "start-node-1757840296673", "start_speaker": "agent", "tools": [ { "headers": {}, "parameter_type": "json", "tool_id": "tool-1757970320005", "method": "GET", "query_params": { "q": "New York,US", "units": "metric", "appid": "a4c3da3f5a7c97acba02069a3401d0f3" }, "args_at_root": false, "timeout_ms": 120000, "name": "Weather", "description": "API to get weather update", "response_variables": { "temperature": "list.main.temp" }, "type": "custom", "url": "https://api.openweathermap.org/data/2.5/forecast" } ], "model_choice": { "type": "cascading", "model": "gpt-4.1" }, "kb_config": { "top_k": 3, "filter_score": 0.6 }, "begin_tag_display_position": { "x": 99.69230769230771, "y": 404.53846153846155 }, "is_published": false }

In the Python SDK for LiveKit, the concepts of nodes, edges, and tools can be brought to life like this: 1. Nodes Think of nodes as the different stages in your agent's processing journey, such as STT (speech-to-text), LLM (language model), and TTS (text-to-speech). You have the flexibility to customize these nodes by overriding specific methods in your agent class. Here's a friendly example:

async def llm_node(self, chat_ctx, tools, model_settings):
    tool_call_detected = False
    async for chunk in super().llm_node(chat_ctx, tools, model_settings):
        if isinstance(chunk, ChatChunk) and chunk.delta.tool_calls and not tool_call_detected:
            self._activity.agent.say("Checking that for you...")
            tool_call_detected = True
        yield chunk

2. Tools Tools are what let your agent connect with external systems. You can define them easily using the

@function_tool

decorator or even add them dynamically. Here's a simple way to do it:

@function_tool()
async def get_weather(location: str) -> str:
    """Get the weather in a specific location."""
    return f"The weather in {location} is sunny."

agent = MyAgent(
    instructions="You are a helpful assistant.",
    tools=[get_weather]
)

3. Edges Regarding edges from your JSON schema, the Python SDK handles the flow between nodes (like STT → LLM → TTS) behind the scenes. While edges aren't explicitly defined, you can guide this flow by: • Overriding nodes such as

llm_node

• Using lifecycle hooks like

on_user_turn_completed()

• Dynamically updating tools with

agent.update_tools()

4. Example Implementation Here’s a warm and practical example of defining a tool and weaving it into your agent:

from livekit.agents import Agent, function_tool

@function_tool()
async def get_weather(location: str) -> dict:
    """Fetch weather data for a location."""
    # Replace with actual API call logic
    return {"temperature": 25, "description": "Sunny"}

class WeatherAgent(Agent):
    async def llm_node(self, chat_ctx, tools, model_settings):
        # Custom logic for LLM node
        async for chunk in super().llm_node(chat_ctx, tools, model_settings):
            yield chunk

agent = WeatherAgent(
    instructions="You are a weather assistant.",
    tools=[get_weather]
)

If you’d like, I’m here to help with any further questions or to provide a more tailored example that fits your JSON schema perfectly! Sources: Tool definition and use | LiveKit Docs | Pipeline nodes and hooks | LiveKit Docs | mcp_client/agent_tools.py | Docs::Agents | Tool definition and use | LiveKit Docs

could you please generate complete code from zero to the end so can just run it

I don't have the answer you're looking for. You could also try asking your question: • in one of the other Slack channels or • to https://deepwiki.com/livekit/livekit_composite which is trained on all LiveKit source code If you find the answer, please post it here to help others!

I am getting error : my code

import asyncio
import logging
import aiohttp
from livekit.agents import JobContext, WorkerOptions, cli, function_tool, Agent, AgentSession
from livekit.plugins import openai, silero
from livekit.agents.llm import ChatChunk
from dotenv import load_dotenv

load_dotenv()
logger = logging.getLogger("weather-agent")

# Configuration from your schema
WEATHER_API_KEY = "a4c3da3f5a7c97acba02069a3401d0f3"
WEATHER_API_URL = "<https://api.openweathermap.org/data/2.5/forecast>"

# Global state to track conversation flow
conversation_state = {
    "current_node": "welcome",
    "temperature": None,
    "location": None
}



class WeatherAgent(Agent):
    """Weather Agent following the conversation flow schema"""

    def __init__(self) -> None:
        # Build instructions based on conversation flow schema
        instructions = """You are a weather forecast assistant following a specific conversation flow.

CONVERSATION FLOW:
1. Welcome Node: Greet user and help them get weather updates
2. Weather Node: When user asks about weather, use get_weather tool
3. Conversation Node: After getting weather, explain the temperature clearly
4. End Node: When user wants to end, use end_call tool

CURRENT BEHAVIOR:
- Start by greeting the user warmly
- When they ask about weather, use the get_weather(location) tool
- After getting weather data, explain the current temperature clearly
- Ask if they need anything else
- When they want to end, use the end_call() tool

Be natural and conversational while following this flow."""

        super().__init__(
            instructions=instructions,
            tools=[self.get_weather, self.end_call]
        )

    @function_tool()
    async def get_weather(self, location: str = "New York,US") -> dict:
        """Get the weather in a specific location - corresponds to Weather tool from schema"""
        <http://logger.info|logger.info>(f"Getting weather for: {location}")

        # Update conversation state
        conversation_state["current_node"] = "weather_function"
        conversation_state["location"] = location

        try:
            # Use exact parameters from your schema
            params = {
                "q": location,
                "units": "metric",
                "appid": WEATHER_API_KEY
            }

            async with aiohttp.ClientSession() as session:
                async with session.get(WEATHER_API_URL, params=params) as response:
                    if response.status == 200:
                        data = await response.json()

                        # Extract temperature using schema path: list.main.temp
                        if data.get("list") and len(data["list"]) > 0:
                            temperature = data["list"][0]["main"]["temp"]
                            description = data["list"][0]["weather"][0]["description"]

                            # Store temperature for template replacement
                            conversation_state["temperature"] = temperature
                            conversation_state["current_node"] = "conversation"

                            return {
                                "temperature": temperature,
                                "description": description,
                                "location": location
                            }
                        else:
                            return {"error": "No weather data available"}
                    else:
                        return {"error": f"API request failed"}
        except Exception as e:
            logger.error(f"Weather API error: {e}")
            return {"error": str(e)}

    @function_tool()
    async def end_call(self,) -> str:
        """Politely end the call - corresponds to End Call node"""
        <http://logger.info|logger.info>("Ending conversation")
        conversation_state["current_node"] = "end"
        return "Thank you for using the weather service! Have a wonderful day!"

    async def llm_node(self, chat_ctx, tools, model_settings):
        """Custom logic for LLM node with flow-based feedback"""
        tool_call_detected = False

        async for chunk in super().llm_node(chat_ctx, tools, model_settings):
            # Provide feedback when tools are called
            if isinstance(chunk, ChatChunk) and chunk.delta.tool_calls and not tool_call_detected:
                current_node = conversation_state["current_node"]

                if current_node == "welcome":
                    # Note: We'll handle feedback through the session
                    pass
                elif current_node == "weather_function":
                    # Note: We'll handle feedback through the session
                    pass

                tool_call_detected = True

            yield chunk


async def entrypoint(ctx: JobContext):
    """Main entry point for the weather agent"""

    <http://logger.info|logger.info>("Starting Weather Forecast Agent")

    # Reset conversation state
    conversation_state["current_node"] = "welcome"
    conversation_state["temperature"] = None
    conversation_state["location"] = None

    # Create agent session
    session = AgentSession(
        stt=openai.STT(),
        llm=openai.LLM(model="gpt-4", temperature=0.7),
        tts=openai.TTS(),
        vad=silero.VAD.load(),
    )

    # Start the session with our weather agent
    await session.start(
        room=ctx.room,
        agent=WeatherAgent(),
    )

    # Send initial greeting (Welcome Node behavior)
    await session.generate_reply(
        instructions=(
            "Hello! I'm your weather forecast assistant. "
            "I can help you get current weather information for any location. "
            "What location would you like to check the weather for?"
        )
    )

    <http://logger.info|logger.info>("Weather agent started and ready")


if __name__ == "__main__":
    # Set up logging
    logging.basicConfig(
        level=<http://logging.INFO|logging.INFO>,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )

    # Run the agent
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
        ),
    )

Error 2025-09-16 045508,747 - DEBUG livekit.agents - http_session(): creating a new httpclient ctx 2025-09-16 045508,778 - asyncio - ERROR - Task exception was never retrieved future: <Task finished name='AgentActivity.pipeline_reply' coro=<AgentActivity._pipeline_reply_task() done, defined at /home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/opentelemetry/util/_decorator.py:68> exception=ValueError('duplicate function name: end_call')> Traceback (most recent call last): File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/opentelemetry/util/_decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/livekit/agents/utils/log.py", line 16, in async_fn_logs return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/livekit/agents/voice/agent_activity.py", line 1544, in _pipeline_reply_task tool_ctx = llm.ToolContext(tools) ^^^^^^^^^^^^^^^^^^^^^^ File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/livekit/agents/llm/tool_context.py", line 219, in init self.update_tools(tools) File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/livekit/agents/llm/tool_context.py", line 247, in update_tools raise ValueError(f"duplicate function name: {info.name}") ValueError: duplicate function name: end_call 2025-09-16 045508,778 - ERROR asyncio - Task exception was never retrieved future: <Task finished name='AgentActivity.pipeline_reply' coro=<AgentActivity._pipeline_reply_task() done, defined at /home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/opentelemetry/util/_decorator.py:68> exception=ValueError('duplicate function name: end_call')> Traceback (most recent call last): File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/opentelemetry/util/_decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/livekit/agents/utils/log.py", line 16, in async_fn_logs return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/livekit/agents/voice/agent_activity.py", line 1544, in _pipeline_reply_task tool_ctx = llm.ToolContext(tools) ^^^^^^^^^^^^^^^^^^^^^^ File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/livekit/agents/llm/tool_context.py", line 219, in init self.update_tools(tools) File "/home/umair/.cache/pypoetry/virtualenvs/livekit-workflow-TFmTJtXu-py3.12/lib/python3.12/site-packages/livekit/agents/llm/tool_context.py", line 247, in update_tools raise ValueError(f"duplicate function name: {info.name}") ValueError: duplicate function name: end_call