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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server with Secret Manager

MCP Server with Secret Manager

Author: Venkata Sudhakar

Hardcoding API keys, database passwords, or service credentials inside an agent is a security risk. Google Cloud Secret Manager provides a central store for secrets with fine-grained IAM access controls and automatic versioning. By exposing Secret Manager operations through an MCP server, your ADK agents can retrieve credentials at runtime without any secrets appearing in source code.

In this tutorial, you will build an MCP server with three tools - get secret, create secret, and list secrets. An ADK agent uses the get secret tool to retrieve a database password before connecting, keeping all credential logic outside the agent instruction.

The MCP server below uses the google-cloud-secret-manager library. Secret values are returned as plain strings so the agent can pass them directly to downstream tool calls.

# secretmanager_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 secretmanager

app = Server("secretmanager-mcp-server")
PROJECT_ID = "your-project-id"

@app.list_tools()
async def list_tools():
    return [
        Tool(name="sm_get_secret", description="Get the latest version of a secret",
            inputSchema={"type": "object", "properties": {
                "secret_id": {"type": "string"},
                "version": {"type": "string"}
            }, "required": ["secret_id"]}),
        Tool(name="sm_create_secret", description="Create a new secret with an initial value",
            inputSchema={"type": "object", "properties": {
                "secret_id": {"type": "string"},
                "value": {"type": "string"}
            }, "required": ["secret_id", "value"]}),
        Tool(name="sm_list_secrets", description="List all secrets in the project",
            inputSchema={"type": "object", "properties": {}, "required": []})
    ]

@app.call_tool()
async def call_tool(name, arguments):
    client = secretmanager.SecretManagerServiceClient()
    if name == "sm_get_secret":
        version = arguments.get("version", "latest")
        secret_name = f"projects/{PROJECT_ID}/secrets/{arguments["secret_id"]}/versions/{version}"
        response = client.access_secret_version(request={"name": secret_name})
        value = response.payload.data.decode("utf-8")
        return [TextContent(type="text", text=value)]
    elif name == "sm_create_secret":
        parent = f"projects/{PROJECT_ID}"
        secret = client.create_secret(request={
            "parent": parent,
            "secret_id": arguments["secret_id"],
            "secret": {"replication": {"automatic": {}}}
        })
        client.add_secret_version(request={
            "parent": secret.name,
            "payload": {"data": arguments["value"].encode("utf-8")}
        })
        return [TextContent(type="text", text="Secret created: " + arguments["secret_id"])]
    elif name == "sm_list_secrets":
        parent = f"projects/{PROJECT_ID}"
        secrets = [s.name.split("/")[-1] for s in client.list_secrets(request={"parent": parent})]
        return [TextContent(type="text", text=json.dumps(secrets))]

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 uses the Secret Manager MCP server to fetch a database password at runtime. This means no credentials are stored in the agent definition or environment variables visible in code.

# secretmanager_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=["secretmanager_server.py"]
        )
    )
    agent = LlmAgent(
        model="gemini-2.0-flash",
        name="secrets_agent",
        instruction="You help retrieve and manage application secrets from Secret Manager. Never expose secret values in explanations.",
        tools=tools
    )
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name="sm_app", user_id="user1")
    runner = Runner(agent=agent, app_name="sm_app", session_service=session_service)
    response = runner.run(
        user_id="user1",
        session_id=session.id,
        new_message=Content(parts=[Part(text="List all available secrets in the project")])
    )
    async for event in response:
        if event.is_final_response():
            print(event.content.parts[0].text)
    await exit_stack.aclose()

asyncio.run(main())

Using Secret Manager through MCP keeps credentials out of your agent code and centralises access control. You can extend the server with tools to rotate secrets, add new versions, or disable old versions as part of automated security workflows.

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