OpenAI Agents SDK

pip install openai-agents requests
export OPENAI_API_KEY=sk-...
export NIMA_API_KEY=nima_...

import asyncio
import os

from agents.mcp import MCPServerStreamableHttp

NIMA_MCP_URL = "https://mcp.getnima.io/mcp"


def nima_server(api_key: str, **kwargs) -> MCPServerStreamableHttp:
    return MCPServerStreamableHttp(
        name="nima",
        params={
            "url": NIMA_MCP_URL,
            "headers": {"Authorization": f"Bearer {api_key}"},
            "timeout": 70,
        },
        cache_tools_list=True,
        # nima gives a tool call up to 60 seconds. The SDK default is 5.
        client_session_timeout_seconds=70,
        **kwargs,
    )


async def main() -> None:
    api_key = os.environ["NIMA_API_KEY"]
    async with nima_server(api_key) as server:
        tools = await server.list_tools()
        for tool in tools:
            print(tool.name)


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

from agents import HostedMCPTool

nima_hosted = HostedMCPTool(
    tool_config={
        "type": "mcp",
        "server_label": "nima",
        "server_url": "https://mcp.getnima.io/mcp",
        "headers": {"Authorization": f"Bearer {api_key}"},
        "require_approval": "never",
        # The tool filter goes here, not in tool_filter.
        "allowed_tools": ["ask_knowledge_graph", "search_code"],
    }
)

import asyncio
import os
import sys

from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp, create_static_tool_filter

NIMA_MCP_URL = "https://mcp.getnima.io/mcp"

REVIEWER_INSTRUCTIONS = """\
Before writing the review, call search_code with the path or symbol of each changed area.
It returns ranked files with a one-line summary, the top symbols and their line ranges, and
per repository the indexed commit and its last sync time. Compare that commit against the one
under review and say so in the review when the index is behind. Then call ask_knowledge_graph
with the feature name: it returns markdown listing matching entities, their facts and
relationships, and unresolved conflicts. Neither tool reads the diff. Quote, do not infer.
"""


async def main(question: str) -> None:
    api_key = os.environ["NIMA_API_KEY"]
    async with MCPServerStreamableHttp(
        name="nima",
        params={
            "url": NIMA_MCP_URL,
            "headers": {"Authorization": f"Bearer {api_key}"},
            "timeout": 70,
        },
        cache_tools_list=True,
        client_session_timeout_seconds=70,
        tool_filter=create_static_tool_filter(
            allowed_tool_names=["ask_knowledge_graph", "search_code"]
        ),
    ) as server:
        agent = Agent(
            name="Reviewer",
            instructions=REVIEWER_INSTRUCTIONS,
            mcp_servers=[server],
        )
        result = await Runner.run(agent, question)
        print(result.final_output)


if __name__ == "__main__":
    asyncio.run(
        main(
            " ".join(sys.argv[1:])
            or "Review a change to CSV import chunking at commit b41c9de2."
        )
    )

{
  "tool": "search_code",
  "arguments": { "q": "where is the csv import chunked", "limit": 3 }
}

{
  "query": "where is the csv import chunked",
  "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/import/stream-reader.ts",
      "score": 0.79,
      "summary": "Reads an uploaded CSV in fixed-size chunks and hands each to the row parser.",
      "matchedChannels": ["vector", "symbol"],
      "match": "semantic",
      "symbols": [
        {
          "name": "readChunks",
          "kind": "function",
          "signature": "readChunks(file: Upload, size: number): AsyncIterable<Row[]>",
          "lineStart": 18,
          "lineEnd": 64
        }
      ]
    }
  ]
}

{
  "tool": "ask_knowledge_graph",
  "arguments": { "q": "why did we drop the legacy importer", "limit": 2 }
}

  Knowledge graph results for: "why did we drop the legacy importer"

  ### Legacy importer (feature) [implemented] · id: 9a2e4c61-0b7d-4f3a-8e15-6c9d2b7f4a08
  References: 14 | Similarity: 0.781
  Summary: First-generation CSV importer, replaced by the streaming importer in March.
  Facts:
    - status: removed (source: SPEC:7d1c3e95-4a26-4b8f-9f02-e6a15b3c8d47, learned 2026-03-11)
    - replaced_by: streaming importer (source: PR:5c7d9e13-2a4b-4e6f-8c1d-3b5a7e9f0c22, learned 2026-03-19)
  ⚠ contradicted: owner "Platform" (SPEC:7d1c3e95-4a26-4b8f-9f02-e6a15b3c8d47) vs "Data" (PR:5c7d9e13-2a4b-4e6f-8c1d-3b5a7e9f0c22), 2026-04-02

import asyncio
import os

from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp, create_static_tool_filter

NIMA_MCP_URL = "https://mcp.getnima.io/mcp"


async def main() -> None:
    api_key = os.environ["NIMA_API_KEY"]  # must carry signals:write
    async with MCPServerStreamableHttp(
        name="nima",
        params={
            "url": NIMA_MCP_URL,
            "headers": {"Authorization": f"Bearer {api_key}"},
            "timeout": 70,
        },
        client_session_timeout_seconds=70,
        # Keep the default of 0 retries: a retried create files a second signal.
        max_retry_attempts=0,
        tool_filter=create_static_tool_filter(allowed_tool_names=["create_signal"]),
    ) as server:
        agent = Agent(
            name="Filer",
            instructions=(
                "File exactly one signal with create_signal. Put the finding in the "
                "reporter's words in content, where it came from in sourceLabel, and "
                "the affected customers and product areas in stakeholders and "
                "productAreas. Report the id nima returns."
            ),
            mcp_servers=[server],
        )
        result = await Runner.run(
            agent,
            "Northwind reports the weekly CSV import dies around 40k rows with a 504. "
            "They split the file by hand. Heard on a customer call.",
        )
        print(result.final_output)


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

{
  "tool": "create_signal",
  "arguments": {
    "content": "On the Northwind call: the weekly CSV import dies around 40k rows with a 504. They are splitting the file by hand.",
    "sourceLabel": "customer call",
    "stakeholders": ["Northwind"],
    "productAreas": ["import"]
  }
}

import asyncio
import json
import os

import requests

from agents import Agent, Runner, function_tool

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


def _headers() -> dict[str, str]:
    return {"Authorization": f"Bearer {os.environ['NIMA_API_KEY']}"}


@function_tool
def ask_knowledge_graph(q: str, limit: int = 8) -> str:
    """Ask nima why something is the way it is and who asked for it.

    Args:
        q: The question in plain English, 2 to 500 characters.
        limit: How many entities to return, 1 to 20.
    """
    response = requests.get(
        f"{NIMA_API}/v1/knowledge-graph/ask",
        headers=_headers(),
        params={"q": q, "limit": limit},
        timeout=70,
    )
    response.raise_for_status()
    # Markdown, not JSON.
    return response.text


@function_tool
def search_code(q: str, limit: int = 8) -> str:
    """Find where something is implemented in the indexed repositories.

    Args:
        q: What to find, in plain English or as an identifier, 2 to 300 characters.
        limit: File hits to return, at most 20.
    """
    response = requests.get(
        f"{NIMA_API}/v1/code-intel/search",
        headers=_headers(),
        params={"q": q, "limit": limit},
        timeout=70,
    )
    response.raise_for_status()
    return json.dumps(response.json())


@function_tool
def create_signal(content: str, source_label: str, product_areas: list[str]) -> str:
    """File one signal in nima. Needs a key with signals:write.

    Args:
        content: The finding in the reporter's own words, at most 10,000 characters.
        source_label: Where it came from, at most 100 characters.
        product_areas: What it touches, at most 10 entries.
    """
    response = requests.post(
        f"{NIMA_API}/v1/signals",
        headers=_headers(),
        json={
            "content": content,
            "sourceLabel": source_label,
            "productAreas": product_areas,
        },
        timeout=70,
    )
    response.raise_for_status()
    return json.dumps({"id": response.json()["id"]})


async def main() -> None:
    agent = Agent(
        name="Reviewer",
        instructions=(
            "Call search_code for each changed area and ask_knowledge_graph for the "
            "feature before answering. Quote, do not infer."
        ),
        tools=[ask_knowledge_graph, search_code],
    )
    result = await Runner.run(agent, "Review a change to CSV import chunking.")
    print(result.final_output)


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