# Directory Structure
```
├── .gitignore
├── .python-version
├── AGENTS.md
├── CONTRIBUTING.md
├── examples
│ └── basic_mission.py
├── pyproject.toml
├── README.md
├── src
│ └── stk_mcp
│ ├── __init__.py
│ ├── app.py
│ ├── cli.py
│ ├── stk_logic
│ │ ├── analysis.py
│ │ ├── config.py
│ │ ├── core.py
│ │ ├── decorators.py
│ │ ├── location.py
│ │ ├── logging_config.py
│ │ ├── objects.py
│ │ ├── satellite.py
│ │ ├── scenario.py
│ │ └── utils.py
│ └── tools
│ ├── __init__.py
│ ├── analysis.py
│ ├── health.py
│ ├── location.py
│ ├── objects.py
│ ├── satellite.py
│ └── scenario.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.12
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
*.whl
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# STK-MCP
[](https://www.python.org/downloads/) [](https://pypi.org/project/mcp/)
STK-MCP is an MCP (Model Context Protocol) server designed to enable Large Language Models (LLMs) or other MCP clients to interact with [Ansys/AGI STK](https://www.ansys.com/products/missions/ansys-stk) (Systems Tool Kit) - the leading Digital Mission Engineering software.
This project allows controlling STK via an MCP server, supporting both STK Desktop (Windows only) and STK Engine (Windows & Linux). It utilizes `FastMCP` from the official [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk).
## Overview
The primary goal of this project is to bridge the gap between programmatic interaction and the powerful simulation capabilities of STK. By exposing STK functionalities through a robust CLI and an MCP server, users can command STK simulations using simple commands or LLM-driven applications.
The MCP application, defined in `src/stk_mcp/app.py`, exposes STK operations as MCP tools, which are dynamically managed by a CLI entry point in `src/stk_mcp/cli.py`.
## Features
* CLI entry point powered by `Typer`.
* Dual mode operation: STK Engine (Windows/Linux) and STK Desktop (Windows).
* OS-aware: Desktop mode auto-disabled on non-Windows platforms.
* Managed lifecycle: STK instance is started/stopped with the MCP server.
* Tool discovery: `list-tools` command enumerates available MCP tools.
* Modular architecture: CLI (`cli.py`), MCP (`app.py`), STK logic (`stk_logic/`), and MCP tools (`tools/`).
## Prerequisites
* **Operating System:** Windows or Linux. STK Desktop mode requires Windows.
* **Python:** Version 3.12 or higher.
* **Ansys/AGI STK:** Version 12.x Desktop or Engine installed.
* **STK Python API:** The `agi.stk12` Python wheel corresponding to your STK installation must be available. Typically found under `CodeSamples\Automation\Python` in your STK install.
## Installation
1. Clone the repository
```bash
git clone <repository-url>
cd stk-mcp
```
2. Create and activate a virtual environment
```bash
# Create the virtual environment
uv venv
# Activate it
# On Windows (in PowerShell/CMD):
# .venv\Scripts\activate
# On Linux (in bash/zsh):
source .venv/bin/activate
```
3. Add dependencies with uv (preferred)
- Add the STK Python wheel from your STK installation (local file):
```bash
uv add ./agi.stk12-12.10.0-py3-none-any.whl
# or: uv add path/to/your/STK/CodeSamples/Automation/Python/agi.stk12-*.whl
# Windows only: COM bridge for Desktop automation
uv add "pywin32; platform_system == 'Windows'"
```
4. Sync the environment (installs deps from `pyproject.toml`)
```bash
uv sync
```
## Usage
This project is a command-line application. Ensure your virtual environment is activated before running commands.
### Listing Available Tools
```bash
uv run -m stk_mcp.cli list-tools
```
Prints a table of tool names and their descriptions.
### Running the MCP Server
Use the `run` command to start the MCP server. The server will automatically start and manage an STK instance.
Run with `uv run` so you don’t need to install the package into site-packages.
**1) STK Engine (recommended for automation, Windows/Linux):**
```bash
uv run -m stk_mcp.cli run --mode engine
```
**2) STK Desktop (Windows only, shows GUI):**
Ensure STK Desktop is closed; the server will launch and manage its own instance.
```bash
uv run -m stk_mcp.cli run --mode desktop
```
The server will start, initialize STK, and listen for MCP connections on `http://127.0.0.1:8765` by default.
**3. Command Options:**
You can see all options with the `--help` flag:
```bash
stk-mcp run --help
```
### Interacting with the Server
Once the server is running, you can connect to it using any MCP client, such as the MCP Inspector.
1. Open the MCP Inspector URL provided in the console (e.g., `http://127.0.0.1:8765`).
2. Find the "STK Control" server in the list.
3. Use the "Tools" section to execute `setup_scenario`, `create_location`, and `create_satellite`.
### Stopping the Server
Press `Ctrl+C` in the terminal where the server is running. The lifecycle manager will automatically close the STK Engine or Desktop instance.
## MCP Tools and Resources
The server exposes the following MCP tools/resources.
| Name | Kind | Description | Desktop (Windows) | Engine (Windows) | Engine (Linux) |
|------------------|----------|-----------------------------------------------------------------------------------------------|-------------------|------------------|----------------|
| `setup_scenario` | Tool | Create/configure an STK Scenario; sets time period and rewinds animation. | Yes | Yes | Yes |
| `create_location`| Tool | Create/update a `Facility` (default) or `Place` at latitude/longitude/altitude (km). | Yes | Yes | Yes |
| `create_satellite`| Tool | Create/configure a satellite from apogee/perigee (km), RAAN, and inclination; TwoBody prop. | Yes | Yes | No |
Notes:
- `create_satellite` on Linux Engine is not yet supported because it relies on COM-specific casts; a Connect-based fallback is planned.
Resources:
| Name | Kind | Description | Desktop (Windows) | Engine (Windows) | Engine (Linux) |
|------|------|-------------|-------------------|------------------|----------------|
| `resource://stk/objects` | Resource | List all objects in the active scenario. Returns JSON records: `{name, type}`. | Yes | Yes | Yes |
| `resource://stk/objects/{type}` | Resource | List objects filtered by `type` (e.g., `satellite`, `facility`, `place`, `sensor`). Returns JSON records. | Yes | Yes | Yes |
| `resource://stk/health` | Resource | Report basic state: mode, scenario name, and object counts. | Yes | Yes | Yes |
| `resource://stk/analysis/access/{object1}/{object2}` | Resource | Compute access intervals between two objects. Provide paths like `Satellite/SatA` and `Facility/FacB` (with or without leading `*/`). | Yes | Yes | Yes |
| `resource://stk/reports/lla/{satellite}` | Resource | Return satellite LLA ephemeris over the scenario start/stop interval. Provide path like `Satellite/SatA` (with or without leading `*/`). | Yes | Yes | Yes |
Examples:
- Read all objects: `resource://stk/objects`
- Read only satellites: `resource://stk/objects/satellite`
- Read ground locations: `resource://stk/objects/location` (alias for facilities and places)
Access and LLA examples:
- Compute access: `resource://stk/analysis/access/Satellite/ISS/Facility/Boulder`
- Get ISS LLA (60 s): `resource://stk/reports/lla/Satellite/ISS` (optional `step_sec` argument)
## Configuration & Logging
Configuration is centralized in `src/stk_mcp/stk_logic/config.py` using `pydantic-settings`.
Defaults can be overridden with environment variables (prefix `STK_MCP_`).
- `STK_MCP_DEFAULT_HOST` (default `127.0.0.1`)
- `STK_MCP_DEFAULT_PORT` (default `8765`)
- `STK_MCP_LOG_LEVEL` (default `INFO`)
- `STK_MCP_DEFAULT_SCENARIO_NAME` (default `MCP_STK_Scenario`)
- `STK_MCP_DEFAULT_START_TIME` (default `20 Jan 2020 17:00:00.000`)
- `STK_MCP_DEFAULT_DURATION_HOURS` (default `48.0`)
Logging is standardized via `src/stk_mcp/stk_logic/logging_config.py`. The CLI uses
this configuration, producing structured logs with timestamps, levels, and context.
## Implementation Notes
- STK access is serialized with a global lock to avoid concurrency issues.
- Common STK-availability checks are handled via decorators in
`src/stk_mcp/stk_logic/decorators.py` (`@require_stk_tool` and `@require_stk_resource`).
- STK Connect commands that may be transiently flaky are executed with retry logic
(`tenacity`) in `src/stk_mcp/stk_logic/utils.py` (`safe_stk_command`).
- Long-running internal operations are timed with `@timed_operation` for diagnostics.
## Dependencies
Managed with `uv`:
* `agi.stk12` (local wheel from your STK install)
* `mcp[cli]>=1.6.0`
* `uvicorn>=0.30` (explicit for CLI server)
* `rich>=13.7` (CLI table output)
* `typer>=0.15.2`
* `pydantic>=2.11.7`
* `pywin32` (Windows only)
Notes:
- On macOS (Darwin), STK Engine/Desktop are not supported. The server will start but STK-dependent tools/resources are unavailable.
- The server serializes STK access via a global lock to avoid concurrency issues with COM/Engine calls.
## Contributing
Contributions are welcome! Please see the [CONTRIBUTING.md](CONTRIBUTING.md) file for guidelines.
```
--------------------------------------------------------------------------------
/AGENTS.md:
--------------------------------------------------------------------------------
```markdown
You are a senior Python software engineer and a domain expert in Orbital Mechanics and Ansys/AGI STK (Satellite/Systems Tool Kit).
You are tasked to work on this project called STK-MCP that allows LLMs to interact with Ansys/AGI STK (Satellite/Systems Tool Kit) using the Model Context Protocol (MCP).
# Guidelines
- always use `uv` to manage the project and dependencies.
- always use `uv add <package_name>` to add packages/dependencies instead of `pip install`.
- to remove a package/dependency, always use `uv remove <package_name>`.
- always use `uv sync` to create/update the virtual environment with the dependencies in `pyproject.toml`.
- always use `uv run` to run the project.
- always aim to make the mcp tools/resources applicalbe with both STK Desktop and STK Engine.
- exceptions should be made for tools/resources that are only applicable to one of the modes.
- when adding a new mcp tool/resource, always:
- add it to the `tools` directory
- add any STK SDK logic code to the `stk_logic` directory
- add the necessary documentation in the `README.md` file (update tools and/or resources table).
```
--------------------------------------------------------------------------------
/CONTRIBUTING.md:
--------------------------------------------------------------------------------
```markdown
# Contributing to STK-MCP
First off, thank you for considering contributing to STK-MCP! Your help is appreciated.
This document provides guidelines for contributing to the project.
## How Can I Contribute?
There are several ways you can contribute:
* **Reporting Bugs:** If you find a bug, please open an issue in the GitHub repository. Describe the bug clearly, including steps to reproduce it, expected behavior, and actual behavior. Include STK version, Python version, and OS information if relevant.
* **Suggesting Enhancements:** Have an idea for a new feature or an improvement to an existing one? Open an issue to discuss it. Provide as much detail as possible about your suggestion.
* **Writing Code:** If you'd like to contribute code, please follow the steps below.
* **Improving Documentation:** If you find errors or areas for improvement in the README or other documentation, feel free to suggest changes.
## Development Process
1. **Fork the repository:** Create your own fork of the `stk-mcp` repository on GitHub.
2. **Clone your fork:** `git clone https://github.com/YOUR_USERNAME/stk-mcp.git`
3. **Create a branch:** `git checkout -b feature/your-feature-name` or `git checkout -b fix/your-bug-fix-name`
4. **Make your changes:** Implement your feature or bug fix.
* Follow the existing code style.
* Add comments for complex logic.
* Ensure your changes work with the required versions of Python and STK.
* Update documentation if necessary.
5. **Test your changes:** Ensure your changes don't break existing functionality. (Add details here if specific tests are set up later).
6. **Commit your changes:** Use clear and descriptive commit messages. `git commit -m "feat: Add feature X"` or `git commit -m "fix: Resolve issue Y"`
7. **Push to your fork:** `git push origin feature/your-feature-name`
8. **Open a Pull Request:** Go to the original `stk-mcp` repository on GitHub and open a pull request from your branch to the `main` branch (or the appropriate target branch).
* Provide a clear title and description for your pull request, explaining the changes and why they are needed.
* Link to any relevant issues.
## Pull Request Guidelines
* Ensure your code adheres to the project's style guidelines (details can be added here, e.g., linting tools like Black, Flake8).
* Keep pull requests focused on a single feature or bug fix.
* Provide a clear description of the changes.
* Be responsive to feedback and questions during the review process.
Thank you for contributing!
```
--------------------------------------------------------------------------------
/src/stk_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""STK-MCP package.
This package contains the MCP server (`app`), CLI (`cli`), tools, and
STK-specific logic under `stk_logic`.
"""
# Optionally expose the server at package level if desired:
# from .app import mcp_server
```
--------------------------------------------------------------------------------
/src/stk_mcp/app.py:
--------------------------------------------------------------------------------
```python
from mcp.server.fastmcp import FastMCP
from .stk_logic.core import StkState
# Define the central MCP server instance.
# The lifespan will be attached dynamically by the CLI based on user selection.
mcp_server = FastMCP[StkState]("STK Control")
# Import the tools module.
# This triggers the execution of the @mcp_server.tool() decorators.
from . import tools
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/__init__.py:
--------------------------------------------------------------------------------
```python
"""Register MCP tools and resources.
Importing modules ensures the @mcp_server.tool() and .resource() decorators
run and register endpoints with the server.
"""
from . import scenario # noqa: F401
from . import satellite # noqa: F401
from . import location # noqa: F401
from . import objects # noqa: F401
from . import health # noqa: F401
from . import analysis # noqa: F401
# You can optionally define an __all__ if needed, but importing is usually sufficient
# for the decorators to register.
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/logging_config.py:
--------------------------------------------------------------------------------
```python
from __future__ import annotations
import logging
import sys
def configure_logging(level: str = "INFO") -> None:
"""Configure structured logging for STK-MCP.
Example format:
2025-01-01 12:00:00 | INFO | stk_mcp.module:function:42 - Message
"""
fmt = (
"%(asctime)s | %(levelname)-8s | "
"%(name)s:%(funcName)s:%(lineno)d - %(message)s"
)
datefmt = "%Y-%m-%d %H:%M:%S"
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter(fmt=fmt, datefmt=datefmt))
root = logging.getLogger()
for h in list(root.handlers):
root.removeHandler(h)
root.addHandler(handler)
lvl = getattr(logging, str(level).upper(), logging.INFO)
root.setLevel(lvl)
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "stk-mcp"
version = "0.2.0"
description = "STK-MCP, an MCP server allowing LLMs to interact with Ansys/AGI STK - Digital Mission Engineering Software"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"mcp[cli]>=1.6.0",
"pydantic>=2.11.7",
"typer>=0.15.2",
"uvicorn>=0.30",
"rich>=13.7",
"pywin32>=306; platform_system == 'Windows'",
"agi-stk12",
"pydantic-settings>=2.10.1",
"tenacity>=9.1.2",
]
[project.scripts]
stk-mcp = "stk_mcp.cli:app" # New CLI entry point
# Optional: Add pywin32 if needed for clarity, though pip usually handles it on Windows
# dependencies = [
# "mcp[cli]>=1.6.0",
# "pywin32>=306; sys_platform == 'win32'" # Example conditional dependency
# ]
[build-system]
requires = ["uv-build>=0.4.0"]
build-backend = "uv_build"
[tool.uv.sources]
agi-stk12 = { path = "agi.stk12-12.10.0-py3-none-any.whl" }
```
--------------------------------------------------------------------------------
/examples/basic_mission.py:
--------------------------------------------------------------------------------
```python
"""
Example: Creating a basic Earth observation mission.
This demonstrates:
1. Setting up a scenario
2. Creating a ground facility
3. Creating a simple satellite (Windows/Desktop/Engine on Windows)
Run the server first, then invoke these tools from your MCP client.
"""
# This file is illustrative. Use an MCP client (e.g., MCP Inspector)
# to call the following tools in order:
#
# 1) setup_scenario(scenario_name="Demo", start_time="20 Jan 2020 17:00:00.000", duration_hours=12)
# 2) create_location(name="Boulder", latitude_deg=40.015, longitude_deg=-105.27, altitude_km=1.656, kind="facility")
# 3) create_satellite(name="DemoSat", apogee_alt_km=420, perigee_alt_km=410, raan_deg=51.6, inclination_deg=51.6)
# 4) resource: resource://stk/analysis/access/Satellite/DemoSat/Facility/Boulder
# 5) resource: resource://stk/reports/lla/Satellite/DemoSat
print(
"Open the MCP client and try the listed tools/resources to run this example."
)
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/config.py:
--------------------------------------------------------------------------------
```python
from __future__ import annotations
"""
Central configuration for STK-MCP.
Values can be overridden via environment variables prefixed with `STK_MCP_`.
For example: `STK_MCP_DEFAULT_PORT=9000`.
"""
from pydantic_settings import BaseSettings, SettingsConfigDict
class StkConfig(BaseSettings):
"""STK-MCP configuration.
Environment variables use the `STK_MCP_` prefix, e.g., `STK_MCP_DEFAULT_PORT`.
"""
# Physical constants / domain
earth_radius_km: float = 6378.137
# Scenario defaults
default_scenario_name: str = "MCP_STK_Scenario"
default_start_time: str = "20 Jan 2020 17:00:00.000"
default_duration_hours: float = 48.0
# Server defaults
default_host: str = "127.0.0.1"
default_port: int = 8765
# Logging
log_level: str = "INFO"
model_config = SettingsConfigDict(
env_prefix="STK_MCP_",
extra="ignore",
)
def get_config() -> StkConfig:
"""Return a cached settings instance."""
# pydantic-settings already caches by default, but we can wrap for clarity
return StkConfig() # type: ignore[call-arg]
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/health.py:
--------------------------------------------------------------------------------
```python
import logging
from collections import Counter
from mcp.server.fastmcp import Context
from mcp.server.fastmcp.exceptions import ResourceError
from ..app import mcp_server
from ..stk_logic.core import StkState, STK_LOCK
from ..stk_logic.decorators import require_stk_resource
from ..stk_logic.objects import list_objects_internal
logger = logging.getLogger(__name__)
@mcp_server.resource(
"resource://stk/health",
name="STK Health",
title="STK Server Health",
description=(
"Report basic STK state: mode, current scenario, and object counts."
),
mime_type="application/json",
)
@require_stk_resource
def health(ctx: Context):
lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
if not lifespan_ctx:
raise ResourceError("No lifespan context set.")
mode = lifespan_ctx.mode.value if lifespan_ctx.mode else None
scenario_name = None
try:
if lifespan_ctx.stk_root and lifespan_ctx.stk_root.CurrentScenario:
scenario_name = lifespan_ctx.stk_root.CurrentScenario.InstanceName
except Exception:
scenario_name = None
objects = []
try:
if lifespan_ctx.stk_root:
with STK_LOCK:
objects = list_objects_internal(lifespan_ctx.stk_root)
except Exception:
objects = []
counts = Counter([o.get("type", "") for o in objects if o.get("type")])
return {
"mode": mode,
"scenario": scenario_name,
"counts": dict(counts),
}
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/utils.py:
--------------------------------------------------------------------------------
```python
from __future__ import annotations
import logging
import time
from functools import wraps
from typing import Any, Callable, TypeVar, ParamSpec
from tenacity import retry, stop_after_attempt, wait_exponential
logger = logging.getLogger(__name__)
P = ParamSpec("P")
T = TypeVar("T")
def timed_operation(func: Callable[P, T]) -> Callable[P, T]:
"""Decorator to log operation duration for diagnostics."""
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
start = time.perf_counter()
try:
result = func(*args, **kwargs)
duration = time.perf_counter() - start
logger.info("%s completed in %.3fs", func.__name__, duration)
return result
except Exception as e: # pragma: no cover - diagnostic path
duration = time.perf_counter() - start
logger.error("%s failed after %.3fs: %s", func.__name__, duration, e)
raise
return wrapper
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), reraise=True)
def safe_stk_command(stk_root: Any, command: str):
"""Execute an STK Connect command with basic retry logic."""
return stk_root.ExecuteCommand(command)
def safe_exec_lines(stk_root: Any, command: str) -> list[str]:
"""Execute a Connect command and return result lines; returns [] on failure."""
try:
res = safe_stk_command(stk_root, command)
return [res.Item(i) for i in range(res.Count)]
except Exception: # pragma: no cover - depends on STK runtime
logger.debug("Connect command failed: %s", command)
return []
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/decorators.py:
--------------------------------------------------------------------------------
```python
from __future__ import annotations
import logging
from functools import wraps
from typing import Callable, Any, TypeVar, ParamSpec
from mcp.server.fastmcp import Context
from mcp.server.fastmcp.exceptions import ResourceError
from .core import stk_available
logger = logging.getLogger(__name__)
P = ParamSpec("P")
T = TypeVar("T")
def require_stk_tool(func: Callable[P, T]) -> Callable[P, T]:
"""Ensure STK is available and initialized for MCP tools.
Returns a user-friendly error string if unavailable.
Expects first parameter to be `ctx: Context`.
"""
@wraps(func)
def wrapper(ctx: Context, *args: P.args, **kwargs: P.kwargs) -> T: # type: ignore[override]
lifespan_ctx = ctx.request_context.lifespan_context
if not stk_available:
return "Error: STK is not available on this system." # type: ignore[return-value]
if not lifespan_ctx or not lifespan_ctx.stk_root:
return "Error: STK Root not available. Initialize via server lifespan." # type: ignore[return-value]
return func(ctx, *args, **kwargs)
return wrapper
def require_stk_resource(func: Callable[P, T]) -> Callable[P, T]:
"""Ensure STK is available and initialized for MCP resources.
Raises ResourceError if unavailable. Expects first parameter `ctx: Context`.
"""
@wraps(func)
def wrapper(ctx: Context, *args: P.args, **kwargs: P.kwargs) -> T: # type: ignore[override]
lifespan_ctx = ctx.request_context.lifespan_context
if not stk_available:
raise ResourceError("STK is not available on this system.")
if not lifespan_ctx or not lifespan_ctx.stk_root:
raise ResourceError("STK Root not available. Initialize via server lifespan.")
return func(ctx, *args, **kwargs)
return wrapper
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/analysis.py:
--------------------------------------------------------------------------------
```python
import logging
from mcp.server.fastmcp import Context
from mcp.server.fastmcp.exceptions import ResourceError
from ..app import mcp_server
from ..stk_logic.core import StkState, STK_LOCK
from ..stk_logic.decorators import require_stk_resource
from ..stk_logic.analysis import (
compute_access_intervals_internal,
get_lla_ephemeris_internal,
)
logger = logging.getLogger(__name__)
@mcp_server.resource(
"resource://stk/analysis/access/{object1}/{object2}",
name="STK Access",
title="Compute Access Intervals",
description=(
"Compute access intervals between two objects."
" Provide paths like 'Satellite/SatA' and 'Facility/FacB'"
" (with or without leading '*/')."
),
mime_type="application/json",
)
@require_stk_resource
def compute_access(ctx: Context, object1: str, object2: str):
lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
if not lifespan_ctx or not lifespan_ctx.stk_root:
raise ResourceError("STK Root unavailable.")
with STK_LOCK:
return compute_access_intervals_internal(lifespan_ctx.stk_root, object1, object2)
@mcp_server.resource(
"resource://stk/reports/lla/{satellite}",
name="STK LLA Ephemeris",
title="Satellite LLA Ephemeris",
description=(
"Return satellite LLA ephemeris over the scenario interval."
" Provide path like 'Satellite/SatA' (with or without leading '*/')."
),
mime_type="application/json",
)
@require_stk_resource
def get_satellite_lla(ctx: Context, satellite: str):
lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
if not lifespan_ctx or not lifespan_ctx.stk_root:
raise ResourceError("STK Root unavailable.")
with STK_LOCK:
return get_lla_ephemeris_internal(lifespan_ctx.stk_root, satellite, 60.0)
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/objects.py:
--------------------------------------------------------------------------------
```python
import logging
from mcp.server.fastmcp import Context
from mcp.server.fastmcp.exceptions import ResourceError
from ..app import mcp_server
from ..stk_logic.core import StkState, STK_LOCK
from ..stk_logic.decorators import require_stk_resource
from ..stk_logic.objects import list_objects_internal
logger = logging.getLogger(__name__)
@mcp_server.resource(
"resource://stk/objects",
name="STK Scenario Objects",
title="List STK Objects",
description=(
"List all objects in the active STK scenario with their name and type. "
"Returns JSON: [{name, type}, ...]."
),
mime_type="application/json",
)
@require_stk_resource
def list_objects(ctx: Context):
"""
MCP Resource: List all scenario objects as JSON records.
"""
lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
try:
with STK_LOCK:
return list_objects_internal(lifespan_ctx.stk_root)
except Exception as e:
raise ResourceError(str(e))
@mcp_server.resource(
"resource://stk/objects/{object_type}",
name="STK Scenario Objects (Filtered)",
title="List STK Objects by Type",
description=(
"List scenario objects filtered by type (e.g., satellite, facility, place, sensor). "
"Returns JSON: [{name, type}, ...]."
),
mime_type="application/json",
)
@require_stk_resource
def list_objects_by_type(ctx: Context, object_type: str):
"""
MCP Resource: List scenario objects filtered by the provided type.
"""
lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
try:
with STK_LOCK:
objects = list_objects_internal(lifespan_ctx.stk_root, filter_type=object_type)
# If the filter was unrecognized, return empty with a hint instead of throwing
if not objects:
# We still return JSON for consistency
return []
return objects
except Exception as e:
raise ResourceError(str(e))
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/scenario.py:
--------------------------------------------------------------------------------
```python
import logging
from mcp.server.fastmcp import Context
# Use relative imports within the package
from ..app import mcp_server # Import the server instance created in server.py
from ..stk_logic.core import StkState, STK_LOCK
from ..stk_logic.decorators import require_stk_tool
from ..stk_logic.config import get_config
from ..stk_logic.scenario import setup_scenario_internal
logger = logging.getLogger(__name__)
@mcp_server.tool() # Decorate with the server instance
@require_stk_tool
def setup_scenario(
ctx: Context,
scenario_name: str | None = None,
start_time: str | None = None, # Default UTCG start
duration_hours: float | None = None # Default duration
) -> str:
"""
MCP Tool: Creates/Configures an STK Scenario. Closes any existing scenario first.
Args:
ctx: The MCP context (provides access to stk_root via lifespan).
scenario_name: Name for the new scenario.
start_time: Scenario start time in STK UTCG format.
duration_hours: Scenario duration in hours.
Returns:
A string indicating success or failure.
"""
logger.info("MCP Tool: setup_scenario '%s'", scenario_name)
lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
# Basic input validation
cfg = get_config()
if scenario_name is None:
scenario_name = cfg.default_scenario_name
if start_time is None:
start_time = cfg.default_start_time
if duration_hours is None:
duration_hours = cfg.default_duration_hours
if not scenario_name or not isinstance(scenario_name, str):
return "Error: scenario_name must be a non-empty string."
if duration_hours <= 0:
return "Error: duration_hours must be positive."
# Call the internal logic function
with STK_LOCK:
success, message, _ = setup_scenario_internal(
stk_root=lifespan_ctx.stk_root,
scenario_name=scenario_name,
start_time=start_time,
duration_hours=duration_hours,
)
return message # Return the status message from the internal function
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/scenario.py:
--------------------------------------------------------------------------------
```python
from __future__ import annotations
import logging
from .core import stk_available, IAgStkObjectRoot, IAgScenario
from .utils import timed_operation, safe_stk_command
logger = logging.getLogger(__name__)
@timed_operation
def setup_scenario_internal(
stk_root: IAgStkObjectRoot,
scenario_name: str,
start_time: str,
duration_hours: float
) -> tuple[bool, str, IAgScenario | None]:
"""
Internal logic to create/configure an STK Scenario.
Returns:
tuple: (success_flag, status_message, scenario_object_or_None)
"""
if not stk_available or not stk_root:
return False, "STK Root object not available.", None
try:
# Close existing scenario if open
if stk_root.Children.Count > 0:
current_scen_name = stk_root.CurrentScenario.InstanceName
logger.info(" Closing existing scenario: %s", current_scen_name)
stk_root.CloseScenario()
# Create new scenario
logger.info(" Creating new scenario: %s", scenario_name)
stk_root.NewScenario(scenario_name)
scenario = stk_root.CurrentScenario
if scenario is None:
raise Exception("Failed to create or get the new scenario object.")
# Set time period
duration_str = f"+{duration_hours} hours"
logger.info(" Setting scenario time: Start='%s', Duration='%s'", start_time, duration_str)
scenario.SetTimePeriod(start_time, duration_str)
# Reset animation time
stk_root.Rewind()
# Optional: Maximize windows
try:
logger.info(" Maximizing STK windows...")
safe_stk_command(stk_root, 'Application / Raise')
safe_stk_command(stk_root, 'Application / Maximize')
# Consider checking for 3D window existence if needed
# stk_root.ExecuteCommand('Window3D * Maximize')
except Exception as cmd_e:
logger.warning(" Could not execute maximize commands: %s", cmd_e)
return True, f"Successfully created and configured scenario: '{scenario_name}'", scenario
except Exception as e:
error_msg = f"Error setting up scenario '{scenario_name}': {e}"
logger.error(" %s", error_msg)
# import traceback
# traceback.print_exc()
return False, error_msg, None
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/location.py:
--------------------------------------------------------------------------------
```python
import logging
from mcp.server.fastmcp import Context
from ..app import mcp_server
from ..stk_logic.core import StkState, STK_LOCK
from ..stk_logic.decorators import require_stk_tool
from ..stk_logic.location import create_location_internal
logger = logging.getLogger(__name__)
@mcp_server.tool()
@require_stk_tool
def create_location(
ctx: Context,
name: str,
latitude_deg: float,
longitude_deg: float,
altitude_km: float = 0.0,
kind: str = "facility",
) -> str:
"""
Create or update a ground location in the active scenario.
Args:
ctx: MCP request context (provides STK lifespan state).
name: Object name (e.g., "Boulder").
latitude_deg: Geodetic latitude in degrees [-90, 90].
longitude_deg: Geodetic longitude in degrees [-180, 180].
altitude_km: Altitude above mean sea level in kilometers.
kind: Object kind: "facility" (default) or "place".
Returns:
Status message indicating success or error details.
Examples:
>>> create_location(ctx, name="Boulder", latitude_deg=40.015, longitude_deg=-105.27, altitude_km=1.656, kind="facility")
"Successfully created facility: 'Boulder'"
"""
lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
try:
scenario = lifespan_ctx.stk_root.CurrentScenario
if scenario is None:
return "Error: No active scenario found. Use 'setup_scenario' first."
except Exception as e:
return f"Error: Could not access current scenario: {e}"
# Input validation
if not (-90.0 <= latitude_deg <= 90.0):
return "Error: latitude_deg must be within [-90, 90] degrees."
if not (-180.0 <= longitude_deg <= 180.0):
return "Error: longitude_deg must be within [-180, 180] degrees."
if altitude_km < -0.5:
return "Error: altitude_km must be >= -0.5 km."
if kind.lower() not in ("facility", "place"):
return "Error: kind must be 'facility' or 'place'."
with STK_LOCK:
ok, msg, _ = create_location_internal(
stk_root=lifespan_ctx.stk_root,
scenario=scenario,
name=name,
latitude_deg=latitude_deg,
longitude_deg=longitude_deg,
altitude_km=altitude_km,
kind=kind,
)
return msg
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/location.py:
--------------------------------------------------------------------------------
```python
from __future__ import annotations
import logging
from typing import Literal
from .core import stk_available, IAgStkObjectRoot, IAgScenario
from .utils import timed_operation, safe_stk_command
logger = logging.getLogger(__name__)
try:
# Enums are available on both Desktop and Engine
from agi.stk12.stkobjects import AgESTKObjectType
enums_available = True
except Exception:
AgESTKObjectType = None # type: ignore[assignment]
enums_available = False
@timed_operation
def create_location_internal(
stk_root: IAgStkObjectRoot,
scenario: IAgScenario,
name: str,
latitude_deg: float,
longitude_deg: float,
altitude_km: float,
kind: Literal["facility", "place"] = "facility",
):
"""
Create or update a ground location in the active scenario.
Uses the STK Object Model for creation and assigns position using object model
where possible; falls back to STK Connect for position assignment to preserve
cross-platform compatibility (Desktop/Engine).
Returns:
tuple[bool, str, object | None]: (success, message, created_or_found_object)
"""
if not stk_available or not stk_root or not scenario:
return False, "STK Root/Scenario is not available.", None
kind = kind.lower().strip()
if kind not in ("facility", "place"):
return False, "Invalid kind. Use 'facility' or 'place'.", None
if not enums_available:
return False, "STK enums not available; cannot create object.", None
obj_type = (
AgESTKObjectType.eFacility if kind == "facility" else AgESTKObjectType.ePlace
)
try:
children = scenario.Children
if not children.Contains(obj_type, name):
obj = children.New(obj_type, name)
created = True
else:
obj = children.Item(name)
created = False
# Assign position via object model if available; otherwise use Connect
try:
# Most STK Python wrappers surface AssignGeodetic directly
obj.Position.AssignGeodetic(latitude_deg, longitude_deg, altitude_km)
except Exception:
# Fallback to STK Connect with retry: Geodetic lat lon alt km
class_name = "Facility" if kind == "facility" else "Place"
cmd = (
f"SetPosition */{class_name}/{name} Geodetic "
f"{latitude_deg} {longitude_deg} {altitude_km} km"
)
safe_stk_command(stk_root, cmd)
action = "created" if created else "updated"
return True, f"Successfully {action} {kind}: '{name}'", obj
except Exception as e:
logger.error("Error creating %s '%s': %s", kind, name, e)
return False, f"Error creating {kind} '{name}': {e}", None
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/analysis.py:
--------------------------------------------------------------------------------
```python
from __future__ import annotations
import logging
from typing import Any
from .core import IAgStkObjectRoot
from .utils import timed_operation
logger = logging.getLogger(__name__)
def _normalize_path(path: str) -> str:
p = (path or "").strip()
if not p:
raise ValueError("Object path must be non-empty.")
if p.startswith("*/"):
return p
if p.startswith("/"):
return f"*{p}"
# Require explicit class/name from users for reliability
return f"*/{p}"
@timed_operation
def compute_access_intervals_internal(
stk_root: IAgStkObjectRoot,
object1_path: str,
object2_path: str,
) -> dict[str, Any]:
"""Compute access intervals between two STK objects using the Object Model.
Returns a dictionary with input paths and a list of {start, stop} intervals.
"""
p1 = _normalize_path(object1_path)
p2 = _normalize_path(object2_path)
from_obj = stk_root.GetObjectFromPath(p1)
to_obj = stk_root.GetObjectFromPath(p2)
access = from_obj.GetAccessToObject(to_obj)
access.ComputeAccess()
intervals = access.AccessIntervals
out: list[dict[str, str]] = []
for i in range(intervals.Count):
ivl = intervals.Item(i)
out.append({"start": ivl.StartTime, "stop": ivl.StopTime})
return {"from": p1, "to": p2, "intervals": out}
@timed_operation
def get_lla_ephemeris_internal(
stk_root: IAgStkObjectRoot,
satellite_path: str,
step_sec: float = 60.0,
) -> dict[str, Any]:
"""Fetch LLA ephemeris for a satellite over the scenario interval using Data Providers.
Returns a dictionary: {satellite, step_sec, records:[{time, lat_deg, lon_deg, alt_km}...]}
"""
p = _normalize_path(satellite_path)
sat = stk_root.GetObjectFromPath(p)
scenario = stk_root.CurrentScenario
if scenario is None:
raise RuntimeError("No active scenario.")
start = scenario.StartTime
stop = scenario.StopTime
# Data provider name and elements are standard for satellites
dp_group = sat.DataProviders.Item("LLA State")
dp = dp_group.Group.Item("Fixed")
res = dp.ExecElements(start, stop, step_sec, ["Time", "Lat", "Lon", "Alt"])
time_vals = list(res.DataSets.GetDataSetByName("Time").GetValues())
lat_vals = list(res.DataSets.GetDataSetByName("Lat").GetValues())
lon_vals = list(res.DataSets.GetDataSetByName("Lon").GetValues())
alt_vals = list(res.DataSets.GetDataSetByName("Alt").GetValues())
records: list[dict[str, float | str]] = []
for i in range(len(time_vals)):
records.append(
{
"time": time_vals[i],
"lat_deg": float(lat_vals[i]),
"lon_deg": float(lon_vals[i]),
"alt_km": float(alt_vals[i]),
}
)
return {"satellite": p, "step_sec": step_sec, "records": records}
```
--------------------------------------------------------------------------------
/src/stk_mcp/tools/satellite.py:
--------------------------------------------------------------------------------
```python
import logging
from mcp.server.fastmcp import Context
# Use relative imports within the package
from ..app import mcp_server # Import the server instance
from ..stk_logic.core import StkState, STK_LOCK
from ..stk_logic.decorators import require_stk_tool
from ..stk_logic.satellite import create_satellite_internal
logger = logging.getLogger(__name__)
@mcp_server.tool() # Decorate with the server instance
@require_stk_tool
def create_satellite(
ctx: Context,
name: str,
apogee_alt_km: float,
perigee_alt_km: float,
raan_deg: float,
inclination_deg: float
) -> str:
"""
MCP Tool: Creates/modifies an STK satellite using Apogee/Perigee altitudes, RAAN, and Inclination.
Assumes a scenario is already open.
Args:
ctx: The MCP context.
name: Desired name for the satellite.
apogee_alt_km: Apogee altitude (km).
perigee_alt_km: Perigee altitude (km).
raan_deg: RAAN (degrees).
inclination_deg: Inclination (degrees).
Returns:
A string indicating success or failure.
Examples:
>>> create_satellite(ctx, "ISS", apogee_alt_km=420, perigee_alt_km=410, raan_deg=51.6, inclination_deg=51.6)
"Successfully created/configured satellite: 'ISS'"
"""
logger.info("MCP Tool: create_satellite '%s'", name)
lifespan_ctx: StkState | None = ctx.request_context.lifespan_context
# Get the current scenario from the STK root object
try:
scenario = lifespan_ctx.stk_root.CurrentScenario
if scenario is None:
return "Error: No active scenario found in STK. Use 'setup_scenario' tool first."
logger.info(" Operating within scenario: %s", scenario.InstanceName)
except Exception as e:
return f"Error accessing current scenario: {e}. Use 'setup_scenario' tool first."
# Input validation
if apogee_alt_km < perigee_alt_km:
return "Error: apogee_alt_km cannot be less than perigee_alt_km."
if not (0.0 <= inclination_deg <= 180.0):
return "Error: inclination_deg must be within [0, 180] degrees."
# RAAN wraps; accept 0..360 inclusive
if not (0.0 <= raan_deg <= 360.0):
return "Error: raan_deg must be within [0, 360] degrees."
if perigee_alt_km < -0.5 or apogee_alt_km < -0.5:
return "Error: perigee/apogee altitudes must be >= -0.5 km."
# Call the internal logic function
try:
with STK_LOCK:
success, message, _ = create_satellite_internal(
stk_root=lifespan_ctx.stk_root,
scenario=scenario,
name=name,
apogee_alt_km=apogee_alt_km,
perigee_alt_km=perigee_alt_km,
raan_deg=raan_deg,
inclination_deg=inclination_deg,
)
return message # Return the message from the internal function
except ValueError as ve:
error_msg = f"Configuration Error for satellite '{name}': {ve}"
logger.error(" %s", error_msg)
return error_msg
except Exception as e:
# Catch potential errors from the internal function (e.g., COM errors)
error_msg = f"Error creating satellite '{name}': {e}"
logger.error(" %s", error_msg)
# import traceback
# traceback.print_exc()
return error_msg
```
--------------------------------------------------------------------------------
/src/stk_mcp/cli.py:
--------------------------------------------------------------------------------
```python
import os
import sys
import logging
import uvicorn
import typer
from rich.console import Console
from rich.table import Table
# --- Check for STK installation early ---
stk_installed = False
try:
from agi.stk12.stkengine import STKEngine # noqa: F401
from agi.stk12.stkdesktop import STKDesktop # noqa: F401
stk_installed = True
except ImportError:
print(
"Warning: Ansys/AGI STK Python API not found. Please install it to use this application.",
file=sys.stderr,
)
# Allow Typer to still show help, but commands will fail.
# --- Local imports (safe regardless of STK availability) ---
from stk_mcp.app import mcp_server
from stk_mcp.stk_logic.core import create_stk_lifespan, StkMode # type: ignore
from stk_mcp.stk_logic.config import get_config
from stk_mcp.stk_logic.logging_config import configure_logging
import anyio
# --- Typer Application Setup ---
app = typer.Typer(
name="stk-mcp",
help="A CLI for running and interacting with the STK-MCP server.",
add_completion=False,
)
console = Console()
def _validate_desktop_mode(mode: StkMode):
"""Callback to ensure 'desktop' mode is only used on Windows."""
if mode == StkMode.DESKTOP and os.name != "nt":
console.print(f"[bold red]Error:[/] STK Desktop mode is only available on Windows. Use '--mode {StkMode.ENGINE.value}'.")
raise typer.Exit(code=1)
return mode
@app.command()
def run(
host: str = typer.Option(None, help="The host to bind the server to."),
port: int = typer.Option(None, help="The port to run the server on."),
mode: StkMode = typer.Option(
StkMode.ENGINE if os.name != "nt" else StkMode.DESKTOP,
"--mode", "-m",
case_sensitive=False,
help="STK execution mode. 'desktop' is only available on Windows.",
callback=_validate_desktop_mode,
),
log_level: str = typer.Option(
"info",
help="Log level: critical, error, warning, info, debug",
),
):
"""
Run the STK-MCP server.
"""
if not stk_installed:
console.print("[bold red]Error:[/] Cannot run server. STK Python API is not installed.")
raise typer.Exit(code=1)
# Configure logging
cfg = get_config()
level = log_level or cfg.log_level
configure_logging(level)
# Resolve host/port from config if not provided
host = host or cfg.default_host
port = int(port or cfg.default_port)
console.print(
f"[green]Starting STK-MCP server in[/] [bold cyan]{mode.value}[/] [green]mode on {host}:{port}...[/]"
)
# Dynamically create the lifespan based on the selected mode
stk_lifespan_manager = create_stk_lifespan(mode)
# Attach the lifespan to the server instance
mcp_server.lifespan = stk_lifespan_manager
# Run the server using uvicorn
uvicorn.run(
mcp_server,
host=host,
port=port,
)
@app.command(name="list-tools")
def list_tools():
"""
List all available MCP tools and their descriptions.
"""
if not stk_installed:
console.print("[bold red]Error:[/] Cannot list tools. STK Python API is not installed.")
raise typer.Exit(code=1)
table = Table(title="[bold blue]STK-MCP Available Tools[/bold blue]")
table.add_column("Tool Name", style="cyan", no_wrap=True)
table.add_column("Description", style="magenta")
tools = anyio.run(mcp_server.list_tools)
if not tools:
console.print("[yellow]No tools have been registered on the server.[/yellow]")
return
for t in tools:
description = (t.description or "No description provided.").strip()
table.add_row(t.name, description)
console.print(table)
if __name__ == "__main__":
app()
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/objects.py:
--------------------------------------------------------------------------------
```python
from __future__ import annotations
import logging
from typing import Optional
from .core import stk_available, IAgStkObjectRoot
from .utils import safe_exec_lines, timed_operation
logger = logging.getLogger(__name__)
def _exec_lines(stk_root: IAgStkObjectRoot, cmd: str) -> list[str]:
"""
Execute an STK Connect command and return the result lines.
Safe helper that returns an empty list on failure.
"""
return safe_exec_lines(stk_root, cmd)
def _parse_all_instance_names(lines: list[str]) -> list[tuple[str, str]]:
"""
Parse lines returned by the "AllInstanceNames" Connect command.
Returns a list of tuples: (class_name, instance_name)
"""
out: list[tuple[str, str]] = []
for raw in lines:
line = (raw or "").strip()
if not line:
continue
# Heuristically skip headers if present
lower = line.lower()
if "number" in lower and ("object" in lower or "instance" in lower):
continue
# Paths often look like "/Class/Name" or "*/Class/Name" or longer
parts = [p for p in line.split('/') if p]
if len(parts) >= 2:
cls = parts[-2]
name = parts[-1]
out.append((cls, name))
else:
# Fallback if the format is unexpected
out.append(("", line))
return out
_TOP_LEVEL_CLASSES = (
# Common top-level object classes in STK
"Satellite",
"Facility",
"Place",
"Aircraft",
"Ship",
"GroundVehicle",
"Missile",
"LaunchVehicle",
"Submarine",
"AreaTarget",
"LineTarget",
)
_SENSOR_PARENTS = (
# Object classes that commonly host Sensors
"Satellite",
"Facility",
"Aircraft",
"Ship",
"GroundVehicle",
"Missile",
"LaunchVehicle",
"Submarine",
)
def _normalize_filter(filter_type: str) -> set[str] | None:
"""
Normalize a user-provided filter type (case-insensitive) to a set of
canonical STK class names. Returns None if the filter is unrecognized.
"""
t = (filter_type or "").strip().lower()
if not t:
return None
# Aliases and plural handling
mapping: dict[str, set[str]] = {
"sat": {"Satellite"},
"satellite": {"Satellite"},
"satellites": {"Satellite"},
"facility": {"Facility"},
"facilities": {"Facility"},
"place": {"Place"},
"places": {"Place"},
"location": {"Facility", "Place"},
"locations": {"Facility", "Place"},
"sensor": {"Sensor"},
"sensors": {"Sensor"},
"aircraft": {"Aircraft"},
"ship": {"Ship"},
"ships": {"Ship"},
"groundvehicle": {"GroundVehicle"},
"groundvehicles": {"GroundVehicle"},
"missile": {"Missile"},
"missiles": {"Missile"},
"areatarget": {"AreaTarget"},
"areatargets": {"AreaTarget"},
"linetarget": {"LineTarget"},
"linetargets": {"LineTarget"},
"launchvehicle": {"LaunchVehicle"},
"launchvehicles": {"LaunchVehicle"},
"submarine": {"Submarine"},
"submarines": {"Submarine"},
}
return mapping.get(t)
@timed_operation
def list_objects_internal(
stk_root: IAgStkObjectRoot,
filter_type: Optional[str] = None,
) -> list[dict[str, str]]:
"""
Enumerate objects in the active scenario and return a list of
{"name": <instance name>, "type": <class>} dictionaries.
Uses STK Connect (AllInstanceNames) for broad compatibility with
both STK Desktop and STK Engine.
"""
if not stk_available or not stk_root:
raise RuntimeError("STK Root is not available.")
# Ensure there is an active scenario
try:
scenario = stk_root.CurrentScenario
if scenario is None:
raise RuntimeError("No active scenario found.")
except Exception as e:
raise RuntimeError(f"Could not access current scenario: {e}")
normalized: set[str] | None = _normalize_filter(filter_type) if filter_type else None
results: list[dict[str, str]] = []
def add_from_cmd(cmd: str, expected_type: str | None = None) -> None:
lines = _exec_lines(stk_root, cmd)
for cls, name in _parse_all_instance_names(lines):
typ = expected_type or (cls if cls else "")
if not typ:
continue
if normalized is not None and typ not in normalized:
continue
results.append({"name": name, "type": typ})
# Top-level classes
for cls in _TOP_LEVEL_CLASSES:
if normalized is not None and cls not in normalized:
continue
add_from_cmd(f"AllInstanceNames */{cls}", expected_type=cls)
# Sensors (nested under multiple parents)
# Only include if no filter, or if filter explicitly asks for sensors
if normalized is None or ("Sensor" in normalized):
for parent in _SENSOR_PARENTS:
add_from_cmd(f"AllInstanceNames */{parent}/*/Sensor", expected_type="Sensor")
return results
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/satellite.py:
--------------------------------------------------------------------------------
```python
import os
import logging
from . import core as core
from .core import IAgStkObjectRoot, IAgScenario
from .utils import timed_operation
from .config import get_config
logger = logging.getLogger(__name__)
# Import STK Objects specific to satellite creation if available (Windows Desktop/Engine)
AgESTKObjectType = None
AgEVePropagatorType = None
AgEClassicalLocation = None
win32com_client = None
satellite_capable = False
if core.stk_available and os.name == 'nt':
try:
from agi.stk12.stkobjects import (
AgESTKObjectType as AgESTKObjectTypeImport,
AgEVePropagatorType as AgEVePropagatorTypeImport,
AgEClassicalLocation as AgEClassicalLocationImport,
)
import win32com.client as win32com_client_import
AgESTKObjectType = AgESTKObjectTypeImport
AgEVePropagatorType = AgEVePropagatorTypeImport
AgEClassicalLocation = AgEClassicalLocationImport
win32com_client = win32com_client_import
satellite_capable = True
except ImportError:
logger.warning("Could not import STK COM types for satellite creation (Windows only feature).")
except Exception as e:
logger.error("Error importing win32com or STK enums: %s", e)
cfg = get_config()
EARTH_RADIUS_KM = cfg.earth_radius_km
@timed_operation
def create_satellite_internal(
stk_root: IAgStkObjectRoot, # Although not directly used, good for context
scenario: IAgScenario,
name: str,
apogee_alt_km: float,
perigee_alt_km: float,
raan_deg: float,
inclination_deg: float,
):
"""
Internal logic to create/configure an STK satellite.
Returns:
tuple: (success_flag, status_message, satellite_object_or_None)
Raises:
ValueError: If input parameters are invalid (e.g., apogee < perigee).
Exception: For COM or other STK errors.
"""
if not core.stk_available or not scenario or win32com_client is None or not satellite_capable:
raise RuntimeError("STK modules, active scenario, or win32com not available/initialized.")
if AgESTKObjectType is None or AgEVePropagatorType is None or AgEClassicalLocation is None:
raise RuntimeError("Required STK Object Enums not imported.")
logger.info(" Attempting internal satellite creation/configuration: %s", name)
if apogee_alt_km < perigee_alt_km:
raise ValueError("Apogee altitude cannot be less than Perigee altitude.")
# --- Calculate Semi-Major Axis (a) and Eccentricity (e) ---
radius_apogee_km = apogee_alt_km + EARTH_RADIUS_KM
radius_perigee_km = perigee_alt_km + EARTH_RADIUS_KM
semi_major_axis_km = (radius_apogee_km + radius_perigee_km) / 2.0
denominator = radius_apogee_km + radius_perigee_km
eccentricity = 0.0 if denominator == 0 else (radius_apogee_km - radius_perigee_km) / denominator
logger.debug(" Calculated Semi-Major Axis (a): %.3f km", semi_major_axis_km)
logger.debug(" Calculated Eccentricity (e): %.6f", eccentricity)
# --- Get or Create Satellite Object ---
scenario_children = scenario.Children
satellite = None
if not scenario_children.Contains(AgESTKObjectType.eSatellite, name):
logger.info(" Creating new Satellite object: %s", name)
satellite = scenario_children.New(AgESTKObjectType.eSatellite, name)
else:
logger.info(" Satellite '%s' already exists. Getting reference.", name)
satellite = scenario_children.Item(name)
if satellite is None:
raise Exception(f"Failed to create or retrieve satellite object '{name}'.")
# --- Set Propagator to TwoBody ---
logger.info(" Setting propagator to TwoBody...")
satellite.SetPropagatorType(AgEVePropagatorType.ePropagatorTwoBody)
propagator = satellite.Propagator
propagator_twobody = win32com_client.CastTo(propagator, "IAgVePropagatorTwoBody")
if propagator_twobody is None:
raise Exception("Failed to cast propagator to IAgVePropagatorTwoBody.")
# --- Define Orbital Elements ---
argp_deg = 0.0 # Assumed
true_anom_deg = 0.0 # Assumed (starts at perigee)
logger.info(" Assigning Classical Elements (J2000):")
# (Print statements omitted for brevity, add back if desired)
orbit_state = propagator_twobody.InitialState.Representation
classical_elements = win32com_client.CastTo(orbit_state, "IAgOrbitStateClassical")
if classical_elements:
classical_elements.AssignClassical(
AgEClassicalLocation.eCoordinateSystemJ2000,
semi_major_axis_km, eccentricity, inclination_deg,
argp_deg, raan_deg, true_anom_deg
)
else:
raise Exception("Failed to cast orbit state to IAgOrbitStateClassical.")
# --- Propagate the Orbit ---
logger.info(" Propagating orbit...")
propagator_twobody.Propagate()
logger.info(" Internal satellite configuration for '%s' complete.", name)
# Return success flag, message, and the object
return True, f"Successfully created/configured satellite: '{satellite.InstanceName}'", satellite
```
--------------------------------------------------------------------------------
/src/stk_mcp/stk_logic/core.py:
--------------------------------------------------------------------------------
```python
import os
import platform
import logging
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from typing import Union
from enum import Enum # <--- IMPORT Enum
from threading import Lock
from pydantic import BaseModel
from typing import NamedTuple, Optional
from mcp.server.fastmcp import FastMCP
logger = logging.getLogger(__name__)
# --- Define shared data types here ---
class StkMode(str, Enum):
"""Enumeration for selecting the STK execution mode."""
DESKTOP = "desktop"
ENGINE = "engine"
# --- Attempt STK Imports ---
stk_available = False
STKApplication = None
STKDesktop = None
STKEngine = None
IAgStkObjectRoot = None
IAgScenario = None
_platform = platform.system()
if _platform == "Windows":
try:
from agi.stk12.stkdesktop import STKDesktop as STKDesktopImport, STKApplication as STKApplicationImport
from agi.stk12.stkengine import STKEngine as STKEngineImport
from agi.stk12.stkobjects import IAgStkObjectRoot as IAgStkObjectRootImport, IAgScenario as IAgScenarioImport
STKDesktop = STKDesktopImport
STKApplication = STKApplicationImport # Desktop App
STKEngine = STKEngineImport
IAgStkObjectRoot = IAgStkObjectRootImport
IAgScenario = IAgScenarioImport
stk_available = True
logger.info("STK Desktop and Engine modules loaded successfully (Windows).")
except ImportError:
logger.warning("Failed to import STK modules on Windows. Functionality disabled.")
elif _platform == "Linux":
try:
from agi.stk12.stkengine import STKEngine as STKEngineImport, IAgStkObjectRoot as IAgStkObjectRootImport
from agi.stk12.stkobjects import IAgScenario as IAgScenarioImport
STKEngine = STKEngineImport
IAgStkObjectRoot = IAgStkObjectRootImport
IAgScenario = IAgScenarioImport
stk_available = True
logger.info("STK Engine modules loaded successfully (Linux). STK Desktop is unavailable.")
except ImportError:
logger.warning("Failed to import STK Engine modules. Functionality disabled.")
elif _platform == "Darwin":
# STK Engine is not supported on macOS
logger.error("Detected macOS (Darwin). STK Engine/Desktop are not supported on this platform.")
else:
logger.warning("Unknown platform '%s'. STK availability undetermined.", _platform)
# A type hint for whichever application object is in use
StkAppType = object | None
class StkState(BaseModel):
"""Holds the state of the STK application connection."""
stk_app: StkAppType = None
stk_root: object | None = None
mode: StkMode | None = None
# Global lock to serialize all STK access across tools/resources
STK_LOCK: Lock = Lock()
class OperationResult(NamedTuple):
"""Standard result type for STK operations."""
success: bool
message: str
data: Optional[dict] = None
def create_stk_lifespan(mode: StkMode):
"""
A factory that returns an async context manager for the STK lifecycle.
"""
@asynccontextmanager
async def stk_lifespan_manager(server: FastMCP) -> AsyncIterator[StkState]:
"""
Manages the STK application lifecycle based on the selected mode.
"""
if not stk_available or IAgStkObjectRoot is None:
logger.warning("STK is not available. MCP server will run without STK functionality.")
yield StkState(mode=mode)
return
logger.info("MCP Server Startup: Initializing STK in '%s' mode...", mode.value)
state = StkState(mode=mode)
try:
if mode == StkMode.DESKTOP:
# --- Desktop Mode Logic ---
logger.info(" Attempting to attach to existing STK instance...")
try:
state.stk_app = STKDesktop.AttachToApplication()
logger.info(" Successfully attached to existing STK instance.")
state.stk_app.Visible = True
except Exception:
logger.info(" Could not attach. Launching new STK instance...")
state.stk_app = STKDesktop.StartApplication(visible=True, userControl=True)
state.stk_root = state.stk_app.Root
# Close any open scenario to start clean
if state.stk_root and state.stk_root.Children.Count > 0:
logger.info(" Closing existing scenario '%s'...", state.stk_root.CurrentScenario.InstanceName)
state.stk_root.CloseScenario()
elif mode == StkMode.ENGINE:
# --- Engine Mode Logic ---
logger.info(" Starting new STK Engine instance...")
state.stk_app = STKEngine.StartApplication(noGraphics=True)
state.stk_root = state.stk_app.NewObjectRoot()
logger.info(" STK Engine instance started.")
if state.stk_root is None:
raise RuntimeError("Failed to obtain STK Root object.")
logger.info("STK Initialized. Providing STK context to tools.")
yield state
except Exception as e:
logger.exception("FATAL: Failed to initialize STK in %s mode: %s", mode.value, e)
yield StkState(mode=mode) # Yield empty state on failure
finally:
logger.info("MCP Server Shutdown: Cleaning up STK (%s mode)...", mode.value)
if state.stk_app:
try:
state.stk_app.Close()
logger.info(" STK Application/Engine Closed.")
except Exception as quit_e:
logger.warning(" Error closing STK: %s", quit_e)
logger.info("STK Cleanup Complete.")
return stk_lifespan_manager
```