Function Call Request

[HamzaRaouf]

Hi I'm trying to integrate function call I'm facing an error ,which is :
{
"description": "Exception in AgentWebSocketClient._process_text",
"message": "'function_name'",
"type": "Exception"
}

when i do normal conversation it is working properly , but when i talk about to any tool which defined , it give me this exception and then disconnect the connection, I'm struggle it with from few days , but error is still persist , any thing is required relevant to this , i can provide, I will be very thankful

[deepgram-community[bot]]

Thanks for asking your question. Please be sure to reply with as much detail as possible so the community can assist you efficiently.
_{Consider joining our Discord community for more opportunity to engage with your fellow Deepgram users. You can earn points which can be redeemed for cool stuff by being active in our communities!}

[deepgram-community[bot]]

Hey there! It looks like you haven't connected your GitHub account to your Deepgram account. You can do this at https://community.deepgram.com - being verified through this process will allow our team to help you in a much more streamlined fashion.

[deepgram-community[bot]]

It looks like we're missing some important information to help debug your issue. Would you mind providing us with the following details in a reply?

• The programming language you are working in (e.g. JavaScript, Python).
• A request ID that triggered your error or issue.

[HamzaRaouf]

programming language is python ,
how i can get request ID ?

[HamzaRaouf]

here is my Function defination :
FUNCTION_DEFINITIONS = [
{
"name": "search_trip",
"description": "Search for trips using the passenger's name to answer the user's relevant trip history query.",
"parameters": {
"type": "object",
"properties": {
"displayName": {
"type": "string",
"description": "Full name of the passenger to search for (e.g. John Smith)."
}
},
"required": ["displayName"]
},

},
{
    "name": "check_eta",
    "description": "Check the ETA and trip status using a passenger's contact number.",
    "parameters": {
        "type": "object",
        "properties": {
            "contact": {
                "type": "string",
                "description": "The user's 10-digit US phone number (digits only, no formatting)."
            }
        },
        "required": ["contact"]
    },

}

]

[deepgram-community[bot]]

You'll want to check your function call response format and JSON parsing and verify event handler registration, validate function call structure, and ensure proper error handling.

1. JSON Parsing Errors: The most common cause is malformed JSON in the WebSocket message. The _process_text method attempts to parse incoming messages with json.loads(message), which can fail if:

   • The message contains invalid JSON syntax
   • The message is empty or contains only whitespace
   • The message has unexpected encoding issues

2. FunctionCallRequest Structure Mismatch: The FunctionCallRequest class expects specific fields:

   @dataclass
   class FunctionCallRequest(BaseResponse):
       type: str
       function_name: str
       function_call_id: str
       input: str

   If the incoming message doesn't match this structure, the from_json() method will fail.

3. Missing Event Handlers: If you haven't properly registered a handler for AgentWebSocketEvents.FunctionCallRequest, the function call request won't be processed correctly.

4. Response Object Creation Failures: When creating FunctionCallResponse objects, ensure you're providing the correct parameters:

   FunctionCallResponse(
       function_call_id=function_call_request.function_call_id,
       output="Your function result here"
   )

Debugging Steps:

1. Enable Verbose Logging: Add debug logging to see the exact message content:

   from deepgram.utils import verboselogs
   verboselogs.set_verbose(True)

2. Check Message Format: Verify the incoming WebSocket message structure matches the expected format for function calls.

3. Validate Event Handler: Ensure you have registered the function call handler:

   dg_connection.on(AgentWebSocketEvents.FunctionCallRequest, on_function_call_request)

4. Handle Exceptions Gracefully: The current implementation catches exceptions and emits error events, but you should also handle them in your application logic.

The error handling in the SDK is designed to catch these issues and emit appropriate error events, but the root cause is typically in the message format or event handler setup.

This message was sent by John Vajda from Deepgram, via our community automation.

[HamzaRaouf]

Thanks for your responce , here is complete code which i'm trying to use :

def main():
try:
# 🔧 Deepgram setup
config = DeepgramClientOptions(options={"keepalive": "true"})
deepgram = DeepgramClient(DEEPGRAM_API_KEY, config)
connection = deepgram.agent.websocket.v("1")

    # ⚙️ Agent settings
    options = SettingsOptions()
    options.audio.input.encoding = "linear16"
    options.audio.input.sample_rate = RATE
    options.audio.output.encoding = "linear16"
    options.audio.output.sample_rate = RATE
    options.audio.output.container = "wav"

    options.agent.language = "en"
    options.agent.listen.provider.type = "deepgram"
    options.agent.listen.provider.model = "nova-3"

    options.agent.think.provider.type = "open_ai"
    options.agent.think.provider.model = "gpt-4o-mini"
    
    # Pass the system prompt dynamically
    options.agent.think.prompt = get_system_prompt()
    options.agent.think.functions = FUNCTION_DEFINITIONS

    options.agent.speak.provider.type = "deepgram"
    options.agent.speak.model = "aura-2-thalia-en"
    options.agent.speak.provider.model = "aura-2-thalia-en"
    options.agent.speak.enabled = True
    options.agent.greeting = "Hi there! How can I help you today?"

    # 🎧 PyAudio stream for input (for user microphone input)
    audio = pyaudio.PyAudio()

    input_stream = audio.open(format=FORMAT,
                              channels=CHANNELS,
                              rate=RATE,
                              input=True,
                              frames_per_buffer=CHUNK)

    output_stream = audio.open(format=FORMAT,
                               channels=CHANNELS,
                               rate=RATE,
                               output=True)

    # 🔁 Keep-alive thread
    def keep_alive():  
        while True:
            time.sleep(5)
            connection.send(str(AgentKeepAlive()))

    threading.Thread(target=keep_alive, daemon=True).start()

    # 📞 Event Handlers
    def on_audio_data(client, data, **kwargs):
        output_stream.write(data)

    def on_conversation_text(self, conversation_text, **kwargs):
        print(f"Conversation Text: {conversation_text}")
        if conversation_text["role"] == "assistant":
            print(f"Assistant said: {conversation_text['content']}")
            return  # Don't process assistant's own responses
        
        # Parse the assistant's response for function calls
        user_input = conversation_text["content"]
        print("content of User is :",user_input)
        tool_calls = []  # This is where we store tool calls, if any
        
        # Check for tool calls and call the appropriate function
        for tool in FUNCTION_DEFINITIONS:
            # if tool["function"]["name"] in user_input:
            if tool["name"] in user_input:
                function_name = tool["name"]
                function_args = self.extract_function_arguments(user_input, function_name)
                result = self.call_function(function_name, function_args)
                tool_calls.append(result)

                # Append the result to the conversation (or use it to respond)
                st.session_state.messages.append({"role": "assistant", "content": result["message"]})
                return
        else:
            print("not found Tool call and converstion text:",user_input)    

        # Default response when no function is called
        st.session_state.messages.append({"role": "assistant", "content": "I didn't understand your request."})

    def on_function_call(client, request):
        print("=====> on function call and request is :",request)
        # request["functions"] is a list – handle each one
        for fc in request["functions"]:
            name = fc["name"]
            args = json.loads(fc["arguments"])
            result = FUNCTION_MAP[name](args)       # <- your own function

            response = {
                "type": "FunctionCallResponse",
                "id": fc["id"],
                "name": name,
                "content": json.dumps(result)       # must be a JSON-encoded string
            }
            connection.send(json.dumps(response))  
            

    def extract_function_arguments(self, user_input, function_name):
        # Placeholder function for extracting arguments (this will depend on your app's logic)
        # Example: If user_input is something like "check eta for phone 12345", extract the contact number
        if function_name == "check_eta":
            contact = user_input.split(" ")[-1]  # Just an example: Extract the last word as contact
            return {"contact": contact}
        else:
            return {}

    def call_function(self, function_name, function_args):
        # Use the FUNCTION_MAP to call the appropriate function
        if function_name in FUNCTION_MAP:
            func = FUNCTION_MAP[function_name]
            return func(function_args)
        else:
            return {"message": "Function not implemented."}

    def on_agent_audio_done(event, **kwargs):
        print("✅ Agent finished speaking.")

    def on_error(client, error, **kwargs):
        print("❌ Error:", error)

    def on_close(client, close, **kwargs):
        print("🔌 Connection closed.")

    def on_user_started_speaking(event, **kwargs):
        print("🗣️ User started speaking.")

    def on_agent_started_speaking(event, **kwargs):
        print(f"🎤 Agent started speaking: {event}")

    # 🧠 Register Events
    connection.on(AgentWebSocketEvents.AudioData, on_audio_data)
    connection.on(AgentWebSocketEvents.ConversationText, on_conversation_text)
    connection.on(AgentWebSocketEvents.AgentAudioDone, on_agent_audio_done)
    connection.on(AgentWebSocketEvents.Error, on_error)
    connection.on(AgentWebSocketEvents.Close, on_close)
    connection.on(AgentWebSocketEvents.UserStartedSpeaking, on_user_started_speaking)
    connection.on(AgentWebSocketEvents.AgentStartedSpeaking, on_agent_started_speaking)
    connection.on(AgentWebSocketEvents.FunctionCallRequest, on_function_call)


    # 🚀 Start connection
    if not connection.start(options):
        print("❌ Failed to start the connection.")
        return

    print("🎤 Connection started. You can start speaking...")

    # 🎙️ Send audio in loop (microphone input)
    def stream_microphone():
        while True:
            try:
                data = input_stream.read(CHUNK, exception_on_overflow=False)
                connection.send(data)
            except Exception as e:
                print("🎙️ Mic error:", e)
                break

    audio_thread = threading.Thread(target=stream_microphone)
    audio_thread.start()

    # 🧍 Keep main thread alive
    while True:
        time.sleep(1)

except Exception as e:
    print("⚠️ Exception:", e)

if name == "main":
main()

[HamzaRaouf]

we verified many times on when system needs to trigger this event 'FunctionCallRequest' , It rise exception which i mention on start of conversation , moreover i check that my system not showing logs ,
def on_function_call(client, request):
print("=====> on function call and request is :",request)

[krisdiallo]

@HamzaRaouf This seems to be an issue when using the deepgram-sdk python library.

Was having a similar issue and then switched to using raw websockets and was able to get functions running

    async def setup(self):
        """Establish WebSocket connection to Deepgram"""
        try:
            print(f"Connecting to Deepgram WebSocket: {self.ws_url}")
            self.ws = await websockets.connect(
                self.ws_url,
                additional_headers={"Authorization": f"Token {self.api_key}"},
            )
            
            # Send initial settings
            settings = self.get_agent_settings()
            print(f"Sending settings to Deepgram...")
            print(f"Tools count: {len(self.tools_dict)}")
            print(f"Tool names: {list(self.tools_dict.keys())}")
            
            await self.ws.send(json.dumps(settings))
            print("Settings sent successfully")
            return True
            
        except Exception as e:
            print(f"Failed to connect to Deepgram: {e}")
            print(f"Traceback: {traceback.format_exc()}")
            return False

🔧 FUNCTION CALL REQUEST RECEIVED
==================================================
Function name: good_bye, ID: call_A33tYyYtc0VCkw1AlQOKhfKc
Function arguments: {'farewell_type': 'general'}
⚡ Function Execution Latency: 0.000s

📤 SENDING FUNCTION RESPONSE
==================================================
Function name: good_bye
Function call ID: call_A33tYyYtc0VCkw1AlQOKhfKc
Function result: {'type': 'InjectAgentMessage', 'message': 'Thanks for calling! Have a great day.'}
Formatted response:
{
  "type": "FunctionCallResponse",
  "id": "call_A33tYyYtc0VCkw1AlQOKhfKc",
  "name": "good_bye",
  "content": "{\"type\": \"InjectAgentMessage\", \"message\": \"Thanks for calling! Have a great day.\"}"
}
✅ Response sent!
==================================================

[HamzaRaouf]

Yes ,@krisdiallo you're right , I did the same thing which works, but Python sdk is not working well with this