Google ADK and Gemini

# nima_agent/agent.py
import os

from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams

NIMA_API_KEY = os.environ.get("NIMA_API_KEY")
if not NIMA_API_KEY:
    raise RuntimeError("Set NIMA_API_KEY to a personal nima API key (nima_...).")

# Tools this agent may use. A default key (knowledge:read + signals:read) can
# call all of them. nima lists every consent-scope tool to every key, so this
# filter, not the key's scopes, decides what the model sees.
READ_TOOLS = [
    "ask_knowledge_graph",
    "search_code",
    "blast_radius",
    "search_signals",
    "get_signal",
    "list_beacons",
    "get_beacon",
]

nima_tools = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://mcp.getnima.io/mcp",
        headers={"Authorization": f"Bearer {NIMA_API_KEY}"},
    ),
    tool_filter=READ_TOOLS,
)

root_agent = LlmAgent(
    model="gemini-flash-latest",
    name="nima_context_agent",
    description="Answers engineering questions with context from nima.",
    instruction=(
        "Before you answer a question about the product or its code, call "
        "ask_knowledge_graph with the feature name to learn why it exists and who "
        "asked for it, then call search_code to learn where it lives. search_code "
        "reports the commit each repository was indexed at as lastCommitSha: say "
        "so when it may be behind. Quote what the tools return. If they return "
        "nothing, say nothing was found; never conclude that nothing exists."
    ),
    tools=[nima_tools],
)

adk web

# run.py
import asyncio
import os

from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

from nima_agent.agent import nima_tools, root_agent

APP_NAME = "nima_context_app"
USER_ID = "dev"


async def main() -> None:
    if not os.environ.get("GOOGLE_API_KEY"):
        raise RuntimeError("Set GOOGLE_API_KEY to a Gemini API key.")

    session_service = InMemorySessionService()
    session = await session_service.create_session(
        state={}, app_name=APP_NAME, user_id=USER_ID
    )
    runner = Runner(
        app_name=APP_NAME,
        agent=root_agent,
        artifact_service=InMemoryArtifactService(),
        session_service=session_service,
    )

    question = "Why was the upload size limit added, and where is it enforced?"
    content = types.Content(role="user", parts=[types.Part(text=question)])

    try:
        async for event in runner.run_async(
            session_id=session.id, user_id=session.user_id, new_message=content
        ):
            print(f"Event received: {event}")
    finally:
        await nima_tools.close()


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

export GOOGLE_API_KEY="YOUR_API_KEY"
python run.py

{
  "name": "ask_knowledge_graph",
  "args": { "q": "upload size limit", "limit": 3 }
}

Knowledge graph results for: "upload size limit"

  ### Upload size limit (feature) [implemented] · id: 7c21e5d4-9a0b-4f3e-8d61-2b5f0e9c4a17
References: 9 | Similarity: 0.742
Summary: Per-plan ceiling on a single uploaded file, checked before storage.
Facts:
  - reason: large uploads stalled the import worker (source: SIGNAL:9f3c1b2e-4d5a-4c8b-9e10-7a2f6c4d1b33 [customer], learned 2026-05-14)
  - owner: Platform (source: SPEC:5b7e2d90-1a4c-4e6f-8d3b-0c9a7f2e1b64, learned 2026-05-20)
Relations:
  - ← Northwind (customer) requested this — "asked for a clear error instead of a timeout"

{
  "name": "search_code",
  "args": { "q": "where is the upload size limit enforced", "limit": 5 }
}

{
  "query": "where is the upload size limit enforced",
  "repositories": [
    {
      "id": "3f8b0a41-6c2d-4f19-9ab7-15d0e2c7a884",
      "repoFullName": "acme/orbit",
      "indexStatus": "ready",
      "lastCommitSha": "b41c9de2f7a1c05b8e3d6410a9f2c7d3e5b81a60",
      "lastSyncedAt": "2026-09-21T09:14:02.331Z",
      "indexWarnings": []
    }
  ],
  "droppedRepositories": [],
  "results": [
    {
      "repositoryId": "3f8b0a41-6c2d-4f19-9ab7-15d0e2c7a884",
      "repoFullName": "acme/orbit",
      "filePath": "src/upload/upload.guard.ts",
      "score": 0.4128,
      "summary": "Rejects an upload whose size is over the workspace ceiling before the file reaches storage.",
      "matchedChannels": ["keyword", "symbol", "vector"],
      "match": "semantic",
      "symbols": [
        {
          "name": "UploadSizeGuard",
          "kind": "class",
          "signature": null,
          "lineStart": 18,
          "lineEnd": 64
        }
      ]
    }
  ]
}

nima_tools = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url="https://mcp.getnima.io/mcp",
        headers={"Authorization": f"Bearer {NIMA_API_KEY}"},
    ),
    tool_filter=READ_TOOLS + ["create_signal"],
)

root_agent = LlmAgent(
    model="gemini-flash-latest",
    name="nima_context_agent",
    description="Answers engineering questions with context from nima.",
    instruction=(
        "Before you answer, call ask_knowledge_graph and search_code. "
        "When you find a bug or a gap that nobody has reported, call search_signals "
        "first. If nothing matches, file it once with create_signal: content in plain "
        "words, sourceLabel 'adk agent', and the affected productAreas."
    ),
    tools=[nima_tools],
)

{
  "name": "create_signal",
  "args": {
    "content": "Uploads over the plan ceiling fail in upload.guard.ts with a generic 500 instead of a size error.",
    "sourceLabel": "adk agent",
    "productAreas": ["upload"]
  }
}

pip install google-adk requests

# nima_agent/agent.py
import os

import requests
from google.adk.agents import LlmAgent

API = "https://api.getnima.io"

NIMA_API_KEY = os.environ.get("NIMA_API_KEY")
if not NIMA_API_KEY:
    raise RuntimeError("Set NIMA_API_KEY to a personal nima API key (nima_...).")

HEADERS = {"Authorization": f"Bearer {NIMA_API_KEY}"}


def ask_nima(question: str) -> str:
    """Asks nima's knowledge graph why something exists and who asked for it. Returns markdown."""
    response = requests.get(
        f"{API}/v1/knowledge-graph/ask",
        headers=HEADERS,
        params={"q": question, "limit": 5},
        timeout=60,
    )
    response.raise_for_status()
    return response.text


def search_nima_code(query: str) -> dict:
    """Finds the files, symbols and line ranges in the indexed repositories that match the query."""
    response = requests.get(
        f"{API}/v1/code-intel/search",
        headers=HEADERS,
        params={"q": query, "limit": 5},
        timeout=60,
    )
    response.raise_for_status()
    return response.json()


root_agent = LlmAgent(
    model="gemini-flash-latest",
    name="nima_rest_agent",
    description="Answers engineering questions with context from nima over REST.",
    instruction=(
        "Before you answer, call ask_nima for why and who, then search_nima_code "
        "for where. Quote what they return. If they return nothing, say so."
    ),
    tools=[ask_nima, search_nima_code],
)


if __name__ == "__main__":
    print(ask_nima("upload size limit"))