mindroot.coreplugins.admin package¶
Submodules¶
mindroot.coreplugins.admin.agent_importer module¶
- mindroot.coreplugins.admin.agent_importer.download_github_files(repo_path: str, tag: str = None) tuple[str, str][source]¶
Download GitHub repo files to temp directory
- mindroot.coreplugins.admin.agent_importer.import_agent(agent_name: str, agent_info: Dict, scope: str) None[source]¶
- mindroot.coreplugins.admin.agent_importer.import_github_agent(repo_path: str, scope: str, tag: str = None) Dict[str, any][source]¶
Import an agent from a GitHub repository
- mindroot.coreplugins.admin.agent_importer.import_persona(persona_name: str, source_dir: Path, scope: str) None[source]¶
mindroot.coreplugins.admin.agent_router module¶
- class mindroot.coreplugins.admin.agent_router.GitHubImportRequest(*, repo_path: str, scope: str, tag: str = None)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- repo_path: str¶
- scope: str¶
- tag: str¶
- class mindroot.coreplugins.admin.agent_router.ScanDirectoryRequest(*, directory: str, scope: str)[source]¶
Bases:
BaseModel- directory: str¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- scope: str¶
- mindroot.coreplugins.admin.agent_router.create_agent(scope: str, agent: str = Form(PydanticUndefined))[source]¶
- mindroot.coreplugins.admin.agent_router.export_agent_zip(scope: str, name: str)[source]¶
Export an agent as a zip file
- mindroot.coreplugins.admin.agent_router.get_agent_ownership_info()[source]¶
Get cached ownership information for all agents
- mindroot.coreplugins.admin.agent_router.get_command_providers()[source]¶
Get all available providers for each command
- async mindroot.coreplugins.admin.agent_router.get_full_agent_data(scope: str, name: str)[source]¶
Get complete agent data including persona
- async mindroot.coreplugins.admin.agent_router.import_agent_zip(scope: str, file: UploadFile = File(PydanticUndefined))[source]¶
Import an agent from a zip file
- mindroot.coreplugins.admin.agent_router.import_github_agent_endpoint(request: GitHubImportRequest)[source]¶
- async mindroot.coreplugins.admin.agent_router.load_agent_data(agent_name: str) dict[source]¶
Load agent data from local or shared directory
- async mindroot.coreplugins.admin.agent_router.load_persona_data(persona_name: str) dict[source]¶
Load persona data from local or shared directory
- mindroot.coreplugins.admin.agent_router.refresh_agent_ownership_cache()[source]¶
Force refresh of the agent ownership cache
- mindroot.coreplugins.admin.agent_router.scan_agent_ownership() Dict[str, Any][source]¶
Scan all agents and build ownership information
- mindroot.coreplugins.admin.agent_router.scan_and_import_agents_endpoint(request: ScanDirectoryRequest)[source]¶
mindroot.coreplugins.admin.asset_manager module¶
- class mindroot.coreplugins.admin.asset_manager.AssetManager(base_dir: str = 'registry_assets')[source]¶
Bases:
objectManages deduplicated asset storage for personas
- decrement_reference_count(file_hash: str) bool[source]¶
Decrement reference count and delete if zero. Returns True if deleted.
mindroot.coreplugins.admin.command_router module¶
- async mindroot.coreplugins.admin.command_router.check_recommended_plugins(agent_name: str)[source]¶
Check which recommended plugins are not installed.
- async mindroot.coreplugins.admin.command_router.command_plugin_mapping()[source]¶
Get mapping of commands to plugins that provide them.
- async mindroot.coreplugins.admin.command_router.install_agent_recommended_plugins(agent_name: str)[source]¶
Install plugins recommended for an agent.
- async mindroot.coreplugins.admin.command_router.install_queued_plugins()[source]¶
Install all queued plugins.
- async mindroot.coreplugins.admin.command_router.missing_commands(agent_name: str)[source]¶
Get commands mentioned in agent instructions but not available.
mindroot.coreplugins.admin.mcp_catalog_routes module¶
- async mindroot.coreplugins.admin.mcp_catalog_routes.get_catalog_server_info(server_name: str)[source]¶
Get detailed information about a specific server from the catalog.
- async mindroot.coreplugins.admin.mcp_catalog_routes.get_mcp_catalog()[source]¶
Get the MCP server catalog/directory.
- async mindroot.coreplugins.admin.mcp_catalog_routes.get_mcp_categories()[source]¶
Get available MCP server categories.
- async mindroot.coreplugins.admin.mcp_catalog_routes.install_from_catalog(server_name: str)[source]¶
Install an MCP server from the catalog.
mindroot.coreplugins.admin.mcp_publish_routes module¶
- class mindroot.coreplugins.admin.mcp_publish_routes.McpCompleteOAuthRequest(*, server_name: str, code: str, state: str | None = None)[source]¶
Bases:
BaseModel- code: str¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- server_name: str¶
- state: str | None¶
- class mindroot.coreplugins.admin.mcp_publish_routes.McpServerPublishRequest(*, name: str, description: str, server_type: str, tools: List[Dict[str, Any]], command: str | None = None, args: List[str] | None = None, env: Dict[str, str] | None = None, url: str | None = None)[source]¶
Bases:
BaseModel- args: List[str] | None¶
- command: str | None¶
- description: str¶
- env: Dict[str, str] | None¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- name: str¶
- server_type: str¶
- tools: List[Dict[str, Any]]¶
- url: str | None¶
- class mindroot.coreplugins.admin.mcp_publish_routes.McpTestLocalRequest(*, name: str, command: str, args: List[str] = [], env: Dict[str, str] = {})[source]¶
Bases:
BaseModel- args: List[str]¶
- command: str¶
- env: Dict[str, str]¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- name: str¶
- class mindroot.coreplugins.admin.mcp_publish_routes.McpTestRemoteRequest(*, url: str, name: str | None = None)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- name: str | None¶
- url: str¶
- async mindroot.coreplugins.admin.mcp_publish_routes.complete_oauth_flow(request: McpCompleteOAuthRequest)[source]¶
Complete OAuth flow for MCP server testing.
- async mindroot.coreplugins.admin.mcp_publish_routes.get_oauth_status(server_name: str)[source]¶
Get OAuth flow status for a server.
- async mindroot.coreplugins.admin.mcp_publish_routes.publish_mcp_server(request: McpServerPublishRequest)[source]¶
Publish an MCP server to the registry.
- async mindroot.coreplugins.admin.mcp_publish_routes.test_direct_connection(url: str)[source]¶
Test direct connection to MCP server without OAuth to check if auth is required.
- async mindroot.coreplugins.admin.mcp_publish_routes.test_local_mcp_server(request: McpTestLocalRequest)[source]¶
Test connection to a local MCP server and list its capabilities.
- async mindroot.coreplugins.admin.mcp_publish_routes.test_remote_mcp_server(request: McpTestRemoteRequest)[source]¶
Test connection to a remote MCP server and list its tools using MCP manager OAuth flow.
mindroot.coreplugins.admin.mcp_registry_routes module¶
MCP Registry Integration Routes
Handles browsing and installing MCP servers from the registry.
- class mindroot.coreplugins.admin.mcp_registry_routes.RegistryBrowseRequest(*, category: str | None = None, search: str | None = None, page: int = 1, limit: int = 20)[source]¶
Bases:
BaseModel- category: str | None¶
- limit: int¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- page: int¶
- search: str | None¶
- class mindroot.coreplugins.admin.mcp_registry_routes.RegistryServerInstallRequest(*, registry_id: str, server_name: str | None = None)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- registry_id: str¶
- server_name: str | None¶
- async mindroot.coreplugins.admin.mcp_registry_routes.browse_registry_servers(category: str | None = None, search: str | None = None, page: int = 1, limit: int = 20)[source]¶
Browse MCP servers from the registry.
- async mindroot.coreplugins.admin.mcp_registry_routes.complete_registry_oauth(server_name: str, code: str, state: str | None = None)[source]¶
Complete OAuth flow for registry server installation.
- async mindroot.coreplugins.admin.mcp_registry_routes.get_installation_status()[source]¶
Get installation status of all registry servers.
- async mindroot.coreplugins.admin.mcp_registry_routes.get_registry_categories()[source]¶
Get available MCP server categories from the registry.
- async mindroot.coreplugins.admin.mcp_registry_routes.get_registry_server_details(registry_id: str)[source]¶
Get detailed information about a specific registry server.
- async mindroot.coreplugins.admin.mcp_registry_routes.install_registry_server(request: RegistryServerInstallRequest)[source]¶
Install an MCP server from the registry.
mindroot.coreplugins.admin.mcp_routes module¶
- class mindroot.coreplugins.admin.mcp_routes.McpConnectRequest(*, server_name: str, secrets: Dict[str, str] | None = None)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- secrets: Dict[str, str] | None¶
- server_name: str¶
- class mindroot.coreplugins.admin.mcp_routes.McpLocalTestRequest(*, name: str, command: str, args: List[str] = [], env: Dict[str, str] = {}, secrets: Dict[str, str] | None = None)[source]¶
Bases:
BaseModel- args: List[str]¶
- command: str¶
- env: Dict[str, str]¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- name: str¶
- secrets: Dict[str, str] | None¶
- class mindroot.coreplugins.admin.mcp_routes.McpOAuthCallbackRequest(*, server_name: str, code: str, state: str | None = None)[source]¶
Bases:
BaseModel- code: str¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- server_name: str¶
- state: str | None¶
- class mindroot.coreplugins.admin.mcp_routes.McpServerAddRequest(*, name: str, description: str, command: str | None = None, args: List[str] = [], env: dict = {}, transport: str = 'stdio', url: str | None = None, provider_url: str | None = None, transport_url: str | None = None, transport_type: str | None = None, auth_type: str = 'none', auth_headers: Dict[str, str] = {}, authorization_server_url: str | None = None, client_id: str | None = None, client_secret: str | None = None, scopes: List[str] = [], redirect_uri: str | None = None)[source]¶
Bases:
BaseModel- args: List[str]¶
- auth_headers: Dict[str, str]¶
- auth_type: str¶
- authorization_server_url: str | None¶
- client_id: str | None¶
- client_secret: str | None¶
- command: str | None¶
- description: str¶
- env: dict¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- name: str¶
- provider_url: str | None¶
- redirect_uri: str | None¶
- scopes: List[str]¶
- transport: str¶
- transport_type: str | None¶
- transport_url: str | None¶
- url: str | None¶
- class mindroot.coreplugins.admin.mcp_routes.McpServerRequest(*, server_name: str)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- server_name: str¶
- async mindroot.coreplugins.admin.mcp_routes.add_mcp_server(server_request: McpServerAddRequest)[source]¶
Add a new MCP server configuration.
- async mindroot.coreplugins.admin.mcp_routes.connect_mcp_server(request: McpConnectRequest)[source]¶
Connect to an MCP server.
- async mindroot.coreplugins.admin.mcp_routes.disconnect_mcp_server(request: McpServerRequest)[source]¶
Disconnect from an MCP server.
- async mindroot.coreplugins.admin.mcp_routes.list_mcp_servers()[source]¶
List all configured MCP servers.
- async mindroot.coreplugins.admin.mcp_routes.remove_mcp_server(request: McpServerRequest)[source]¶
Remove an MCP server configuration.
mindroot.coreplugins.admin.mod module¶
mindroot.coreplugins.admin.oauth_callback_router module¶
OAuth callback router for MCP servers.
This router contains public routes that do not require authentication, as external OAuth providers need to be able to redirect to these endpoints.
mindroot.coreplugins.admin.persona_handler module¶
- mindroot.coreplugins.admin.persona_handler.handle_persona_import(persona_data: dict, scope: str, owner: str = None) str[source]¶
Handle importing a persona from embedded data in agent configuration. Returns the persona name to be used in agent configuration.
- Parameters:
persona_data – Dictionary containing persona data or string with persona name
scope – ‘local’ or ‘shared’
- Returns:
Name of the persona to reference in agent config
- Return type:
str
mindroot.coreplugins.admin.persona_router module¶
- mindroot.coreplugins.admin.persona_router.create_persona(scope: str, persona: str = Form(PydanticUndefined), faceref: UploadFile = File(None), avatar: UploadFile = File(None))[source]¶
- mindroot.coreplugins.admin.persona_router.create_registry_persona(persona: str = Form(PydanticUndefined), owner: str = Form(PydanticUndefined))[source]¶
Create a registry persona with owner namespace
- mindroot.coreplugins.admin.persona_router.create_registry_persona_with_assets(persona: str = Form(PydanticUndefined), owner: str = Form(PydanticUndefined), faceref: UploadFile = File(None), avatar: UploadFile = File(None))[source]¶
Create a registry persona with deduplicated asset storage
- mindroot.coreplugins.admin.persona_router.read_persona_by_path(persona_path: str)[source]¶
Read persona by full path (supports registry/owner/name format)
mindroot.coreplugins.admin.plugin_manager module¶
- class mindroot.coreplugins.admin.plugin_manager.DirectoryRequest(*, directory: str)[source]¶
Bases:
BaseModel- directory: str¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- class mindroot.coreplugins.admin.plugin_manager.GitHubPluginRequest(*, plugin: str, url: str | None = None, github_url: str | None = None)[source]¶
Bases:
BaseModel- github_url: str | None¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- plugin: str¶
- url: str | None¶
- class mindroot.coreplugins.admin.plugin_manager.InstallFromIndexRequest(*, plugin: str, index_name: str)[source]¶
Bases:
BaseModel- index_name: str¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- plugin: str¶
- class mindroot.coreplugins.admin.plugin_manager.PluginMetadata(*, description: str | None = None, commands: List[str] | None = None, services: List[str] | None = None, dependencies: List[str] | None = None)[source]¶
Bases:
BaseModel- commands: List[str] | None¶
- dependencies: List[str] | None¶
- description: str | None¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- services: List[str] | None¶
- class mindroot.coreplugins.admin.plugin_manager.PluginRequest(*, plugin: str)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- plugin: str¶
- class mindroot.coreplugins.admin.plugin_manager.StreamInstallRequest(*, plugin: str, source: str, source_path: str = None)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- plugin: str¶
- source: str¶
- source_path: str¶
- class mindroot.coreplugins.admin.plugin_manager.TogglePluginRequest(*, plugin: str, enabled: bool)[source]¶
Bases:
BaseModel- enabled: bool¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- plugin: str¶
- async mindroot.coreplugins.admin.plugin_manager.install_from_index(request: InstallFromIndexRequest)[source]¶
- async mindroot.coreplugins.admin.plugin_manager.install_github_plugin(request: GitHubPluginRequest)[source]¶
- async mindroot.coreplugins.admin.plugin_manager.install_local_plugin(request: PluginRequest)[source]¶
- async mindroot.coreplugins.admin.plugin_manager.publish_plugin_from_github(repo: str, registry_token: str, registry_url: str)[source]¶
Fetches plugin_info.json from a GitHub repo and publishes it to the registry.
- async mindroot.coreplugins.admin.plugin_manager.scan_directory(request: DirectoryRequest)[source]¶
- async mindroot.coreplugins.admin.plugin_manager.stream_install_plugin(request: StreamInstallRequest)[source]¶
Stream the installation process of a plugin using SSE (POST method).
- async mindroot.coreplugins.admin.plugin_manager.stream_install_plugin_get(request: Request)[source]¶
Stream the installation process of a plugin using SSE (GET method).
- async mindroot.coreplugins.admin.plugin_manager.toggle_plugin(request: TogglePluginRequest)[source]¶
mindroot.coreplugins.admin.plugin_router module¶
- class mindroot.coreplugins.admin.plugin_router.PluginUpdateRequest(*, plugins: dict)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- plugins: dict¶
- mindroot.coreplugins.admin.plugin_router.update_plugins(request: PluginUpdateRequest)[source]¶
mindroot.coreplugins.admin.plugin_router_fixed module¶
mindroot.coreplugins.admin.plugin_router_new_not_working module¶
mindroot.coreplugins.admin.plugin_routes module¶
- class mindroot.coreplugins.admin.plugin_routes.GithubPublishRequest(*, repo: str, registry_url: str | None = None)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- registry_url: str | None¶
- repo: str¶
- class mindroot.coreplugins.admin.plugin_routes.PluginUpdateRequest(*, plugins: dict)[source]¶
Bases:
BaseModel- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- plugins: dict¶
- async mindroot.coreplugins.admin.plugin_routes.publish_plugin_from_github(request: GithubPublishRequest, authorization: str | None = Header(None))[source]¶
Publish a plugin from GitHub repository to the registry.
This endpoint allows publishing a plugin by simply providing the GitHub repository in the format ‘username/repo’. It will fetch the plugin_info.json from the repository and publish it to the configured registry.
The registry token can be provided via: 1. Authorization header: “Bearer <token>” 2. REGISTRY_TOKEN environment variable 3. registry_token in data/registry_settings.json
- mindroot.coreplugins.admin.plugin_routes.update_plugins(request: PluginUpdateRequest)[source]¶
Update plugin enabled/disabled status.
mindroot.coreplugins.admin.registry_settings_routes module¶
- async mindroot.coreplugins.admin.registry_settings_routes.clear_registry_token()[source]¶
Clear the stored registry token.
- async mindroot.coreplugins.admin.registry_settings_routes.get_registry_settings()[source]¶
Get registry settings including token status.
mindroot.coreplugins.admin.router module¶
- async mindroot.coreplugins.admin.router.get_model_preferences_v2_html()[source]¶
Serve the new Model Preferences V2 page
mindroot.coreplugins.admin.server_router module¶
mindroot.coreplugins.admin.service_models module¶
- async mindroot.coreplugins.admin.service_models.cached_get_service_models()[source]¶
Retrieve service models from the cache or fetch them if not available. :returns:
- {
- “service_name”: {
“provider_name”: [“model1”, “model2”, …], …
}
- Return type:
Dictionary mapping services to providers to models
- async mindroot.coreplugins.admin.service_models.get_service_models_from_providers(timeout: float = 500.0, context=None) Dict[str, Dict[str, List[str]]][source]¶
Gather service models from all providers.
- Parameters:
timeout – Maximum time in seconds to wait for each provider response
- Returns:
- {
- “service_name”: {
“provider_name”: [“model1”, “model2”, …], …
}
- Return type:
Dictionary mapping services to providers to models
mindroot.coreplugins.admin.settings_router module¶
- class mindroot.coreplugins.admin.settings_router.Setting(*, service_or_command_name: str, flag: str, model: str)[source]¶
Bases:
BaseModel- flag: str¶
- model: str¶
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- service_or_command_name: str¶
- async mindroot.coreplugins.admin.settings_router.get_settings_v2()[source]¶
Get preferences in new v2 format: {service: [[provider, model], …]}
- async mindroot.coreplugins.admin.settings_router.migrate_settings()[source]¶
One-time migration from old format to new v2 format.
- async mindroot.coreplugins.admin.settings_router.save_settings_v2(request: Request)[source]¶
Save preferences in new v2 format.