需求分析公司希望用Workbuddy连接即有的数据供用户通过对话模式得到报表等数据。即有资源分析Workbuddy购买了企业版安全性有一定保障且能得到官方的支持。源头系统的数据服务器不对外全部部署在内网环境且数据需要严格控制不能让AI工具随意获取数据。现有服务器有空余资源可以在不新购服务器的情况下部署MCP服务。内部人力众所周知ALL IN 1系统架构架构图数据服务器新增用户、密码赋予查询某个表的权限MCP服务器服务器配置白名单加入MCP服务的对外端口选择自己的技术栈部署MCP服务这里举例用python来进行MCP服务部署python安装mcp包以及其他依赖包pycharm同时安装AI工具通过AI进行初步的代码生成历经超过30次的调试MCP服务调通最终的示例代码如下#codingutf-8importasyncioimportloggingimportoracledbfromtypingimportAny,Dict,Listfrommcp.serverimportServerfrommcp.typesimportTool,TextContentfromstarlette.applicationsimportStarlettefromstarlette.routingimportRoutefromstarlette.requestsimportRequestfromstarlette.responsesimportResponsefrommcp.server.sseimportSseServerTransportimportuvicornimportos# 配置日志logging.basicConfig(levellogging.INFO)loggerlogging.getLogger(MCP-Oracle-Server)# 环境变量读取注意这里配置环境变量更安全hostos.environ.get(host,localhost)portint(os.environ.get(port,))oracleuseros.environ.get(oracleuser,)oraclepasswordos.environ.get(oraclepassword,)service_nameos.environ.get(service_name,)# 启动前校验避免带空配置运行ifnotall([oracleuser,oraclepassword,service_name]):logger.warning(⚠️ Missing required environment variables (oracleuser, oraclepassword, service_name). Server may fail to connect.)# 数据库配置DB_CONFIG{host:host,port:port,service_name:service_name,user:oracleuser,password:oraclepassword}# 初始化 MCP ServerappServer(oracle-data-connector)# 白名单ALLOWED_TABLES{实例.表名:[X,XX,XXX,XXXX]}# 全局连接池poolNonedefinit_oracle_client():初始化 Oracle Thick 模式try:# 使用客户端路径client_dirr有效路径oracledb.init_oracle_client(lib_dirclient_dir)logger.info(f✅ Oracle Thick mode initialized successfully using client:{client_dir})exceptExceptionase:logger.error(f❌ Failed to initialize Oracle Thick mode:{e})raisedefinit_connection_pool():初始化 Oracle 连接池提升查询性能globalpoolifpoolisNone:try:dsnf{DB_CONFIG[host]}:{DB_CONFIG[port]}/{DB_CONFIG[service_name]}pooloracledb.create_pool(userDB_CONFIG[user],passwordDB_CONFIG[password],dsndsn,min1,max4,increment1,getmodeoracledb.POOL_GETMODE_WAIT)logger.info(✅ Oracle Connection Pool initialized successfully.)exceptExceptionase:logger.error(f❌ Failed to initialize Oracle Connection Pool:{e})raisedefget_db_connection():从连接池获取连接ifpoolisNone:init_connection_pool()returnpool.acquire()app.list_tools()asyncdeflist_tools()-List[Tool]:return[Tool(namelist_allowed_tables,description列出当前 MCP 服务允许查询的所有表名,inputSchema{type:object,properties:{},required:[]}),Tool(namequery_oracle_table,description查询指定 Oracle 表的数据。仅支持预定义白名单中的表和列。,inputSchema{type:object,properties:{table_name:{type:string,description:要查询的表名必须在大写白名单中},columns:{type:array,items:{type:string},description:要查询的列名列表。如果为空或包含*则返回所有允许列。},limit:{type:integer,description:限制返回行数默认 10最大 100,default:10},filter_column:{type:string,description:可选用于过滤的列名},filter_value:{type:string,description:可选过滤值}},required:[table_name]})]app.call_tool()asyncdefcall_tool(name:str,arguments:Dict[str,Any])-List[TextContent]:try:ifnamelist_allowed_tables:tableslist(ALLOWED_TABLES.keys())return[TextContent(typetext,textfAllowed tables:{, .join(tables)})]elifnamequery_oracle_table:table_namearguments.get(table_name,).upper()requested_columnsarguments.get(columns,[])limitmin(arguments.get(limit,10),100)filter_colarguments.get(filter_column)filter_valarguments.get(filter_value)iftable_namenotinALLOWED_TABLES:return[TextContent(typetext,textfError: Table {table_name} is not allowed.)]allowed_colsALLOWED_TABLES[table_name]ifnotrequested_columnsor*inrequested_columns:select_colsallowed_colselse:invalid_cols[colforcolinrequested_columnsifcol.upper()notinallowed_cols]ifinvalid_cols:return[TextContent(typetext,textfError: Columns{invalid_cols}are not allowed.)]select_cols[col.upper()forcolinrequested_columns]col_str, .join(select_cols)# 修复使用字典传参更安全可靠params{}base_sqlfSELECT{col_str}FROM{table_name}iffilter_colandfilter_valisnotNone:iffilter_col.upper()notinallowed_cols:return[TextContent(typetext,textfError: Filter column {filter_col} is not allowed.)]base_sqlf WHERE{filter_col.upper()} :bind_valparams[:bind_val]filter_val# Oracle 11g 兼容写法sqlfSELECT * FROM ({base_sql}) WHERE ROWNUM {limit}connget_db_connection()cursorconn.cursor()try:cursor.execute(sql,params)columns[desc[0]fordescincursor.description]rowscursor.fetchall()result_textfQuery Result for{table_name}:\nresult_text| | .join(columns) |\nresult_text||.join([---for_incolumns])|\nforrowinrows:formatted_row[str(item)ifitemisnotNoneelseNULLforiteminrow]result_text| | .join(formatted_row) |\nreturn[TextContent(typetext,textresult_text)]finally:# 无论是否报错必须关闭游标并将连接归还连接池cursor.close()conn.close()else:return[TextContent(typetext,textfUnknown tool:{name})]exceptExceptionase:logger.error(fError executing tool{name}:{e})return[TextContent(typetext,textfError:{str(e)})]defmain():# 1. 强制启用 Thick 模式并指定路径init_oracle_client()# 2. 初始化连接池try:init_connection_pool()exceptException:logger.error(❌ Database connection pool failed to initialize. MCP Server will not start.)returnsseSseServerTransport(/messages/)asyncdefhandle_sse(request:Request):asyncwithsse.connect_sse(request.scope,request.receive,request._send)as(read_stream,write_stream):awaitapp.run(read_stream,write_stream,app.create_initialization_options())asyncdefhandle_messages(request:Request):awaitsse.handle_post_message(request.scope,request.receive,request._send)returnResponse(status_code200)starlette_appStarlette(debugTrue,routes[Route(/sse,endpointhandle_sse),Route(/messages/,endpointhandle_messages,methods[POST]),Route(/messages,endpointhandle_messages,methods[POST]),],)logger.info( Starting Oracle MCP Server on port 白名单端口号...)uvicorn.run(starlette_app,host0.0.0.0,port白名单端口号)if__name____main__:main()Workbuddy端操作路径截图MCP配置代码{mcpServers:{Oracle-Connector:{transport:sse,url:http://地址:端口/sse,disabled:false}}}结论通过上述案例可以将系统的数据放在内网通过物理网络数据库单独账号控制权限同时MCP服务器单独部署也进一步杜绝了网络安全问题以及AI查数据不可控的情况。
需求分析公司希望用Workbuddy连接即有的数据供用户通过对话模式得到报表等数据。即有资源分析Workbuddy购买了企业版安全性有一定保障且能得到官方的支持。源头系统的数据服务器不对外全部部署在内网环境且数据需要严格控制不能让AI工具随意获取数据。现有服务器有空余资源可以在不新购服务器的情况下部署MCP服务。内部人力众所周知ALL IN 1系统架构架构图数据服务器新增用户、密码赋予查询某个表的权限MCP服务器服务器配置白名单加入MCP服务的对外端口选择自己的技术栈部署MCP服务这里举例用python来进行MCP服务部署python安装mcp包以及其他依赖包pycharm同时安装AI工具通过AI进行初步的代码生成历经超过30次的调试MCP服务调通最终的示例代码如下#codingutf-8importasyncioimportloggingimportoracledbfromtypingimportAny,Dict,Listfrommcp.serverimportServerfrommcp.typesimportTool,TextContentfromstarlette.applicationsimportStarlettefromstarlette.routingimportRoutefromstarlette.requestsimportRequestfromstarlette.responsesimportResponsefrommcp.server.sseimportSseServerTransportimportuvicornimportos# 配置日志logging.basicConfig(levellogging.INFO)loggerlogging.getLogger(MCP-Oracle-Server)# 环境变量读取注意这里配置环境变量更安全hostos.environ.get(host,localhost)portint(os.environ.get(port,))oracleuseros.environ.get(oracleuser,)oraclepasswordos.environ.get(oraclepassword,)service_nameos.environ.get(service_name,)# 启动前校验避免带空配置运行ifnotall([oracleuser,oraclepassword,service_name]):logger.warning(⚠️ Missing required environment variables (oracleuser, oraclepassword, service_name). Server may fail to connect.)# 数据库配置DB_CONFIG{host:host,port:port,service_name:service_name,user:oracleuser,password:oraclepassword}# 初始化 MCP ServerappServer(oracle-data-connector)# 白名单ALLOWED_TABLES{实例.表名:[X,XX,XXX,XXXX]}# 全局连接池poolNonedefinit_oracle_client():初始化 Oracle Thick 模式try:# 使用客户端路径client_dirr有效路径oracledb.init_oracle_client(lib_dirclient_dir)logger.info(f✅ Oracle Thick mode initialized successfully using client:{client_dir})exceptExceptionase:logger.error(f❌ Failed to initialize Oracle Thick mode:{e})raisedefinit_connection_pool():初始化 Oracle 连接池提升查询性能globalpoolifpoolisNone:try:dsnf{DB_CONFIG[host]}:{DB_CONFIG[port]}/{DB_CONFIG[service_name]}pooloracledb.create_pool(userDB_CONFIG[user],passwordDB_CONFIG[password],dsndsn,min1,max4,increment1,getmodeoracledb.POOL_GETMODE_WAIT)logger.info(✅ Oracle Connection Pool initialized successfully.)exceptExceptionase:logger.error(f❌ Failed to initialize Oracle Connection Pool:{e})raisedefget_db_connection():从连接池获取连接ifpoolisNone:init_connection_pool()returnpool.acquire()app.list_tools()asyncdeflist_tools()-List[Tool]:return[Tool(namelist_allowed_tables,description列出当前 MCP 服务允许查询的所有表名,inputSchema{type:object,properties:{},required:[]}),Tool(namequery_oracle_table,description查询指定 Oracle 表的数据。仅支持预定义白名单中的表和列。,inputSchema{type:object,properties:{table_name:{type:string,description:要查询的表名必须在大写白名单中},columns:{type:array,items:{type:string},description:要查询的列名列表。如果为空或包含*则返回所有允许列。},limit:{type:integer,description:限制返回行数默认 10最大 100,default:10},filter_column:{type:string,description:可选用于过滤的列名},filter_value:{type:string,description:可选过滤值}},required:[table_name]})]app.call_tool()asyncdefcall_tool(name:str,arguments:Dict[str,Any])-List[TextContent]:try:ifnamelist_allowed_tables:tableslist(ALLOWED_TABLES.keys())return[TextContent(typetext,textfAllowed tables:{, .join(tables)})]elifnamequery_oracle_table:table_namearguments.get(table_name,).upper()requested_columnsarguments.get(columns,[])limitmin(arguments.get(limit,10),100)filter_colarguments.get(filter_column)filter_valarguments.get(filter_value)iftable_namenotinALLOWED_TABLES:return[TextContent(typetext,textfError: Table {table_name} is not allowed.)]allowed_colsALLOWED_TABLES[table_name]ifnotrequested_columnsor*inrequested_columns:select_colsallowed_colselse:invalid_cols[colforcolinrequested_columnsifcol.upper()notinallowed_cols]ifinvalid_cols:return[TextContent(typetext,textfError: Columns{invalid_cols}are not allowed.)]select_cols[col.upper()forcolinrequested_columns]col_str, .join(select_cols)# 修复使用字典传参更安全可靠params{}base_sqlfSELECT{col_str}FROM{table_name}iffilter_colandfilter_valisnotNone:iffilter_col.upper()notinallowed_cols:return[TextContent(typetext,textfError: Filter column {filter_col} is not allowed.)]base_sqlf WHERE{filter_col.upper()} :bind_valparams[:bind_val]filter_val# Oracle 11g 兼容写法sqlfSELECT * FROM ({base_sql}) WHERE ROWNUM {limit}connget_db_connection()cursorconn.cursor()try:cursor.execute(sql,params)columns[desc[0]fordescincursor.description]rowscursor.fetchall()result_textfQuery Result for{table_name}:\nresult_text| | .join(columns) |\nresult_text||.join([---for_incolumns])|\nforrowinrows:formatted_row[str(item)ifitemisnotNoneelseNULLforiteminrow]result_text| | .join(formatted_row) |\nreturn[TextContent(typetext,textresult_text)]finally:# 无论是否报错必须关闭游标并将连接归还连接池cursor.close()conn.close()else:return[TextContent(typetext,textfUnknown tool:{name})]exceptExceptionase:logger.error(fError executing tool{name}:{e})return[TextContent(typetext,textfError:{str(e)})]defmain():# 1. 强制启用 Thick 模式并指定路径init_oracle_client()# 2. 初始化连接池try:init_connection_pool()exceptException:logger.error(❌ Database connection pool failed to initialize. MCP Server will not start.)returnsseSseServerTransport(/messages/)asyncdefhandle_sse(request:Request):asyncwithsse.connect_sse(request.scope,request.receive,request._send)as(read_stream,write_stream):awaitapp.run(read_stream,write_stream,app.create_initialization_options())asyncdefhandle_messages(request:Request):awaitsse.handle_post_message(request.scope,request.receive,request._send)returnResponse(status_code200)starlette_appStarlette(debugTrue,routes[Route(/sse,endpointhandle_sse),Route(/messages/,endpointhandle_messages,methods[POST]),Route(/messages,endpointhandle_messages,methods[POST]),],)logger.info( Starting Oracle MCP Server on port 白名单端口号...)uvicorn.run(starlette_app,host0.0.0.0,port白名单端口号)if__name____main__:main()Workbuddy端操作路径截图MCP配置代码{mcpServers:{Oracle-Connector:{transport:sse,url:http://地址:端口/sse,disabled:false}}}结论通过上述案例可以将系统的数据放在内网通过物理网络数据库单独账号控制权限同时MCP服务器单独部署也进一步杜绝了网络安全问题以及AI查数据不可控的情况。
