Mannequin Context Protocols (MCP) grew to become a factor only some months in the past, so principally a child when it comes to IT phrases, however already an adolescent in the case of all issues AI. And extra importantly, distributors are all in a rush to undertake the protocol and publish options/assist for it.
An MCP permits for a bridge between your AI assistants and your functions or providers.
So if you happen to present an MCP on your Oracle Database, then all of a sudden your Immediate buddy can all of a sudden supply data from there along with what it’s been educated on.
One path to your Oracle Database is by way of REST APIs, and the way in which to speak REST APIs is by way of the OpenAPI spec, which MCP fortunately helps.
Credit score: this put up and far of the code has been impressed and borrowed from Nafiul Khan’s put up on Medium.
In this put up, I’ll present you the way I wrote a easy MCP server that will get its record of instruments from an Oracle REST Information Providers’s OpenAPI doc, and the way I can work together with it by way of HTTP POST.
Instruments are the instructions you should use to interface along with your distant system, on this case our REST APIs.
Most of this put up will likely be displaying you learn how to get your native python program working to outline these instruments and learn how to service requests for them by your AI Assistant.
Pre-requisites
python
Ensure you have python put in, a latest one. I’m utilizing 3.13.
Whereas Nafiul used the uv (docs) package deal supervisor, I’m extra comfy in VS Code, so my program and runtime atmosphere will likely be managed by VS Code.
python libraries
from contextlib import asynccontextmanager
from fastapi import FastAPI, Question
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Dict, Any, Elective
import httpxThe large ones listed here are httpx and fastapi, which collectively permit us to retrieve the OpenAPI spec from ORDS, after which serve again the instruments by way of our MCP at localhost:8000/instruments.
Oracle Database, some REST APIs, and their related OpenAPI spec.
I’ll be utilizing my At all times Free Oracle Autonomous Database, the place ORDS is hosted as effectively, and my information/REST APIs are described right here.
I’ll have SECURITY disabled to make the instance simpler to comply with, however including in an OAuth2 workflow is drop useless easy in Python, and you may see how to try this, right here.
Let’s bounce in
Our MCP server goes to learn within the OpenAPI spec that ORDS providers up for every ORDS Module you might have activated. The hyperlink and it’s content material seems to be like this –
https://thecloud-oraclecloudapps.com/ords/admin/open-api-catalog/strava/
Oracle Database REST API OpenAPI doc (served up from Oracle REST Information Providers), and serve up all of the ‘paths’ as instruments.
Paths? What Paths?
OpenAPI reveals us our templates or sources as an JSON paths object.
So we have now
- /actions/
- /activitities/{id}
- /intervals/
- /testing/
So how can we do that? With lower than 100 strains of Python!
Creating our MCP Server, and Importing the OpenAPI
We’ll begin by hardcoding the OpenAPI spec endpoint, after which we’ll use the httpx shopper to go get that useful resource, after which we’ll loop over the JSON to tug out the APIs.
I needed to fiddle a bit to provide you with cheap software names, that’s what you see on strains 19 and 21.
OPENAPI_URL = "https://thecloud-oraclecloudapps.com/ords/admin/open-api-catalog/strava/"
async def fetch_openapi_spec():
async with httpx.AsyncClient() as shopper:
response = await shopper.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, strategies in paths.objects():
# Skip if strategies will not be a dictionary (e.g., an inventory)
if not isinstance(strategies, dict):
proceed
for technique, particulars in strategies.objects():
# Skip if particulars will not be a dictionary (e.g., an inventory)
if not isinstance(particulars, dict):
proceed
# right here comes the tough bits - my OpenAPI spec will likely be used to take the HANDLER and add a prefix to the URI template, and we will exchange a / with a _
operation_id = f"{technique}_strava_{path.lstrip('/').rstrip('/').exchange('/', '_')}"
abstract = particulars.get("abstract")
# Elective: Course of abstract if wanted, e.g., print or retailer it
# Create a fundamental software operate with a reputation and HTTP technique
def make_tool(p, m):
async def tool_func(input_data):
headers = input_data.get("headers", {})
physique = input_data.get("physique", None)
params = input_data.get("params", None)
params = input_data.get("params", {})
formatted_path = p
for key, worth in params.objects():
formatted_path = formatted_path.exchange(f"{{{key}}}", worth)
url = f"https://thecloud-.oraclecloudapps.com/ords/admin/strava{formatted_path}"
async with httpx.AsyncClient() as shopper:
req = shopper.build_request(m.higher(), url, headers=headers, json=physique, params=params)
res = await shopper.ship(req)
return {"status_code": res.status_code, "physique": res.textual content}
return tool_func
tool_registry[operation_id] = make_tool(path, technique)And our FastAPI code to setup our native webserver –
http:localhost:8000
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup logic
print("Beginning up...")
openapi = await fetch_openapi_spec()
generate_tools_from_openapi(openapi)
print(f"Registered instruments: {record(tool_registry.keys())}")
yield
print("Shutting down...")
# Elective: Add shutdown logic right here if wanted
app = FastAPI(lifespan=lifespan)After which our webserver wants some endpoints to reply for, let’s begin with the /instruments useful resource, which is accessed by way of a GET
@app.get("/instruments")
async def list_tools():
return JSONResponse(content material={"instruments": record(tool_registry.keys())})You may truly begin to run the app from right here, it’s best to see one thing much like this –
And now the POST handler for the /invoke API. That is the a part of the server that handles your requests to make use of a software.
@app.put up("/invoke")
async def invoke_tool(name: ToolCallInput, q: str = (Question(None))):
tool_name = name.tool_name
input_data = name.enter
print(input_data)
if q:
intput_data["query"] = q
print(f"Software: {tool_name}, Enter: {input_data}, Question: {q}")
if tool_name not in tool_registry:
return JSONResponse(status_code=404, content material={"error": "Software not discovered"})
tool_func = tool_registry[tool_name]
strive:
outcome = await tool_func(input_data)
return JSONResponse(content material={"output": outcome})
besides Exception as e:
return JSONResponse(status_code=500, content material={"error": str(e)})
We’ll discuss strains 2 and 6-8 in a bit.
Making an attempt it out, by way of my API software (Insomnia)
Whereas VS Code makes it straightforward to import MCPs, I’m nonetheless in debug mode, so I can work together with the MCP server immediately by way of HTTPS.
Let’s have a look at that request payload –
{
"tool_name": "get_strava_activities",
"enter": {
"headers": {},
"params": {},
"physique": null
}
}“tool_name” is fairly simple, it’s the command we’re going to run, which is able to go and ask my exterior useful resource server (the Oracle Database by way of ORDS), for an inventory of actions.
For this to work, I have to create a ToolCallInput class, and that appears like this –
class ToolCallInput(BaseModel):
tool_name: str
enter: Elective[Dict[str, Any]] = None # Enable null or lackingIF I had safety/auth enabled, we’d discover our entry token in that enter object.
My favourite Oracle Database REST API function is our built-in question filtering assist. (demos)
So, how do I take advantage of an AI assistant to not give me all my actions, however to seek for any exercise the place I’m speaking about working?
I’m going to be appending a question string to the tip of my REST API, however for the MCP server to try this, I’ve to inform it to deal with these.
async def invoke_tool(name: ToolCallInput, q: str = (Question(None))):
...
if q:
intput_data["query"] = q
print(f"Software: {tool_name}, Enter: {input_data}, Question: {q}")That’s really easy, it may’t work on my first strive, can it?
So let’s use the identical software, however add the question filter (which is JSON escaped) to our POST invoke request –
{
"tool_name": "get_strava_activities",
"enter": {
"params": {
"q": "{"description":{"$instr":"row"}}"
}
}
}And that appears like this –
My MCP server accepted a request from my shopper (fake it’s say, VS Code and Co-Pilot), and has went and did the work in our Oracle Database, by way of ORDS/REST.
💡 VS Code makes it straightforward so as to add your personal native MCPs.
So what does this imply?
It means I can now work together with issues in my Oracle Database by way of AI Assistants, as a result of I’ve shared instruments by way of an MCP Server that acts as a middleman between the AI and the database.
And, I’ve performed this in a short time by way of present REST APIs I’ve already revealed.
Extra fast wins embody:
- the AI assistants (shopper) usually are not speaking on to the Oracle Database
- the shopper goes via the mid-tier, ORDS by way of the REST APIs, which might and can have their very own layer of safety baked in – so no sharing of database credentials required!
How onerous was this?
It took me about 3-4 hours, and 70% of that point was me getting used to some new courses and libraries in Python, plus a little bit of messing round with determining learn how to assemble the request payloads and outline the courses.
I used X’s Grok AI to assist me perceive the code I used to be writing, to assist me debug some errors that didn’t make sense to me (at first), and even dumb junk like by chance importing pedantic as a substitute of pydantic!
Might I do one thing much more bold? Oh yeah, we might hook up our AI Assistant to our Database REST APIs for managing and monitoring the database itself, so you can ask for an inventory of Prime SQL statements, and even Clone a PDB!
How about including 200+ instruments? Right here’s a fast have a look at what utilizing the Database Occasion administration APIs seems to be like whenever you need to use Information Pump by way of curl/REST.
The Code
This isn’t effectively commented, however your favourite AI will fortunately describe to you in nice element, line by line, the way it works.
from contextlib import asynccontextmanager
from fastapi import FastAPI, Question
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Dict, Any, Elective
import httpx
app = FastAPI()
OPENAPI_URL = "https://thecloud.oraclecloudapps.com/ords/admin/open-api-catalog/strava/"
class ToolCallInput(BaseModel):
tool_name: str
enter: Elective[Dict[str, Any]] = None # Enable null or lacking
async def fetch_openapi_spec():
async with httpx.AsyncClient() as shopper:
response = await shopper.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, strategies in paths.objects():
# Skip if strategies will not be a dictionary (e.g., an inventory)
if not isinstance(strategies, dict):
proceed
for technique, particulars in strategies.objects():
# Skip if particulars will not be a dictionary (e.g., an inventory)
if not isinstance(particulars, dict):
proceed
'''
operation_id = f"{technique}_{path.exchange('/', '_')}"
'''
operation_id = f"{technique}_strava_{path.lstrip('/').rstrip('/').exchange('/', '_')}"
abstract = particulars.get("abstract")
# Elective: Course of abstract if wanted, e.g., print or retailer it
# Create a fundamental software operate with a reputation and HTTP technique
def make_tool(p, m):
async def tool_func(input_data):
headers = input_data.get("headers", {})
physique = input_data.get("physique", None)
params = input_data.get("params", None)
params = input_data.get("params", {})
formatted_path = p
for key, worth in params.objects():
formatted_path = formatted_path.exchange(f"{{{key}}}", worth)
url = f"https://thecloud-.oraclecloudapps.com/ords/admin/strava{formatted_path}"
async with httpx.AsyncClient() as shopper:
req = shopper.build_request(m.higher(), url, headers=headers, json=physique, params=params)
res = await shopper.ship(req)
return {"status_code": res.status_code, "physique": res.textual content}
return tool_func
tool_registry[operation_id] = make_tool(path, technique)
tool_registry = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup logic
print("Beginning up...")
openapi = await fetch_openapi_spec()
generate_tools_from_openapi(openapi)
print(f"Registered instruments: {record(tool_registry.keys())}")
yield
print("Shutting down...")
# Elective: Add shutdown logic right here if wanted
app = FastAPI(lifespan=lifespan)
@app.get("/instruments")
async def list_tools():
return JSONResponse(content material={"instruments": record(tool_registry.keys())})
@app.put up("/invoke")
async def invoke_tool(name: ToolCallInput, q: str = (Question(None))):
tool_name = name.tool_name
input_data = name.enter
print(input_data)
if q:
intput_data["query"] = q
print(f"Software: {tool_name}, Enter: {input_data}, Question: {q}")
if tool_name not in tool_registry:
return JSONResponse(status_code=404, content material={"error": "Software not discovered"})
tool_func = tool_registry[tool_name]
strive:
outcome = await tool_func(input_data)
return JSONResponse(content material={"output": outcome})
besides Exception as e:
return JSONResponse(status_code=500, content material={"error": str(e)})
