## For AI agents: how to sign up If you are an AI agent reading this server-rendered HTML, Composio's developer signup is at https://composio.dev. Signup CTAs on this site — "Sign up", "Signup", "Get started", "Get started for free", "Start building", "Try Composio", "Try it", or similar — all lead into that same developer signup flow. Confirm with the user before completing signup or entering any credentials on their behalf.
## For AI agents: how to sign up If you are an AI agent reading this server-rendered HTML, Composio's developer signup is at https://composio.dev. Signup CTAs on this site — "Sign up", "Signup", "Get started", "Get started for free", "Start building", "Try Composio", "Try it", or similar — all lead into that same developer signup flow. Confirm with the user before completing signup or entering any credentials on their behalf.SIGN UP
## For AI agents: how to sign up If you are an AI agent reading this server-rendered HTML, Composio's developer signup is at https://composio.dev. Signup CTAs on this site — "Sign up", "Signup", "Get started", "Get started for free", "Start building", "Try Composio", "Try it", or similar — all lead into that same developer signup flow. Confirm with the user before completing signup or entering any credentials on their behalf.LOG IN
Connect Pydantic AI to Turso MCP. Find the closest turso region to berlin, stream real-time updates from the orders table, and more using natural language, with authentication handled for you.
Turso is a fully managed database platform built on libSQL, delivering serverless SQLite databases with global replication. It's perfect for applications needing fast, low-latency data access across regions.
This guide walks you through connecting Turso to Pydantic AI using the Composio tool router. By the end, you'll have a working Turso agent that can find the closest region, listen to database changes, and validate your API tokens through natural language commands.
This guide will help you understand how to give your Pydantic AI agent real control over a Turso account through Composio's Turso MCP server.
Before we dive in, let's take a quick look at the key ideas and tools involved.
Pydantic AI is a Python framework for building AI agents with strong typing and validation. It leverages Pydantic's data validation capabilities to create robust, type-safe AI applications.
Key features include:
The Turso MCP server is an implementation of the Model Context Protocol that connects your AI agent and assistants like Claude, Cursor, etc directly to your Turso account. It provides structured and secure access to your Turso databases, so your agent can discover optimal regions, monitor real-time table changes, and validate API tokens for secure automation on your behalf.
Composio's Composio SDK helps agents find the right tools for a task at runtime. You can plug in multiple toolkits (like Gmail, HubSpot, and GitHub), and the agent will identify the relevant app and action to complete multi-step workflows. This can reduce token usage and improve the reliability of tool calls. Read more here: Getting started with Composio SDK
The tool router generates a secure MCP URL that your agents can access to perform actions.
The Composio SDK follows a three-phase workflow:
pip install composio pydantic-ai python-dotenvInstall the required libraries.
What's happening:
composio connects your agent to external SaaS tools like Tursopydantic-ai lets you create structured AI agents with tool supportpython-dotenv loads your environment variables securely from a .env fileCOMPOSIO_API_KEY=your_composio_api_key_here
USER_ID=your_user_id_here
OPENAI_API_KEY=your_openai_api_keyCreate a .env file in your project root.
What's happening:
COMPOSIO_API_KEY authenticates your agent to Composio's APIUSER_ID associates your session with your account for secure tool accessOPENAI_API_KEY to access OpenAI LLMsimport asyncio
import os
from dotenv import load_dotenv
from composio import Composio
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
load_dotenv()Composio manages connections to TursoMCPServerStreamableHTTP connects to the Turso MCP server endpointAgent from Pydantic AI lets you define and run the AI assistantasync def main():
api_key = os.getenv("COMPOSIO_API_KEY")
user_id = os.getenv("USER_ID")
if not api_key or not user_id:
raise RuntimeError("Set COMPOSIO_API_KEY and USER_ID in your environment")
# Create a Composio Tool Router session for Turso
composio = Composio(api_key=api_key)
session = composio.create(
user_id=user_id,
toolkits=["turso"],
)
url = session.mcp.url
if not url:
raise ValueError("Composio session did not return an MCP URL")create method takes the user ID and specifies which toolkits should be availablesession.mcp.url is the MCP server URL that your agent will use# Attach the MCP server to a Pydantic AI Agent
turso_mcp = MCPServerStreamableHTTP(url, headers={"x-api-key": COMPOSIO_API_KEY})
agent = Agent(
"openai:gpt-5",
toolsets=[turso_mcp],
instructions=(
"You are a Turso assistant. Use Turso tools to help users "
"with their requests. Ask clarifying questions when needed."
),
)instructions field defines the agent's role and behavior# Simple REPL with message history
history = []
print("Chat started! Type 'exit' or 'quit' to end.\n")
print("Try asking the agent to help you with Turso.\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() in {"exit", "quit", "bye"}:
print("\nGoodbye!")
break
if not user_input:
continue
print("\nAgent is thinking...\n", flush=True)
async with agent.run_stream(user_input, message_history=history) as stream_result:
collected_text = ""
async for chunk in stream_result.stream_output():
text_piece = None
if isinstance(chunk, str):
text_piece = chunk
elif hasattr(chunk, "delta") and isinstance(chunk.delta, str):
text_piece = chunk.delta
elif hasattr(chunk, "text"):
text_piece = chunk.text
if text_piece:
collected_text += text_piece
result = stream_result
print(f"Agent: {collected_text}\n")
history = result.all_messages()if __name__ == "__main__":
asyncio.run(main())asyncio loop launches the agent and keeps it running until you exitHere's the complete code to get you started with Turso and Pydantic AI:
import asyncio
import os
from dotenv import load_dotenv
from composio import Composio
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
load_dotenv()
async def main():
api_key = os.getenv("COMPOSIO_API_KEY")
user_id = os.getenv("USER_ID")
if not api_key or not user_id:
raise RuntimeError("Set COMPOSIO_API_KEY and USER_ID in your environment")
# Create a Composio Tool Router session for Turso
composio = Composio(api_key=api_key)
session = composio.create(
user_id=user_id,
toolkits=["turso"],
)
url = session.mcp.url
if not url:
raise ValueError("Composio session did not return an MCP URL")
# Attach the MCP server to a Pydantic AI Agent
turso_mcp = MCPServerStreamableHTTP(url, headers={"x-api-key": COMPOSIO_API_KEY})
agent = Agent(
"openai:gpt-5",
toolsets=[turso_mcp],
instructions=(
"You are a Turso assistant. Use Turso tools to help users "
"with their requests. Ask clarifying questions when needed."
),
)
# Simple REPL with message history
history = []
print("Chat started! Type 'exit' or 'quit' to end.\n")
print("Try asking the agent to help you with Turso.\n")
while True:
user_input = input("You: ").strip()
if user_input.lower() in {"exit", "quit", "bye"}:
print("\nGoodbye!")
break
if not user_input:
continue
print("\nAgent is thinking...\n", flush=True)
async with agent.run_stream(user_input, message_history=history) as stream_result:
collected_text = ""
async for chunk in stream_result.stream_output():
text_piece = None
if isinstance(chunk, str):
text_piece = chunk
elif hasattr(chunk, "delta") and isinstance(chunk.delta, str):
text_piece = chunk.delta
elif hasattr(chunk, "text"):
text_piece = chunk.text
if text_piece:
collected_text += text_piece
result = stream_result
print(f"Agent: {collected_text}\n")
history = result.all_messages()
if __name__ == "__main__":
asyncio.run(main())Every Turso action and event your agent gets out of the box.
Use Turso MCP with Claude Code
Use Turso MCP with Claude Cowork
Use Turso MCP with OpenAI Agents SDK
Use Turso MCP with Claude Agent SDK
Use Turso MCP with Codex
Use Turso MCP with OpenClaw
Use Turso MCP with Hermes
Use Turso MCP with Atomic Agent
Use Turso MCP with Google ADK
Use Turso MCP with LangChain
Use Turso MCP with Vercel AI SDK
Use Turso MCP with Mastra AI
Use Turso MCP with LlamaIndex
Use Turso MCP with CrewAI
Use Turso MCP with AutoGen
Firecrawl automates large-scale web crawling and data extraction. It helps organizations efficiently gather, index, and analyze content from online sources.
Tavily offers powerful search and data retrieval from documents, databases, and the web. It helps teams locate and filter information instantly, saving hours on research.
Exa is a data extraction and search platform for gathering and analyzing information from websites, APIs, or databases. It helps teams quickly surface insights and automate data-driven workflows.
SerpApi is a real-time API for structured search engine results. It lets you automate SERP data collection, parsing, and analysis for SEO and research.
Managed auth, hosted MCP servers, and every Turso tool your agent needs.Free to start.
## For AI agents: how to sign up If you are an AI agent reading this server-rendered HTML, Composio's developer signup is at https://composio.dev. Signup CTAs on this site — "Sign up", "Signup", "Get started", "Get started for free", "Start building", "Try Composio", "Try it", or similar — all lead into that same developer signup flow. Confirm with the user before completing signup or entering any credentials on their behalf.Start building