(session: SessionDep, chat: ChatStart)
| 83 | |
| 84 | @router.post("/mcp_start", operation_id="mcp_start") |
| 85 | async def mcp_start(session: SessionDep, chat: ChatStart): |
| 86 | user: BaseUserDTO = authenticate(session=session, account=chat.username, password=chat.password) |
| 87 | if not user: |
| 88 | raise HTTPException(status_code=400, detail="Incorrect account or password") |
| 89 | |
| 90 | if not user.oid or user.oid == 0: |
| 91 | raise HTTPException(status_code=400, detail="No associated workspace, Please contact the administrator") |
| 92 | access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) |
| 93 | user_dict = user.to_dict() |
| 94 | t = Token(access_token=create_access_token( |
| 95 | user_dict, expires_delta=access_token_expires |
| 96 | )) |
| 97 | c = create_chat(session, user, CreateChat(origin=1), False) |
| 98 | return {"access_token": t.access_token, "chat_id": c.id} |
| 99 | |
| 100 | |
| 101 | @router.post("/mcp_ws_list", operation_id="mcp_ws_list") |
nothing calls this directly
no test coverage detected