-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathapp.py
More file actions
69 lines (56 loc) 路 2.06 KB
/
Copy pathapp.py
File metadata and controls
69 lines (56 loc) 路 2.06 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
# Copyright 2025 Clivern
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import os
from fastapi import FastAPI, HTTPException, Header, Depends
from pydantic import BaseModel
from crawl4ai import AsyncWebCrawler
app = FastAPI()
API_KEY = os.getenv("API_KEY")
async def verify_api_key(x_api_key: str = Header(None)):
"""
Optional API key verification.
"""
if API_KEY:
if x_api_key is None:
raise HTTPException(
status_code=401,
detail="API key required. Please provide X-API-Key header.",
)
if x_api_key != API_KEY:
raise HTTPException(status_code=403, detail="Invalid API key.")
return True
class URLRequest(BaseModel):
url: str
@app.post("/crawl")
async def crawl_url(request: URLRequest, _: bool = Depends(verify_api_key)):
"""
Accepts a POST request with a URL and returns the content in markdown format.
"""
try:
async with AsyncWebCrawler() as crawler:
result = await crawler.arun(url=request.url)
if result.markdown:
return {"markdown": result.markdown}
else:
raise HTTPException(
status_code=500, detail="Failed to extract markdown content"
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error crawling URL: {str(e)}")
@app.get("/")
async def root(_: bool = Depends(verify_api_key)):
return {"message": "OK"}
@app.get("/_health")
async def health(_: bool = Depends(verify_api_key)):
return {"message": "OK"}