	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with Firestore

MCP Server with Firestore

Author: Venkata Sudhakar

Firestore is a serverless NoSQL document database from Google Cloud that organises data into collections and documents. ADK agents that need to store conversation context, user preferences, or business records can benefit from a dedicated MCP server that wraps Firestore operations as tools, keeping database logic separate from agent logic.

In this tutorial, you will build an MCP server that exposes Firestore CRUD operations - create document, get document, update document, delete document, and query collection. An ADK agent then uses these tools to store and retrieve structured data on demand.

The MCP server below uses the google-cloud-firestore library. Documents are returned as JSON strings so the agent can parse and reason over them easily.

# firestore_server.py
import asyncio
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from google.cloud import firestore

app = Server("firestore-mcp-server")
db = firestore.Client()

@app.list_tools()
async def list_tools():
    return [
        Tool(name="fs_create_document", description="Create a Firestore document",
            inputSchema={"type": "object", "properties": {
                "collection": {"type": "string"},
                "document_id": {"type": "string"},
                "data": {"type": "object"}
            }, "required": ["collection", "document_id", "data"]}),
        Tool(name="fs_get_document", description="Get a Firestore document by ID",
            inputSchema={"type": "object", "properties": {
                "collection": {"type": "string"},
                "document_id": {"type": "string"}
            }, "required": ["collection", "document_id"]}),
        Tool(name="fs_update_document", description="Update fields in a Firestore document",
            inputSchema={"type": "object", "properties": {
                "collection": {"type": "string"},
                "document_id": {"type": "string"},
                "data": {"type": "object"}
            }, "required": ["collection", "document_id", "data"]}),
        Tool(name="fs_delete_document", description="Delete a Firestore document",
            inputSchema={"type": "object", "properties": {
                "collection": {"type": "string"},
                "document_id": {"type": "string"}
            }, "required": ["collection", "document_id"]}),
        Tool(name="fs_query_collection", description="Query documents in a collection",
            inputSchema={"type": "object", "properties": {
                "collection": {"type": "string"},
                "field": {"type": "string"},
                "operator": {"type": "string"},
                "value": {"type": "string"}
            }, "required": ["collection", "field", "operator", "value"]})
    ]

@app.call_tool()
async def call_tool(name, arguments):
    col = arguments["collection"]
    doc_id = arguments.get("document_id")
    if name == "fs_create_document":
        db.collection(col).document(doc_id).set(arguments["data"])
        return [TextContent(type="text", text="Document created: " + doc_id)]
    elif name == "fs_get_document":
        doc = db.collection(col).document(doc_id).get()
        if doc.exists:
            return [TextContent(type="text", text=json.dumps(doc.to_dict()))]
        return [TextContent(type="text", text="Document not found")]
    elif name == "fs_update_document":
        db.collection(col).document(doc_id).update(arguments["data"])
        return [TextContent(type="text", text="Document updated: " + doc_id)]
    elif name == "fs_delete_document":
        db.collection(col).document(doc_id).delete()
        return [TextContent(type="text", text="Document deleted: " + doc_id)]
    elif name == "fs_query_collection":
        docs = db.collection(col).where(arguments["field"], arguments["operator"], arguments["value"]).stream()
        results = [d.to_dict() for d in docs]
        return [TextContent(type="text", text=json.dumps(results))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

The ADK agent below connects to the Firestore MCP server and uses it to store and retrieve customer records. The agent uses natural language to drive document operations without the caller needing to know any Firestore API details.

# firestore_agent.py
import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai.types import Content, Part

async def main():
    tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command="python", args=["firestore_server.py"]
        )
    )
    agent = LlmAgent(
        model="gemini-2.0-flash",
        name="firestore_agent",
        instruction="You manage customer records in Firestore. Use the available tools to create, read, update and delete documents.",
        tools=tools
    )
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name="fs_app", user_id="user1")
    runner = Runner(agent=agent, app_name="fs_app", session_service=session_service)
    response = runner.run(
        user_id="user1",
        session_id=session.id,
        new_message=Content(parts=[Part(text="Create a customer record for Alice with email alice@example.com and tier gold")])
    )
    async for event in response:
        if event.is_final_response():
            print(event.content.parts[0].text)
    await exit_stack.aclose()

asyncio.run(main())

Firestore MCP servers work well for agents that need persistent state across sessions. You can extend this server with batch write tools, subcollection support, or real-time listener patterns depending on your use case.

Send your comments, suggestions or queries regarding this site to [email protected].