Ollama API Integration
2025-03-1828 turns66,175 charsgpt-4o, gpt-4o-mini
Summary
User debugging FastAPI MCP server integration with Ollama API, fixing endpoint routing and request parameter issues.
Messages
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import pytesseract
import whisper
import json
from bs4 import BeautifulSoup
import os
app = FastAPI()
OLLAMA_API_URL = "http://localhost:11434/api/generate" # Ollama's local API
# Define request body schema
class OllamaRequest(BaseModel):
prompt: str
model: str = "mistral"
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
payload = {"model": request.model, "prompt": request.prompt, "stream": False}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response_data = response.json()
return {"response": response_data.get("response", "No response")}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
This FastAPI app defines an endpoint `/mcp/ollama` that takes a prompt and a model name (defaulting to `"mistral"`) and sends it to a locally running Ollama instance. It then returns the generated response.
### Potential Improvements:
1. **Exception Handling:**
- Handle `requests` exceptions more gracefully (`requests.exceptions.RequestException`).
- Ensure `response_data.get("response", "No response")` works correctly if `response.json()` fails.
2. **Environment Variables for Configurations:**
- Instead of hardcoding `OLLAMA_API_URL`, use `os.getenv()` to allow flexibility.
3. **Logging for Debugging:**
- Use `logging` to capture errors rather than raising raw exceptions.
4. **Add CORS Middleware (if needed):**
- If the API is accessed from a frontend in a different domain, FastAPI's `CORSMiddleware` can help.
Would you like to extend this to support streaming responses from Ollama or add authentication?
gpt-4o
import requests
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage",
json={"url": "https://www.danielkliewer.com"}
)
print(response.json()) {
"name": "ollama_mcp",
"description": "A suite of MCP tools that extend Ollama with practical web and system utilities.",
"tools": [
{
"name": "web_search",
"description": "Performs a web search and retrieves relevant results.",
"parameters": [
{ "name": "query", "type": "string", "description": "Search query." }
],
"api_endpoint": "http://localhost:8000/mcp/web_search"
},
{
"name": "fetch_webpage",
"description": "Fetches and summarizes the content of a webpage.",
"parameters": [
{ "name": "url", "type": "string", "description": "The URL of the webpage to fetch." }
],
"api_endpoint": "http://localhost:8000/mcp/fetch_webpage"
},
{
"name": "read_file",
"description": "Reads and extracts text from a local file (TXT, JSON, PDF).",
"parameters": [
{ "name": "file_path", "type": "string", "description": "Path to the file to read." }
],
"api_endpoint": "http://localhost:8000/mcp/read_file"
},
{
"name": "ocr_image",
"description": "Extracts text from an image using OCR.",
"parameters": [
{ "name": "image_path", "type": "string", "description": "Path to the image file." }
],
"api_endpoint": "http://localhost:8000/mcp/ocr_image"
},
{
"name": "speech_to_text",
"description": "Converts audio speech to text.",
"parameters": [
{ "name": "audio_path", "type": "string", "description": "Path to the audio file." }
],
"api_endpoint": "http://localhost:8000/mcp/speech_to_text"
}
]
}
It looks like you're designing a **FastAPI-based MCP (Multi-Component Processing) system** that integrates **Ollama** with practical web and system utilities like **web scraping, OCR, speech-to-text, and file reading**. Here’s a breakdown of what you're building and some recommendations:
---
### **Current Components**
1. **Web Search (`web_search`)**
- Allows searching the web for relevant information.
- Likely requires a search API or scraping method.
2. **Fetch Webpage (`fetch_webpage`)**
- Fetches and summarizes content from a given URL.
- Could use `requests` + `BeautifulSoup` for scraping.
3. **Read File (`read_file`)**
- Reads local text-based files (TXT, JSON, PDF).
- Needs logic for handling different file formats.
4. **OCR Image (`ocr_image`)**
- Extracts text from an image using **OCR (Tesseract)**.
- Ensure `pytesseract` is correctly installed.
5. **Speech-to-Text (`speech_to_text`)**
- Converts audio files into text using **Whisper**.
- Ensure Whisper’s model is properly set up.
---
### **Key Enhancements**
1. **Ensure Proper Exception Handling**
- Each tool should return clear error messages instead of crashing the API.
2. **Use Streaming for Web Scraping & Speech-to-Text**
- If pages are large, streaming the response will improve performance.
- Whisper's transcription should handle long files without memory issues.
3. **Optimize Web Search with DuckDuckGo or Google API**
- **Google Custom Search API** (paid, but good results).
- **DuckDuckGo API** (free, but limited).
4. **Security Considerations**
- **Limit web requests** to prevent abuse.
- **Sanitize file paths** to avoid path traversal attacks.
- **Rate-limit API endpoints** to prevent overuse.
---
### **Next Steps**
- Implement each API route in FastAPI.
- Add tests to ensure robustness.
- Optionally, **containerize with Docker** for easy deployment.
Would you like help implementing any of these APIs in FastAPI? 🚀
gpt-4o
venv) venvdanielkliewer@Daniels-MacBook-Pro mcp01 % python3 test.py
{'detail': 'Not Found'}
The error message `{'detail': 'Not Found'}` indicates that the FastAPI endpoint you're trying to access isn't found. Here are a few things to check:
### 1. **Verify the FastAPI App is Running**
Make sure your FastAPI server is running. If you haven't started the server yet, you can do so with the following command:
```bash
uvicorn main:app --reload
```
Make sure you're running this from the correct directory where your FastAPI app (`main.py` or the equivalent file) is located.
### 2. **Verify the URL and Endpoint**
Check that the URL and endpoint in your `test.py` file match the route defined in your FastAPI application. In your example, you're calling:
```python
http://localhost:8000/mcp/fetch_webpage
```
Make sure your FastAPI app defines the route like this:
```python
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
# your logic here
```
If you see a mismatch, correct the URL or the FastAPI route.
### 3. **Ensure the Route Exists**
Double-check that the `"/mcp/fetch_webpage"` endpoint is defined in your FastAPI application. If it's not defined, you’ll get a `Not Found` error.
For example, the FastAPI route should look like this:
```python
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
# Fetch and process the webpage
pass
```
### 4. **Check the HTTP Method**
Ensure that you're sending a POST request from `test.py` and the endpoint is expecting a POST request. If you’re trying to access it with the wrong HTTP method (GET vs POST), it will return a `404 Not Found` or method mismatch error.
---
### Sample Test Code (`test.py`)
Ensure the test code is correct like this:
```python
import requests
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage",
json={"url": "https://www.danielkliewer.com"}
)
print(response.json())
```
### 5. **Check Server Logs for Details**
Check the FastAPI server's logs in the terminal. If there's a mismatch or any other issues with the route, the logs will show helpful information.
---
Once you confirm the app is running and routes are set up correctly, it should return a valid response instead of `{'detail': 'Not Found'}`.
Let me know if you need help debugging further!
gpt-4o-mini
venv) danielkliewer@Daniels-MacBook-Pro mcp01 % python3 test.py
{'detail': [{'type': 'missing', 'loc': ['query', 'url'], 'msg': 'Field required', 'input': None}]}
The error message you received indicates that the `url` field is missing from the request body, which is required according to the schema for your FastAPI endpoint. Specifically, the error states:
```
{'detail': [{'type': 'missing', 'loc': ['query', 'url'], 'msg': 'Field required', 'input': None}]}
```
This suggests that FastAPI expects a query parameter named `url`, but you're sending it in the body of the request.
### Solution
To fix this, ensure that the FastAPI route is expecting the `url` in the **request body** instead of the query string.
#### 1. **Check the Request Model**
First, make sure the endpoint in FastAPI is designed to accept a JSON body and not a query parameter. Here’s an example of how you can define the route:
```python
from pydantic import BaseModel
from fastapi import FastAPI
app = FastAPI()
class WebPageRequest(BaseModel):
url: str # This is the field you're sending in the body
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
url = request.url
# Logic to fetch the webpage using the URL
return {"url": url}
```
### 2. **Update Your `test.py`**
Now, in your `test.py` file, make sure you're passing the `url` in the body of the request, like this:
```python
import requests
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage",
json={"url": "https://www.danielkliewer.com"} # URL in the body
)
print(response.json())
```
### Key Points:
- The `url` should be passed in the body of the POST request.
- The FastAPI route should expect `url` as part of the request body (not as a query parameter).
- Your `WebPageRequest` model should have the `url` field.
### If You Want to Use Query Parameters
If for some reason you want to use query parameters instead of the request body, modify the FastAPI route like this:
```python
from fastapi import FastAPI, Query
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(url: str = Query(...)): # Using Query for query parameter
return {"url": url}
```
In this case, you would call it like this in `test.py`:
```python
import requests
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage?url=https://www.danielkliewer.com"
)
print(response.json())
```
---
Let me know if that works or if you need further clarification!
gpt-4o-mini
(venv) danielkliewer@Daniels-MacBook-Pro mcp01 % python3 test.py
{'detail': [{'type': 'missing', 'loc': ['query', 'url'], 'msg': 'Field required', 'input': None}]} from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import pytesseract
import whisper
import json
from bs4 import BeautifulSoup
import os
app = FastAPI()
OLLAMA_API_URL = "http://localhost:11434/api/generate" # Ollama's local API
# Define request body schema
class OllamaRequest(BaseModel):
prompt: str
model: str = "mistral"
class WebPageRequest(BaseModel):
url: str # This is the field you're sending in the body
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
payload = {"model": request.model, "prompt": request.prompt, "stream": False}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response_data = response.json()
return {"response": response_data.get("response", "No response")}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/web_search")
async def web_search(query: str):
search_url = f"https://www.google.com/search?q={query}"
try:
response = requests.get(search_url)
soup = BeautifulSoup(response.text, 'html.parser')
results = []
for g in soup.find_all('div', class_='g'):
anchors = g.find_all('a')
if anchors:
link = anchors[0]['href']
title = g.find('h3').text
item = {"title": title, "link": link}
results.append(item)
return {"results": results}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
url = request.url
try:
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
text = soup.get_text()
return {"content": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/read_file")
async def read_file(file_path: str):
try:
with open(file_path, 'r') as file:
content = file.read()
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/ocr_image")
async def ocr_image(image_path: str):
try:
image = Image.open(image_path)
text = pytesseract.image_to_string(image)
return {"text": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/speech_to_text")
async def speech_to_text(audio_path: str):
try:
with open(audio_path, 'rb') as file:
response = requests.post("https://api.assemblyai.com/v2/upload", headers={"authorization": "YOUR_ASSEMBLYAI_API_KEY"}, files={"file": file})
audio_url = response.json()["upload_url"]
response = requests.post("https://api.assemblyai.com/v2/transcript", headers={"authorization": "YOUR_ASSEMBLYAI_API_KEY"}, json={"audio_url": audio_url})
transcript_id = response.json()["id"]
while True:
response = requests.get(f"https://api.assemblyai.com/v2/transcript/{transcript_id}", headers={"authorization": "YOUR_ASSEMBLYAI_API_KEY"})
if response.json()["status"] == "completed":
text = response.json()["text"]
return {"text": text}
elif response.json()["status"] == "error":
raise HTTPException(status_code=500, detail="Transcription failed")
time.sleep(5)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000) {
"name": "ollama_mcp",
"description": "A suite of MCP tools that extend Ollama with practical web and system utilities.",
"tools": [
{
"name": "web_search",
"description": "Performs a web search and retrieves relevant results.",
"parameters": [
{ "name": "query", "type": "string", "description": "Search query." }
],
"api_endpoint": "http://localhost:8000/mcp/web_search"
},
{
"name": "fetch_webpage",
"description": "Fetches and summarizes the content of a webpage.",
"parameters": [
{ "name": "url", "type": "string", "description": "The URL of the webpage to fetch." }
],
"api_endpoint": "http://localhost:8000/mcp/fetch_webpage"
},
{
"name": "read_file",
"description": "Reads and extracts text from a local file (TXT, JSON, PDF).",
"parameters": [
{ "name": "file_path", "type": "string", "description": "Path to the file to read." }
],
"api_endpoint": "http://localhost:8000/mcp/read_file"
},
{
"name": "ocr_image",
"description": "Extracts text from an image using OCR.",
"parameters": [
{ "name": "image_path", "type": "string", "description": "Path to the image file." }
],
"api_endpoint": "http://localhost:8000/mcp/ocr_image"
},
{
"name": "speech_to_text",
"description": "Converts audio speech to text.",
"parameters": [
{ "name": "audio_path", "type": "string", "description": "Path to the audio file." }
],
"api_endpoint": "http://localhost:8000/mcp/speech_to_text"
}
]
}
It looks like you're trying to test your FastAPI app, but you are still getting the error:
```
{'detail': [{'type': 'missing', 'loc': ['query', 'url'], 'msg': 'Field required', 'input': None}]}
```
This issue typically happens when the request is not properly formed. Based on the current implementation of your `fetch_webpage` endpoint, you're expecting a **JSON** body with the `url` parameter. The `test.py` script should send the `url` parameter in the **body** of the request, not as a query parameter.
### Let's fix this by ensuring the request is correctly structured.
### 1. **Ensure Your FastAPI Endpoint Accepts a JSON Body**
Your endpoint is expecting a **JSON body** with a `url` key as follows:
```python
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
url = request.url # Expecting url in the body
try:
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
text = soup.get_text()
return {"content": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
```
### 2. **Correct the `test.py` Request**
In the `test.py` script, ensure you are passing the `url` **inside the JSON body** and not as a query parameter. Here's how you should modify it:
```python
import requests
# Ensure you're passing the URL as a JSON body
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage",
json={"url": "https://www.danielkliewer.com"} # URL in the body
)
print(response.json()) # Print the response from FastAPI
```
### Key Details:
- In your current FastAPI endpoint, the `url` is part of the body, so `json={"url": "https://www.danielkliewer.com"}` is correct.
- The issue you faced earlier occurred because the request was sent incorrectly or didn't match the FastAPI expected input structure.
### 3. **Test Again**
Once you've made the correction to the `test.py` script, try running it again:
```bash
python3 test.py
```
If everything is set up correctly, you should now receive a response that contains the content of the webpage at `https://www.danielkliewer.com`.
Let me know how it goes!
gpt-4o-mini
import requests
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage",
json={"url": "https://www.danielkliewer.com"}
)
print(response.json()) --- I still get the same error
The issue you're facing could be due to how the request is being handled in the FastAPI endpoint. Let me walk through a couple of things to ensure everything is working as expected.
1. **Check Your Endpoint Definition**: Your `fetch_webpage` endpoint is expecting a request body defined by the `WebPageRequest` model. The model is defined to expect `url` as a string:
```python
class WebPageRequest(BaseModel):
url: str # This is the field you're sending in the body
```
You are correctly sending the `url` as a JSON key, but FastAPI expects the request body to conform to the structure specified in `WebPageRequest`.
### Solution: Modify Your Endpoint to Handle the Request Body Correctly
You should be accepting the body as a `WebPageRequest`, not directly as individual parameters. Here's the corrected version of your endpoint to ensure it properly processes the body:
```python
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
# Here, FastAPI will use WebPageRequest to validate and extract the `url` from the body
url = request.url # This should now correctly get the `url` from the JSON body
try:
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
text = soup.get_text()
return {"content": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
```
This ensures that FastAPI expects a body that matches the `WebPageRequest` model, and FastAPI will automatically extract the `url` from the body as expected.
### 2. **Check the Request Body in Your `test.py` Script**
In your `test.py`, the request is already structured correctly (you're sending a JSON with the `url` key). So, no changes are needed there:
```python
import requests
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage",
json={"url": "https://www.danielkliewer.com"}
)
print(response.json()) # Check if the response contains the webpage content
```
### 3. **Ensure FastAPI Is Running**
Make sure the FastAPI app is running properly before you execute `test.py`. You can do this by running the server using `uvicorn`:
```bash
uvicorn main:app --reload
```
Ensure that the server is running on port `8000` and accessible.
### 4. **Check FastAPI Logs for More Information**
If you continue to get the error, check the FastAPI logs for any additional information. It might show more about what the issue is.
Let me know if this resolves the issue!
gpt-4o-mini
I already did all of that and get the same message
Thanks for confirming that you've already checked those steps. Let's dive deeper into some possible causes and how to resolve them.
### 1. **Check FastAPI Request Body Parsing**
FastAPI should be able to parse the request body into the `WebPageRequest` model correctly. Let's ensure that your request body is indeed being parsed as expected.
In your `fetch_webpage` endpoint, print out the `request` object to make sure it's coming through correctly:
```python
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
print(request) # This will print the parsed `url` field
url = request.url
try:
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
text = soup.get_text()
return {"content": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
```
### 2. **Test Direct JSON with FastAPI's Interactive Docs**
You can test the endpoint directly from FastAPI's interactive docs, which you can access at `http://localhost:8000/docs`.
- Navigate to the `/mcp/fetch_webpage` endpoint in the Swagger UI.
- Enter the `url` parameter in the request body and submit it.
- Check if you get the expected response.
### 3. **Check for HTTP Headers Issues**
Sometimes, there might be issues related to the HTTP headers, especially `Content-Type`. Make sure the request is sent with the correct content type (`application/json`).
Modify your request code to explicitly specify the content type:
```python
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage",
json={"url": "https://www.danielkliewer.com"},
headers={"Content-Type": "application/json"} # Explicitly set the content type
)
print(response.json())
```
### 4. **Recheck FastAPI Logs**
Check the FastAPI server logs when you send the request. Specifically, check for any validation errors. If FastAPI can't parse the body properly, it might log some error about the model validation.
### 5. **Ensure You're Sending the Body as JSON**
Make sure that you're sending the request body as a valid JSON object. The error message you're receiving indicates that FastAPI is expecting a `url` field, but it's not being found. This could happen if the body isn't being sent properly.
### 6. **Check FastAPI Version**
There may be a bug or discrepancy with FastAPI's version you're using. Try upgrading FastAPI and Pydantic to the latest versions:
```bash
pip install --upgrade fastapi pydantic
```
Let me know if you're still encountering the issue after these steps!
gpt-4o-mini
now rewrite test.py to test all of the functions: import requests
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage",
json={"url": "https://www.danielkliewer.com"}
)
print(response.json())
Sure! Below is a rewritten `test.py` script that tests all the functions you've defined in your FastAPI server. This script will send requests to each of the endpoints and print the results:
```python
import requests
# Test fetch_webpage
def test_fetch_webpage():
response = requests.post(
"http://localhost:8000/mcp/fetch_webpage",
json={"url": "https://www.danielkliewer.com"}
)
print("fetch_webpage response:", response.json())
# Test web_search
def test_web_search():
response = requests.post(
"http://localhost:8000/mcp/web_search",
json={"query": "Daniel Kliewer"}
)
print("web_search response:", response.json())
# Test ollama_generate
def test_ollama_generate():
response = requests.post(
"http://localhost:8000/mcp/ollama",
json={"prompt": "Tell me a joke.", "model": "mistral"}
)
print("ollama_generate response:", response.json())
# Test read_file
def test_read_file():
response = requests.post(
"http://localhost:8000/mcp/read_file",
json={"file_path": "/path/to/your/file.txt"} # Adjust with an actual file path
)
print("read_file response:", response.json())
# Test ocr_image
def test_ocr_image():
response = requests.post(
"http://localhost:8000/mcp/ocr_image",
json={"image_path": "/path/to/your/image.png"} # Adjust with an actual image path
)
print("ocr_image response:", response.json())
# Test speech_to_text
def test_speech_to_text():
response = requests.post(
"http://localhost:8000/mcp/speech_to_text",
json={"audio_path": "/path/to/your/audio.wav"} # Adjust with an actual audio path
)
print("speech_to_text response:", response.json())
# Call all test functions
if __name__ == "__main__":
test_fetch_webpage()
test_web_search()
test_ollama_generate()
test_read_file()
test_ocr_image()
test_speech_to_text()
```
### Key Points:
1. **Paths and URLs**:
- For `test_read_file`, `test_ocr_image`, and `test_speech_to_text`, replace the placeholder file paths (`/path/to/your/...`) with actual paths to test.
2. **Responses**:
- The script will send requests to each endpoint and print the responses, so you can verify whether each endpoint is working as expected.
### How to Run:
1. Ensure your FastAPI server is running on `http://localhost:8000`.
2. Save this script as `test.py`.
3. Run it from the terminal:
```bash
python3 test.py
```
This will sequentially test all the endpoints and print their responses.
Let me know if you need help with the file paths or encounter any issues!
gpt-4o-mini
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import pytesseract
import whisper
import json
from bs4 import BeautifulSoup
import os
app = FastAPI()
OLLAMA_API_URL = "http://localhost:11434/api/generate" # Ollama's local API
# Define request body schema
class OllamaRequest(BaseModel):
prompt: str
model: str = "mistral"
class WebPageRequest(BaseModel):
url: str # This is the field you're sending in the body
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
payload = {"model": request.model, "prompt": request.prompt, "stream": False}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response_data = response.json()
return {"response": response_data.get("response", "No response")}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/web_search")
async def web_search(query: str):
search_url = f"https://www.google.com/search?q={query}"
try:
response = requests.get(search_url)
soup = BeautifulSoup(response.text, 'html.parser')
results = []
for g in soup.find_all('div', class_='g'):
anchors = g.find_all('a')
if anchors:
link = anchors[0]['href']
title = g.find('h3').text
item = {"title": title, "link": link}
results.append(item)
return {"results": results}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
print(request) # This will print the parsed `url` field
url = request.url
try:
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
text = soup.get_text()
return {"content": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/read_file")
async def read_file(file_path: str):
try:
with open(file_path, 'r') as file:
content = file.read()
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/ocr_image")
async def ocr_image(image_path: str):
try:
image = Image.open(image_path)
text = pytesseract.image_to_string(image)
return {"text": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/speech_to_text")
async def speech_to_text(audio_path: str):
try:
with open(audio_path, 'rb') as file:
response = requests.post("https://api.assemblyai.com/v2/upload", headers={"authorization": "YOUR_ASSEMBLYAI_API_KEY"}, files={"file": file})
audio_url = response.json()["upload_url"]
response = requests.post("https://api.assemblyai.com/v2/transcript", headers={"authorization": "YOUR_ASSEMBLYAI_API_KEY"}, json={"audio_url": audio_url})
transcript_id = response.json()["id"]
while True:
response = requests.get(f"https://api.assemblyai.com/v2/transcript/{transcript_id}", headers={"authorization": "YOUR_ASSEMBLYAI_API_KEY"})
if response.json()["status"] == "completed":
text = response.json()["text"]
return {"text": text}
elif response.json()["status"] == "error":
raise HTTPException(status_code=500, detail="Transcription failed")
time.sleep(5)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Your FastAPI server is set up with multiple endpoints, including OCR, speech-to-text, web scraping, and interacting with the Ollama API. It looks mostly correct, but I noticed a few potential issues and suggestions to improve:
### Key Issues & Suggestions:
1. **`Image` import missing for `ocr_image`**:
- You should import `Image` from the `PIL` library to load the image for OCR.
```python
from PIL import Image
```
2. **`time.sleep` missing import in `speech_to_text`**:
- For the `time.sleep(5)` to work in the `speech_to_text` function, you need to import the `time` module.
```python
import time
```
3. **AssemblyAI API Key**:
- It looks like you left `"YOUR_ASSEMBLYAI_API_KEY"` as a placeholder. Make sure to replace it with your actual API key.
4. **Error Handling**:
- Error handling is good, but you might want to provide more specific error messages or logging for debugging in production.
Here's the corrected version of your code:
### Corrected FastAPI Code:
```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import pytesseract
import whisper
import json
from bs4 import BeautifulSoup
import os
from PIL import Image # Import Image for OCR
import time # Import time for sleep in speech_to_text
app = FastAPI()
OLLAMA_API_URL = "http://localhost:11434/api/generate" # Ollama's local API
# Define request body schema
class OllamaRequest(BaseModel):
prompt: str
model: str = "mistral"
class WebPageRequest(BaseModel):
url: str # This is the field you're sending in the body
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
payload = {"model": request.model, "prompt": request.prompt, "stream": False}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response_data = response.json()
return {"response": response_data.get("response", "No response")}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/web_search")
async def web_search(query: str):
search_url = f"https://www.google.com/search?q={query}"
try:
response = requests.get(search_url)
soup = BeautifulSoup(response.text, 'html.parser')
results = []
for g in soup.find_all('div', class_='g'):
anchors = g.find_all('a')
if anchors:
link = anchors[0]['href']
title = g.find('h3').text
item = {"title": title, "link": link}
results.append(item)
return {"results": results}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
print(request) # This will print the parsed `url` field
url = request.url
try:
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
text = soup.get_text()
return {"content": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/read_file")
async def read_file(file_path: str):
try:
with open(file_path, 'r') as file:
content = file.read()
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/ocr_image")
async def ocr_image(image_path: str):
try:
image = Image.open(image_path) # Open image for OCR
text = pytesseract.image_to_string(image)
return {"text": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/speech_to_text")
async def speech_to_text(audio_path: str):
try:
with open(audio_path, 'rb') as file:
response = requests.post("https://api.assemblyai.com/v2/upload", headers={"authorization": "YOUR_ASSEMBLYAI_API_KEY"}, files={"file": file})
audio_url = response.json()["upload_url"]
response = requests.post("https://api.assemblyai.com/v2/transcript", headers={"authorization": "YOUR_ASSEMBLYAI_API_KEY"}, json={"audio_url": audio_url})
transcript_id = response.json()["id"]
while True:
response = requests.get(f"https://api.assemblyai.com/v2/transcript/{transcript_id}", headers={"authorization": "YOUR_ASSEMBLYAI_API_KEY"})
if response.json()["status"] == "completed":
text = response.json()["text"]
return {"text": text}
elif response.json()["status"] == "error":
raise HTTPException(status_code=500, detail="Transcription failed")
time.sleep(5) # Sleep to avoid overloading API requests
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
```
### Key Changes:
1. **Added missing imports**:
- `from PIL import Image` for OCR functionality.
- `import time` to use `time.sleep()` in the `speech_to_text` function.
2. **Improved error handling**:
- If there is an issue with any endpoint, it raises an HTTPException with a detailed message for easier debugging.
### How to Test:
- **For `ocr_image`**, ensure that the `image_path` points to a valid image file that `pytesseract` can read.
- **For `speech_to_text`**, replace `"YOUR_ASSEMBLYAI_API_KEY"` with your actual AssemblyAI API key and provide an actual audio file path for testing.
Run the FastAPI app using:
```bash
uvicorn your_filename:app --reload
```
Let me know if you encounter any other issues!
gpt-4o-mini
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import pytesseract
import json
from bs4 import BeautifulSoup
import os
from PIL import Image
import time
app = FastAPI()
OLLAMA_API_URL = "http://localhost:11434/api/generate" # Ollama's local API
# Define request body schema
class OllamaRequest(BaseModel):
prompt: str
model: str = "mistral"
class WebPageRequest(BaseModel):
url: str # This is the field you're sending in the body
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
payload = {"model": request.model, "prompt": request.prompt, "stream": False}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response_data = response.json()
return {"response": response_data.get("response", "No response")}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/web_search")
async def web_search(query: str):
search_url = f"https://www.google.com/search?q={query}"
try:
response = requests.get(search_url)
soup = BeautifulSoup(response.text, 'html.parser')
results = []
for g in soup.find_all('div', class_='g'):
anchors = g.find_all('a')
if anchors:
link = anchors[0]['href']
title = g.find('h3').text
item = {"title": title, "link": link}
results.append(item)
return {"results": results}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
print(request) # This will print the parsed `url` field
url = request.url
try:
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
text = soup.get_text()
return {"content": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/read_file")
async def read_file(file_path: str):
try:
with open(file_path, 'r') as file:
content = file.read()
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/ocr_image")
async def ocr_image(image_path: str):
try:
image = Image.open(image_path)
text = pytesseract.image_to_string(image)
return {"text": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Your FastAPI application looks good and seems well-structured, but I noticed a couple of things that might be useful to address:
1. **Imports**:
- You have `json` imported, but it's not being used in the code. You can remove it if unnecessary.
- You import `time` but it doesn't seem to be used in the current code either. You may want to remove it unless it's planned for future use.
2. **Error Handling**:
- The error handling for each route is wrapped in a generic `try-except` block which is good for catching unexpected errors. However, it might be helpful to customize the error messages a bit more for better debugging or logging in production environments.
3. **Requests Timeout**:
- For the `requests.get` calls (e.g., `web_search` and `fetch_webpage`), you may want to add a timeout to avoid hanging indefinitely. A small timeout (e.g., 5 seconds) can ensure that requests don’t take too long.
4. **Image Path Validation in `/mcp/ocr_image`**:
- The `ocr_image` route expects an image path as input, but it doesn't check if the path is valid or if the file is actually an image. It might be a good idea to check whether the file exists and is indeed an image before trying to open it.
Here are some suggested changes based on the above points:
```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import pytesseract
from bs4 import BeautifulSoup
from PIL import Image
import os
app = FastAPI()
OLLAMA_API_URL = "http://localhost:11434/api/generate" # Ollama's local API
# Define request body schema
class OllamaRequest(BaseModel):
prompt: str
model: str = "mistral"
class WebPageRequest(BaseModel):
url: str # This is the field you're sending in the body
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
payload = {"model": request.model, "prompt": request.prompt, "stream": False}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response.raise_for_status() # Raise an exception for HTTP errors
response_data = response.json()
return {"response": response_data.get("response", "No response")}
except requests.exceptions.RequestException as e:
raise HTTPException(status_code=500, detail=f"Request failed: {str(e)}")
@app.post("/mcp/web_search")
async def web_search(query: str):
search_url = f"https://www.google.com/search?q={query}"
try:
response = requests.get(search_url, timeout=5)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
results = []
for g in soup.find_all('div', class_='g'):
anchors = g.find_all('a')
if anchors:
link = anchors[0]['href']
title = g.find('h3').text
item = {"title": title, "link": link}
results.append(item)
return {"results": results}
except requests.exceptions.RequestException as e:
raise HTTPException(status_code=500, detail=f"Request failed: {str(e)}")
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
url = request.url
try:
response = requests.get(url, timeout=5)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
text = soup.get_text()
return {"content": text}
except requests.exceptions.RequestException as e:
raise HTTPException(status_code=500, detail=f"Request failed: {str(e)}")
@app.post("/mcp/read_file")
async def read_file(file_path: str):
try:
if not os.path.exists(file_path):
raise HTTPException(status_code=400, detail="File not found")
with open(file_path, 'r') as file:
content = file.read()
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/ocr_image")
async def ocr_image(image_path: str):
try:
if not os.path.exists(image_path):
raise HTTPException(status_code=400, detail="Image file not found")
image = Image.open(image_path)
text = pytesseract.image_to_string(image)
return {"text": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
```
### Changes made:
1. Added `response.raise_for_status()` to catch HTTP errors (e.g., 404 or 500) from the requests.
2. Added timeouts to the `requests.get` calls to ensure they don't hang indefinitely.
3. Checked if the image file exists before attempting OCR.
4. Added error handling for file not found when reading a file.
This should improve the robustness and efficiency of your application. Let me know if you need further adjustments!
gpt-4o-mini
is this MCP?
The code you've provided is for a FastAPI application that exposes multiple endpoints, including one for interacting with Ollama's API, a web search, fetching webpages, reading files, and performing OCR on images. It doesn't specifically mention MCP (Multi-Channel Processing or Multi-Channel Pipeline), which is a term that can refer to various types of systems that handle multiple input/output channels (e.g., web, files, speech, etc.).
If by "MCP" you're referring to a system designed to integrate and manage multiple forms of data (such as text from web scraping, file reading, OCR, and audio transcription), then yes, this could be considered part of an MCP-like system because it integrates different types of processing tasks (web scraping, OCR, speech-to-text).
### To align it more with the "MCP" concept:
- **MCP** systems often refer to pipelines that handle data from multiple input types or channels (e.g., text, audio, images) and produce unified outputs.
- This code can be seen as a system that processes data from multiple "channels" (web scraping, OCR from images, reading files) and outputs results in a consistent format (JSON).
If you're working to integrate this into an **MCP framework**, you would typically extend the system to include more advanced management of these channels
gpt-4o-mini
no I mean model context protocol
Ah, got it! You're referring to **Model Context Protocol (MCP)**, not multi-channel processing. In that case, no, the code you provided doesn't directly implement a "Model Context Protocol" (MCP).
A **Model Context Protocol (MCP)** is typically a protocol or framework that manages the context for models in a way that the models' outputs can be controlled, monitored, and adjusted based on the context of the task. In machine learning and AI systems, this can involve managing the state or environment in which the model operates, including input data, prior knowledge, or any relevant parameters that the model needs to consider.
Your code includes multiple endpoints related to interacting with models (like `ollama_generate` for the Ollama model) and other tasks like web scraping, OCR, and speech-to-text processing. However, it doesn't seem to specifically implement an MCP framework where the context is dynamically managed across different tasks or models in a structured protocol.
If you're aiming to implement an MCP, you would likely need to:
1. **Define context management**: Create a system that tracks and updates the context that different models (like Ollama, OCR, etc.) operate in, allowing them to adapt based on the specific task.
2. **Model-specific interactions**: Ensure that when you're calling models, you feed them with the context necessary to guide their responses in a meaningful way.
3. **Context-switching**: Manage the flow of data and context between different processing tasks in the system (e.g., web scraping, OCR, speech-to-text).
In summary, while your system handles multiple tasks that might be part of an MCP-like framework, it doesn't explicitly implement **MCP** in terms of managing the model context across different processing steps.
Would you like help on how to implement MCP principles into your FastAPI setup?
gpt-4o-mini
1. Define context management: Create a system that tracks and updates the context that different models (like Ollama, OCR, etc.) operate in, allowing them to adapt based on the specific task.
2. Model-specific interactions: Ensure that when you’re calling models, you feed them with the context necessary to guide their responses in a meaningful way.
3. Context-switching: Manage the flow of data and context between different processing tasks in the system (e.g., web scraping, OCR, speech-to-text).
To implement a **Model Context Protocol (MCP)** in your FastAPI-based system, we'll break down the three main components you've mentioned: **context management**, **model-specific interactions**, and **context-switching**. Let's go through each of these steps and how you can implement them:
### 1. **Define Context Management**
Context management involves tracking the state or environment that different models (such as Ollama, OCR, or speech-to-text) need to operate in. This will ensure that models have the necessary information to generate meaningful results.
**Implementation Steps:**
- Create a **context object** that will hold shared data that is relevant across tasks. This could include information like the user’s current session, input data (like text or images), previous model outputs, or any user-defined variables.
- Use a **context manager** to update and access the context dynamically as different models are invoked.
**Example Context Management Implementation:**
```python
from typing import Dict, Any
class ContextManager:
def __init__(self):
self.context: Dict[str, Any] = {}
def update_context(self, key: str, value: Any):
"""Update the context with a new value for a given key."""
self.context[key] = value
def get_context(self, key: str) -> Any:
"""Retrieve a value from the context."""
return self.context.get(key)
def reset_context(self):
"""Reset the context."""
self.context = {}
# Create an instance of ContextManager
context_manager = ContextManager()
```
### 2. **Model-Specific Interactions**
Each model (like Ollama for text generation, OCR for extracting text from images) needs to receive the right context in order to function correctly. The context you provide should inform the model on how to process the input.
For example:
- **Ollama**: The model may need context about the prompt, model settings, and the intended outcome (e.g., generating creative text).
- **OCR**: The model may need context about the image being processed and any prior text that needs to be extracted.
- **Speech-to-Text**: Context can include the audio source, transcription settings, and any contextual information about the conversation or speech.
**Example Model-Specific Interaction:**
```python
import requests
# Example of Ollama generation, using context from ContextManager
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
# Get context (for example, any prior data needed for Ollama)
previous_text = context_manager.get_context("previous_text")
# Create payload including context information
payload = {
"model": request.model,
"prompt": f"{request.prompt} {previous_text if previous_text else ''}",
"stream": False
}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response_data = response.json()
result = response_data.get("response", "No response")
# Update the context with the new result for future use
context_manager.update_context("previous_text", result)
return {"response": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
```
In this case, the model is fed not only with the current prompt but also with the **previous response** as part of the context, ensuring the model can generate a coherent and context-aware response.
### 3. **Context-Switching**
Context-switching is about managing the flow of data and context across different tasks. If a user starts a task like a web search and then switches to an OCR task, the system needs to handle the context transitions between tasks smoothly.
**Implementation Steps:**
- Keep track of the task the user is currently engaged in. You might have context flags or markers (e.g., "currently_in_task: OCR").
- Dynamically update the context when switching between tasks so that the right models have the right information.
**Example Context-Switching:**
```python
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
url = request.url
try:
# Fetch the webpage content
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
content = soup.get_text()
# Update the context with the content of the fetched webpage
context_manager.update_context("webpage_content", content)
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/ocr_image")
async def ocr_image(image_path: str):
try:
image = Image.open(image_path)
text = pytesseract.image_to_string(image)
# If OCR is part of the ongoing task, store the OCR text in the context
context_manager.update_context("ocr_text", text)
return {"text": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
```
In this case:
- When the user fetches a webpage, the content is stored in the context.
- When the user switches to the OCR task, the system updates the context with the OCR text.
- The system can then reference the appropriate context for each task as needed, ensuring a smooth transition between tasks.
### Putting it All Together
```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import pytesseract
from PIL import Image
from bs4 import BeautifulSoup
app = FastAPI()
# Context Manager to hold shared context between tasks
context_manager = ContextManager()
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
payload = {"model": request.model, "prompt": request.prompt, "stream": False}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response_data = response.json()
result = response_data.get("response", "No response")
context_manager.update_context("previous_text", result)
return {"response": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
url = request.url
try:
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
content = soup.get_text()
context_manager.update_context("webpage_content", content)
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/ocr_image")
async def ocr_image(image_path: str):
try:
image = Image.open(image_path)
text = pytesseract.image_to_string(image)
context_manager.update_context("ocr_text", text)
return {"text": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# Your other tasks (like speech-to-text, file reading, etc.) would similarly interact with context_manager to switch between contexts and maintain task-specific data.
```
### Summary
- **Context Management**: The `ContextManager` class tracks the state and updates context as needed, storing shared data that models and tasks use.
- **Model-Specific Interactions**: Each task (e.g., Ollama, OCR) accesses the right context and updates it as necessary, ensuring relevant information is always available.
- **Context-Switching**: The system can switch between tasks and adjust context accordingly, maintaining a seamless flow for users.
With this structure, your system can manage multiple models and tasks in a context-sensitive manner, and switch between them dynamically. Would you like to dive deeper into any of these steps?
gpt-4o-mini
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import requests
import pytesseract
from bs4 import BeautifulSoup
from PIL import Image
import os
from typing import Dict, Any
app = FastAPI()
OLLAMA_API_URL = "http://localhost:11434/api/generate" # Ollama's local API
# Define request body schema
class OllamaRequest(BaseModel):
prompt: str
model: str = "mistral"
class WebPageRequest(BaseModel):
url: str # This is the field you're sending in the body
class ContextManager:
def __init__(self):
self.context: Dict[str, Any] = {}
def update_context(self, key: str, value: Any):
"""Update the context with a new value for a given key."""
self.context[key] = value
def get_context(self, key: str) -> Any:
"""Retrieve a value from the context."""
return self.context.get(key)
def reset_context(self):
"""Reset the context."""
self.context = {}
# Create an instance of ContextManager
context_manager = ContextManager()
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
# Get context (for example, any prior data needed for Ollama)
previous_text = context_manager.get_context("previous_text")
# Create payload including context information
payload = {
"model": request.model,
"prompt": f"{request.prompt} {previous_text if previous_text else ''}",
"stream": False
}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response_data = response.json()
result = response_data.get("response", "No response")
# Update the context with the new result for future use
context_manager.update_context("previous_text", result)
return {"response": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/web_search")
async def web_search(query: str):
search_url = f"https://www.google.com/search?q={query}"
try:
response = requests.get(search_url, timeout=5)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
results = []
for g in soup.find_all('div', class_='g'):
anchors = g.find_all('a')
if anchors:
link = anchors[0]['href']
title = g.find('h3').text
item = {"title": title, "link": link}
results.append(item)
return {"results": results}
except requests.exceptions.RequestException as e:
raise HTTPException(status_code=500, detail=f"Request failed: {str(e)}")
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
url = request.url
try:
# Fetch the webpage content
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
content = soup.get_text()
# Update the context with the content of the fetched webpage
context_manager.update_context("webpage_content", content)
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/read_file")
async def read_file(file_path: str):
try:
if not os.path.exists(file_path):
raise HTTPException(status_code=400, detail="File not found")
with open(file_path, 'r') as file:
content = file.read()
return {"content": content}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/ocr_image")
async def ocr_image(image_path: str):
try:
image = Image.open(image_path)
text = pytesseract.image_to_string(image)
# If OCR is part of the ongoing task, store the OCR text in the context
context_manager.update_context("ocr_text", text)
return {"text": text}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Your FastAPI implementation is well-structured, and it covers various models like Ollama, web scraping, file reading, and OCR. Here’s a quick overview of what your code does, and I’ll also suggest a few enhancements for robustness and readability:
### Overview:
1. **Ollama Integration** (`/mcp/ollama`):
- It uses the Ollama API to generate text based on a prompt and optionally includes previous context for continuity in responses.
2. **Web Search** (`/mcp/web_search`):
- Uses Google search to fetch links and titles from the results page. It scrapes the data using BeautifulSoup.
3. **Fetch Webpage** (`/mcp/fetch_webpage`):
- Fetches the raw content of a webpage (text only) and updates the context with the extracted content.
4. **File Reading** (`/mcp/read_file`):
- Reads the content of a text file from a specified path and returns the content.
5. **OCR Image** (`/mcp/ocr_image`):
- Performs OCR (Optical Character Recognition) on an image to extract text, and stores the extracted text in context.
### Enhancements:
1. **Error Handling**:
- It's great that you’re catching exceptions, but you could make it more granular. For example, in file reading, check specifically if the file is empty or if the content is in the expected format.
2. **Context Updates**:
- Since you’re dealing with multiple tasks (Ollama, OCR, file reading), it might be beneficial to ensure that context updates are cleared or checked for unnecessary overwrites. You could allow resetting the context for a particular task or user session.
3. **Type Annotations**:
- You can use `Optional` from `typing` for variables that may not always be set, e.g., when no previous text is available for Ollama.
4. **Timeouts**:
- For external requests like web scraping, setting explicit timeouts for network requests can prevent the app from hanging if the network is slow or unavailable.
5. **Better Logging**:
- It could be useful to log error details (especially HTTP errors, like status codes), which would help diagnose issues faster in production environments.
### Suggested Updates:
```python
import logging
from typing import Optional
# Setup logging
logging.basicConfig(level=logging.INFO)
class ContextManager:
def __init__(self):
self.context: Dict[str, Any] = {}
def update_context(self, key: str, value: Any):
"""Update the context with a new value for a given key."""
logging.info(f"Updating context: {key} = {value}")
self.context[key] = value
def get_context(self, key: str) -> Optional[Any]:
"""Retrieve a value from the context."""
return self.context.get(key)
def reset_context(self):
"""Reset the context."""
logging.info("Resetting context.")
self.context = {}
context_manager = ContextManager()
@app.post("/mcp/ollama")
async def ollama_generate(request: OllamaRequest):
previous_text = context_manager.get_context("previous_text") or ""
payload = {
"model": request.model,
"prompt": f"{request.prompt} {previous_text}",
"stream": False
}
try:
response = requests.post(OLLAMA_API_URL, json=payload)
response.raise_for_status() # This will trigger an exception for bad status codes
response_data = response.json()
result = response_data.get("response", "No response")
context_manager.update_context("previous_text", result)
return {"response": result}
except requests.exceptions.RequestException as e:
logging.error(f"Ollama request failed: {e}")
raise HTTPException(status_code=500, detail="Error contacting Ollama API.")
except Exception as e:
logging.error(f"Unexpected error in Ollama: {e}")
raise HTTPException(status_code=500, detail="Unexpected error.")
@app.post("/mcp/web_search")
async def web_search(query: str):
search_url = f"https://www.google.com/search?q={query}"
try:
response = requests.get(search_url, timeout=5)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
results = [{"title": g.find('h3').text, "link": anchors[0]['href']}
for g in soup.find_all('div', class_='g') if (anchors := g.find_all('a'))]
return {"results": results}
except requests.exceptions.RequestException as e:
logging.error(f"Web search failed: {e}")
raise HTTPException(status_code=500, detail="Error during web search.")
except Exception as e:
logging.error(f"Unexpected error in web search: {e}")
raise HTTPException(status_code=500, detail="Unexpected error.")
@app.post("/mcp/fetch_webpage")
async def fetch_webpage(request: WebPageRequest):
url = request.url
try:
response = requests.get(url, timeout=5)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
content = soup.get_text()
context_manager.update_context("webpage_content", content)
return {"content": content}
except requests.exceptions.RequestException as e:
logging.error(f"Failed to fetch webpage: {e}")
raise HTTPException(status_code=500, detail="Error fetching webpage.")
except Exception as e:
logging.error(f"Unexpected error in fetching webpage: {e}")
raise HTTPException(status_code=500, detail="Unexpected error.")
@app.post("/mcp/read_file")
async def read_file(file_path: str):
try:
if not os.path.exists(file_path):
raise HTTPException(status_code=400, detail="File not found")
with open(file_path, 'r') as file:
content = file.read()
return {"content": content}
except FileNotFoundError:
logging.error(f"File not found: {file_path}")
raise HTTPException(status_code=400, detail="File not found.")
except Exception as e:
logging.error(f"Error reading file: {e}")
raise HTTPException(status_code=500, detail="Error reading file.")
@app.post("/mcp/ocr_image")
async def ocr_image(image_path: str):
try:
if not os.path.exists(image_path):
raise HTTPException(status_code=400, detail="Image not found")
image = Image.open(image_path)
text = pytesseract.image_to_string(image)
context_manager.update_context("ocr_text", text)
return {"text": text}
except FileNotFoundError:
logging.error(f"Image file not found: {image_path}")
raise HTTPException(status_code=400, detail="Image file not found.")
except Exception as e:
logging.error(f"Error in OCR: {e}")
raise HTTPException(status_code=500, detail="Error processing image.")
```
### Key Improvements:
1. **Logging**: Errors and context updates are logged to help with diagnostics.
2. **Timeouts**: Explicit timeouts are added to network requests to prevent the system from hanging.
3. **Error Handling**: More specific exception handling for cases like missing files or HTTP errors.
You can now start your FastAPI app and handle a variety of tasks while maintaining context across the different tasks. Let me know if you'd like to explore any specific aspect further!
gpt-4o-mini