from fastapi import FastAPI, Body, HTTPException, Header
from pydantic import BaseModel
app = FastAPI()
class InputData(BaseModel):
point: str
params: dict = {}
@app.post("/api/dify/receive")
async def dify_receive(data: InputData = Body(...), authorization: str = Header(None)):
"""
Receive API query data from Dify.
"""
expected_api_key = "123456" # TODO Your API key of this API
auth_scheme, _, api_key = authorization.partition(' ')
if auth_scheme.lower() != "bearer" or api_key != expected_api_key:
raise HTTPException(status_code=401, detail="Unauthorized")
point = data.point
# for debug
print(f"point: {point}")
if point == "ping":
return {
"result": "pong"
}
if point == "app.external_data_tool.query":
return handle_app_external_data_tool_query(params=data.params)
# elif point == "{point name}":
# TODO other point implementation here
raise HTTPException(status_code=400, detail="Not implemented")
def handle_app_external_data_tool_query(params: dict):
app_id = params.get("app_id")
tool_variable = params.get("tool_variable")
inputs = params.get("inputs")
query = params.get("query")
# for debug
print(f"app_id: {app_id}")
print(f"tool_variable: {tool_variable}")
print(f"inputs: {inputs}")
print(f"query: {query}")
# TODO your external data tool query implementation here,
# return must be a dict with key "result", and the value is the query result
if inputs.get("location") == "London":
return {
"result": "City: London\nTemperature: 10°C\nRealFeel®: 8°C\nAir Quality: Poor\nWind Direction: ENE\nWind "
"Speed: 8 km/h\nWind Gusts: 14 km/h\nPrecipitation: Light rain"
}
else:
return {"result": "Unknown city"}
启动 API 服务，默认端口为 8000，API 完整地址为：http://127.0.0.1:8000/api/dify/receive，配置的 API Key 为 123456。
