victorious-monitor-50234
08/14/2025, 11:41 AMtall-airline-16756
08/14/2025, 11:41 AMtall-airline-16756
08/14/2025, 11:41 AMpip install livekit-agents[mcp]~=1.2
2. Import MCP Module: Next, import the MCP-related classes in your agent file:
from livekit.agents import mcp
3. Set Up MCP Server Connection: Create an MCPServerHTTP
instance using your MCP server URL:
mcp_server_url = os.environ.get("YOUR_MCP_SERVER_URL")
mcp_server = mcp.MCPServerHTTP(url=mcp_server_url)
4. Pass MCP Server to AgentSession: When you initialize the AgentSession
, just include the MCP server like this:
session = AgentSession(
# Other session parameters
mcp_servers=[mcp_server]
)
Here’s a quick example to help you see it all in action:
import os
from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, cli, mcp
from livekit.plugins import deepgram, openai, silero
class MyAgent(Agent):
def __init__(self):
super().__init__(
instructions="You are a helpful assistant using an MCP server.",
stt=deepgram.STT(),
llm=openai.LLM(model="gpt-4o"),
tts=openai.TTS(),
vad=silero.VAD.load(),
allow_interruptions=True
)
async def entrypoint(ctx: JobContext):
mcp_server_url = os.environ.get("YOUR_MCP_SERVER_URL")
mcp_server = mcp.MCPServerHTTP(url=mcp_server_url)
session = AgentSession(
vad=silero.VAD.load(),
stt=deepgram.STT(),
llm=openai.LLM(model="gpt-4o"),
tts=openai.TTS(),
mcp_servers=[mcp_server]
)
await session.start(agent=MyAgent(), room=ctx.room)
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))
Just a quick tip: make sure your MCP server URL is stored securely, like in a .env
file:
YOUR_MCP_SERVER_URL=<https://your-mcp-server.com>
If you have any questions or need a hand with anything else, feel free to ask—I’m here to help!victorious-monitor-50234
08/14/2025, 11:46 AMtall-airline-16756
08/14/2025, 11:46 AMvictorious-monitor-50234
08/14/2025, 11:48 AMtall-airline-16756
08/14/2025, 11:48 AMMCPSEVER:NOT_GIVEN
usually means the MCP server configuration is missing or hasn’t been set up quite right. No worries though, here’s a simple way to fix it:
1. Make sure the MCP Dependency is Installed: First, check that the MCP package is installed by running:
pip install livekit-agents[mcp]
2. Provide MCP Server Details: When you create the AgentSession
, be sure to include the MCP server info like this:
from livekit.agents import mcp
session = AgentSession(
mcp_servers=[
mcp.MCPServerHTTP(url="<https://your-mcp-server.com>")
]
)
3. Double-check the MCP Server URL: It’s a good idea to verify that the URL and any credentials you’re using are correct.
4. Optional Setup: If the MCP server isn’t required for your setup, just make sure you’re not using any features that depend on it.
Here’s a quick example if you want to handle MCP configuration optionally:
mcp_servers = []
if os.getenv("MCP_SERVER_URL"):
mcp_servers.append(mcp.MCPServerHTTP(url=os.getenv("MCP_SERVER_URL")))
session = AgentSession(
mcp_servers=mcp_servers,
# Other configurations
)
Feel free to reach out if you want me to walk you through this or if you hit any other bumps along the way!victorious-monitor-50234
08/14/2025, 11:50 AMtall-airline-16756
08/14/2025, 11:50 AMzsh: no matches found: livekit-agents[mcp]
happens because zsh thinks the square brackets []
are special characters for matching filenames. To fix this, you just need to put the package name in quotes.
Here are a few ways you can do it:
• Using double quotes:
pip install "livekit-agents[mcp]"
• Using single quotes:
pip install 'livekit-agents[mcp]'
• Or by escaping the brackets:
pip install livekit-agents\[mcp\]
If you run into any other issues or have questions, just let me know—I'm here to help!
Sources: Tool definition and use | LiveKit Docs | livekit-agents/livekit/agents/py.typedvictorious-monitor-50234
08/14/2025, 11:54 AMfrom typing import Any
import httpx
import datetime
from mcp.server.fastmcp import FastMCP
from genralscraper import get_data
from fastscraper import scraper
from googlecal import calendar_details, list_events, create_event ,Event
mcp = FastMCP("GoogleMCP")
@mcp.tool()
async def search(query: str) -> Any:
"""
This tool is used to perform web search and retrieve data.
:param query: string
:return: any
"""
return await get_data(query)
@mcp.tool()
async def scrape(url: list) -> Any:
"""
This function can be used to scrape a list of urls.
:param url: list
:return: any
"""
try:
return await scraper(url)
except Exception as e:
return f"Error occurred: {e}"
@mcp.tool()
async def get_current_time() -> str:
"""
This function returns the current date, time and day.
:return: string
"""
try:
current_datetime = datetime.datetime.now()
friendly_format = current_datetime.strftime("%A, %B %d, %Y at %I:%M %p")
return friendly_format
except Exception as e:
return f"Error occurred: {e}"
@mcp.tool()
async def get_calendar_events(time,maxresults):
"""
This function lists the upcoming events on your calendar.
:param time: Needs a time in ISO format. For example: 2023-01-01T00:00:00Z.
:param maxresults: Number of events to be returned. For example: 10.
:return: list of events.
"""
try:
events = list_events(time,maxresults)
return events
except Exception as e:
return f"Error occurred: {e}"
@mcp.tool()
async def create_event_in_calendar(event)-> str:
'''
This function creates an event on your calendar.
Example Input Schema:
"""
Example Input Schema:
{
"summary": "Test Event",
"location": "800 Howard St., San Francisco, CA 94103",
"description": "A chance to hear more about Google's developer products.",
"start": {
"dateTime": "2025-08-07T11:00:00+05:30",
"timeZone": "Asia/Kolkata"
},
"end": {
"dateTime": "2025-08-07T12:00:00+05:30",
"timeZone": "Asia/Kolkata"
},
"attendees": [
{"email": "<mailto:lpage@example.com|lpage@example.com>"},
{"email": "<mailto:sbrin@example.com|sbrin@example.com>"}
],
"reminders": {
"useDefault": false,
"overrides": [
{"method": "email", "minutes": 1440},
{"method": "popup", "minutes": 10}
]
}
}
"""
:param event: Event object.
:return: String containing the link to the event.
'''
try:
return create_event(event)
except Exception as e:
return f"Error occurred: {e}"
@mcp.tool()
async def get_calendar_details():
""""
This function returns the details of your calendar.
"""
try:
return calendar_details()
except Exception as e:
return f"Error occurred: {e}"
if __name__ == "__main__":
mcp.settings.host = "127.0.0.1" # or "0.0.0.0"
mcp.settings.port = 8001
mcp.settings.streamable_http_path = "/mcp" # default is /mcp
# Start Streamable HTTP transport
mcp.run(transport="streamable-http")
tall-airline-16756
08/14/2025, 11:54 AM@mcp.tool()
decorator to define your tools and run the server with mcp.run(transport="streamable-http")
, which fits nicely with what LiveKit expects.
• Functionality: The tools you’ve put together—like web search, scraping, and calendar events—are set up solidly and should work smoothly when LiveKit agents access them.
• Configuration: Running the server on 127.0.0.1
at port 8001
and exposing /mcp
is a pretty standard and good setup.
To keep things running smoothly, here are a few friendly reminders:
1. Make sure the MCP server URL (like YOUR_MCP_SERVER_URL
) is correctly passed to your LiveKit agent.
2. Double-check that all required dependencies (like FastMCP) are installed and up-to-date.
3. Try testing the server endpoints on their own to see if they respond as expected.
If you hit any bumps or have questions while integrating, just give me a shout—I’m here to help!victorious-monitor-50234
08/14/2025, 12:03 PMtall-airline-16756
08/14/2025, 12:04 PM@mcp.tool()
decorator. If they’re not registered, they won’t show up in the list.
4. Empty Tool List: If the server’s list_tools()
method is coming back empty, that means no tools are being exposed. Let’s ensure the server is sharing those tools correctly.
5. Cache Issues: Sometimes caching can cause trouble. If your server caches tools, check that the cache isn’t outdated or corrupted. If needed, try clearing or invalidating it.
6. Logs: Take a look at both the MCP server and agent logs for any error messages related to loading tools or connection problems. These logs often hold the key to what’s going on.
To dig a little deeper, you might want to:
• Test the MCP server on its own to see if it’s exposing tools properly.
• Check your agent’s connection to the MCP server and scan the logs for any red flags.
• Make sure the tools are correctly defined and exposed in your MCP server code.
If you’re still stuck after trying these, feel free to share more details about your setup—like your MCP server configuration, how you’re connecting, and any error messages you’re seeing in the logs. I’m here to help you get to the bottom of this!
Sources: examples/voice_agents/zapier_mcp_integration.py | README.mdvictorious-monitor-50234
08/14/2025, 12:06 PMtall-airline-16756
08/14/2025, 12:06 PMon_metrics_collected
event in LiveKit fires whenever new metrics are gathered during an active session. It’s a great way to keep an eye on your agent’s performance and usage in real-time.
Here’s how it works:
• Event Subscription: Just subscribe to this event with session.on("metrics_collected", on_metrics_collected)
. This way, your function will run every time new metrics come in.
• Logging Metrics: You can use the handy metrics.log_metrics()
function to log the metrics in a neat, structured way for easy analysis.
• Aggregating Metrics: The UsageCollector
class helps you gather metrics throughout the session, so you can track API usage and even estimate costs.
Here’s a quick example:
from livekit.agents import metrics, AgentSession
usage_collector = metrics.UsageCollector()
@session.on("metrics_collected")
def _on_metrics_collected(ev):
metrics.log_metrics(ev.metrics)
usage_collector.collect(ev.metrics)
async def log_usage():
summary = usage_collector.get_summary()
print(f"Usage: {summary}")
# Add a shutdown callback to log the summary
ctx.add_shutdown_callback(log_usage)
This event is super helpful for keeping tabs on performance, summarizing usage, and making your agent run even better. If you want me to explain anything in more detail, just let me know!victorious-monitor-50234
08/14/2025, 12:23 PMtall-airline-16756
08/14/2025, 12:25 PMfrom livekit.agents import mcp
mcp_servers = [
mcp.MCPServerHTTP("YOUR_MCP_SERVER_URL")
]
2. List Tools: You can grab the tools from the MCP servers using the list_tools()
method. Here’s a handy function you can use:
async def list_mcp_tools(agent):
activity = agent._get_activity_or_raise()
mcp_servers = activity.mcp_servers
if not mcp_servers:
return []
all_tools = []
for server in mcp_servers:
if not server.initialized:
await server.initialize()
tools = await server.list_tools()
all_tools.extend(tools)
return all_tools
3. Check Initialization: Just a quick reminder to make sure your MCP servers are initialized before you list tools. Like this:
if not server.initialized:
await server.initialize()
4. Debugging: If you don’t see any tools listed, try checking the server logs. Also, double-check that your tools are properly registered with the @mcp.tool()
decorator in your MCP server code.
Feel free to reach out if you want me to help you troubleshoot or if you have any other questions!
Sources: README.md