intersystems-community/fhir-ai-hackathon-kit/iris-ai-agents

import iris, requests
from requests.auth import HTTPBasicAuth
CONNECTION_ARGS = {"hostname": "localhost", "port": 1972, "namespace": "FHIRSERVER",
                   "username": "SuperUser", "password": "SYS"}
def search_superheros() -> list:
    """Returns first name, last name and age of every superhero in Sample.Person."""
    conn = iris.connect(**CONNECTION_ARGS)
    cursor = conn.cursor()
    try:
        cursor.execute("SELECT FirstName, LastName, Age FROM Sample.Person")
        return cursor.fetchall()
    finally:
        cursor.close()
        conn.close()
def fhir_query(endpoint: str, search_params: str) -> str:
    """Query the FHIR R4 server: GET /<endpoint>?<search_params>. Returns the FHIR JSON bundle as text."""
    url = "http://localhost:52773/fhir/r4/" + endpoint + "?" + search_params
    res = requests.get(url, headers={"Accept": "application/fhir+json"},
                       auth=HTTPBasicAuth("SuperUser", "SYS"))
    res.raise_for_status()
    return res.text

from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain_openai import ChatOpenAI
load_dotenv()
model = ChatOpenAI(model="gpt-5-nano", reasoning_effort="medium")
agent = create_agent(
    model=model,
    tools=[fhir_query, search_superheros, vector_search_diabetes_text],
    system_prompt=("You are an InterSystems IRIS data agent. Use the tools to find data in IRIS "
                   "and answer only from tool results."),
)
result = agent.invoke({"messages": [HumanMessage("How tall is Tish Lemke? Use FHIR data")]})
for m in result["messages"]:
    m.pretty_print()          # shows tool calls + tool outputs, useful for debugging
answer = result["messages"][-1].content

from fastmcp import FastMCP
mcp = FastMCP("IRIS Tools")
@mcp.tool
def sql_query(query: str) -> dict:
    """Run a read-only SELECT against IRIS (namespace FHIRSERVER). Returns up to 50 rows."""
    if not query.lstrip().upper().startswith("SELECT"):
        raise ValueError("Only SELECT statements are allowed")
    conn = iris.connect(**CONNECTION_ARGS)
    cursor = conn.cursor()
    try:
        cursor.execute(query)
        rows = [list(r) for r in cursor.fetchmany(50)]
        truncated = cursor.fetchone() is not None
        return {"columns": [d[0] for d in cursor.description], "data": rows,
                "row_count": len(rows), "truncated": truncated}
    finally:
        cursor.close()
        conn.close()
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="127.0.0.1", port=8001)   # endpoint http://127.0.0.1:8001/mcp
    # or: mcp.run(transport="stdio")  - client launches the process itself

{
  "mcpServers": {
    "FHIR-Server": {"type": "http", "url": "http://localhost:8001/mcp"},
    "VectorSearch": {"type": "stdio", "command": "uv", "args": ["run", "mcp-vector-stdio.py"]}
  }
}

from langchain.agents import create_agent
from langchain.mcp import MCPAdapter      # beta API in langchain 1.x, may change
mcp_client = MCPAdapter("http://localhost:8001/mcp")
async def ask(prompt: str):
    async with mcp_client:
        tools = await mcp_client.list_tools()
        agent = create_agent(model="openai:gpt-5-nano", tools=tools)
        return await agent.ainvoke({"messages": [{"role": "user", "content": prompt}]})   # MCP tools require async