Build an MCP to connect AI to Oracle Database w/OpenAPI

Model Context Protocols (MCP) became a thing only a few months ago, so basically a baby in terms of IT terms, but already an adolescent when it comes to all things AI. And more importantly, vendors are all in a rush to adopt the protocol and publish solutions/support for it.

An MCP allows for a bridge between your AI assistants and your applications or services.

So if you provide an MCP for your Oracle Database, then suddenly your Prompt buddy can suddenly source information from there in addition to what it’s been trained on.

One path to your Oracle Database is via REST APIs, and the way to communicate REST APIs is via the OpenAPI spec, which MCP happily supports.

Credit: this post and much of the code has been inspired and borrowed from Nafiul Khan’s post on Medium.

In this post, I’ll show you how I wrote a simple MCP server that gets its list of tools from an Oracle REST Data Services’s OpenAPI doc, and how I can interact with it via HTTP POST.

Tools are the commands you can use to interface with your remote system, in this case our REST APIs.

On the left our OpenAPI spec for our REST APIs, and on the right, our MCP’s list of imported tools.

Most of this post will be showing you how to get your local python program running to define these tools and how to service requests for them by your AI Assistant.

Pre-requisites

python

Make sure you have python installed, a recent one. I’m using 3.13.

While Nafiul used the uv (docs) package manager, I’m more comfortable in VS Code, so my program and runtime environment will be managed by VS Code.

python libraries

Bash

from contextlib import asynccontextmanager
from fastapi import FastAPI, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Dict, Any, Optional
import httpx

The big ones here are httpx and fastapi, which together allow us to retrieve the OpenAPI spec from ORDS, and then serve back the tools via our MCP at localhost:8000/tools.

Oracle Database, some REST APIs, and their associated OpenAPI spec.

I’ll be using my Always Free Oracle Autonomous Database, where ORDS is hosted as well, and my data/REST APIs are described here.

I will have SECURITY disabled to make the example easier to follow, but adding in an OAuth2 workflow is drop dead simple in Python, and you can see how to do that, here.

Let’s jump in

Our MCP server is going to read in the OpenAPI spec that ORDS services up for each ORDS Module you have activated. The link and it’s content looks like this –

https://thecloud-oraclecloudapps.com/ords/admin/open-api-catalog/strava/

ORDS. spits out JSON version, but editor.swagger.io happily converts to YAML if you’re in that club.

Oracle Database REST API OpenAPI doc (served up from Oracle REST Data Services), and serve up all the ‘paths’ as tools.

Paths? What Paths?

OpenAPI shows us our templates or resources as an JSON paths object.

So we have

/activities/
/activitities/{id}
/intervals/
/testing/

So how can we do this? With less than 100 lines of Python!

Creating our MCP Server, and Importing the OpenAPI

We’ll start by hardcoding the OpenAPI spec endpoint, and then we’ll use the httpx client to go get that resource, and then we’ll loop over the JSON to pull out the APIs.

I had to mess around a bit to come up with reasonable tool names, that’s what you see on lines 19 and 21.

Python

OPENAPI_URL = "https://thecloud-oraclecloudapps.com/ords/admin/open-api-catalog/strava/"

async def fetch_openapi_spec():
    async with httpx.AsyncClient() as client:
        response = await client.get(OPENAPI_URL)
        response.raise_for_status()
        return response.json()
    
def generate_tools_from_openapi(openapi: dict[str, any]):
    paths = openapi.get("paths", {})
    for path, methods in paths.items():
        # Skip if methods is not a dictionary (e.g., a list)
        if not isinstance(methods, dict):
            continue
        for method, details in methods.items():
            # Skip if details is not a dictionary (e.g., a list)
            if not isinstance(details, dict):
                continue
            # here comes the tricky bits - my OpenAPI spec will be used to take the HANDLER and add a prefix to the URI template, and we're going to replace a / with a _ 
            
            operation_id = f"{method}_strava_{path.lstrip('/').rstrip('/').replace('/', '_')}"

            summary = details.get("summary")
            # Optional: Process summary if needed, e.g., print or store it

            # Create a basic tool function with a name and HTTP method
            def make_tool(p, m):
                async def tool_func(input_data):
                    headers = input_data.get("headers", {})
                    body = input_data.get("body", None)
                    params = input_data.get("params", None)
                    params = input_data.get("params", {})
                    formatted_path = p
                    for key, value in params.items():
                        formatted_path = formatted_path.replace(f"{{{key}}}", value)
                    url = f"https://thecloud-.oraclecloudapps.com/ords/admin/strava{formatted_path}"
                    async with httpx.AsyncClient() as client:
                        req = client.build_request(m.upper(), url, headers=headers, json=body, params=params)
                        res = await client.send(req)
                        return {"status_code": res.status_code, "body": res.text}
                return tool_func

            tool_registry[operation_id] = make_tool(path, method)

And our FastAPI code to setup our local webserver –
http:localhost:8000

Python

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup logic
    print("Starting up...")
    openapi = await fetch_openapi_spec()
    generate_tools_from_openapi(openapi)
    print(f"Registered tools: {list(tool_registry.keys())}")
    yield
    print("Shutting down...")

    # Optional: Add shutdown logic here if needed

app = FastAPI(lifespan=lifespan)

And then our webserver needs some endpoints to answer for, let’s start with the /tools resource, which is accessed via a GET

Python

@app.get("/tools")
async def list_tools():
    return JSONResponse(content={"tools": list(tool_registry.keys())})

You can actually start to run the app from here, you should see something similar to this –

And now the POST handler for the /invoke API. This is the part of the server that handles your requests to use a tool.

Python

@app.post("/invoke")
async def invoke_tool(call: ToolCallInput, q: str = (Query(None))):
    tool_name = call.tool_name
    input_data = call.input
    print(input_data)
    if q:
        intput_data["query"] = q
    print(f"Tool: {tool_name}, Input: {input_data}, Query: {q}")
    if tool_name not in tool_registry:
        return JSONResponse(status_code=404, content={"error": "Tool not found"})
    tool_func = tool_registry[tool_name]
    try:
        result = await tool_func(input_data)
        return JSONResponse(content={"output": result})
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})

We’ll talk about lines 2 and 6-8 in a bit.

Trying it out, via my API tool (Insomnia)

While VS Code makes it easy to import MCPs, I’m still in debug mode, so I can interact with the MCP server directly via HTTPS.

Let’s look at that request payload –

JSON

{
 "tool_name": "get_strava_activities",
	"input": {
    "headers": {},
    "params": {},
    "body": null
  }
}

“tool_name” is pretty straightforward, it’s the command we’re going to run, which will go and ask my external resource server (the Oracle Database via ORDS), for a list of activities.

For this to work, I need to create a ToolCallInput class, and that looks like this –

Python

class ToolCallInput(BaseModel):
    tool_name: str
    input: Optional[Dict[str, Any]] = None  # Allow null or missing

IF I had security/auth enabled, we’d find our access token in that input object.

My favorite Oracle Database REST API feature is our built-in query filtering support. (demos)

So, how do I use an AI assistant to not give me all my activities, but to search for any activity where I’m talking about running?

I’m going to be appending a query string to the end of my REST API, but for the MCP server to do that, I have to tell it to handle those.

Python

async def invoke_tool(call: ToolCallInput, q: str = (Query(None))):
...
if q:
    intput_data["query"] = q
print(f"Tool: {tool_name}, Input: {input_data}, Query: {q}")

That’s so easy, it can’t work on my first try, can it?

So let’s use the same tool, but add the query filter (which is JSON escaped) to our POST invoke request –

JSON

{
  "tool_name": "get_strava_activities",
  "input": {
    "params": {
      "q": "{\"description\":{\"$instr\":\"row\"}}"
    }
  }
}

And that looks like this –

My MCP server accepted a request from my client (pretend it’s say, VS Code and Co-Pilot), and has went and did the work in our Oracle Database, via ORDS/REST.

💡 VS Code makes it easy to add your own local MCPs.

So what does this mean?

It means I can now interact with things in my Oracle Database via AI Assistants, because I have shared tools via an MCP Server that acts as a intermediary between the AI and the database.

And, I’ve done this very quickly via existing REST APIs I’ve already published.

Additional quick wins include:

the AI assistants (client) are not talking directly to the Oracle Database
the client is going thru the mid-tier, ORDS via the REST APIs, which can and will have their own layer of security baked in – so no sharing of database credentials required!

How hard was this?

It took me about 3-4 hours, and 70% of that time was me getting used to some new classes and libraries in Python, plus a bit of messing around with figuring out how to construct the request payloads and define the classes.

I used X’s Grok AI to help me understand the code I was writing, to help me debug some errors that didn’t make sense to me (at first), and even dumb junk like accidentally importing pedantic instead of pydantic!

Could I do something even more ambitious? Oh yeah, we could hook up our AI Assistant to our Database REST APIs for managing and monitoring the database itself, so you could ask for a list of Top SQL statements, or even Clone a PDB!

How about adding 200+ tools? Here’s a quick look at what using the Database Instance management APIs looks like when you want to use Data Pump via curl/REST.

The Code

This isn’t well commented, but your favorite AI will happily describe to you in great detail, line by line, how it works.

Python

from contextlib import asynccontextmanager
from fastapi import FastAPI, Query
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Dict, Any, Optional
import httpx
app = FastAPI()

OPENAPI_URL = "https://thecloud.oraclecloudapps.com/ords/admin/open-api-catalog/strava/"

class ToolCallInput(BaseModel):
    tool_name: str
    input: Optional[Dict[str, Any]] = None  # Allow null or missing



async def fetch_openapi_spec():
    async with httpx.AsyncClient() as client:
        response = await client.get(OPENAPI_URL)
        response.raise_for_status()
        return response.json()
    
def generate_tools_from_openapi(openapi: dict[str, any]):
    paths = openapi.get("paths", {})
    for path, methods in paths.items():
        # Skip if methods is not a dictionary (e.g., a list)
        if not isinstance(methods, dict):
            continue
        for method, details in methods.items():
            # Skip if details is not a dictionary (e.g., a list)
            if not isinstance(details, dict):
                continue
            '''
            operation_id = f"{method}_{path.replace('/', '_')}"
            '''
            operation_id = f"{method}_strava_{path.lstrip('/').rstrip('/').replace('/', '_')}"

            summary = details.get("summary")
            # Optional: Process summary if needed, e.g., print or store it

            # Create a basic tool function with a name and HTTP method
            def make_tool(p, m):
                async def tool_func(input_data):
                    headers = input_data.get("headers", {})
                    body = input_data.get("body", None)
                    params = input_data.get("params", None)
                    params = input_data.get("params", {})
                    formatted_path = p
                    for key, value in params.items():
                        formatted_path = formatted_path.replace(f"{{{key}}}", value)
                    url = f"https://thecloud-.oraclecloudapps.com/ords/admin/strava{formatted_path}"
                    async with httpx.AsyncClient() as client:
                        req = client.build_request(m.upper(), url, headers=headers, json=body, params=params)
                        res = await client.send(req)
                        return {"status_code": res.status_code, "body": res.text}
                return tool_func

            tool_registry[operation_id] = make_tool(path, method)

tool_registry = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup logic
    print("Starting up...")
    openapi = await fetch_openapi_spec()
    generate_tools_from_openapi(openapi)
    print(f"Registered tools: {list(tool_registry.keys())}")
    yield
    print("Shutting down...")

    # Optional: Add shutdown logic here if needed

app = FastAPI(lifespan=lifespan)

@app.get("/tools")
async def list_tools():
    return JSONResponse(content={"tools": list(tool_registry.keys())})

@app.post("/invoke")
async def invoke_tool(call: ToolCallInput, q: str = (Query(None))):
    tool_name = call.tool_name
    input_data = call.input
    print(input_data)
    if q:
        intput_data["query"] = q
    print(f"Tool: {tool_name}, Input: {input_data}, Query: {q}")
    if tool_name not in tool_registry:
        return JSONResponse(status_code=404, content={"error": "Tool not found"})

    tool_func = tool_registry[tool_name]
    try:
        result = await tool_func(input_data)
        return JSONResponse(content={"output": result})
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})