Google ADK Integration#

Integrate FirstOps with the Google Agent Development Kit (ADK) to govern your Google ADK agents.

Overview#

Google ADK agents connect to external tools via MCP. FirstOps sits between the agent and those tools, providing identity, policy enforcement, credential brokering, and audit.

Setup#

1. Create an agent principal#

Create an autonomous agent principal in the FirstOps dashboard. You receive an agent ID and a DPoP private key PEM file.

2. Register connections#

fo connection add --name notion --url https://mcp.notion.com/mcp --auth bearer --token ntn_secret_... --agent <agent-name>

Note the connection ID returned — you need it in your agent code.

3. Install dependencies#

pip install google-adk firstops

4. Configure and run the agent#

import asyncio
import os
import firstops
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
from google.genai import types as genai_types

PROXY_PORT = 9322
NOTION_CONN_ID = "<notion-conn-id>"
GATEWAY_URL = os.environ.get("FIRSTOPS_GATEWAY_URL", "https://api.firstops.dev")

# Start the FirstOps proxy — handles DPoP authentication per request
firstops.init(
    agent_id="<uuid>",
    private_key_pem=open("key.pem").read(),
    port=PROXY_PORT,
    gateway_url=GATEWAY_URL,
)


async def run():
    # Connect to Notion MCP server through the FirstOps governance proxy
    notion_toolset = McpToolset(
        connection_params=StreamableHTTPConnectionParams(
            url=f"http://127.0.0.1:{PROXY_PORT}/mcp/proxy/{NOTION_CONN_ID}",
        ),
    )

    # Create an ADK agent with the governed MCP tools
    agent = LlmAgent(
        model="gemini-2.0-flash",
        name="my_adk_agent",
        instruction="You are a helpful assistant with access to Notion.",
        tools=[notion_toolset],
    )

    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="my-app", user_id="user-1",
    )
    runner = Runner(
        app_name="my-app", agent=agent, session_service=session_service,
    )

    message = genai_types.Content(
        role="user",
        parts=[genai_types.Part(text="Summarize the latest project updates from Notion")],
    )

    async for event in runner.run_async(
        user_id="user-1", session_id=session.id, new_message=message,
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    print(part.text)

    await notion_toolset.close()


asyncio.run(run())

# Clean up on shutdown
firstops.shutdown()

Every tool call flows through FirstOps:

A fresh DPoP proof is generated per request
Policies are evaluated against the request content
Upstream credentials are injected by the proxy
The full interaction is logged in the audit trail

What gets enforced#

All policies attached to the agent's access group apply:

Deny rules: Dangerous operations are blocked before reaching the upstream tool
Modify rules: PII is redacted before reaching upstream tools
Flag rules: Suspicious patterns are logged for review
Outbound set: Requests to tools outside the permitted set are blocked