mcp: MCP server calling application use cases
This commit is contained in:
335
crates/mcp/src/main.rs
Normal file
335
crates/mcp/src/main.rs
Normal file
@@ -0,0 +1,335 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use domain::ports::{IMediaProvider, IProviderRegistry, ProviderCapabilities, StreamingProtocol};
|
||||
use domain::{DomainError, DomainResult, MediaFilter, MediaItemId, MediaItem, ScheduleEngineService};
|
||||
use domain::ports::StreamQuality;
|
||||
use infra_wiring::DbPool;
|
||||
use tracing::info;
|
||||
use uuid::Uuid;
|
||||
|
||||
mod error;
|
||||
mod server;
|
||||
mod tools;
|
||||
|
||||
use server::KTvMcpServer;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> anyhow::Result<()> {
|
||||
let _ = dotenvy::dotenv();
|
||||
|
||||
tracing_subscriber::fmt()
|
||||
.with_writer(std::io::stderr)
|
||||
.with_env_filter(
|
||||
tracing_subscriber::EnvFilter::from_default_env()
|
||||
.add_directive("mcp=info".parse().unwrap()),
|
||||
)
|
||||
.init();
|
||||
|
||||
let database_url = std::env::var("DATABASE_URL")
|
||||
.unwrap_or_else(|_| "sqlite:data.db?mode=rwc".to_string());
|
||||
|
||||
let owner_id: Uuid = std::env::var("MCP_USER_ID")
|
||||
.map_err(|_| anyhow::anyhow!("MCP_USER_ID env var is required (UUID of the user)"))?
|
||||
.parse()
|
||||
.map_err(|_| anyhow::anyhow!("MCP_USER_ID must be a valid UUID"))?;
|
||||
|
||||
info!("Connecting to database: {}", database_url);
|
||||
|
||||
let pool = DbPool::connect(&database_url).await?;
|
||||
pool.run_migrations().await?;
|
||||
|
||||
let wire = wire_repositories(&pool)?;
|
||||
|
||||
let provider_registry = build_provider_registry().await;
|
||||
|
||||
let event_bus = Arc::new(adapter_event_publisher::ChannelEventBus::new(64));
|
||||
let event_publisher: Arc<dyn domain::ports::EventPublisher> = event_bus.clone();
|
||||
|
||||
let schedule_engine = Arc::new(ScheduleEngineService::new(
|
||||
provider_registry.clone(),
|
||||
wire.channel_query.clone(),
|
||||
wire.schedule_query.clone(),
|
||||
wire.schedule_command.clone(),
|
||||
));
|
||||
|
||||
let channel_cmd_deps = Arc::new(application::channels::ChannelCommandDeps {
|
||||
channel_command: wire.channel_command.clone(),
|
||||
channel_query: wire.channel_query.clone(),
|
||||
event_publisher: event_publisher.clone(),
|
||||
});
|
||||
|
||||
let channel_query_deps = Arc::new(application::channels::ChannelQueryDeps {
|
||||
channel_query: wire.channel_query.clone(),
|
||||
});
|
||||
|
||||
let schedule_deps = Arc::new(application::schedule::ScheduleDeps {
|
||||
schedule_engine,
|
||||
channel_query: wire.channel_query.clone(),
|
||||
schedule_query: wire.schedule_query.clone(),
|
||||
schedule_command: wire.schedule_command.clone(),
|
||||
event_publisher: event_publisher.clone(),
|
||||
});
|
||||
|
||||
let library_query_deps = Arc::new(application::library::LibraryQueryDeps {
|
||||
library_query: wire.library_query.clone(),
|
||||
});
|
||||
|
||||
let server = KTvMcpServer {
|
||||
channel_cmd_deps,
|
||||
channel_query_deps,
|
||||
schedule_deps,
|
||||
library_query_deps,
|
||||
owner_id,
|
||||
};
|
||||
|
||||
info!("K-TV MCP server starting (stdio transport), owner_id={}", owner_id);
|
||||
|
||||
use rmcp::ServiceExt;
|
||||
let service = server
|
||||
.serve(rmcp::transport::stdio())
|
||||
.await
|
||||
.inspect_err(|e| tracing::error!("MCP server error: {e}"))?;
|
||||
|
||||
service.waiting().await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
struct WireOutput {
|
||||
channel_command: Arc<dyn domain::ports::ChannelCommand>,
|
||||
channel_query: Arc<dyn domain::ports::ChannelQuery>,
|
||||
schedule_command: Arc<dyn domain::ports::ScheduleCommand>,
|
||||
schedule_query: Arc<dyn domain::ports::ScheduleQuery>,
|
||||
library_query: Arc<dyn domain::ports::LibraryQuery>,
|
||||
}
|
||||
|
||||
fn wire_repositories(pool: &DbPool) -> anyhow::Result<WireOutput> {
|
||||
match pool {
|
||||
#[cfg(feature = "sqlite")]
|
||||
DbPool::Sqlite(sqlite_pool) => {
|
||||
let w = adapter_sqlite::wire(sqlite_pool.clone());
|
||||
Ok(WireOutput {
|
||||
channel_command: w.channel_command,
|
||||
channel_query: w.channel_query,
|
||||
schedule_command: w.schedule_command,
|
||||
schedule_query: w.schedule_query,
|
||||
library_query: w.library_query,
|
||||
})
|
||||
}
|
||||
#[cfg(feature = "postgres")]
|
||||
DbPool::Postgres(pg_pool) => {
|
||||
let w = adapter_postgres::wire(pg_pool.clone());
|
||||
Ok(WireOutput {
|
||||
channel_command: w.channel_command,
|
||||
channel_query: w.channel_query,
|
||||
schedule_command: w.schedule_command,
|
||||
schedule_query: w.schedule_query,
|
||||
library_query: w.library_query,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn build_provider_registry() -> Arc<dyn IProviderRegistry> {
|
||||
let mut providers: Vec<(String, Arc<dyn IMediaProvider>)> = Vec::new();
|
||||
|
||||
#[cfg(feature = "jellyfin")]
|
||||
if let (Some(url), Some(api_key), Some(user_id)) = (
|
||||
std::env::var("JELLYFIN_BASE_URL").ok(),
|
||||
std::env::var("JELLYFIN_API_KEY").ok(),
|
||||
std::env::var("JELLYFIN_USER_ID").ok(),
|
||||
) {
|
||||
info!("Media provider: Jellyfin at {}", url);
|
||||
providers.push((
|
||||
"jellyfin".to_string(),
|
||||
Arc::new(adapter_jellyfin::JellyfinMediaProvider::new(
|
||||
adapter_jellyfin::JellyfinConfig {
|
||||
base_url: url,
|
||||
api_key,
|
||||
user_id,
|
||||
},
|
||||
)),
|
||||
));
|
||||
}
|
||||
|
||||
if providers.is_empty() {
|
||||
tracing::warn!("No media provider configured. Set JELLYFIN_BASE_URL.");
|
||||
providers.push(("noop".to_string(), Arc::new(NoopMediaProvider)));
|
||||
}
|
||||
|
||||
Arc::new(SimpleProviderRegistry::new(providers))
|
||||
}
|
||||
|
||||
struct NoopMediaProvider;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl IMediaProvider for NoopMediaProvider {
|
||||
fn capabilities(&self) -> ProviderCapabilities {
|
||||
ProviderCapabilities {
|
||||
collections: false,
|
||||
series: false,
|
||||
genres: false,
|
||||
tags: false,
|
||||
decade: false,
|
||||
search: false,
|
||||
streaming_protocol: StreamingProtocol::DirectFile,
|
||||
rescan: false,
|
||||
transcode: false,
|
||||
}
|
||||
}
|
||||
|
||||
async fn fetch_items(&self, _: &MediaFilter) -> DomainResult<Vec<MediaItem>> {
|
||||
Err(DomainError::InfrastructureError(
|
||||
"No media provider configured.".into(),
|
||||
))
|
||||
}
|
||||
|
||||
async fn fetch_by_id(&self, _: &MediaItemId) -> DomainResult<Option<MediaItem>> {
|
||||
Err(DomainError::InfrastructureError(
|
||||
"No media provider configured.".into(),
|
||||
))
|
||||
}
|
||||
|
||||
async fn get_stream_url(&self, _: &MediaItemId, _: &StreamQuality) -> DomainResult<String> {
|
||||
Err(DomainError::InfrastructureError(
|
||||
"No media provider configured.".into(),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
struct SimpleProviderRegistry {
|
||||
providers: Vec<(String, Arc<dyn IMediaProvider>)>,
|
||||
}
|
||||
|
||||
impl SimpleProviderRegistry {
|
||||
fn new(providers: Vec<(String, Arc<dyn IMediaProvider>)>) -> Self {
|
||||
Self { providers }
|
||||
}
|
||||
|
||||
fn get(&self, id: &str) -> Option<&Arc<dyn IMediaProvider>> {
|
||||
self.providers.iter().find(|(k, _)| k == id).map(|(_, v)| v)
|
||||
}
|
||||
|
||||
fn primary(&self) -> Option<&Arc<dyn IMediaProvider>> {
|
||||
self.providers.first().map(|(_, v)| v)
|
||||
}
|
||||
|
||||
fn extract_provider_id(item_id: &str) -> Option<&str> {
|
||||
item_id.find("::").map(|pos| &item_id[..pos])
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl IProviderRegistry for SimpleProviderRegistry {
|
||||
async fn fetch_items(
|
||||
&self,
|
||||
provider_id: &str,
|
||||
filter: &MediaFilter,
|
||||
) -> DomainResult<Vec<MediaItem>> {
|
||||
let id = if provider_id.is_empty() {
|
||||
self.providers.first().map(|(k, _)| k.as_str()).unwrap_or("")
|
||||
} else {
|
||||
provider_id
|
||||
};
|
||||
let provider = self
|
||||
.get(id)
|
||||
.ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?;
|
||||
provider.fetch_items(filter).await
|
||||
}
|
||||
|
||||
async fn fetch_by_id(&self, item_id: &MediaItemId) -> DomainResult<Option<MediaItem>> {
|
||||
let id_str = item_id.value();
|
||||
if let Some(pid) = Self::extract_provider_id(id_str)
|
||||
&& let Some(provider) = self.get(pid)
|
||||
{
|
||||
return provider.fetch_by_id(item_id).await;
|
||||
}
|
||||
if let Some(provider) = self.primary() {
|
||||
provider.fetch_by_id(item_id).await
|
||||
} else {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_stream_url(
|
||||
&self,
|
||||
item_id: &MediaItemId,
|
||||
quality: &StreamQuality,
|
||||
) -> DomainResult<String> {
|
||||
let id_str = item_id.value();
|
||||
if let Some(pid) = Self::extract_provider_id(id_str)
|
||||
&& let Some(provider) = self.get(pid)
|
||||
{
|
||||
return provider.get_stream_url(item_id, quality).await;
|
||||
}
|
||||
if let Some(provider) = self.primary() {
|
||||
provider.get_stream_url(item_id, quality).await
|
||||
} else {
|
||||
Err(DomainError::InfrastructureError(
|
||||
"No provider available".into(),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
fn provider_ids(&self) -> Vec<String> {
|
||||
self.providers.iter().map(|(k, _)| k.clone()).collect()
|
||||
}
|
||||
|
||||
fn primary_id(&self) -> &str {
|
||||
self.providers
|
||||
.first()
|
||||
.map(|(k, _)| k.as_str())
|
||||
.unwrap_or("")
|
||||
}
|
||||
|
||||
fn capabilities(&self, provider_id: &str) -> Option<ProviderCapabilities> {
|
||||
self.get(provider_id).map(|p| p.capabilities())
|
||||
}
|
||||
|
||||
async fn list_collections(
|
||||
&self,
|
||||
provider_id: &str,
|
||||
) -> DomainResult<Vec<domain::ports::Collection>> {
|
||||
let id = if provider_id.is_empty() {
|
||||
self.primary_id()
|
||||
} else {
|
||||
provider_id
|
||||
};
|
||||
let provider = self
|
||||
.get(id)
|
||||
.ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?;
|
||||
provider.list_collections().await
|
||||
}
|
||||
|
||||
async fn list_series(
|
||||
&self,
|
||||
provider_id: &str,
|
||||
collection_id: Option<&str>,
|
||||
) -> DomainResult<Vec<domain::ports::SeriesSummary>> {
|
||||
let id = if provider_id.is_empty() {
|
||||
self.primary_id()
|
||||
} else {
|
||||
provider_id
|
||||
};
|
||||
let provider = self
|
||||
.get(id)
|
||||
.ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?;
|
||||
provider.list_series(collection_id).await
|
||||
}
|
||||
|
||||
async fn list_genres(
|
||||
&self,
|
||||
provider_id: &str,
|
||||
content_type: Option<&domain::ContentType>,
|
||||
) -> DomainResult<Vec<String>> {
|
||||
let id = if provider_id.is_empty() {
|
||||
self.primary_id()
|
||||
} else {
|
||||
provider_id
|
||||
};
|
||||
let provider = self
|
||||
.get(id)
|
||||
.ok_or_else(|| DomainError::InfrastructureError(format!("Unknown provider: {id}")))?;
|
||||
provider.list_genres(content_type).await
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user