OpenWebUI/mcpo
1
0
Fork 0
mirror of https://github.com/open-webui/mcpo synced 2025-06-26 18:26:58 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

refac

Browse Source
This commit is contained in:
Timothy Jaeryang Baek 2025-04-06 15:05:00 -07:00
parent 615c333e7f
commit 334089b406
1 changed files with 41 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
src/mcpo/main.py
View File

@ -32,6 +32,7 @@ def get_python_type(param_type: str):
return str # Fallback
# Expand as needed. PRs welcome!
def process_tool_response(result: CallToolResult) -> list:
"""Universal response processor for all tool endpoints"""
response = []
@ -52,6 +53,7 @@ def process_tool_response(result: CallToolResult) -> list:
response.append("Embedded resource not supported yet.")
return response
async def create_dynamic_endpoints(app: FastAPI, api_dependency=None):
session = app.state.session
if not session:
@ -91,20 +93,29 @@ async def create_dynamic_endpoints(app: FastAPI, api_dependency=None):
if model_fields:
FormModel = create_model(f"{endpoint_name}_form_model", **model_fields)
def make_endpoint_func(endpoint_name: str, FormModel, session: ClientSession): # Parameterized endpoint
async def tool_endpoint(form_data: FormModel):
def make_endpoint_func(
endpoint_name: str, FormModel, session: ClientSession
): # Parameterized endpoint
async def tool(form_data: FormModel):
args = form_data.model_dump(exclude_none=True)
result = await session.call_tool(endpoint_name, arguments=args)
return process_tool_response(result)
return tool_endpoint
return tool
tool_handler = make_endpoint_func(endpoint_name, FormModel, session)
else:
def make_endpoint_func_no_args(endpoint_name: str, session: ClientSession): # Parameterless endpoint
async def tool_endpoint(): # No parameters
result = await session.call_tool(endpoint_name, arguments={}) # Empty dict
def make_endpoint_func_no_args(
endpoint_name: str, session: ClientSession
): # Parameterless endpoint
async def tool(): # No parameters
result = await session.call_tool(
endpoint_name, arguments={}
) # Empty dict
return process_tool_response(result) # Same processor
return tool_endpoint
return tool
tool_handler = make_endpoint_func_no_args(endpoint_name, session)
@ -170,7 +181,12 @@ async def run(
path_prefix = kwargs.get("path_prefix") or "/"
main_app = FastAPI(
title=name, description=description, version=version, ssl_certfile=ssl_certfile, ssl_keyfile=ssl_keyfile, lifespan=lifespan
title=name,
description=description,
version=version,
ssl_certfile=ssl_certfile,
ssl_keyfile=ssl_keyfile,
lifespan=lifespan,
)
main_app.add_middleware(
@ -217,11 +233,20 @@ async def run(
sub_app.state.api_dependency = api_dependency
main_app.mount(f"{path_prefix}{server_name}", sub_app)
main_app.description += f"\n - [{server_name}](http://{host}:{port}/{server_name}/docs)"
main_app.description += (
f"\n - [{server_name}](http://{host}:{port}/{server_name}/docs)"
)
else:
raise ValueError("You must provide either server_command or config.")
config = uvicorn.Config(app=main_app, host=host, port=port, ssl_certfile=ssl_certfile, ssl_keyfile=ssl_keyfile, log_level="info")
config = uvicorn.Config(
app=main_app,
host=host,
port=port,
ssl_certfile=ssl_certfile,
ssl_keyfile=ssl_keyfile,
log_level="info",
)
server = uvicorn.Server(config)
await server.serve()